shangfangjian
786f10aa11
- crypto.py: API Key 加密/解密工具 - runtime_config.py: 运行时动态配置管理 - log_buffer.py: 内存日志缓冲区 - config.py: 新增加密配置项 - http_client.py: 增强重试和错误处理
105 lines
3.6 KiB
Python
105 lines
3.6 KiB
Python
"""安全原语:对称加密(Fernet/AES)与密码哈希(scrypt)。
|
|
|
|
- API Key 等需要原文调用的敏感值:入库前用 SECRET_KEY 派生的 Fernet 密钥加密,
|
|
存储格式 `enc:v1:<token>`;读取时解密。SECRET_KEY 只存于部署机 .env,不入库。
|
|
- 管理员密码:只存 scrypt 哈希(单向,不可逆),验证用,永远不需要还原原文。
|
|
|
|
`enc:v1:` 前缀 + 透传设计使旧明文数据无需停机即可共存,由启动迁移一次性加密。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac as _hmac
|
|
import logging
|
|
import secrets
|
|
|
|
from cryptography.fernet import Fernet, InvalidToken
|
|
|
|
from src.core.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_ENC_PREFIX = "enc:v1:"
|
|
|
|
# scrypt 参数(OWASP 推荐: n=2^17 更强,取 n=2^15 平衡 NAS CPU)
|
|
_SCRYPT_N = 2**15
|
|
_SCRYPT_R = 8
|
|
_SCRYPT_P = 1
|
|
# OpenSSL 默认 maxmem 限制约 32MB,显式放宽到 128MB
|
|
_SCRYPT_MAXMEM = 128 * 1024 * 1024
|
|
|
|
|
|
def _fernet() -> Fernet:
|
|
"""由 SECRET_KEY 确定性派生 Fernet 密钥(任意字符串输入均可)。
|
|
|
|
SECRET_KEY 未配置时回落派生自 DATABASE_URL(仅为不让开发环境崩溃;
|
|
生产必须显式配置,否则加密强度受限 —— 启动时会打 warning)。
|
|
"""
|
|
raw = settings.SECRET_KEY
|
|
if not raw:
|
|
logger.warning(
|
|
"SECRET_KEY 未设置,加密密钥回落派生自 DATABASE_URL。"
|
|
"请在 .env 配置强随机 SECRET_KEY(openssl rand -base64 32)。"
|
|
)
|
|
raw = f"fallback:{settings.DATABASE_URL}"
|
|
digest = hashlib.sha256(raw.encode()).digest()
|
|
return Fernet(base64.urlsafe_b64encode(digest))
|
|
|
|
|
|
def encrypt_value(plaintext: str) -> str:
|
|
"""加密敏感值,带版本前缀;空值原样返回。"""
|
|
if not plaintext:
|
|
return plaintext
|
|
token = _fernet().encrypt(plaintext.encode()).decode()
|
|
return f"{_ENC_PREFIX}{token}"
|
|
|
|
|
|
def decrypt_value(stored: str) -> str:
|
|
"""解密 `enc:v1:` 前缀的值;无前缀(旧明文)原样返回,便于平滑迁移。"""
|
|
if not stored or not stored.startswith(_ENC_PREFIX):
|
|
return stored
|
|
token = stored[len(_ENC_PREFIX):]
|
|
try:
|
|
return _fernet().decrypt(token.encode()).decode()
|
|
except InvalidToken:
|
|
# 密钥不匹配(通常是 SECRET_KEY 变了):报错而非静默返回错误数据
|
|
raise ValueError(
|
|
"敏感配置解密失败:SECRET_KEY 与加密时不一致。"
|
|
"恢复原 SECRET_KEY 或在后台重新保存对应配置项。"
|
|
) from None
|
|
|
|
|
|
def is_encrypted(stored: str) -> bool:
|
|
return bool(stored) and stored.startswith(_ENC_PREFIX)
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
"""scrypt 哈希,存储格式 scrypt$N$r$p$salt_hex$dk_hex。"""
|
|
salt = secrets.token_bytes(16)
|
|
dk = hashlib.scrypt(
|
|
password.encode(), salt=salt, n=_SCRYPT_N, r=_SCRYPT_R, p=_SCRYPT_P,
|
|
dklen=32, maxmem=_SCRYPT_MAXMEM,
|
|
)
|
|
return f"scrypt${_SCRYPT_N}${_SCRYPT_R}${_SCRYPT_P}${salt.hex()}${dk.hex()}"
|
|
|
|
|
|
def verify_password(password: str, stored: str) -> bool:
|
|
"""校验密码与存储的 scrypt 哈希是否匹配。"""
|
|
try:
|
|
algo, n, r, p, salt_hex, dk_hex = stored.split("$")
|
|
if algo != "scrypt":
|
|
return False
|
|
dk = hashlib.scrypt(
|
|
password.encode(),
|
|
salt=bytes.fromhex(salt_hex),
|
|
n=int(n),
|
|
r=int(r),
|
|
p=int(p),
|
|
dklen=len(bytes.fromhex(dk_hex)),
|
|
maxmem=_SCRYPT_MAXMEM,
|
|
)
|
|
return _hmac.compare_digest(dk, bytes.fromhex(dk_hex))
|
|
except (ValueError, TypeError):
|
|
return False
|