mtsd/backend/app/core/request_payload.py

"""统一解析接口请求载荷（明文 / 密文信封）。"""

from __future__ import annotations

import logging
from typing import Any, Union

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


def resolve_request_payload(
    request_body: dict,
    current_user: dict,
    *,
    source: str = "unknown",
) -> Union[dict, list]:
    """兼容明文与密文信封，返回业务层 payload（多为 dict；批量数组接口解密后为 list）。"""
    logger = logging.getLogger("sqxp")
    payload = request_body if isinstance(request_body, dict) else {}
    if int(payload.get("enc", 0) or 0) == 1 and isinstance(payload.get("edata"), str):
        logger.warning(
            "%s received encrypted envelope in route handler; middleware may be bypassed",
            source,
        )
        try:
            user_id_for_crypto = str(current_user.get("user_id") or "")
            ts = int(payload.get("ts"))
            nonce = str(payload.get("nonce") or "")
            if user_id_for_crypto and nonce and settings.API_ENCRYPT_FIXED_SECRET:
                key = api_crypto.derive_key(
                    fixed_secret=settings.API_ENCRYPT_FIXED_SECRET,
                    user_id=user_id_for_crypto,
                    ts=ts,
                    nonce=nonce,
                )
                plain = api_crypto.decrypt_json(str(payload.get("edata")), key)
                payload = plain if isinstance(plain, dict) else {}
                logger.warning(
                    "%s decrypted payload via route fallback; please verify ApiCryptoMiddleware",
                    source,
                )
        except Exception as exc:
            logger.warning("%s resolve_request_payload decrypt failed: err=%s", source, exc)
            return {}
    if isinstance(payload, dict):
        inner = payload.get("data")
        if isinstance(inner, (dict, list)):
            return inner
    return payload if isinstance(payload, dict) else {}