feat(identity): Huawei Account Kit login via OIDC JWKS verification

POST /auth/huawei：JWKS 验签（kid 缓存/轮换）+ claims 严格校验 +
UnionID 映射 + T4 双令牌签发（FR-ACC-01；真实联调待 AGC 配置）。

- packages/identity/huawei.py（新建）：HuaweiIdTokenVerifier——RS256 验签，
  JWKS 内存缓存（TTL 可配 + 单调钟；kid 未命中强制刷新一次重试，锁内用
  拉取代数防止强制刷新被新鲜缓存短路）；aud/iss/exp/iat 全查，audience
  空配置直接拒绝而非跳过；union_id 缺失拒绝（降级 open_id 策略不做，
  留联调期决定）；错误映射 40102/40101/60000，无新错误码。
  HuaweiLoginRequest/HuaweiClaims 一并定义于此，未动 schemas.py。
- apps/api/routers/auth.py：追加 /huawei 端点；get_huawei_verifier 依赖
  建于路由模块内（缓存到 app.state，可预注入替换），未动 deps.py；
  复用 T4 的 get_or_create_by_external_identity + issue_session +
  TokenPairResponse 与既有 _token_pair/_access_token。
- tests/unit/test_huawei_verifier.py：cryptography 自造 RSA + 本地 JWKS
  （零外网），19 例覆盖合法 claims/错误签名/过期/aud/iss/iat 缺失/
  audience 未配置/union_id 缺失/HS256 混淆/JWKS 拉取失败/kid 轮换/缓存
  TTL。
- tests/integration/test_auth_huawei.py：FakeFetcher 预注入 app.state，
  真实 PG 验证首次建用户 + 双令牌、同 union_id 复登同 user_id、新 access
  调 /quota、refresh 可轮换、伪造 token 40102。

检查：ruff/format/mypy（T6 列表）/lint-imports 全绿；单测 257 passed
（基线 238 + 新增 19）；集成 18 passed（含新增 2）。

偏差说明：无

apps/api/routers/auth.py

@@ -5,17 +5,22 @@
 且接受的设计权衡（要即时失效就得引入黑名单或极短 TTL，V1 不做）——
 「登出后旧令牌立即失效」的验收对象是 refresh 令牌与会话。
 
-明确不做（防止后人误判「漏了」）：华为账号登录（T-P1-12，等资质）、手机验证码
-（T-P1-12b，等短信通道）、游客/合并/注销。dev-token 端点不是正式鉴权设计，
-仅为纵向切片与联调服务：``DEV_AUTH_ENABLED`` + 本地环境双开关才放行。
+华为账号登录（T-P1-12）已实现（``/auth/huawei``）：Account Kit id_token 的
+OIDC 验签与 UnionID 映射在 :mod:`packages.identity.huawei`，代码先行——真实
+联调待 AGC 配置（见该模块 docstring）。明确不做（防止后人误判「漏了」）：
+手机验证码（T-P1-12b，等短信通道）、游客/合并/注销。dev-token 端点不是正式
+鉴权设计，仅为纵向切片与联调服务：``DEV_AUTH_ENABLED`` + 本地环境双开关才放行。
 """
 
 from __future__ import annotations
 
-from fastapi import APIRouter
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Request
 
 from apps.api.deps import CurrentUserId, IdentityServiceDep, SettingsDep
 from apps.api.schemas import DevTokenRequest
+from packages.identity.huawei import HuaweiIdTokenVerifier, HuaweiLoginRequest
 from packages.identity.schemas import LogoutRequest, RefreshRequest, TokenPairResponse
 from packages.identity.service import DEV_PROVIDER, SessionIssued
 from packages.shared.config import Settings
@@ -92,3 +97,48 @@ async def logout(
     revoked = await identity.revoke(payload.refresh_token)
     # 幂等登出：查无此 token 或已撤销都返回 revoked=false，不报错
     return ok({"revoked": revoked})
+
+
+def get_huawei_verifier(request: Request, settings: SettingsDep) -> HuaweiIdTokenVerifier:
+    """从配置构造华为验签器，缓存到 app.state（JWKS 缓存跨请求复用）。
+
+    依赖建在路由模块内而非 deps.py——deps.py 是本批共享禁改文件。
+    预注入 ``app.state.huawei_verifier`` 可整体替换（集成测试借这条缝
+    注入 FakeFetcher，全链路零外网）。
+    """
+    verifier: HuaweiIdTokenVerifier | None = getattr(request.app.state, "huawei_verifier", None)
+    if verifier is None:
+        verifier = HuaweiIdTokenVerifier(
+            jwks_url=settings.huawei_jwks_url,
+            audience=settings.huawei_audience,
+            issuer=settings.huawei_issuer,
+            cache_seconds=settings.huawei_jwks_cache_seconds,
+        )
+        request.app.state.huawei_verifier = verifier
+    return verifier
+
+
+HuaweiVerifierDep = Annotated[HuaweiIdTokenVerifier, Depends(get_huawei_verifier)]
+
+
+@router.post("/huawei", summary="华为账号授权登录（FR-ACC-01）")
+async def huawei_login(
+    payload: HuaweiLoginRequest,
+    settings: SettingsDep,
+    identity: IdentityServiceDep,
+    verifier: HuaweiVerifierDep,
+) -> dict[str, object]:
+    """Account Kit id_token → JWKS 验签 → UnionID 映射 → T4 双令牌。
+
+    audience 未配置（AGC 未就绪）时 verify 直接 40102，端点天然不可用——
+    这是有意的：联调前放行等于允许任意自签 token 登录。
+    """
+    claims = await verifier.verify(payload.id_token)
+    user = await identity.get_or_create_by_external_identity(
+        provider="huawei",
+        union_id=claims.union_id,
+        open_id=claims.open_id,
+        register_source="huawei",
+    )
+    issued = await identity.issue_session(user.user_id)
+    return ok(_token_pair(settings, _access_token(settings, user.user_id), issued))

packages/identity/huawei.py

@@ -0,0 +1,236 @@
+"""华为账号登录：Account Kit id_token 的 OIDC 验签（T10，FR-ACC-01）。
+
+**代码先行，真实联调后置**。联调前置（《前置开通清单》§3）：AGC 创建应用拿到
+App ID、调试/发布**两套证书**的 SHA256 指纹都录入 AGC（漏配发布指纹的表现是
+"调试包登录正常、发布包登录失败"）、开通 Account Kit；随后在 `.env` 填
+``HUAWEI_AUDIENCE``（= App ID）即可上线，代码零改动。
+
+验签策略（联调前后都严格，不因未联调而放松）：
+
+- RS256 验签：先读未验签 header 锁定 ``alg`` 与 ``kid``，再从 JWKS 按 kid
+  取公钥；JWKS 内存缓存（TTL 可配），kid 未命中视为密钥轮换信号，强制刷新
+  一次后重找，仍找不到才判无效；
+- claims 全查：``aud``（必须等于配置 audience）、``iss``、``exp``、``iat``。
+  **audience 空配置时直接拒绝而不是跳过校验**——联调前一旦放行，等于任何
+  自签 token 都能登录；
+- ``union_id`` 缺失拒绝。华为子账号场景可能无 union_id，"降级用 open_id"
+  的策略**不做**：是否降级涉及账号唯一性语义，留给联调期拿真实账号形态再
+  决定，宁可现在拒掉。
+
+请求模型 :class:`HuaweiLoginRequest` 放在本模块而非 schemas.py——那本是 T11
+（短信登录）的改动面，避免两分支相遇。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import time
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from typing import Any
+
+import httpx
+import jwt
+from pydantic import BaseModel, Field
+
+from packages.shared.errors import AppError, ErrorCode
+
+#: Account Kit 只接受 RS256；header 声明其它算法（含 none/HS256 混淆攻击）一律拒绝。
+_EXPECTED_ALG = "RS256"
+
+#: http_fetch / now 的类型：均可注入，单测零外网、时钟可控。
+HttpFetch = Callable[[str], Awaitable[str]]
+Clock = Callable[[], float]
+
+
+async def _default_http_fetch(url: str) -> str:
+    """默认 JWKS 拉取（httpx 已是项目依赖，不算新增依赖）。
+
+    JWKS 访问极低频（TTL 默认一天），用短生命周期客户端即可——identity 域
+    没有 aclose 生命周期钩子，持有长连接反而无处释放。
+    """
+    async with httpx.AsyncClient(timeout=httpx.Timeout(10.0)) as client:
+        response = await client.get(url)
+        response.raise_for_status()
+        return response.text
+
+
+class HuaweiLoginRequest(BaseModel):
+    """``POST /auth/huawei`` 请求体（模型归 huawei.py，见模块 docstring）。"""
+
+    #: Account Kit 签发的 JWT 三段式，长度下限只挡明显误传，语义校验交给验签器。
+    id_token: str = Field(min_length=16, max_length=8192)
+
+
+@dataclass(frozen=True, slots=True)
+class HuaweiClaims:
+    """验签通过后留下的 claims——只保留登录映射所需字段，其余不透传。"""
+
+    union_id: str
+    open_id: str | None
+    exp: int
+    iat: int
+
+
+class HuaweiIdTokenVerifier:
+    """华为 id_token 验签器（JWKS 拉取经 ``http_fetch`` 注入，可测试）。"""
+
+    def __init__(
+        self,
+        jwks_url: str,
+        audience: str,
+        issuer: str,
+        cache_seconds: int,
+        *,
+        http_fetch: HttpFetch | None = None,
+        now: Clock | None = None,
+    ) -> None:
+        self._jwks_url = jwks_url
+        self._audience = audience
+        self._issuer = issuer
+        self._cache_seconds = cache_seconds
+        self._http_fetch: HttpFetch = http_fetch or _default_http_fetch
+        # TTL 用单调钟：墙钟跳变（NTP 校时）不应把一天缓存放大或清零。
+        self._now: Clock = now or time.monotonic
+        self._jwks: dict[str, Any] | None = None
+        self._fetched_at: float | None = None
+        self._generation = 0
+        self._lock = asyncio.Lock()
+
+    async def verify(self, id_token: str) -> HuaweiClaims:
+        """验签 + claims 校验，失败抛 :class:`AppError`（40102/40101/60000）。"""
+        if not self._audience:
+            # 空配置 = 未联调状态。拒绝而不是跳过 aud 校验（见模块 docstring）。
+            raise AppError(
+                ErrorCode.TOKEN_INVALID, message="华为登录未配置 audience（HUAWEI_AUDIENCE）"
+            )
+
+        header = self._unverified_header(id_token)
+        if header.get("alg") != _EXPECTED_ALG:
+            raise AppError(ErrorCode.TOKEN_INVALID, message="id_token 算法不允许")
+        kid = header.get("kid")
+        if not isinstance(kid, str) or not kid:
+            raise AppError(ErrorCode.TOKEN_INVALID, message="id_token 缺少 kid")
+
+        jwk = await self._locate_jwk(kid)
+        payload = self._decode(id_token, jwk)
+        return self._to_claims(payload)
+
+    # --- 内部分步 ---------------------------------------------------------------
+
+    def _unverified_header(self, id_token: str) -> dict[str, Any]:
+        """只读 header 取 kid/alg，不做信任判断——信任来自后面的验签。"""
+        try:
+            header = jwt.get_unverified_header(id_token)
+        except jwt.InvalidTokenError as exc:
+            raise AppError(ErrorCode.TOKEN_INVALID, message="id_token 格式非法") from exc
+        return dict(header)
+
+    async def _locate_jwk(self, kid: str) -> dict[str, Any]:
+        jwk = self._find_kid(await self._get_jwks(), kid)
+        if jwk is None:
+            # 缓存里没有这个 kid：大概率是上游刚轮换密钥，强制刷新一次再找。
+            jwk = self._find_kid(await self._get_jwks(force_refresh=True), kid)
+        if jwk is None:
+            raise AppError(ErrorCode.TOKEN_INVALID, message="id_token 的 kid 不在 JWKS 中")
+        return jwk
+
+    async def _get_jwks(self, *, force_refresh: bool = False) -> dict[str, Any]:
+        if not force_refresh and self._cache_fresh():
+            jwks = self._jwks
+            if jwks is not None:
+                return jwks
+        # 记下进入前的拉取代数：锁内只有「别人确实刷新过」才复用——不能只看
+        # TTL 新鲜，否则强制刷新（kid 轮换重试）会被新鲜但过时的缓存短路。
+        seen_generation = self._generation
+        async with self._lock:
+            if self._generation != seen_generation:
+                jwks = self._jwks
+                if jwks is not None:
+                    return jwks
+            jwks = await self._fetch_jwks()
+            self._jwks = jwks
+            self._fetched_at = self._now()
+            self._generation += 1
+            return jwks
+
+    def _cache_fresh(self) -> bool:
+        return (
+            self._jwks is not None
+            and self._fetched_at is not None
+            and self._now() - self._fetched_at < self._cache_seconds
+        )
+
+    async def _fetch_jwks(self) -> dict[str, Any]:
+        try:
+            raw = await self._http_fetch(self._jwks_url)
+            jwks = json.loads(raw)
+        except AppError:
+            raise
+        except Exception as exc:
+            raise AppError(ErrorCode.UPSTREAM_ERROR, message="华为 JWKS 拉取失败") from exc
+        if not isinstance(jwks, dict) or not isinstance(jwks.get("keys"), list):
+            # 拉到了但结构不是 JWKS：上游数据损坏，与拉取失败同级。
+            raise AppError(ErrorCode.UPSTREAM_ERROR, message="华为 JWKS 格式异常")
+        return jwks
+
+    @staticmethod
+    def _find_kid(jwks: dict[str, Any], kid: str) -> dict[str, Any] | None:
+        for jwk in jwks["keys"]:
+            if isinstance(jwk, dict) and jwk.get("kid") == kid:
+                return jwk
+        return None
+
+    def _decode(self, id_token: str, jwk: dict[str, Any]) -> dict[str, Any]:
+        try:
+            key = jwt.PyJWK(jwk).key
+        except Exception as exc:
+            # kid 匹配但公钥材料损坏：坏的是上游 JWKS，不是这枚 token。
+            raise AppError(ErrorCode.UPSTREAM_ERROR, message="华为 JWKS 公钥解析失败") from exc
+        try:
+            return jwt.decode(
+                id_token,
+                key,
+                algorithms=[_EXPECTED_ALG],
+                audience=self._audience,
+                issuer=self._issuer,
+                options={"require": ["exp", "iat", "aud", "iss"]},
+            )
+        except jwt.ExpiredSignatureError as exc:
+            raise AppError(ErrorCode.TOKEN_EXPIRED) from exc
+        except jwt.InvalidTokenError as exc:
+            # 验签失败 / aud 或 iss 不符 / 必需 claim 缺失，统一按无效令牌处理。
+            raise AppError(ErrorCode.TOKEN_INVALID, message="id_token 验签未通过") from exc
+
+    def _to_claims(self, payload: dict[str, Any]) -> HuaweiClaims:
+        union_id = _first_str(payload, "union_id", "unionID", "unionId")
+        if union_id is None:
+            # 拒绝理由与"是否降级 open_id"的取舍见模块 docstring。
+            raise AppError(ErrorCode.TOKEN_INVALID, message="id_token 缺少 union_id")
+        return HuaweiClaims(
+            union_id=union_id,
+            open_id=_first_str(payload, "open_id", "openID", "openId"),
+            exp=_int_claim(payload, "exp"),
+            iat=_int_claim(payload, "iat"),
+        )
+
+
+def _first_str(payload: dict[str, Any], *names: str) -> str | None:
+    """按候选名取第一个非空字符串 claim。
+
+    华为文档 snake/camel 两种拼法都见过（union_id / unionID），代码先行阶段
+    两种都收；联调拿到真实样本后在此收紧。
+    """
+    for name in names:
+        value = payload.get(name)
+        if isinstance(value, str) and value:
+            return value
+    return None
+
+
+def _int_claim(payload: dict[str, Any], name: str) -> int:
+    try:
+        return int(payload[name])
+    except (KeyError, TypeError, ValueError) as exc:
+        raise AppError(ErrorCode.TOKEN_INVALID, message=f"id_token 的 {name} 非法") from exc

tests/integration/test_auth_huawei.py

@@ -0,0 +1,140 @@
+"""POST /v1/auth/huawei 集成——真实 PostgreSQL + FakeFetcher 本地 JWKS，零外网。
+
+覆盖（FR-ACC-01）：首次登录自动建用户 + T4 双令牌、同 union_id 复登映射到
+同一 user_id、新 access 可调 /quota、伪造 id_token 被拒（40102）。
+验签细节（claims/kid 轮换）由 tests/unit/test_huawei_verifier.py 穷举。
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+import uuid
+from collections.abc import AsyncIterator
+from typing import Any
+
+import httpx
+import jwt
+import pytest
+from cryptography.hazmat.primitives.asymmetric import rsa
+from jwt.algorithms import RSAAlgorithm
+
+from packages.identity.huawei import HuaweiIdTokenVerifier
+
+AUDIENCE = "app-int-1008"
+ISSUER = "https://accounts.huawei.com"
+JWKS_URL = "https://jwks.local/certs"
+KID = "kid-int"
+
+# 模块级固定密钥对：同一用例内多次登录共享 JWKS，避免每个 token 重新生成密钥。
+_SIGNING_KEY = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+_WRONG_KEY = 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
+
+
+async def _local_jwks_fetch(url: str) -> str:
+    assert url == JWKS_URL
+    return json.dumps({"keys": [_jwk(KID, _SIGNING_KEY)]})
+
+
+def _id_token(*, union_id: str | None = None, key: rsa.RSAPrivateKey = _SIGNING_KEY) -> str:
+    """自签 id_token：union_id 缺省随机（新用户），指定则复现同一外部身份。"""
+    now = int(time.time())
+    payload: dict[str, Any] = {
+        "iss": ISSUER,
+        "aud": AUDIENCE,
+        "iat": now - 60,
+        "exp": now + 600,
+        "union_id": union_id or f"union-{uuid.uuid4().hex[:12]}",
+        "open_id": f"open-{uuid.uuid4().hex[:12]}",
+    }
+    return jwt.encode(payload, key, algorithm="RS256", headers={"kid": KID})
+
+
+def _infra_available() -> bool:
+    return bool(os.getenv("MEIDA_RUN_INTEGRATION"))
+
+
+pytestmark = [
+    pytest.mark.integration,
+    pytest.mark.skipif(
+        not _infra_available(),
+        reason="set MEIDA_RUN_INTEGRATION=1 with PostgreSQL and Redis running",
+    ),
+]
+
+
+@pytest.fixture
+async def client() -> AsyncIterator[httpx.AsyncClient]:
+    """预注入 FakeFetcher 验签器到 app.state。
+
+    get_huawei_verifier 优先读 app.state 缓存，因此注入后整条链路不触网；
+    verifier 自带 audience/issuer，也不依赖 .env 是否已配 AGC 凭证。
+    """
+    from apps.api.main import create_app
+    from packages.shared.config import get_settings
+
+    app = create_app(get_settings())
+    app.state.huawei_verifier = HuaweiIdTokenVerifier(
+        jwks_url=JWKS_URL,
+        audience=AUDIENCE,
+        issuer=ISSUER,
+        cache_seconds=86400,
+        http_fetch=_local_jwks_fetch,
+    )
+    transport = httpx.ASGITransport(app=app)
+    async with (
+        httpx.AsyncClient(transport=transport, base_url="http://test") as http,
+        app.router.lifespan_context(app),
+    ):
+        yield http
+
+
+async def _login(client: httpx.AsyncClient, id_token: str) -> dict[str, Any]:
+    resp = await client.post("/v1/auth/huawei", json={"id_token": id_token})
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["code"] == 0
+    return body["data"]
+
+
+async def test_huawei_login_issues_pair_relogin_maps_same_user(client: httpx.AsyncClient) -> None:
+    union_id = f"union-{uuid.uuid4().hex[:12]}"
+
+    # 首次登录：自动建用户 + T4 双令牌（契约字段与 dev-token/refresh 一致）
+    first = await _login(client, _id_token(union_id=union_id))
+    assert first["access_token"]
+    assert first["refresh_token"]
+    assert first["user_id"]
+    assert first["token_type"] == "Bearer"
+    assert first["refresh_expires_in"] > 0
+
+    # 同 union_id 复登：映射唯一（FR-ACC-01），换发全新会话
+    second = await _login(client, _id_token(union_id=union_id))
+    assert second["user_id"] == first["user_id"]
+    assert second["refresh_token"] != first["refresh_token"]
+
+    # 新 access 直接可调需鉴权接口
+    quota = await client.get(
+        "/v1/quota", headers={"Authorization": f"Bearer {first['access_token']}"}
+    )
+    assert quota.status_code == 200
+    assert quota.json()["code"] == 0
+
+    # 华为签发的 refresh 同样进 T4 轮换链路
+    rotated = await client.post("/v1/auth/refresh", json={"refresh_token": first["refresh_token"]})
+    assert rotated.status_code == 200
+    assert rotated.json()["data"]["refresh_token"] != first["refresh_token"]
+
+
+async def test_forged_id_token_rejected(client: httpx.AsyncClient) -> None:
+    # 同 kid 但另一把私钥签发：JWKS 命中 kid、验签失败 → 40102
+    resp = await client.post("/v1/auth/huawei", json={"id_token": _id_token(key=_WRONG_KEY)})
+    assert resp.status_code == 401
+    assert resp.json()["code"] == 40102

tests/unit/test_huawei_verifier.py

@@ -0,0 +1,296 @@
+"""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