meida_backend/tests/integration/test_auth_huawei.py

"""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