From 6c77ff49898f0a7f627e554a1d3c100e05d0d775 Mon Sep 17 00:00:00 2001 From: xsl Date: Fri, 14 Aug 2026 19:58:48 +0800 Subject: [PATCH] feat(identity): Huawei Account Kit login via OIDC JWKS verification MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 58 ++++- packages/identity/huawei.py | 236 ++++++++++++++++++++ tests/integration/test_auth_huawei.py | 140 ++++++++++++ tests/unit/test_huawei_verifier.py | 296 ++++++++++++++++++++++++++ 4 files changed, 726 insertions(+), 4 deletions(-) create mode 100644 packages/identity/huawei.py create mode 100644 tests/integration/test_auth_huawei.py create mode 100644 tests/unit/test_huawei_verifier.py diff --git a/apps/api/routers/auth.py b/apps/api/routers/auth.py index 4495835..20698e9 100755 --- a/apps/api/routers/auth.py +++ b/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)) diff --git a/packages/identity/huawei.py b/packages/identity/huawei.py new file mode 100644 index 0000000..b6f4ec5 --- /dev/null +++ b/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 diff --git a/tests/integration/test_auth_huawei.py b/tests/integration/test_auth_huawei.py new file mode 100644 index 0000000..12b06d5 --- /dev/null +++ b/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 diff --git a/tests/unit/test_huawei_verifier.py b/tests/unit/test_huawei_verifier.py new file mode 100644 index 0000000..faf705e --- /dev/null +++ b/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