72 lines
2.8 KiB
Python
72 lines
2.8 KiB
Python
"""统一解析接口请求载荷（明文 / 密文信封）。"""
|
|
|
|
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 {}
|
|
dw = source.startswith("delivery_warning")
|
|
if isinstance(payload, dict):
|
|
inner = payload.get("data")
|
|
if isinstance(inner, (dict, list)):
|
|
if dw:
|
|
if isinstance(inner, dict):
|
|
logger.info(
|
|
"%s unwrap: using inner dict keys=%s note_len=%s remark_len=%s",
|
|
source,
|
|
sorted(inner.keys()),
|
|
len(str(inner.get("note") or "")),
|
|
len(str(inner.get("remark") or "")),
|
|
)
|
|
else:
|
|
logger.info("%s unwrap: inner is list len=%s", source, len(inner))
|
|
return inner
|
|
if dw:
|
|
logger.info(
|
|
"%s no_inner_data: payload_keys=%s (no dict/list data wrapper)",
|
|
source,
|
|
sorted(payload.keys()),
|
|
)
|
|
elif dw:
|
|
logger.info("%s payload_not_dict type=%s", source, type(payload).__name__)
|
|
return payload if isinstance(payload, dict) else {}
|
|
|