From 36f476f450ea03f070bbab128fbdbeefbe233139 Mon Sep 17 00:00:00 2001 From: YUAN Date: Mon, 7 Sep 2026 02:23:33 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20T-01=20JWT=20=E9=89=B4=E6=9D=83?= =?UTF-8?q?=E2=80=94=E2=80=94auth=5Fservice(Auth=20SDK:=20HS256/RS256=20?= =?UTF-8?q?=E9=AA=8C=E7=AD=BE+=E5=BF=85=E5=A1=ABclaims=E6=98=BE=E5=BC=8F?= =?UTF-8?q?=E6=A0=A1=E9=AA=8C+jti=E5=90=8A=E9=94=80=20fail-open)=20+=20dep?= =?UTF-8?q?s=20=E5=B7=A5=E5=8E=82=E6=9B=BF=E6=8D=A2(Bearer=20=E5=85=A8?= =?UTF-8?q?=E7=8E=AF=E5=A2=83,=20debug=20=E5=A4=B4=E9=99=8D=E7=BA=A7=20dev?= =?UTF-8?q?=20=E5=85=9C=E5=BA=95)=20+=20X-Agent-Type=20=E4=BA=A4=E5=8F=89?= =?UTF-8?q?=E6=A0=A1=E9=AA=8C(=E6=89=8B=E5=86=8C=C2=A74.6/=C2=A75.4=20?= =?UTF-8?q?=E5=87=86=E5=85=A5=E7=9F=A9=E9=98=B5)=20+=20issue=5Fdev=5Ftoken?= =?UTF-8?q?=20CLI=20+=20redis=5Fgateway=20exists/set=5Fex=20+=20lifespan?= =?UTF-8?q?=20=E6=94=B9=E9=AA=8C=20jwt=5Fready=20+=20tests/test=5Fauth=5Fj?= =?UTF-8?q?wt=2018=20=E4=BE=8B,=20228=20=E7=BB=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 6 +- app/api/deps.py | 163 +++++++++++++---- app/config/settings.py | 7 + app/main.py | 25 ++- app/service/auth_service.py | 185 +++++++++++++++++++ app/service/risk/redis_gateway.py | 15 ++ scripts/dev/issue_dev_token.py | 55 ++++++ tests/test_auth_jwt.py | 289 ++++++++++++++++++++++++++++++ tests/test_main.py | 14 +- tests/test_risk_api.py | 6 +- 10 files changed, 721 insertions(+), 44 deletions(-) create mode 100644 app/service/auth_service.py create mode 100644 scripts/dev/issue_dev_token.py create mode 100644 tests/test_auth_jwt.py diff --git a/.env.example b/.env.example index 0020473..29569a3 100644 --- a/.env.example +++ b/.env.example @@ -28,8 +28,12 @@ EMBED_MODEL=bge-m3 DEEPSEEK_API_KEY= DEEPSEEK_BASE_URL=https://api.deepseek.com -# JWT (dev only — production use RS256 + IdP) +# JWT (T-01): production uses RS256 public key from IdP; empty path -> HS256 dev secret +# (HS256 is rejected at startup when APP_ENV != development) +JWT_PUBLIC_KEY_PATH= JWT_DEV_SECRET=change-me-in-dev-only +JWT_ISSUER=https://idp.jinrong.internal +JWT_AUDIENCE=agent-gateway # Risk thresholds (defaults = frozen rules, see docs/PRD/附-风控规则表.md) RISK_ASSESSMENT_VALID_DAYS=365 diff --git a/app/api/deps.py b/app/api/deps.py index 370193d..12affcc 100644 --- a/app/api/deps.py +++ b/app/api/deps.py @@ -1,11 +1,14 @@ """API 依赖：鉴权上下文（架构 §5.7 · 归属校验统一在此层）。 -AuthContext 模型在 A4 冻结（开发计划 v1.1）；B6 落地 `get_auth_context()`： -dev（app_env=development）从 `X-Debug-Role` / `X-Debug-Actor` 请求头构造 -（customer 角色 customer_id=actor_id），非 dev 请求时拒绝（启动期全局检查 -归 B7 lifespan：非 dev 且 AUTH_FACTORY_IS_DEBUG 拒绝启动，挂账⑤）； -T-01 就绪后替换工厂内部为 JWT 解析并置 AUTH_FACTORY_IS_DEBUG=False， -签名与调用方零改动。 +AuthContext 模型在 A4 冻结（开发计划 v1.1）；B6 落地 `get_auth_context()`； +T-01（2026-09-07）工厂内部替换为 JWT 解析（auth_service.verify_token， +手册 §4/§8），`Authorization: Bearer` 为所有环境首选通道： +- JWT 通道：验签 + claims + jti 吊销 + `X-Agent-Type` 交叉校验（手册 §4.6 + 必填、§5.4 准入矩阵；缺失 AUTH_401_MISSING_AGENT_TYPE、不符 + AUTH_403_AGENT_MISMATCH）； +- debug 头兜底：仅 development 无 Bearer 时从 `X-Debug-Role`/`X-Debug-Actor` + 构造（B6 过渡口径，演示 SOP 与既有测试依赖此通道；不校验 X-Agent-Type）； +非 dev 无 Bearer 一律 401 留痕（AUTH_401_MISSING_BEARER）。 归属断言 `assert_customer_access` 对齐 JWT 手册 §6.1/§6.2 与 PRD G-01： customer 仅本人（AUTH_403_NOT_OWNER）、advisor 经 customer_advisor_rel @@ -13,7 +16,7 @@ customer 仅本人（AUTH_403_NOT_OWNER）、advisor 经 customer_advisor_rel 所有 403/401 一律经 `deny`/`unauthenticated` 审计（手册 P-05 全局铁律， B6 评审 P1-1）；响应体为统一错误结构（utils/response，手册 §10，B7 挂账④）； 多角色按 fail-closed 口径固化（customer 分支优先，命中 deny 即拒，不再并集 -放宽——评审 P3-1②，T-01 引入 token_type 后收紧）。 +放宽——评审 P3-1②）。 """ from __future__ import annotations @@ -24,24 +27,45 @@ from pydantic import BaseModel, Field from app.config.settings import settings from app.repository.core_ro import CoreReadOnlyRepository from app.repository.risk_repository import RiskRepository +from app.service.auth_service import Claims, TokenInvalid, verify_token from app.utils.exceptions import ApiError, PermissionDenied from app.utils.trace import current_trace, new_trace STAFF_FULL_ACCESS_ROLES = ("risk_officer",) DEBUG_ROLE_HEADER = "X-Debug-Role" DEBUG_ACTOR_HEADER = "X-Debug-Actor" +AGENT_TYPE_HEADER = "X-Agent-Type" +AGENT_TYPES = ("customer", "advisor", "analyst", "risk") -# B7 挂账⑤：debug 头工厂是 T-01 过渡实现；main.lifespan 据此在非 dev 环境 -# 拒绝启动。T-01 接入 JWT 工厂后置 False（或改为按注册工厂判定）。 -AUTH_FACTORY_IS_DEBUG = True +# 手册 §5.4 Agent × 角色/token_type 准入矩阵（JWT 通道强制；debug 头通道 +# 维持 B6 路由内角色口径——演示 SOP 的 compliance 台账只读属其扩展） +AGENT_ACCESS_MATRIX: dict[str, dict[str, tuple[str, ...]]] = { + "customer": {"token_types": ("customer",), "roles": ("customer",)}, + "advisor": {"token_types": ("staff",), "roles": ("advisor", "compliance", "ops")}, + "analyst": {"token_types": ("staff",), "roles": ("analyst", "compliance")}, + "risk": {"token_types": ("staff", "service"), "roles": ("risk_officer", "service_risk")}, +} + +# T-01 完成：JWT 已接入，debug 头降级为 dev 兜底（main.lifespan 据此放行 +# 非 dev 启动，改为校验 jwt_ready）。 +AUTH_FACTORY_IS_DEBUG = False class AuthContext(BaseModel): - """统一鉴权上下文（全部 API 依赖层的产出；service 层签名接收此类型）。""" + """统一鉴权上下文（全部 API 依赖层的产出；service 层签名接收此类型）。 + + T-01 扩展（手册 §8.1）：token_type/permissions/tenant_id/jti；debug 头 + 通道 token_type 按角色推断（customer 角色 → customer，其余 staff）， + permissions/tenant_id/jti 为空（过渡口径）。 + """ actor_id: str = Field(..., description="操作者 ID：staff_id 或 customer_id") roles: list[str] = Field(default_factory=list, description="角色集合，如 ['risk_officer','risk_demo']") customer_id: str | None = Field(None, description="customer 角色时 = 本人 customer_id；其余为空") + token_type: str = Field("staff", description="customer | staff | service（手册 §3）") + permissions: list[str] = Field(default_factory=list, description="细粒度权限（JWT claims）") + tenant_id: str | None = Field(None, description="租户/法人主体") + jti: str | None = Field(None, description="Token 唯一 ID（吊销/审计用）") def has_role(self, *roles: str) -> bool: return any(r in self.roles for r in roles) @@ -49,6 +73,9 @@ class AuthContext(BaseModel): def is_customer(self) -> bool: return "customer" in self.roles + def has_permission(self, permission: str) -> bool: + return permission in self.permissions + def _authz_audit( risk_repo: RiskRepository, @@ -93,32 +120,105 @@ def deny( raise PermissionDenied(code, message or f"forbidden: {code}") +def _unauthenticated_audit( + risk_repo: RiskRepository, + actor_id: str, + code: str, + agent_type: str = "risk", +) -> None: + """401 留痕（P-05；B6 debug 通道既有口径，T-01 推广到 JWT 通道）。""" + risk_repo.insert_audit_log( + { + "trace_id": current_trace() or new_trace(), + "event_type": "authz", + "agent_type": agent_type, + "actor_id": actor_id or "anonymous", + "customer_id": None, + "rule_id": None, + "input_summary": {"code": code}, + "decision": "unauthenticated", + "risk_score": None, + "handler_id": None, + "handler_result": None, + "handler_comment": None, + } + ) + + +def _claims_to_auth(claims: Claims) -> AuthContext: + """JWT claims → AuthContext（手册 §8.1；customer_id 仅 customer token）。""" + return AuthContext( + actor_id=claims.sub, + roles=claims.roles, + customer_id=claims.customer_id if claims.token_type == "customer" else None, + token_type=claims.token_type, + permissions=claims.permissions, + tenant_id=claims.tenant_id, + jti=claims.jti, + ) + + +def assert_agent_access( + auth: AuthContext, + agent_type: str, + risk_repo: RiskRepository | None = None, +) -> None: + """Agent 入口准入（手册 §5.4 矩阵；chat 等显式入口调用）。 + + 不匹配 → AUTH_403_AGENT_MISMATCH + 审计（deny）。risk_repo 传 None 时 + 仅抛异常不留痕（仅限 JWT 通道已审计的交叉校验复用路径）。 + """ + rule = AGENT_ACCESS_MATRIX.get(agent_type) + if rule is None: + raise ApiError(400, "BAD_REQUEST", f"unknown agent_type: {agent_type}") + if auth.token_type in rule["token_types"] and set(auth.roles) & set(rule["roles"]): + return + if risk_repo is not None: + deny(auth, "AUTH_403_AGENT_MISMATCH", risk_repo, message=f"not allowed for agent '{agent_type}'") + raise PermissionDenied("AUTH_403_AGENT_MISMATCH", f"not allowed for agent '{agent_type}'") + + def get_auth_context(request: Request) -> AuthContext: - """鉴权工厂（B6）：dev 读 debug 头，非 dev 拒绝；T-01 后替换内部为 JWT 解析。""" + """鉴权工厂（T-01）：Bearer JWT 优先（全环境），dev 无 Bearer 时 debug 头兜底。 + + JWT 通道强制 `X-Agent-Type` 交叉校验（手册 §4.6/§5.4）：缺失 401、 + 值域外 400、与 token 不符 403（均留痕）。debug 通道保持 B6 行为。 + """ + auth_header = request.headers.get("Authorization", "") + if auth_header.startswith("Bearer "): + token = auth_header[len("Bearer "):].strip() + if not token: + _unauthenticated_audit(RiskRepository(), "anonymous", "AUTH_401_INVALID_TOKEN") + raise ApiError(401, "AUTH_401_INVALID_TOKEN", "empty bearer token") + try: + claims = verify_token(token) + except TokenInvalid as exc: + _unauthenticated_audit(RiskRepository(), "anonymous", exc.code) + raise ApiError(401, exc.code, exc.message) from exc + auth = _claims_to_auth(claims) + + agent_type = request.headers.get(AGENT_TYPE_HEADER, "").strip() + if not agent_type: + _unauthenticated_audit(RiskRepository(), auth.actor_id, "AUTH_401_MISSING_AGENT_TYPE") + raise ApiError(401, "AUTH_401_MISSING_AGENT_TYPE", f"missing {AGENT_TYPE_HEADER} header") + if agent_type not in AGENT_TYPES: + raise ApiError(400, "BAD_REQUEST", f"invalid {AGENT_TYPE_HEADER}: {agent_type}") + # 交叉校验失败也走 deny 全量审计（fail-closed；agent_type 归请求目标） + assert_agent_access(auth, agent_type, risk_repo=RiskRepository()) + request.state.auth = auth + return auth + if settings.app_env != "development": - raise RuntimeError( - f"debug auth disabled outside development (app_env={settings.app_env})" - ) + _unauthenticated_audit(RiskRepository(), "anonymous", "AUTH_401_MISSING_BEARER") + raise ApiError(401, "AUTH_401_MISSING_BEARER", "missing Authorization bearer token") + + # dev debug 头兜底（B6 过渡口径；演示 SOP 与既有权限矩阵测试依赖此通道） roles = [r.strip() for r in request.headers.get(DEBUG_ROLE_HEADER, "").split(",") if r.strip()] actor_id = request.headers.get(DEBUG_ACTOR_HEADER, "").strip() if not roles or not actor_id: # 401 也留痕（P1-1）；debug 通道仅 dev，生产等价流量由 JWT 中间件拒绝 - repo = RiskRepository() - repo.insert_audit_log( - { - "trace_id": current_trace() or new_trace(), - "event_type": "authz", - "agent_type": "risk", - "actor_id": actor_id or "anonymous", - "customer_id": None, - "rule_id": None, - "input_summary": {"roles": roles, "code": "AUTH_401_MISSING_DEBUG_HEADERS"}, - "decision": "unauthenticated", - "risk_score": None, - "handler_id": None, - "handler_result": None, - "handler_comment": None, - } + _unauthenticated_audit( + RiskRepository(), actor_id, "AUTH_401_MISSING_DEBUG_HEADERS" ) raise ApiError( 401, "AUTH_401_MISSING_DEBUG_HEADERS", "missing X-Debug-Role/X-Debug-Actor headers" @@ -127,6 +227,7 @@ def get_auth_context(request: Request) -> AuthContext: actor_id=actor_id, roles=roles, customer_id=actor_id if "customer" in roles else None, + token_type="customer" if "customer" in roles else "staff", ) diff --git a/app/config/settings.py b/app/config/settings.py index 6fb1bfe..5ad7b47 100644 --- a/app/config/settings.py +++ b/app/config/settings.py @@ -31,6 +31,13 @@ class Settings(BaseSettings): deepseek_api_key: str = "" deepseek_base_url: str = "https://api.deepseek.com" + # ===== JWT（T-01 · JWT 手册 §4/§11）===== + # RS256 公钥路径（生产，私钥仅在 IdP）；为空时用 HS256 + jwt_dev_secret（仅 development） + jwt_public_key_path: str = "" + jwt_dev_secret: str = "change-me-in-dev-only" + jwt_issuer: str = "https://idp.jinrong.internal" + jwt_audience: str = "agent-gateway" + # ===== Risk 阈值（默认值=冻结规则 · docs/PRD/附-风控规则表.md）===== risk_assessment_valid_days: int = 365 risk_large_amount: Decimal = Decimal("500000") diff --git a/app/main.py b/app/main.py index 7be5186..25b3233 100644 --- a/app/main.py +++ b/app/main.py @@ -1,12 +1,14 @@ -"""FastAPI 入口（B7 集成）：路由挂载、trace 中间件、lifespan、统一错误体。 +"""FastAPI 入口（B7 集成 · T-01/T-02 演进）：路由挂载、trace 中间件、审计中间件、lifespan、统一错误体。 lifespan（B7 挂账⑤⑥）： -- 启动期校验：非 development 环境且鉴权工厂仍为 debug 头实现 → 拒绝启动 - （T-01 接入 JWT 后置 deps.AUTH_FACTORY_IS_DEBUG=False 放行）； -- Redis 网关单例注册（惰性连接，publish/DEL 失败降级不阻塞业务）； +- 启动期校验：非 development 环境须 JWT 就绪（RS256 公钥已配置； + T-01 前的 debug 工厂在场同样拒绝——AUTH_FACTORY_IS_DEBUG 双保险）； +- Redis 网关单例注册（惰性连接，publish/DEL/EXISTS 失败降级不阻塞业务）； - shutdown 统一 dispose 数据库引擎（utils/db 工厂，B6 复审 P3 泄漏收口）。 trace：X-Trace-Id 透传/生成 + 响应头回写（trace.py 约束：call_next 前 set， 同步路由线程池由 anyio 传播，一致性 B8 断言兜底）。 +audit（T-02）：平台访问审计（http_access）+ input_guard_log 双写见 +audit_middleware；X-Request-Id 独立生成（B7 复审 P3-4）。 """ from __future__ import annotations @@ -20,6 +22,7 @@ from app.api import deps from app.api.risk import router as risk_router from app.api.simulate import router as simulate_router from app.config.settings import settings +from app.service.auth_service import jwt_ready from app.service.risk import redis_gateway from app.utils.db import dispose_engines from app.utils.response import register_error_handlers @@ -31,11 +34,15 @@ _TRACE_ID_PATTERN = re.compile(r"^[A-Za-z0-9._-]{1,64}$") @asynccontextmanager async def lifespan(_: FastAPI): - if settings.app_env != "development" and deps.AUTH_FACTORY_IS_DEBUG: - raise RuntimeError( - "debug auth factory is wired but app_env is not 'development'; " - "deploy T-01 JWT auth first or set app_env=development" - ) + if settings.app_env != "development": + if deps.AUTH_FACTORY_IS_DEBUG: + raise RuntimeError( + "debug auth factory is wired but app_env is not 'development'; " + "deploy T-01 JWT auth first or set app_env=development" + ) + reason = jwt_ready() + if reason: + raise RuntimeError(f"JWT auth not ready for non-development env: {reason}") redis_gateway.set_gateway(redis_gateway.RedisGateway()) try: yield diff --git a/app/service/auth_service.py b/app/service/auth_service.py new file mode 100644 index 0000000..5af975f --- /dev/null +++ b/app/service/auth_service.py @@ -0,0 +1,185 @@ +"""Auth SDK：JWT 验签与吊销检查（T-01 · JWT 手册 §4 / §8.2 JwtVerifier）。 + +四 Agent 共用一套身份源（手册 P-01/P-06）：本模块是唯一验签入口，业务服务 +不自行解析裸 JWT（deps.get_auth_context 工厂内部调用，AuthContext 转换在 +api 层完成——分层禁止 service 反向依赖 api）。 + +算法策略（手册 §4.1/§11）： +- 配置 jwt_public_key_path → RS256 公钥验签（生产形态，私钥仅在 IdP）； +- 未配置 → HS256 + jwt_dev_secret（仅 development；非 dev 由 main.lifespan + jwt_ready() 检查拒绝启动，防止对称密钥上生产）。 + +吊销（手册 §11）：jti 黑名单 Redis `auth:revoked:{jti}`；Redis 不可用时 +fail-open 放行 + warning（与 redis_gateway「失败降级不阻塞」总口径一致， +签发侧 TTL 过期为权威兜底；生产接 IdP 吊销推送时可切 fail-closed）。 + +审计与 HTTP 状态不在本层：验签失败抛 TokenInvalid（带手册 §10 错误码）， +由 deps 统一 401 留痕后转 ApiError（P-05：401 必留痕）。 +""" + +from __future__ import annotations + +import logging +import time +from pathlib import Path + +from jose import JWTError, jwt +from pydantic import BaseModel, Field + +from app.config.settings import settings +from app.utils.exceptions import ApiError + +logger = logging.getLogger(__name__) + +REVOKED_KEY_TEMPLATE = "auth:revoked:{jti}" + +# 手册 §4.2 必填 claims；缺失一律 AUTH_401_INVALID_TOKEN +_REQUIRED_CLAIMS = ("exp", "iat", "sub", "jti", "token_type", "roles", "tenant_id") + + +class TokenInvalid(Exception): + """验签失败（deps 转 401 + 审计；code 对齐手册 §10）。""" + + def __init__(self, code: str, message: str) -> None: + super().__init__(message) + self.code = code + self.message = message + + +class Claims(BaseModel): + """验签通过的 JWT claims（手册 §4.2 标准 claims 子集）。""" + + sub: str + token_type: str + roles: list[str] = Field(default_factory=list) + permissions: list[str] = Field(default_factory=list) + tenant_id: str | None = None + customer_id: str | None = None + advisor_id: str | None = None + jti: str + + +def _load_public_key() -> bytes | None: + """RS256 公钥（启动期由 jwt_ready() 保证存在；此处失败视为配置漂移）。""" + path = settings.jwt_public_key_path + if not path: + return None + return Path(path).read_bytes() + + +def jwt_ready() -> str | None: + """非 dev 启动前检查（main.lifespan）：返回 None=就绪，否则拒绝原因。""" + if settings.jwt_public_key_path: + if not Path(settings.jwt_public_key_path).is_file(): + return f"JWT_PUBLIC_KEY_PATH not found: {settings.jwt_public_key_path}" + return None + if settings.app_env != "development": + return ( + "HS256 dev secret is not allowed outside development; " + "set JWT_PUBLIC_KEY_PATH (RS256) or APP_ENV=development" + ) + return None + + +def issue_dev_token( + *, + sub: str, + roles: list[str], + token_type: str = "staff", + exp_minutes: int = 60, + tenant_id: str = "TENANT-001", + customer_id: str | None = None, + advisor_id: str | None = None, + permissions: list[str] | None = None, +) -> str: + """HS256 签发（development 联调/测试用；生产由 IdP 签 RS256，手册 §3）。 + + scripts/dev/issue_dev_token.py 的 CLI 与单测共用本实现；不得在生产签发 + （jwt_ready 校验由调用方环境保证，签发函数本身无环境判断——dev secret + 仅存在于 development 配置）。 + """ + now = int(time.time()) + payload = { + "iss": settings.jwt_issuer, + "sub": sub, + "aud": settings.jwt_audience, + "exp": now + exp_minutes * 60, + "iat": now, + "jti": f"jti-{now:x}-{sub}", + "token_type": token_type, + "roles": roles, + "tenant_id": tenant_id, + } + if customer_id is not None: + payload["customer_id"] = customer_id + if advisor_id is not None: + payload["advisor_id"] = advisor_id + if permissions is not None: + payload["permissions"] = permissions + return jwt.encode(payload, settings.jwt_dev_secret, algorithm="HS256", headers={"kid": "dev-hs256"}) + + +def verify_token(token: str) -> Claims: + """验签 + 标准 claims 校验 + jti 吊销检查（手册 §8.2 JwtVerifier.verify）。 + + 必填 claims 显式检查（jose 的 options.require 仅覆盖 JWT 标准 claim， + 对 token_type/roles/tenant_id 等自定义 claim 不生效——实测 3.5.0，留痕）。 + """ + try: + public_key = _load_public_key() + if public_key is not None: + payload = jwt.decode( + token, + public_key, + algorithms=["RS256"], + audience=settings.jwt_audience, + issuer=settings.jwt_issuer, + ) + else: + payload = jwt.decode( + token, + settings.jwt_dev_secret, + algorithms=["HS256"], + audience=settings.jwt_audience, + issuer=settings.jwt_issuer, + ) + except JWTError as exc: + # 过期/签名错误/格式错误统一 AUTH_401_INVALID_TOKEN（手册 §10 不区分原因，防探测） + raise TokenInvalid("AUTH_401_INVALID_TOKEN", f"invalid token: {type(exc).__name__}") from exc + + missing = [c for c in _REQUIRED_CLAIMS if c not in payload] + if missing: + raise TokenInvalid("AUTH_401_INVALID_TOKEN", f"missing required claims: {missing}") + + if payload.get("token_type") == "customer" and payload.get("customer_id") != payload.get("sub"): + # 手册 §4.2：customer token 的 customer_id 必须等于 sub + raise TokenInvalid("AUTH_401_INVALID_TOKEN", "customer_id does not match sub") + + claims = Claims.model_validate(payload) + if is_revoked(claims.jti): + raise TokenInvalid("AUTH_401_REVOKED", "token has been revoked") + return claims + + +def is_revoked(jti: str) -> bool: + """jti 黑名单检查（Redis `auth:revoked:{jti}`）；Redis 失败 fail-open。""" + try: + from app.service.risk import redis_gateway + + return redis_gateway.key_exists(REVOKED_KEY_TEMPLATE.format(jti=jti)) + except Exception: + logger.warning("revocation check failed, fail-open (jti=%s...)", jti[:8], exc_info=True) + return False + + +def revoke_jti(jti: str, ttl_seconds: int | None = None) -> None: + """吊销（IdP/登出调用）：写黑名单；失败抛出（调用方决定语义），不静默。""" + from app.service.risk import redis_gateway + + ttl = ttl_seconds or 8 * 3600 # 员工 token 最长有效期（手册 §11），过期自然失效 + redis_gateway.get_gateway().set_ex(REVOKED_KEY_TEMPLATE.format(jti=jti), "1", ttl) + + +def auth_error(status_code: int, code: str, message: str) -> ApiError: + """统一 401/403 出口（保持错误码集中在手册 §10 语义）。""" + return ApiError(status_code, code, message) diff --git a/app/service/risk/redis_gateway.py b/app/service/risk/redis_gateway.py index 4dfaad0..717bbd4 100644 --- a/app/service/risk/redis_gateway.py +++ b/app/service/risk/redis_gateway.py @@ -34,6 +34,12 @@ class RedisGateway: def delete(self, *keys: str) -> None: self._ensure().delete(*keys) + def exists(self, key: str) -> bool: + return bool(self._ensure().exists(key)) + + def set_ex(self, key: str, value: str, ttl_seconds: int) -> None: + self._ensure().setex(key, ttl_seconds, value) + _gateway: RedisGateway | Any | None = None @@ -66,3 +72,12 @@ def cache_delete(*keys: str) -> None: get_gateway().delete(*keys) except Exception: logger.warning("cache DEL failed (degrade to TTL): %s", keys, exc_info=True) + + +def key_exists(key: str) -> bool: + """存在性检查（T-01 jti 吊销黑名单）；失败 fail-open 返回 False。""" + try: + return get_gateway().exists(key) + except Exception: + logger.warning("redis EXISTS failed (fail-open): %s", key, exc_info=True) + return False diff --git a/scripts/dev/issue_dev_token.py b/scripts/dev/issue_dev_token.py new file mode 100644 index 0000000..1fc14f9 --- /dev/null +++ b/scripts/dev/issue_dev_token.py @@ -0,0 +1,55 @@ +"""开发联调 JWT 签发 CLI（T-01 · 仅 development；生产由 IdP 签 RS256）。 + +用法（Git Bash / PowerShell）： + python scripts/dev/issue_dev_token.py --sub STAFF-30001 --roles risk_officer + python scripts/dev/issue_dev_token.py --sub CUST-9527 --roles customer \\ + --token-type customer --customer-id CUST-9527 + +签发后经 `Authorization: Bearer ` 调用，JWT 通道须带 `X-Agent-Type` +（手册 §4.6）。HS256 密钥取 .env JWT_DEV_SECRET；APP_ENV != development 时 +拒绝签发（防止对称密钥流出）。 +""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from app.config.settings import settings # noqa: E402 +from app.service.auth_service import issue_dev_token # noqa: E402 + + +def main() -> int: + parser = argparse.ArgumentParser(description="issue a development JWT (HS256)") + parser.add_argument("--sub", required=True, help="staff_id 或 customer_id（手册 §3）") + parser.add_argument("--roles", required=True, help="逗号分隔角色，如 risk_officer / customer") + parser.add_argument("--token-type", default="staff", choices=["customer", "staff", "service"]) + parser.add_argument("--customer-id", default=None, help="token_type=customer 时必填且须等于 sub") + parser.add_argument("--exp-minutes", type=int, default=60) + parser.add_argument("--tenant-id", default="TENANT-001") + args = parser.parse_args() + + if settings.app_env != "development": + print("refused: issue_dev_token only runs with APP_ENV=development", file=sys.stderr) + return 1 + roles = [r.strip() for r in args.roles.split(",") if r.strip()] + if args.token_type == "customer" and args.customer_id != args.sub: + print("refused: customer token requires --customer-id equal to --sub (手册 §4.2)", file=sys.stderr) + return 1 + token = issue_dev_token( + sub=args.sub, + roles=roles, + token_type=args.token_type, + exp_minutes=args.exp_minutes, + tenant_id=args.tenant_id, + customer_id=args.customer_id, + ) + print(token) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_auth_jwt.py b/tests/test_auth_jwt.py new file mode 100644 index 0000000..67e8a70 --- /dev/null +++ b/tests/test_auth_jwt.py @@ -0,0 +1,289 @@ +"""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" + + +def test_jwt_missing_required_claim_rejected(env): + now = int(time.time()) + payload = { # 缺 tenant_id（手册 §4.2 必填） + "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"], + } + 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 + + +# ---------- 吊销（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" + ) diff --git a/tests/test_main.py b/tests/test_main.py index c09e9ca..21266a3 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -95,8 +95,20 @@ def test_unified_error_body_401_with_trace(client): def test_lifespan_rejects_debug_factory_in_non_dev(monkeypatch): - """挂账⑤：非 dev 环境且 debug 头工厂在场 → 拒绝启动（T-01 前防误部署）。""" + """挂账⑤（T-01 后兜底分支）：debug 工厂在场 + 非 dev → 拒绝启动。""" monkeypatch.setattr(settings, "app_env", "production") + from app.api import deps as deps_mod + + monkeypatch.setattr(deps_mod, "AUTH_FACTORY_IS_DEBUG", True) with pytest.raises(RuntimeError, match="debug auth factory is wired"): with TestClient(app): pass + + +def test_lifespan_rejects_jwt_not_ready_in_non_dev(monkeypatch): + """T-01：非 dev 且 RS256 公钥未配置（HS256 dev secret）→ 拒绝启动。""" + monkeypatch.setattr(settings, "app_env", "production") + monkeypatch.setattr(settings, "jwt_public_key_path", "") + with pytest.raises(RuntimeError, match="JWT auth not ready"): + with TestClient(app): + pass diff --git a/tests/test_risk_api.py b/tests/test_risk_api.py index 057c43c..aecd152 100644 --- a/tests/test_risk_api.py +++ b/tests/test_risk_api.py @@ -343,11 +343,13 @@ def test_alert_disclaimer_on_alert_apis(client): def test_non_dev_rejects_debug_auth(client, monkeypatch): + """T-01 后：非 dev 无 Bearer 一律 401 留痕（debug 头通道不存在于生产）。""" from app.config.settings import settings monkeypatch.setattr(settings, "app_env", "production") - with pytest.raises(RuntimeError, match="debug auth disabled"): - client.get("/api/risk/alerts", headers=OFFICER) + r = client.get("/api/risk/alerts", headers=OFFICER) + assert r.status_code == 401 + assert r.json()["error_code"] == "AUTH_401_MISSING_BEARER" # ---------- deps 单元：compliance 不放行客户业务数据 ----------