53 lines
1.6 KiB
Python
53 lines
1.6 KiB
Python
"""外部系统凭据列加密。
|
||
|
||
与早期微信绑定不同，这里没有明文降级：未配置 master key 时拒绝创建和调用。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import hashlib
|
||
import os
|
||
|
||
from cryptography.fernet import Fernet, InvalidToken
|
||
|
||
_PREFIX = "v1:"
|
||
_ENV = "ZCBOT_CREDENTIAL_MASTER_KEY"
|
||
|
||
|
||
def configured() -> bool:
|
||
return len(os.getenv(_ENV, "").strip()) >= 32
|
||
|
||
|
||
def _fernet() -> Fernet:
|
||
raw = os.getenv(_ENV, "").strip()
|
||
if not raw:
|
||
raise RuntimeError(f"{_ENV} 未配置，不能保存或使用外部系统凭据")
|
||
if len(raw) < 32:
|
||
raise RuntimeError(f"{_ENV} 至少需要 32 个字符")
|
||
digest = hashlib.sha256(raw.encode("utf-8")).digest()
|
||
return Fernet(base64.urlsafe_b64encode(digest))
|
||
|
||
|
||
def encrypt_secret(value: str) -> str:
|
||
if not isinstance(value, str) or not value:
|
||
raise ValueError("credential value must be a non-empty string")
|
||
return _PREFIX + _fernet().encrypt(value.encode("utf-8")).decode("ascii")
|
||
|
||
|
||
def decrypt_secret(value: str) -> str:
|
||
if not isinstance(value, str) or not value.startswith(_PREFIX):
|
||
raise RuntimeError("外部系统凭据格式无效")
|
||
try:
|
||
return _fernet().decrypt(value[len(_PREFIX):].encode("ascii")).decode("utf-8")
|
||
except InvalidToken as exc:
|
||
raise RuntimeError("外部系统凭据无法解密，master key 可能已变化") from exc
|
||
|
||
|
||
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:]
|