MeshSate/app/security/signing.py

"""API 请求签名与响应签名。

请求签名规范（HMAC-SHA256）
---------------------------
客户端需在请求头中携带：

- ``X-MS-Client-Id``    调用方标识
- ``X-MS-Timestamp``    Unix 秒级时间戳
- ``X-MS-Nonce``        随机串，32 位十六进制，服务端保证一次性
- ``X-MS-Body-SHA256``  请求体字节的 SHA256（无请求体时为空串的 SHA256）
- ``X-MS-Signature``    签名值

待签名串由以下部分按固定顺序拼接，换行符为 ``\\n``::

    METHOD
    PATH
    CANONICAL_QUERY
    BODY_SHA256
    TIMESTAMP
    NONCE
    CLIENT_ID

签名为 ``HMAC-SHA256(请求签名密钥, 待签名串)`` 的小写十六进制。

响应签名
--------
服务端在响应头中返回：

- ``X-MS-Response-Timestamp``
- ``X-MS-Response-Signature``: ``HMAC-SHA256(响应签名密钥, 响应体SHA256 + 时间戳)``

客户端用相同密钥派生规则即可独立验签，从而防止中间人篡改结论。
"""

from __future__ import annotations

import hashlib
import hmac
import json
from typing import Any, Dict, Mapping, Optional

EMPTY_BODY_SHA256 = hashlib.sha256(b"").hexdigest()


def canonical_query(params: Optional[Mapping[str, Any]]) -> str:
    """将查询参数规范化为 ``k=v&k2=v2`` 形式（按 key 升序）。"""
    if not params:
        return ""
    items = []
    for key in sorted(params.keys()):
        value = params[key]
        if value is None:
            continue
        if isinstance(value, (list, tuple)):
            for item in value:
                items.append(f"{key}={item}")
        else:
            items.append(f"{key}={value}")
    return "&".join(items)


def body_sha256(body: bytes) -> str:
    return hashlib.sha256(body).hexdigest()


def build_signing_string(
    method: str,
    path: str,
    query: str,
    body_hash: str,
    timestamp: str,
    nonce: str,
    client_id: str,
) -> str:
    return "\n".join(
        [method.upper(), path, query, body_hash, str(timestamp), nonce, client_id]
    )


def sign_request(
    secret: str,
    client_id: str,
    method: str,
    path: str,
    query: Optional[Mapping[str, Any]] = None,
    body: bytes = b"",
    timestamp: Optional[int] = None,
    nonce: Optional[str] = None,
) -> Dict[str, str]:
    """生成完整的请求签名头集合。"""
    import os
    import time

    ts = str(timestamp if timestamp is not None else int(time.time()))
    nc = nonce or os.urandom(16).hex()
    bh = body_sha256(body)
    qs = canonical_query(query)
    payload = build_signing_string(method, path, qs, bh, ts, nc, client_id)
    key = _signing_key(secret, client_id)
    signature = hmac.new(key, payload.encode("utf-8"), hashlib.sha256).hexdigest()
    return {
        "X-MS-Client-Id": client_id,
        "X-MS-Timestamp": ts,
        "X-MS-Nonce": nc,
        "X-MS-Body-SHA256": bh,
        "X-MS-Signature": signature,
    }


def verify_request_signature(
    secret: str,
    client_id: str,
    method: str,
    path: str,
    query: Optional[Mapping[str, Any]],
    body: bytes,
    timestamp: str,
    nonce: str,
    provided_body_hash: str,
    provided_signature: str,
    *,
    skip_body_hash_check: bool = False,
) -> bool:
    """服务端侧验签。

    skip_body_hash_check=True 时（multipart/form-data 场景，服务端拿不到原始 body），
    直接信任客户端提供的 provided_body_hash，只校验 HMAC 签名部分。
    """
    if skip_body_hash_check:
        expected_body_hash = provided_body_hash or ""
    else:
        expected_body_hash = body_sha256(body)
        if not hmac.compare_digest(expected_body_hash, provided_body_hash or ""):
            return False
    payload = build_signing_string(
        method, path, canonical_query(query), expected_body_hash, timestamp, nonce, client_id
    )
    key = _signing_key(secret, client_id)
    expected = hmac.new(key, payload.encode("utf-8"), hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, provided_signature or "")


def sign_response(secret: str, client_id: str, body_bytes: bytes, timestamp: Optional[int] = None) -> Dict[str, str]:
    """为响应体生成签名头。"""
    import time

    ts = str(timestamp if timestamp is not None else int(time.time()))
    key = _response_key(secret, client_id)
    payload = body_sha256(body_bytes) + ts
    signature = hmac.new(key, payload.encode("utf-8"), hashlib.sha256).hexdigest()
    return {"X-MS-Response-Timestamp": ts, "X-MS-Response-Signature": signature}


def verify_response_signature(
    secret: str, client_id: str, body_bytes: bytes, timestamp: str, signature: str
) -> bool:
    key = _response_key(secret, client_id)
    payload = body_sha256(body_bytes) + timestamp
    expected = hmac.new(key, payload.encode("utf-8"), hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature or "")


def canonical_json_bytes(payload: Any) -> bytes:
    """将响应对象序列化为稳定的 JSON 字节串（供签名使用）。"""
    return json.dumps(
        payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")
    ).encode("utf-8")


def _signing_key(secret: str, client_id: str) -> bytes:
    from app.security.crypto import derive_client_signing_key

    return derive_client_signing_key(secret, client_id)


def _response_key(secret: str, client_id: str) -> bytes:
    from app.security.crypto import derive_client_response_key

    return derive_client_response_key(secret, client_id)