"""T-01 JWT 鉴权（auth_service 验签/吊销 + deps 工厂 + Agent 准入矩阵）。 通道矩阵：JWT（全环境，强制 X-Agent-Type 交叉校验）/ dev debug 头兜底 / 非 dev 无 Bearer 401 留痕。401/403 全部经 deps 审计（P-05），sqlite 断言。 RS256 用 cryptography 现场生成密钥对（生产形态走通）；HS256 走 dev secret。 """ from __future__ import annotations import time from pathlib import Path import pytest from fastapi.testclient import TestClient from jose import jwt as jose_jwt from sqlalchemy import text from _ddl import create_sqlite_engine from app.api import deps as deps_mod from app.api import risk as risk_api from app.config.settings import settings from app.main import app from app.repository.risk_repository import RiskRepository from app.service import auth_service from app.service.risk import redis_gateway class FakeGateway: """Redis fake：吊销黑名单语义（exists/set_ex 记录）。""" def __init__(self): self.revoked: set[str] = set() self.messages: list = [] self.deletes: list = [] def publish(self, channel, payload): self.messages.append((channel, payload)) def delete(self, *keys): self.deletes.append(keys) def exists(self, key: str) -> bool: return key in self.revoked def set_ex(self, key, value, ttl): self.revoked.add(key) @pytest.fixture() def env(monkeypatch): engine = create_sqlite_engine() repo = RiskRepository(engine=engine) monkeypatch.setattr(risk_api, "_repo", lambda: repo) monkeypatch.setattr(deps_mod, "RiskRepository", lambda: repo) monkeypatch.setattr(redis_gateway, "_gateway", FakeGateway()) yield {"client": TestClient(app, raise_server_exceptions=False), "repo": repo, "engine": engine} engine.dispose() def _bearer(tok: str, agent: str = "risk") -> dict: return {"Authorization": f"Bearer {tok}", "X-Agent-Type": agent} def _audit_rows(engine, decision: str) -> list: with engine.connect() as conn: return conn.execute( text("SELECT event_type, actor_id, decision, input_summary FROM audit_log WHERE decision = :d"), {"d": decision}, ).mappings().all() # ---------- JWT 通道基本链路 ---------- def test_jwt_risk_officer_access_alerts(env): tok = auth_service.issue_dev_token(sub="STAFF-30001", roles=["risk_officer"]) r = env["client"].get("/api/risk/alerts", headers=_bearer(tok)) assert r.status_code == 200 assert r.json()["disclaimer"] def test_jwt_missing_agent_type_401_with_audit(env): tok = auth_service.issue_dev_token(sub="STAFF-30001", roles=["risk_officer"]) r = env["client"].get("/api/risk/alerts", headers={"Authorization": f"Bearer {tok}"}) assert r.status_code == 401 assert r.json()["error_code"] == "AUTH_401_MISSING_AGENT_TYPE" assert _audit_rows(env["engine"], "unauthenticated") def test_jwt_invalid_agent_type_400(env): tok = auth_service.issue_dev_token(sub="STAFF-30001", roles=["risk_officer"]) r = env["client"].get("/api/risk/alerts", headers=_bearer(tok, agent="backoffice")) assert r.status_code == 400 and r.json()["error_code"] == "BAD_REQUEST" def test_jwt_customer_token_cannot_enter_risk_agent(env): """手册 §5.4：customer token 只能进 customer agent（交叉校验 403 + 审计）。""" tok = auth_service.issue_dev_token( sub="CUST-9527", roles=["customer"], token_type="customer", customer_id="CUST-9527" ) r = env["client"].get("/api/risk/alerts", headers=_bearer(tok)) assert r.status_code == 403 assert r.json()["error_code"] == "AUTH_403_AGENT_MISMATCH" assert _audit_rows(env["engine"], "forbidden") def test_jwt_advisor_cannot_enter_risk_agent(env): tok = auth_service.issue_dev_token(sub="STAFF-10086", roles=["advisor"]) r = env["client"].get("/api/risk/alerts", headers=_bearer(tok)) assert r.status_code == 403 and r.json()["error_code"] == "AUTH_403_AGENT_MISMATCH" # ---------- 验签失败矩阵（统一 401 INVALID_TOKEN + 留痕） ---------- def test_jwt_tampered_signature_rejected(env): tok = auth_service.issue_dev_token(sub="STAFF-30001", roles=["risk_officer"]) head, payload, sig = tok.split(".") r = env["client"].get("/api/risk/alerts", headers=_bearer(f"{head}.{payload}.AAAA{sig[4:]}")) assert r.status_code == 401 and r.json()["error_code"] == "AUTH_401_INVALID_TOKEN" assert _audit_rows(env["engine"], "unauthenticated") def test_jwt_expired_rejected(env): now = int(time.time()) payload = { "iss": settings.jwt_issuer, "sub": "STAFF-30001", "aud": settings.jwt_audience, "exp": now - 10, "iat": now - 100, "jti": "jti-exp", "token_type": "staff", "roles": ["risk_officer"], "tenant_id": "TENANT-001", } tok = jose_jwt.encode(payload, settings.jwt_dev_secret, algorithm="HS256") r = env["client"].get("/api/risk/alerts", headers=_bearer(tok)) assert r.status_code == 401 and r.json()["error_code"] == "AUTH_401_INVALID_TOKEN" def test_jwt_wrong_issuer_rejected(env): now = int(time.time()) payload = { "iss": "https://evil.example", "sub": "STAFF-30001", "aud": settings.jwt_audience, "exp": now + 600, "iat": now, "jti": "jti-iss", "token_type": "staff", "roles": ["risk_officer"], "tenant_id": "TENANT-001", } tok = jose_jwt.encode(payload, settings.jwt_dev_secret, algorithm="HS256") r = env["client"].get("/api/risk/alerts", headers=_bearer(tok)) assert r.status_code == 401 def test_jwt_customer_id_mismatch_rejected(env): """手册 §4.2：customer token 的 customer_id 必须等于 sub（防冒名）。""" tok = auth_service.issue_dev_token( sub="CUST-9527", roles=["customer"], token_type="customer", customer_id="CUST-1001" ) r = env["client"].get("/api/risk/alerts", headers=_bearer(tok)) assert r.status_code == 401 and r.json()["error_code"] == "AUTH_401_INVALID_TOKEN" @pytest.mark.parametrize("missing", ["sub", "jti", "token_type", "roles", "tenant_id", "exp", "iat"]) def test_jwt_missing_required_claim_rejected(env, missing): """手册 §4.2 必填 claims 逐一缺失 → 401（显式校验，评审 P3-14 参数化）。""" now = int(time.time()) payload = { "iss": settings.jwt_issuer, "sub": "STAFF-30001", "aud": settings.jwt_audience, "exp": now + 600, "iat": now, "jti": "jti-noclaim", "token_type": "staff", "roles": ["risk_officer"], "tenant_id": "TENANT-001", } payload.pop(missing) tok = jose_jwt.encode(payload, settings.jwt_dev_secret, algorithm="HS256") r = env["client"].get("/api/risk/alerts", headers=_bearer(tok)) assert r.status_code == 401 and r.json()["error_code"] == "AUTH_401_INVALID_TOKEN" def test_jwt_none_algorithm_rejected(env): """alg=none 手工构造 unsigned JWT → 401（jose 白名单库级防护 + 显式测试固化）。""" import base64 import json as _json now = int(time.time()) header = base64.urlsafe_b64encode(_json.dumps({"alg": "none", "typ": "JWT"}).encode()).rstrip(b"=") payload = base64.urlsafe_b64encode(_json.dumps({ "iss": settings.jwt_issuer, "sub": "STAFF-30001", "aud": settings.jwt_audience, "exp": now + 600, "iat": now, "jti": "jti-none", "token_type": "staff", "roles": ["risk_officer"], "tenant_id": "TENANT-001", }).encode()).rstrip(b"=") tok = f"{header.decode()}.{payload.decode()}." # 空签名 r = env["client"].get("/api/risk/alerts", headers=_bearer(tok)) assert r.status_code == 401 def test_auth_audit_db_failure_keeps_401(env, monkeypatch): """P2-5 显式化：留痕写库失败降级（拒绝语义不漂移为 500）。""" def _boom(): raise RuntimeError("audit db down") monkeypatch.setattr(deps_mod, "RiskRepository", _boom) r = env["client"].get("/api/risk/alerts") assert r.status_code == 401 assert r.json()["error_code"] == "AUTH_401_MISSING_DEBUG_HEADERS" # ---------- 吊销（jti 黑名单） ---------- def test_jwt_revoked_jti_rejected(env): tok = auth_service.issue_dev_token(sub="STAFF-30001", roles=["risk_officer"]) claims = auth_service.verify_token(tok) # 先取 jti redis_gateway._gateway.revoked.add(f"auth:revoked:{claims.jti}") r = env["client"].get("/api/risk/alerts", headers=_bearer(tok)) assert r.status_code == 401 and r.json()["error_code"] == "AUTH_401_REVOKED" def test_revocation_fail_open_when_redis_down(monkeypatch): """Redis 不可用 fail-open（redis_gateway 总口径；TTL 为权威兜底）。""" monkeypatch.setattr(redis_gateway, "_gateway", None) monkeypatch.setattr( redis_gateway.RedisGateway, "exists", lambda self, key: (_ for _ in ()).throw(ConnectionError()) ) assert auth_service.is_revoked("jti-x") is False # ---------- debug 头兜底（dev）与非 dev 行为 ---------- def test_dev_debug_headers_still_work(env): r = env["client"].get( "/api/risk/alerts", headers={"X-Debug-Role": "risk_officer", "X-Debug-Actor": "STAFF-30001"} ) assert r.status_code == 200 # B6 演示 SOP 通道不回归 def test_non_dev_without_bearer_401(env, monkeypatch): monkeypatch.setattr(settings, "app_env", "production") r = env["client"].get( "/api/risk/alerts", headers={"X-Debug-Role": "risk_officer", "X-Debug-Actor": "STAFF-30001"} ) assert r.status_code == 401 and r.json()["error_code"] == "AUTH_401_MISSING_BEARER" assert _audit_rows(env["engine"], "unauthenticated") def test_non_dev_jwt_still_accepted(env, monkeypatch): """JWT 是唯一生产通道：非 dev 下有效 JWT + 交叉头照常放行。""" monkeypatch.setattr(settings, "app_env", "production") tok = auth_service.issue_dev_token(sub="STAFF-30001", roles=["risk_officer"]) r = env["client"].get("/api/risk/alerts", headers=_bearer(tok)) assert r.status_code == 200 def test_non_dev_hs256_secret_refused_inline(env, monkeypatch): """HS256 上生产防护：jwt_ready 必须给出拒绝原因（lifespan 消费）。""" monkeypatch.setattr(settings, "app_env", "production") monkeypatch.setattr(settings, "jwt_public_key_path", "") assert auth_service.jwt_ready() is not None # ---------- RS256（生产形态） ---------- _RSA_CACHE: dict = {} def _rsa_keypair() -> tuple[bytes, bytes]: if not _RSA_CACHE: from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import rsa key = rsa.generate_private_key(public_exponent=65537, key_size=2048) _RSA_CACHE["priv"] = key.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption(), ) _RSA_CACHE["pub"] = key.public_key().public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo ) return _RSA_CACHE["priv"], _RSA_CACHE["pub"] def test_rs256_public_key_roundtrip(env, monkeypatch, tmp_path: Path): priv, pub = _rsa_keypair() key_file = tmp_path / "idp_pub.pem" key_file.write_bytes(pub) monkeypatch.setattr(settings, "jwt_public_key_path", str(key_file)) assert auth_service.jwt_ready() is None now = int(time.time()) payload = { "iss": settings.jwt_issuer, "sub": "STAFF-30001", "aud": settings.jwt_audience, "exp": now + 600, "iat": now, "jti": "jti-rs256", "token_type": "staff", "roles": ["risk_officer"], "tenant_id": "TENANT-001", "permissions": ["risk:alert:write"], } tok = jose_jwt.encode(payload, priv, algorithm="RS256", headers={"kid": "2026-09-key-1"}) r = env["client"].get("/api/risk/alerts", headers=_bearer(tok)) assert r.status_code == 200 # HS256 token 在 RS256 模式下必须被拒（算法白名单，防 alg 混淆） hs_tok = auth_service.issue_dev_token(sub="STAFF-30001", roles=["risk_officer"]) r2 = env["client"].get("/api/risk/alerts", headers=_bearer(hs_tok)) assert r2.status_code == 401 # ---------- assert_agent_access 纯函数矩阵 ---------- def test_agent_access_matrix(): from app.api.deps import AGENT_ACCESS_MATRIX, assert_agent_access for agent, rule in AGENT_ACCESS_MATRIX.items(): for tt in rule["token_types"]: for role in rule["roles"]: assert_agent_access( deps_mod.AuthContext(actor_id="X", roles=[role], token_type=tt), agent ) with pytest.raises(deps_mod.PermissionDenied): assert_agent_access( deps_mod.AuthContext(actor_id="X", roles=["ops"], token_type="staff"), "analyst" ) with pytest.raises(deps_mod.PermissionDenied): assert_agent_access( deps_mod.AuthContext(actor_id="X", roles=["risk_officer"], token_type="customer"), "risk" )