66 lines
1.8 KiB
Python
66 lines
1.8 KiB
Python
"""Web API 请求/响应加解密工具。"""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import time
|
|
from typing import Any, Dict
|
|
|
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
|
|
|
|
|
class ApiCryptoError(Exception):
|
|
"""API 加解密错误。"""
|
|
|
|
|
|
def derive_key(fixed_secret: str, user_id: str, ts: int, nonce: str) -> bytes:
|
|
seed = f"{fixed_secret}:{user_id}:{ts}:{nonce}"
|
|
return hashlib.sha256(seed.encode("utf-8")).digest()
|
|
|
|
|
|
def _b64url_encode(data: bytes) -> str:
|
|
return base64.urlsafe_b64encode(data).decode("utf-8").rstrip("=")
|
|
|
|
|
|
def _b64url_decode(data: str) -> bytes:
|
|
pad = "=" * (-len(data) % 4)
|
|
return base64.urlsafe_b64decode((data + pad).encode("utf-8"))
|
|
|
|
|
|
def encrypt_json(payload: Dict[str, Any], key: bytes) -> str:
|
|
iv = os.urandom(12)
|
|
aes = AESGCM(key)
|
|
raw = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
|
cipher = aes.encrypt(iv, raw, None)
|
|
return _b64url_encode(iv + cipher)
|
|
|
|
|
|
def decrypt_json(cipher_text: str, key: bytes) -> Dict[str, Any]:
|
|
try:
|
|
data = _b64url_decode(cipher_text)
|
|
if len(data) < 13:
|
|
raise ApiCryptoError("密文长度非法")
|
|
iv = data[:12]
|
|
cipher = data[12:]
|
|
aes = AESGCM(key)
|
|
plain = aes.decrypt(iv, cipher, None)
|
|
obj = json.loads(plain.decode("utf-8"))
|
|
if not isinstance(obj, dict):
|
|
raise ApiCryptoError("解密后报文格式非法")
|
|
return obj
|
|
except ApiCryptoError:
|
|
raise
|
|
except Exception as exc:
|
|
raise ApiCryptoError("密文解析失败") from exc
|
|
|
|
|
|
def gen_nonce_hex() -> str:
|
|
return os.urandom(12).hex()
|
|
|
|
|
|
def is_ts_in_window(ts: int, skew_sec: int) -> bool:
|
|
now_ms = int(time.time() * 1000)
|
|
return abs(now_ms - int(ts)) <= max(30, int(skew_sec)) * 1000
|