118 lines
4.2 KiB
Python
118 lines
4.2 KiB
Python
"""外部系统凭据列加密。
|
||
|
||
与早期微信绑定不同，这里没有明文降级：未配置 master key 时拒绝创建和调用。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import secrets
|
||
|
||
from cryptography.exceptions import InvalidTag
|
||
from cryptography.fernet import Fernet, InvalidToken
|
||
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
||
|
||
_LEGACY_PREFIX = "v1:"
|
||
_PREFIX = "v2:"
|
||
_ENV = "ZCBOT_CREDENTIAL_MASTER_KEY"
|
||
_KEY_ID_ENV = "ZCBOT_CREDENTIAL_KEY_ID"
|
||
_PREVIOUS_KEYS_ENV = "ZCBOT_CREDENTIAL_PREVIOUS_KEYS"
|
||
|
||
|
||
def configured() -> bool:
|
||
return len(os.getenv(_ENV, "").strip()) >= 32
|
||
|
||
|
||
def _keyring() -> tuple[str, dict[str, str]]:
|
||
current = os.getenv(_ENV, "").strip()
|
||
if len(current) < 32:
|
||
raise RuntimeError(f"{_ENV} 至少需要 32 个字符")
|
||
current_id = os.getenv(_KEY_ID_ENV, "primary").strip() or "primary"
|
||
if ":" in current_id or len(current_id) > 64:
|
||
raise RuntimeError(f"{_KEY_ID_ENV} 格式无效")
|
||
keys = {current_id: current}
|
||
raw_previous = os.getenv(_PREVIOUS_KEYS_ENV, "").strip()
|
||
if raw_previous:
|
||
try:
|
||
previous = json.loads(raw_previous)
|
||
except ValueError as exc:
|
||
raise RuntimeError(f"{_PREVIOUS_KEYS_ENV} 必须是 JSON 对象") from exc
|
||
if not isinstance(previous, dict):
|
||
raise RuntimeError(f"{_PREVIOUS_KEYS_ENV} 必须是 JSON 对象")
|
||
for key_id, secret in previous.items():
|
||
key_id = str(key_id).strip()
|
||
secret = str(secret).strip()
|
||
if not key_id or ":" in key_id or len(key_id) > 64 or len(secret) < 32:
|
||
raise RuntimeError(f"{_PREVIOUS_KEYS_ENV} 包含无效密钥")
|
||
keys.setdefault(key_id, secret)
|
||
return current_id, keys
|
||
|
||
|
||
def _aes_key(secret: str) -> bytes:
|
||
return hashlib.sha256(secret.encode("utf-8")).digest()
|
||
|
||
|
||
def encrypt_secret(value: str, *, aad: str = "") -> str:
|
||
if not isinstance(value, str) or not value:
|
||
raise ValueError("credential value must be a non-empty string")
|
||
key_id, keys = _keyring()
|
||
nonce = secrets.token_bytes(12)
|
||
ciphertext = AESGCM(_aes_key(keys[key_id])).encrypt(
|
||
nonce,
|
||
value.encode("utf-8"),
|
||
aad.encode("utf-8"),
|
||
)
|
||
encoded = base64.urlsafe_b64encode(nonce + ciphertext).decode("ascii")
|
||
return f"{_PREFIX}{key_id}:{encoded}"
|
||
|
||
|
||
def decrypt_secret(value: str, *, aad: str = "") -> str:
|
||
if not isinstance(value, str):
|
||
raise RuntimeError("外部系统凭据格式无效")
|
||
if value.startswith(_PREFIX):
|
||
try:
|
||
_, key_id, encoded = value.split(":", 2)
|
||
_, keys = _keyring()
|
||
secret = keys[key_id]
|
||
payload = base64.urlsafe_b64decode(encoded.encode("ascii"))
|
||
plaintext = AESGCM(_aes_key(secret)).decrypt(
|
||
payload[:12],
|
||
payload[12:],
|
||
aad.encode("utf-8"),
|
||
)
|
||
return plaintext.decode("utf-8")
|
||
except (KeyError, ValueError, InvalidTag, UnicodeDecodeError) as exc:
|
||
raise RuntimeError(
|
||
"外部系统凭据无法解密，密钥或绑定上下文可能已变化"
|
||
) from exc
|
||
if not value.startswith(_LEGACY_PREFIX):
|
||
raise RuntimeError("外部系统凭据格式无效")
|
||
# 0027 前的 Fernet 密文没有 key id/AAD；仅用于滚动迁移时读取。
|
||
try:
|
||
_, keys = _keyring()
|
||
for secret in keys.values():
|
||
digest = hashlib.sha256(secret.encode("utf-8")).digest()
|
||
try:
|
||
return (
|
||
Fernet(base64.urlsafe_b64encode(digest))
|
||
.decrypt(value[len(_LEGACY_PREFIX) :].encode("ascii"))
|
||
.decode("utf-8")
|
||
)
|
||
except InvalidToken:
|
||
continue
|
||
except (RuntimeError, UnicodeDecodeError) as exc:
|
||
raise RuntimeError("外部系统凭据无法解密，master key 可能已变化") from exc
|
||
raise RuntimeError("外部系统凭据无法解密，master key 可能已变化")
|
||
|
||
|
||
def mask_username(username: str) -> str:
|
||
username = (username or "").strip()
|
||
if not username:
|
||
return "***"
|
||
if len(username) <= 2:
|
||
return username[0] + "*"
|
||
return username[:2] + "***" + username[-1:]
|