一、系统环境与基础工具准备
本指南基于Ubuntu 22.04 LTS服务器环境,所有操作均通过命令行完成。请确保你拥有root权限或sudo权限。
1.1 操作系统与依赖安装
首先更新系统并安装必要的软件包:
```bash
sudo apt update && sudo apt upgrade -y
sudo apt install -y openssl gpg gnupg2 git python3 python3-pip postgresql postgresql-contrib nginx
```
1.2 加密工具配置
GPG是档案加密的核心工具,生成密钥对:
```bash
gpg --full-generate-key
```
在交互式界面中选择以下选项:
- 密钥类型:RSA and RSA (default)
- 密钥长度:4096
- 有效期:0 (永不过期)
- 输入真实姓名和邮箱
- 设置强密码(至少16位,包含大小写字母、数字、特殊字符)
导出公钥用于后续系统集成:
```bash
gpg --armor --export your-email@example.com > public_key.asc
```
二、档案存储架构设计
2.1 目录结构规划
创建标准化的档案存储目录:
```bash
sudo mkdir -p /opt/archives/{incoming,processing,encrypted,decrypted_temp,logs,backup}
sudo chown -R $USER:$USER /opt/archives
sudo chmod -R 750 /opt/archives
```
设置目录权限:
- incoming:接收原始档案,权限750
- processing:处理中的档案,权限700
- encrypted:加密后存储,权限700
- decrypted_temp:临时解密区域,每次使用后自动清空
- logs:操作日志,权限750
- backup:备份文件,权限700
2.2 数据库设计
创建PostgreSQL数据库和用户:
```bash
sudo -u postgres psql -c "CREATE USER archive_admin WITH PASSWORD 'StrongPassword123!';"
sudo -u postgres psql -c "CREATE DATABASE archives_db OWNER archive_admin;"
```
创建档案元数据表:
```sql
CREATE TABLE archive_metadata (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
original_filename VARCHAR(500) NOT NULL,
encrypted_filename VARCHAR(500) NOT NULL,
file_hash VARCHAR(128) NOT NULL,
encryption_key_id VARCHAR(50) NOT NULL,
file_size BIGINT NOT NULL,
mime_type VARCHAR(100),
classification_level INTEGER CHECK (classification_level BETWEEN 1 AND 5),
retention_period DATE NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_accessed TIMESTAMP,
access_count INTEGER DEFAULT 0,
is_deleted BOOLEAN DEFAULT FALSE,
deletion_scheduled DATE
);
CREATE INDEX idx_file_hash ON archive_metadata(file_hash);
CREATE INDEX idx_retention_period ON archive_metadata(retention_period);
CREATE INDEX idx_classification ON archive_metadata(classification_level);
```
三、自动化处理脚本开发
3.1 档案加密脚本
创建/opt/archives/scripts/encrypt_archive.py:
```python
!/usr/bin/env python3
import os
import hashlib
import subprocess
import psycopg2
from datetime import datetime, timedelta
from pathlib import Path
def calculate_file_hash(filepath):
"""计算文件SHA-512哈希值"""
sha512 = hashlib.sha512()
with open(filepath, 'rb') as f:
while chunk := f.read(8192):
sha512.update(chunk)
return sha512.hexdigest()
def encrypt_file(input_path, output_path, recipient_email):
"""使用GPG加密文件"""
cmd = [
'gpg', '--encrypt', '--recipient', recipient_email,
'--output', output_path, '--trust-model', 'always',
input_path
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise Exception(f"加密失败: {result.stderr}")
return True
def main():
配置参数
incoming_dir = Path('/opt/archives/incoming')
encrypted_dir = Path('/opt/archives/encrypted')
processing_dir = Path('/opt/archives/processing')
recipient_email = 'archive-system@yourdomain.com'
处理所有待处理文件
for file_path in incoming_dir.glob(''):
if file_path.is_file():
try:
移动到处理目录
processing_path = processing_dir / file_path.name
file_path.rename(processing_path)
计算哈希
file_hash = calculate_file_hash(processing_path)
加密文件
encrypted_filename = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{file_path.name}.gpg"
encrypted_path = encrypted_dir / encrypted_filename
encrypt_file(str(processing_path), str(encrypted_path), recipient_email)
记录到数据库
conn = psycopg2.connect(
host="localhost",
database="archives_db",
user="archive_admin",
password="StrongPassword123!"
)
cursor = conn.cursor()
cursor.execute("""
INSERT INTO archive_metadata
(original_filename, encrypted_filename, file_hash,
encryption_key_id, file_size, retention_period)
VALUES (%s, %s, %s, %s, %s, %s)
""", (
file_path.name,
encrypted_filename,
file_hash,
recipient_email,
processing_path.stat().st_size,
datetime.now() + timedelta(days=3657) 默认保存7年
))
conn.commit()
cursor.close()
conn.close()
删除处理中的原始文件
processing_path.unlink()
print(f"成功处理: {file_path.name}")
except Exception as e:
print(f"处理失败 {file_path.name}: {str(e)}")
将失败文件移回incoming目录
processing_path.rename(file_path)
if __name__ == "__main__":
main()
```
设置脚本权限并安装依赖:
```bash
chmod +x /opt/archives/scripts/encrypt_archive.py
pip3 install psycopg2-binary
```
3.2 档案检索与解密脚本
创建/opt/archives/scripts/retrieve_archive.py:
```python
!/usr/bin/env python3
import psycopg2
import subprocess
import hashlib
from pathlib import Path
import tempfile
import shutil
def verify_decryption(original_hash, decrypted_path):
"""验证解密文件的完整性"""
current_hash = calculate_file_hash(decrypted_path)
return original_hash == current_hash
def retrieve_archive(file_hash, output_dir):
"""根据哈希值检索并解密档案"""
conn = psycopg2.connect(
host="localhost",
database="archives_db",
user="archive_admin",
password="StrongPassword123!"
)
cursor = conn.cursor()
cursor.execute("""
SELECT encrypted_filename, original_filename, file_hash
FROM archive_metadata
WHERE file_hash = %s AND is_deleted = FALSE
""", (file_hash,))
result = cursor.fetchone()
if not result:
return None
encrypted_filename, original_filename, stored_hash = result
更新访问记录
cursor.execute("""
UPDATE archive_metadata
SET last_accessed = CURRENT_TIMESTAMP,
access_count = access_count + 1
WHERE file_hash = %s
""", (file_hash,))
conn.commit()
cursor.close()
conn.close()
解密文件
encrypted_path = Path('/opt/archives/encrypted') / encrypted_filename
temp_dir = Path('/opt/archives/decrypted_temp')
temp_dir.mkdir(exist_ok=True)
decrypted_path = temp_dir / original_filename
使用GPG解密
cmd = [
'gpg', '--decrypt',
'--output', str(decrypted_path),
str(encrypted_path)
]
result = subprocess.run(cmd, capture_output=True, text=True, input='\n')
if result.returncode == 0:
验证文件完整性
if verify_decryption(stored_hash, decrypted_path):
复制到输出目录
output_path = Path(output_dir) / original_filename
shutil.copy2(decrypted_path, output_path)
清理临时文件
decrypted_path.unlink()
return str(output_path)
return None
```
四、访问控制与审计系统
4.1 用户权限管理
创建用户角色表:
```sql
CREATE TABLE archive_users (
user_id SERIAL PRIMARY KEY,
username VARCHAR(100) UNIQUE NOT NULL,
email VARCHAR(255) UNIQUE NOT NULL,
gpg_key_id VARCHAR(50),
access_level INTEGER CHECK (access_level BETWEEN 1 AND 3),
is_active BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE access_logs (
log_id BIGSERIAL PRIMARY KEY,
user_id INTEGER REFERENCES archive_users(user_id),
action VARCHAR(50) NOT NULL,
file_hash VARCHAR(128),
ip_address INET,
user_agent TEXT,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
success BOOLEAN NOT NULL
);
CREATE INDEX idx_access_logs_user ON access_logs(user_id, timestamp);
CREATE INDEX idx_access_logs_file ON access_logs(file_hash, timestamp);
```
4.2 审计日志脚本
创建/opt/archives/scripts/log_access.py:
```python
!/usr/bin/env python3
import psycopg2
from datetime import datetime
def log_access(user_id, action, file_hash=None, ip_address=None,
user_agent=None, success=True):
"""记录访问日志"""
conn = psycopg2.connect(
host="localhost",
database="archives_db",
user="archive_admin",
password="StrongPassword123!"
)
cursor = conn.cursor()
cursor.execute("""
INSERT INTO access_logs
(user_id, action, file_hash, ip_address, user_agent, success)
VALUES (%s, %s, %s, %s, %s, %s)
""", (user_id, action, file_hash, ip_address, user_agent, success))
conn.commit()
cursor.close()
conn.close()
```
五、自动化维护任务
5.1 定期备份脚本

创建/opt/archives/scripts/backup_archives.py:
```python
!/usr/bin/env python3
import subprocess
from datetime import datetime
from pathlib import Path
def backup_archives():
"""执行完整备份"""
backup_dir = Path('/opt/archives/backup')
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
备份加密文件
encrypted_backup = backup_dir / f"encrypted_{timestamp}.tar.gz"
subprocess.run([
'tar', '-czf', str(encrypted_backup),
'-C', '/opt/archives', 'encrypted'
], check=True)
备份数据库
db_backup = backup_dir / f"database_{timestamp}.dump"
subprocess.run([
'pg_dump', '-U', 'archive_admin', '-d', 'archives_db',
'-f', str(db_backup)
], check=True)
加密备份文件
subprocess.run([
'gpg', '--encrypt', '--recipient', 'backup-admin@yourdomain.com',
'--output', f"{encrypted_backup}.gpg", str(encrypted_backup)
], check=True)
subprocess.run([
'gpg', '--encrypt', '--recipient', 'backup-admin@yourdomain.com',
'--output', f"{db_backup}.gpg", str(db_backup)
], check=True)
删除未加密的备份
encrypted_backup.unlink()
db_backup.unlink()
清理30天前的备份
for backup_file in backup_dir.glob('.gpg'):
if (datetime.now() - datetime.fromtimestamp(backup_file.stat().st_mtime)).days > 30:
backup_file.unlink()
```
5.2 设置Cron定时任务
编辑crontab:
```bash
crontab -e
```
添加以下内容:
```bash
每小时处理新档案
0 /usr/bin/python3 /opt/archives/scripts/encrypt_archive.py >> /opt/archives/logs/encryption.log 2>&1
每天凌晨2点执行备份
0 2 /usr/bin/python3 /opt/archives/scripts/backup_archives.py >> /opt/archives/logs/backup.log 2>&1
每月1号清理临时文件
0 0 1 /bin/rm -rf /opt/archives/decrypted_temp/
每天检查保留期限
0 3 /usr/bin/psql -U archive_admin -d archives_db -c "UPDATE archive_metadata SET deletion_scheduled = CURRENT_DATE + INTERVAL '30 days' WHERE retention_period < CURRENT_DATE AND deletion_scheduled IS NULL;"
```
六、安全加固措施
6.1 文件完整性监控
创建完整性检查脚本:
```python
!/usr/bin/env python3
import hashlib
import psycopg2
from pathlib import Path
def verify_archive_integrity():
"""验证所有档案的完整性"""
conn = psycopg2.connect(
host="localhost",
database="archives_db",
user="archive_admin",
password="StrongPassword123!"
)
cursor = conn.cursor()
cursor.execute("""
SELECT encrypted_filename, file_hash
FROM archive_metadata
WHERE is_deleted = FALSE
""")
issues = []
for encrypted_filename, stored_hash in cursor.fetchall():
file_path = Path('/opt/archives/encrypted') / encrypted_filename
if not file_path.exists():
issues.append(f"文件丢失: {encrypted_filename}")
continue
计算当前哈希
sha512 = hashlib.sha512()
with open(file_path, 'rb') as f:
while chunk := f.read(8192):
sha512.update(chunk)
current_hash = sha512.hexdigest()
if current_hash != stored_hash:
issues.append(f"哈希不匹配: {encrypted_filename}")
cursor.close()
conn.close()
if issues:
with open('/opt/archives/logs/integrity_issues.log', 'a') as f:
for issue in issues:
f.write(f"{datetime.now()}: {issue}\n")
return len(issues) == 0
```
6.2 SSH加固配置
编辑/etc/ssh/sshd_config:
```bash
Port 2222 修改默认端口
PermitRootLogin no
PasswordAuthentication no
PubkeyAuthentication yes
AllowUsers archive_admin
MaxAuthTries 3
ClientAliveInterval 300
ClientAliveCountMax 2
```
重启SSH服务:
```bash
sudo systemctl restart sshd
```
七、故障排查与恢复
7.1 常见问题解决
问题1:GPG加密失败
检查密钥是否存在:
```bash
gpg --list-keys
```
导入缺失的公钥:
```bash
gpg --import public_key.asc
```
问题2:数据库连接失败
检查PostgreSQL服务状态:
```bash
sudo systemctl status postgresql
```
验证连接:
```bash
psql -U archive_admin -d archives_db -c "SELECT 1;"
```
问题3:文件权限错误
修复目录权限:
```bash
sudo chown -R archive_admin:archive_admin /opt/archives
sudo chmod -R 750 /opt/archives
sudo chmod 700 /opt/archives/{encrypted,decrypted_temp,processing}
```
7.2 数据恢复流程
从备份恢复数据库:
```bash
解密备份文件
gpg --decrypt --output database_backup.dump database_20240101_020000.dump.gpg
停止相关服务
sudo systemctl stop nginx
恢复数据库
pg_restore -U postgres -d archives_db database_backup.dump
恢复加密文件
gpg --decrypt --output encrypted_backup.tar.gz encrypted_20240101_020000.tar.gz.gpg
tar -xzf encrypted_backup.tar.gz -C /opt/archives/
重启服务