mtsd/backend/app/core/api_crypto_middleware.py

"""Web API 加解密中间件。"""
from __future__ import annotations

import asyncio
import json
import logging
import time
from typing import Any, Optional, Tuple
from urllib.parse import urlencode

import redis.asyncio as redis
from jose import jwt
from redis import exceptions as redis_exceptions
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

from app.core import api_crypto
from app.core.config import settings
from app.core.response import error_response

logger = logging.getLogger("sqxp")

_redis_client: Optional[redis.Redis] = None
_local_nonce_cache: dict[str, float] = {}


def _nonce_key(user_id: str, nonce: str) -> str:
    return f"api-enc:nonce:{user_id}:{nonce}"


def _should_skip_path(path: str) -> bool:
    excludes = [x.strip() for x in (settings.API_ENCRYPT_EXCLUDE_PATHS or "").split(",") if x.strip()]
    return path in excludes


def _is_crypto_api(path: str) -> bool:
    """仅对运营端与小程序业务 API 应用统一密文协议。"""
    prefix = settings.API_V1_PREFIX
    return path.startswith(f"{prefix}/web/") or path.startswith(f"{prefix}/mini/")


def _claims_from_auth(auth_header: Optional[str]) -> Tuple[Optional[str], Optional[str]]:
    if not auth_header or not auth_header.startswith("Bearer "):
        return None, None
    token = auth_header[7:].strip()
    if not token:
        return None, None
    try:
        payload = jwt.decode(
            token,
            settings.SECRET_KEY,
            algorithms=[settings.ALGORITHM],
            options={"verify_signature": False, "verify_exp": False},
        )
        return payload.get("sub"), token
    except Exception:
        return None, token


def _get_redis_client() -> Optional[redis.Redis]:
    global _redis_client
    if not settings.REDIS_HOST:
        return None
    if _redis_client is None:
        _redis_client = redis.Redis(
            host=settings.REDIS_HOST,
            port=settings.REDIS_PORT,
            db=settings.REDIS_DB,
            password=settings.REDIS_PASSWORD or None,
            socket_timeout=settings.REDIS_TIMEOUT_SEC,
            ssl=settings.REDIS_SSL_ENABLED,
            decode_responses=True,
        )
    return _redis_client


async def _mark_nonce_once(user_id: str, nonce: str) -> bool:
    ttl_sec = max(30, int(settings.API_ENCRYPT_NONCE_TTL_SEC))
    key = _nonce_key(user_id, nonce)
    client = _get_redis_client()
    if client is not None:
        try:
            ok = await asyncio.wait_for(client.set(key, "1", ex=ttl_sec, nx=True), timeout=2.0)
            return bool(ok)
        except (asyncio.TimeoutError, redis_exceptions.RedisError):
            logger.warning("api crypto nonce redis unavailable, fallback to local memory")

    now = time.time()
    expire_before = now - ttl_sec
    stale_keys = [k for k, v in _local_nonce_cache.items() if v < expire_before]
    for stale in stale_keys:
        _local_nonce_cache.pop(stale, None)
    if key in _local_nonce_cache:
        return False
    _local_nonce_cache[key] = now
    return True


def _json_error_response(code: int, message: str) -> Response:
    body = error_response(code, message)
    return Response(content=json.dumps(body, ensure_ascii=False), media_type="application/json", status_code=200)


class ApiCryptoMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        if not settings.API_ENCRYPT_ENABLED:
            return await call_next(request)
        if not settings.API_ENCRYPT_FIXED_SECRET:
            if settings.API_ENCRYPT_FAIL_OPEN:
                logger.warning("api crypto bypassed: API_ENCRYPT_FIXED_SECRET is empty")
                return await call_next(request)
            return _json_error_response(500, "服务端加密配置缺失")
        path = request.url.path
        if not _is_crypto_api(path) or _should_skip_path(path):
            return await call_next(request)

        content_type = (request.headers.get("content-type") or "").lower()
        if "multipart/form-data" in content_type:
            return await call_next(request)

        body_bytes = await request.body()
        req_json = {}
        if body_bytes:
            try:
                req_json = json.loads(body_bytes.decode("utf-8"))
                if not isinstance(req_json, dict):
                    req_json = {}
            except Exception:
                req_json = {}

        is_encrypted_req = int(req_json.get("enc", 0) or 0) == 1
        if not is_encrypted_req:
            logger.info("api crypto bypassed plaintext request: %s", path)
            return await call_next(request)

        user_id, _ = _claims_from_auth(request.headers.get("authorization"))
        if not user_id:
            return _json_error_response(401, "未授权或 token 无效")

        ts = req_json.get("ts")
        nonce = req_json.get("nonce")
        edata = req_json.get("edata")
        if ts is None or not nonce or not isinstance(edata, str):
            logger.warning("api crypto invalid envelope: path=%s", path)
            return _json_error_response(400, "加密报文格式错误")
        if not api_crypto.is_ts_in_window(int(ts), settings.API_ENCRYPT_TS_SKEW_SEC):
            logger.warning("api crypto ts expired: path=%s user_id=%s", path, user_id)
            return _json_error_response(400, "请求已过期，请重试")
        if not await _mark_nonce_once(str(user_id), str(nonce)):
            logger.warning("api crypto replay detected: path=%s user_id=%s nonce=%s", path, user_id, nonce)
            return _json_error_response(409, "重复请求")

        key = api_crypto.derive_key(
            fixed_secret=settings.API_ENCRYPT_FIXED_SECRET,
            user_id=str(user_id),
            ts=int(ts),
            nonce=str(nonce),
        )
        try:
            plain = api_crypto.decrypt_json(edata, key)
        except Exception as exc:
            logger.warning("api crypto decrypt failed: path=%s user_id=%s err=%s", path, user_id, exc)
            return _json_error_response(400, "密文解密失败")

        override_method = request.headers.get("x-http-method-override", "").upper()
        payload = plain.get("data", plain)
        if not isinstance(payload, dict):
            payload = {}

        if override_method == "GET":
            request.scope["method"] = "GET"
            request.scope["query_string"] = urlencode(payload, doseq=True).encode("utf-8")
            body_for_route = b""
        else:
            body_for_route = json.dumps(payload, ensure_ascii=False).encode("utf-8")

        async def receive():
            return {"type": "http.request", "body": body_for_route, "more_body": False}

        request = Request(request.scope, receive)
        response = await call_next(request)

        content_type = (response.headers.get("content-type") or "").lower()
        is_json_response = "application/json" in content_type or response.media_type == "application/json"
        if not is_json_response:
            return response

        raw_resp = b""
        if hasattr(response, "body_iterator"):
            chunks: list[bytes] = []
            async for chunk in response.body_iterator:
                chunks.append(chunk)
            raw_resp = b"".join(chunks)
        elif getattr(response, "body", None) is not None:
            raw_resp = response.body
        if not raw_resp:
            return response

        try:
            resp_obj = json.loads(raw_resp.decode("utf-8"))
            if not isinstance(resp_obj, dict):
                return response
        except Exception:
            return response

        try:
            resp_ts = int(time.time() * 1000)
            resp_nonce = api_crypto.gen_nonce_hex()
            resp_key = api_crypto.derive_key(
                fixed_secret=settings.API_ENCRYPT_FIXED_SECRET,
                user_id=str(user_id),
                ts=resp_ts,
                nonce=resp_nonce,
            )
            resp_edata = api_crypto.encrypt_json(resp_obj, resp_key)
            wrapped = {
                "code": resp_obj.get("code", 200),
                "encrypted": True,
                "ts": resp_ts,
                "nonce": resp_nonce,
                "edata": resp_edata,
            }
        except Exception as exc:
            logger.warning("api crypto response encrypt failed: path=%s user_id=%s err=%s", path, user_id, exc)
            if settings.API_ENCRYPT_FAIL_OPEN:
                return response
            return _json_error_response(500, "响应加密失败")
        headers = {k: v for k, v in response.headers.items() if k.lower() != "content-length"}
        return Response(
            content=json.dumps(wrapped, ensure_ascii=False),
            status_code=response.status_code,
            headers=headers,
            media_type="application/json",
            background=response.background,
        )