xsl
6c77ff4989
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）。 偏差说明：无
237 lines
9.7 KiB
Python
237 lines
9.7 KiB
Python
"""华为账号登录：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
|