mtsd/backend/app/core/api_crypto.py

"""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