feat(security): 增强密码加密安全机制

- 新增 ENCRYPTION_KEY_RAW 环境变量支持,可直接使用 Fernet 密钥
- 添加密钥丢失保护机制,防止在有加密数据时意外生成新密钥
- 新增 verify_encryption_key() 函数用于启动时验证密钥
- docker-compose.yml 改为从 .env 文件读取敏感配置
- 新增 crypto_utils.py 文件挂载,支持热更新

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Yu Yon
2026-01-21 09:31:15 +08:00
parent 10be464265
commit dfc93bce2e
2 changed files with 116 additions and 7 deletions

View File

@@ -4,9 +4,15 @@
加密工具模块
用于加密存储敏感信息(如第三方账号密码)
使用Fernet对称加密
安全增强版本 - 2026-01-21
- 支持 ENCRYPTION_KEY_RAW 直接使用 Fernet 密钥
- 增加密钥丢失保护机制
- 增加启动时密钥验证
"""
import os
import sys
import base64
from pathlib import Path
from cryptography.fernet import Fernet
@@ -47,27 +53,89 @@ def _derive_key(password: bytes, salt: bytes) -> bytes:
return base64.urlsafe_b64encode(kdf.derive(password))
def _check_existing_encrypted_data() -> bool:
"""
检查是否存在已加密的数据
用于防止在有加密数据的情况下意外生成新密钥
"""
try:
import sqlite3
db_path = os.environ.get('DB_FILE', 'data/app_data.db')
if not Path(db_path).exists():
return False
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM accounts WHERE password LIKE 'gAAAAA%'")
count = cursor.fetchone()[0]
conn.close()
return count > 0
except Exception as e:
logger.warning(f"检查加密数据时出错: {e}")
return False
def get_encryption_key():
"""获取加密密钥(优先环境变量,否则从文件读取或生成)"""
# 优先从环境变量读取
"""
获取加密密钥
优先级:
1. ENCRYPTION_KEY_RAW - 直接使用 Fernet 密钥(推荐用于 Docker 部署)
2. ENCRYPTION_KEY - 通过 PBKDF2 派生密钥
3. 从文件读取
4. 生成新密钥(仅在无现有加密数据时)
"""
# 优先级 1: 直接使用 Fernet 密钥(推荐)
raw_key = os.environ.get('ENCRYPTION_KEY_RAW')
if raw_key:
logger.info("使用环境变量 ENCRYPTION_KEY_RAW 作为加密密钥")
return raw_key.encode() if isinstance(raw_key, str) else raw_key
# 优先级 2: 从环境变量派生密钥
env_key = os.environ.get('ENCRYPTION_KEY')
if env_key:
# 使用环境变量中的密钥派生Fernet密钥
logger.info("使用环境变量 ENCRYPTION_KEY 派生加密密钥")
salt = _get_or_create_salt()
return _derive_key(env_key.encode(), salt)
# 从文件读取
# 优先级 3: 从文件读取
key_path = Path(ENCRYPTION_KEY_FILE)
if key_path.exists():
logger.info(f"从文件 {ENCRYPTION_KEY_FILE} 读取加密密钥")
with open(key_path, 'rb') as f:
return f.read()
# 优先级 4: 生成新密钥(带保护检查)
# 安全检查:如果已有加密数据,禁止生成新密钥
if _check_existing_encrypted_data():
error_msg = (
"\n" + "=" * 60 + "\n"
"[严重错误] 检测到数据库中存在已加密的密码数据,但加密密钥文件丢失!\n"
"\n"
"这将导致所有已加密的密码无法解密!\n"
"\n"
"解决方案:\n"
"1. 恢复 data/encryption_key.bin 文件(如有备份)\n"
"2. 或在 docker-compose.yml 中设置 ENCRYPTION_KEY_RAW 环境变量\n"
"3. 如果密钥确实丢失,需要重新录入所有账号密码\n"
"\n"
"设置 ALLOW_NEW_KEY=true 环境变量可强制生成新密钥(不推荐)\n"
+ "=" * 60
)
logger.error(error_msg)
# 检查是否强制允许生成新密钥
if os.environ.get('ALLOW_NEW_KEY', '').lower() != 'true':
print(error_msg, file=sys.stderr)
raise RuntimeError("加密密钥丢失且存在已加密数据,请检查配置")
# 生成新的密钥
key = Fernet.generate_key()
os.makedirs(key_path.parent, exist_ok=True)
with open(key_path, 'wb') as f:
f.write(key)
logger.info(f"已生成新的加密密钥并保存到 {ENCRYPTION_KEY_FILE}")
logger.warning("请立即备份此密钥文件,并建议设置 ENCRYPTION_KEY_RAW 环境变量!")
return key
@@ -120,8 +188,11 @@ def decrypt_password(encrypted_password: str) -> str:
decrypted = fernet.decrypt(encrypted_password.encode('utf-8'))
return decrypted.decode('utf-8')
except Exception as e:
# 解密失败,可能是旧的明文密码
logger.warning(f"密码解密失败,可能是未加密的旧数据: {e}")
# 解密失败,可能是旧的明文密码或密钥不匹配
if is_encrypted(encrypted_password):
logger.error(f"密码解密失败(密钥可能不匹配): {e}")
else:
logger.warning(f"密码解密失败,可能是未加密的旧数据: {e}")
return encrypted_password
@@ -138,7 +209,6 @@ def is_encrypted(password: str) -> bool:
"""
if not password:
return False
# Fernet加密的数据是base64编码以'gAAAAA'开头
return password.startswith('gAAAAA')
@@ -157,6 +227,39 @@ def migrate_password(password: str) -> str:
return encrypt_password(password)
def verify_encryption_key() -> bool:
"""
验证当前密钥是否能解密现有数据
用于启动时检查
Returns:
bool: 密钥是否有效
"""
try:
import sqlite3
db_path = os.environ.get('DB_FILE', 'data/app_data.db')
if not Path(db_path).exists():
return True
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("SELECT password FROM accounts WHERE password LIKE 'gAAAAA%' LIMIT 1")
row = cursor.fetchone()
conn.close()
if not row:
return True
# 尝试解密
fernet = _get_fernet()
fernet.decrypt(row[0].encode('utf-8'))
logger.info("加密密钥验证成功")
return True
except Exception as e:
logger.error(f"加密密钥验证失败: {e}")
return False
if __name__ == '__main__':
# 测试加密解密
test_password = "test_password_123"
@@ -169,3 +272,6 @@ if __name__ == '__main__':
print(f"加密解密成功: {test_password == decrypted}")
print(f"是否已加密: {is_encrypted(encrypted)}")
print(f"明文是否加密: {is_encrypted(test_password)}")
# 验证密钥
print(f"\n密钥验证: {verify_encryption_key()}")

View File

@@ -15,6 +15,7 @@ services:
- ./templates:/app/templates # 模板文件(实时更新)
- ./app.py:/app/app.py # 主程序(实时更新)
- ./database.py:/app/database.py # 数据库模块(实时更新)
- ./crypto_utils.py:/app/crypto_utils.py # 加密工具(实时更新)
dns:
- 223.5.5.5
- 114.114.114.114
@@ -37,6 +38,8 @@ services:
- MAX_CONCURRENT_PER_ACCOUNT=1
- MAX_CONCURRENT_CONTEXTS=100
# 安全配置
# 加密密钥配置(重要!防止容器重建时丢失密钥)
- ENCRYPTION_KEY_RAW=${ENCRYPTION_KEY_RAW}
- SESSION_LIFETIME_HOURS=24
- SESSION_COOKIE_SECURE=false
- MAX_CAPTCHA_ATTEMPTS=5