meida_backend/tests/unit/test_huawei_verifier.py

"""HuaweiIdTokenVerifier 纯逻辑单测——自造 RSA 密钥对 + 本地 JWKS，零外网。

覆盖：合法 claims 解析、错误签名、过期、aud/iss 不符、audience 未配置、
union_id 缺失、非 RS256 算法、JWKS 拉取失败、kid 轮换刷新与缓存 TTL。
路由与 DB 路径由 tests/integration/test_auth_huawei.py 验证。
"""

from __future__ import annotations

import json
import time
from collections.abc import Awaitable, Callable
from typing import Any

import jwt
import pytest
from cryptography.hazmat.primitives.asymmetric import rsa
from jwt.algorithms import RSAAlgorithm

from packages.identity.huawei import HuaweiIdTokenVerifier
from packages.shared.errors import AppError, ErrorCode

AUDIENCE = "app-1008"
ISSUER = "https://accounts.huawei.com"
JWKS_URL = "https://jwks.local/certs"

Fetch = Callable[[str], Awaitable[str]]


class FakeFetcher:
    """本地 JWKS 源：可热替换内容（模拟轮换）、统计拉取次数、可抛错。"""

    def __init__(self, jwks: dict[str, Any]) -> None:
        self.jwks = jwks
        self.calls = 0
        self.fail = False

    async def __call__(self, url: str) -> str:
        assert url == JWKS_URL
        self.calls += 1
        if self.fail:
            msg = "network down"
            raise RuntimeError(msg)
        return json.dumps(self.jwks)


def _keypair() -> rsa.RSAPrivateKey:
    return rsa.generate_private_key(public_exponent=65537, key_size=2048)


def _jwk(kid: str, private_key: rsa.RSAPrivateKey) -> dict[str, Any]:
    jwk: dict[str, Any] = json.loads(RSAAlgorithm.to_jwk(private_key.public_key()))
    jwk.update({"kid": kid, "alg": "RS256", "use": "sig"})
    return jwk


def _token(
    private_key: rsa.RSAPrivateKey | str,
    *,
    kid: str = "kid-1",
    alg: str = "RS256",
    aud: str = AUDIENCE,
    iss: str = ISSUER,
    exp_in: int = 600,
    iat_in: int = -60,
    payload_overrides: dict[str, Any] | None = None,
    drop_claims: tuple[str, ...] = (),
) -> str:
    """自签 id_token。时间参数以当前时刻为基准（exp_in/iat_in 为相对秒）。"""
    now = int(time.time())
    payload: dict[str, Any] = {
        "iss": iss,
        "aud": aud,
        "iat": now + iat_in,
        "exp": now + exp_in,
        "union_id": "union-1",
        "open_id": "open-1",
    }
    payload.update(payload_overrides or {})
    for claim in drop_claims:
        payload.pop(claim, None)
    return jwt.encode(payload, private_key, algorithm=alg, headers={"kid": kid})


def _verifier(
    fetcher: Fetch,
    *,
    audience: str = AUDIENCE,
    cache_seconds: int = 86400,
) -> HuaweiIdTokenVerifier:
    return HuaweiIdTokenVerifier(
        jwks_url=JWKS_URL,
        audience=audience,
        issuer=ISSUER,
        cache_seconds=cache_seconds,
        http_fetch=fetcher,
    )


def _assert_code(exc_info: pytest.ExceptionInfo[AppError], code: ErrorCode) -> None:
    assert exc_info.value.code == code, (
        f"期望 {code}({int(code)})，实际 {exc_info.value.code}({int(exc_info.value.code)})"
    )


# --- 合法路径 ---------------------------------------------------------------


async def test_valid_token_parses_claims() -> None:
    key = _keypair()
    fetcher = FakeFetcher({"keys": [_jwk("kid-1", key)]})
    before = int(time.time())
    claims = await _verifier(fetcher).verify(_token(key))
    after = int(time.time())

    assert claims.union_id == "union-1"
    assert claims.open_id == "open-1"
    # 时间断言容忍秒边界（token 在 before/after 之间签发）
    assert before - 60 <= claims.iat <= after - 59
    assert before + 600 <= claims.exp <= after + 601


async def test_valid_token_without_open_id_still_maps_by_union_id() -> None:
    # open_id 可缺（union_id 才是映射键），缺了不应拒绝登录
    key = _keypair()
    fetcher = FakeFetcher({"keys": [_jwk("kid-1", key)]})
    claims = await _verifier(fetcher).verify(_token(key, drop_claims=("open_id",)))
    assert claims.union_id == "union-1"
    assert claims.open_id is None


async def test_union_id_camel_case_claim_is_accepted() -> None:
    # 联调前对 snake/camel 两种拼法都收（见 huawei._first_str），此处锁行为
    key = _keypair()
    fetcher = FakeFetcher({"keys": [_jwk("kid-1", key)]})
    token = _token(key, payload_overrides={"unionID": "union-c"}, drop_claims=("union_id",))
    claims = await _verifier(fetcher).verify(token)
    assert claims.union_id == "union-c"


# --- token 侧拒绝（40102 / 40101）------------------------------------------


async def test_wrong_signing_key_rejected() -> None:
    # kid 相同但用另一把私钥签 → JWKS 命中 kid 却验签失败 → 40102
    signing_key, jwks_key = _keypair(), _keypair()
    fetcher = FakeFetcher({"keys": [_jwk("kid-1", jwks_key)]})
    with pytest.raises(AppError) as exc_info:
        await _verifier(fetcher).verify(_token(signing_key, kid="kid-1"))
    _assert_code(exc_info, ErrorCode.TOKEN_INVALID)


async def test_expired_token_rejected() -> None:
    key = _keypair()
    fetcher = FakeFetcher({"keys": [_jwk("kid-1", key)]})
    with pytest.raises(AppError) as exc_info:
        await _verifier(fetcher).verify(_token(key, exp_in=-30, iat_in=-3600))
    _assert_code(exc_info, ErrorCode.TOKEN_EXPIRED)


async def test_wrong_audience_rejected() -> None:
    key = _keypair()
    fetcher = FakeFetcher({"keys": [_jwk("kid-1", key)]})
    with pytest.raises(AppError) as exc_info:
        await _verifier(fetcher).verify(_token(key, aud="someone-else"))
    _assert_code(exc_info, ErrorCode.TOKEN_INVALID)


async def test_missing_audience_claim_rejected() -> None:
    key = _keypair()
    fetcher = FakeFetcher({"keys": [_jwk("kid-1", key)]})
    with pytest.raises(AppError) as exc_info:
        await _verifier(fetcher).verify(_token(key, drop_claims=("aud",)))
    _assert_code(exc_info, ErrorCode.TOKEN_INVALID)


async def test_wrong_issuer_rejected() -> None:
    key = _keypair()
    fetcher = FakeFetcher({"keys": [_jwk("kid-1", key)]})
    with pytest.raises(AppError) as exc_info:
        await _verifier(fetcher).verify(_token(key, iss="https://evil.example.com"))
    _assert_code(exc_info, ErrorCode.TOKEN_INVALID)


async def test_missing_iat_rejected() -> None:
    key = _keypair()
    fetcher = FakeFetcher({"keys": [_jwk("kid-1", key)]})
    with pytest.raises(AppError) as exc_info:
        await _verifier(fetcher).verify(_token(key, drop_claims=("iat",)))
    _assert_code(exc_info, ErrorCode.TOKEN_INVALID)


async def test_audience_unconfigured_rejected() -> None:
    # AGC 未配置（HUAWEI_AUDIENCE 空）时必须拒绝而不是跳过 aud 校验
    key = _keypair()
    fetcher = FakeFetcher({"keys": [_jwk("kid-1", key)]})
    with pytest.raises(AppError) as exc_info:
        await _verifier(fetcher, audience="").verify(_token(key))
    _assert_code(exc_info, ErrorCode.TOKEN_INVALID)
    # 且应在拉取 JWKS 之前就拒绝（未联调状态不该打上游）
    assert fetcher.calls == 0


async def test_missing_union_id_rejected() -> None:
    key = _keypair()
    fetcher = FakeFetcher({"keys": [_jwk("kid-1", key)]})
    with pytest.raises(AppError) as exc_info:
        await _verifier(fetcher).verify(_token(key, drop_claims=("union_id",)))
    _assert_code(exc_info, ErrorCode.TOKEN_INVALID)


async def test_hs256_token_rejected_before_verification() -> None:
    # alg 混淆攻击：拿对称密钥伪造 header，必须在取 JWKS 前拒绝
    # （密钥给足 32 字节，避免 PyJWT 的短密钥告警干扰输出）
    key = _keypair()
    fetcher = FakeFetcher({"keys": [_jwk("kid-1", key)]})
    with pytest.raises(AppError) as exc_info:
        await _verifier(fetcher).verify(
            _token("hmac-secret-0123456789abcdef0123456789abcdef", alg="HS256", kid="kid-1")
        )
    _assert_code(exc_info, ErrorCode.TOKEN_INVALID)
    assert fetcher.calls == 0


async def test_garbage_token_rejected() -> None:
    fetcher = FakeFetcher({"keys": []})
    with pytest.raises(AppError) as exc_info:
        await _verifier(fetcher).verify("not-a-jwt-at-all")
    _assert_code(exc_info, ErrorCode.TOKEN_INVALID)


# --- JWKS 侧（60000 / kid 轮换 / 缓存）--------------------------------------


async def test_jwks_fetch_failure_maps_upstream_error() -> None:
    key = _keypair()
    fetcher = FakeFetcher({"keys": [_jwk("kid-1", key)]})
    fetcher.fail = True
    with pytest.raises(AppError) as exc_info:
        await _verifier(fetcher).verify(_token(key))
    _assert_code(exc_info, ErrorCode.UPSTREAM_ERROR)


async def test_jwks_payload_not_jwks_shape_maps_upstream_error() -> None:
    key = _keypair()
    fetcher = FakeFetcher({"keys": [_jwk("kid-1", key)]})
    fetcher.jwks = {"unexpected": True}
    with pytest.raises(AppError) as exc_info:
        await _verifier(fetcher).verify(_token(key))
    _assert_code(exc_info, ErrorCode.UPSTREAM_ERROR)


async def test_unknown_kid_rejected_after_one_forced_refresh() -> None:
    key = _keypair()
    other = _keypair()
    fetcher = FakeFetcher({"keys": [_jwk("kid-1", key)]})
    with pytest.raises(AppError) as exc_info:
        await _verifier(fetcher).verify(_token(other, kid="kid-unknown"))
    _assert_code(exc_info, ErrorCode.TOKEN_INVALID)
    # 未命中 → 强制刷新一次 → 仍没有 → 拒绝：恰好两次拉取
    assert fetcher.calls == 2


async def test_kid_rotation_refreshes_cache_and_retry_succeeds() -> None:
    old_key, new_key = _keypair(), _keypair()
    fetcher = FakeFetcher({"keys": [_jwk("kid-1", old_key)]})
    verifier = _verifier(fetcher)
    claims = await verifier.verify(_token(old_key, kid="kid-1"))
    assert claims.union_id == "union-1"
    assert fetcher.calls == 1  # TTL 内第二次验证应走缓存

    # 上游轮换：JWKS 换成 kid-2，本地缓存仍是旧 keys
    fetcher.jwks = {"keys": [_jwk("kid-2", new_key)]}
    claims = await verifier.verify(
        _token(new_key, kid="kid-2", payload_overrides={"union_id": "union-2"})
    )
    assert claims.union_id == "union-2"
    assert fetcher.calls == 2  # kid 未命中触发一次强制刷新，随后验签成功


async def test_cache_reuses_jwks_within_ttl() -> None:
    key = _keypair()
    fetcher = FakeFetcher({"keys": [_jwk("kid-1", key)]})
    verifier = _verifier(fetcher)
    for _ in range(3):
        await verifier.verify(_token(key))
    assert fetcher.calls == 1


async def test_cache_expires_when_ttl_is_zero() -> None:
    key = _keypair()
    fetcher = FakeFetcher({"keys": [_jwk("kid-1", key)]})
    verifier = _verifier(fetcher, cache_seconds=0)
    await verifier.verify(_token(key))
    await verifier.verify(_token(key))
    assert fetcher.calls == 2