chore: establish stable application baseline
This commit is contained in:
+71
@@ -0,0 +1,71 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import os
|
||||
from typing import Any
|
||||
|
||||
from cryptography.fernet import Fernet, InvalidToken
|
||||
|
||||
|
||||
PASSWORD_SCRYPT_N = 2**14
|
||||
PASSWORD_SCRYPT_R = 8
|
||||
PASSWORD_SCRYPT_P = 1
|
||||
|
||||
|
||||
class SecretVault:
|
||||
def __init__(self, key: str) -> None:
|
||||
try:
|
||||
self._fernet = Fernet(key.encode("ascii"))
|
||||
except (ValueError, TypeError) as exc:
|
||||
raise ValueError("APP_ENCRYPTION_KEY 格式无效。") from exc
|
||||
|
||||
@staticmethod
|
||||
def generate_key() -> str:
|
||||
return Fernet.generate_key().decode("ascii")
|
||||
|
||||
def encrypt_json(self, payload: dict[str, Any]) -> str:
|
||||
raw = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
||||
return self._fernet.encrypt(raw).decode("ascii")
|
||||
|
||||
def decrypt_json(self, token: str) -> dict[str, Any]:
|
||||
if not token:
|
||||
return {}
|
||||
try:
|
||||
payload = json.loads(self._fernet.decrypt(token.encode("ascii")).decode("utf-8"))
|
||||
except (InvalidToken, UnicodeDecodeError, json.JSONDecodeError) as exc:
|
||||
raise ValueError("账号加密数据无法解密,请检查 APP_ENCRYPTION_KEY。") from exc
|
||||
if not isinstance(payload, dict):
|
||||
raise ValueError("账号加密数据格式无效。")
|
||||
return payload
|
||||
|
||||
|
||||
def hash_password(password: str, salt: bytes | None = None) -> tuple[str, str]:
|
||||
raw_salt = salt or os.urandom(16)
|
||||
digest = hashlib.scrypt(
|
||||
password.encode("utf-8"),
|
||||
salt=raw_salt,
|
||||
n=PASSWORD_SCRYPT_N,
|
||||
r=PASSWORD_SCRYPT_R,
|
||||
p=PASSWORD_SCRYPT_P,
|
||||
dklen=32,
|
||||
)
|
||||
return (
|
||||
base64.urlsafe_b64encode(raw_salt).decode("ascii"),
|
||||
base64.urlsafe_b64encode(digest).decode("ascii"),
|
||||
)
|
||||
|
||||
|
||||
def verify_password(password: str, salt_text: str, expected_hash: str) -> bool:
|
||||
try:
|
||||
salt = base64.urlsafe_b64decode(salt_text.encode("ascii"))
|
||||
_, actual_hash = hash_password(password, salt)
|
||||
except (ValueError, TypeError):
|
||||
return False
|
||||
return hmac.compare_digest(actual_hash, expected_hash)
|
||||
|
||||
|
||||
def token_hash(token: str) -> str:
|
||||
return hashlib.sha256(token.encode("utf-8")).hexdigest()
|
||||
Reference in New Issue
Block a user