"""平台认证(Phase 2 Task 2):login/logout/change-password + 认证依赖。 经 app_client(jiancai Host)打真实路由——Task 3 前 /api/platform/* 仍过 租户 Host 解析,jiancai 是已知域名故直达(计划注:本任务即挂路由,TDD 循环才能闭环)。conftest 的 lifespan bootstrap 超管 admin/admin123 (must_change=True)每测试全新临时平台库(Task 1 就绪)。 改密后 token 语义(本实现选定,见 auth_routes.change_password docstring): 当前 token 不失效、must_change 就地清零——改完密无需重登。 跨测试隔离:conftest autouse _reset_platform_auth 复位内存 token 表与 登录限流计数(照抄 rate_limit 纪律)。 """ import pytest from fastapi import HTTPException from app.platform.auth import ( UserContext, login_limiter, require_super, require_tenant, ) LOGIN = "/api/platform/auth/login" LOGOUT = "/api/platform/auth/logout" CHANGE_PW = "/api/platform/auth/change-password" def _login(client, username="admin", password="admin123"): return client.post(LOGIN, json={"username": username, "password": password}) def _auth(token): return {"Authorization": f"Bearer {token}"} # ---- 登录 --------------------------------------------------------------------- def test_login_ok_returns_token(app_client): r = _login(app_client) assert r.status_code == 200 data = r.json() assert set(data) == {"token", "role", "tenant_slug", "must_change_password"} assert data["role"] == "super" assert data["tenant_slug"] is None assert data["must_change_password"] is True # 引导密码须首登改掉 assert data["token"] def test_login_bad_password_401(app_client): r = _login(app_client, password="wrong") assert r.status_code == 401 # 防枚举: 用户不存在与密码错同一响应(状态+文案都不区分) r2 = _login(app_client, username="nosuchuser") assert r2.status_code == 401 assert r2.json() == r.json() def test_login_lockout_after_5_fails(app_client): for _ in range(5): assert _login(app_client, password="wrong").status_code == 401 # 第 6 次 429——即使密码正确(锁定检查先于验证) assert _login(app_client).status_code == 429 # 窗口恢复: 已记录失败拨到窗外(等效 15 分钟过后,免真实等待)→ 可再登录 key = ("admin", "testclient") login_limiter._fails[key] = [ t - login_limiter.window_seconds - 1 for t in login_limiter._fails[key]] assert _login(app_client).status_code == 200 class TestClientIpXffLast: """spec 2026-10-19 §6：限流 IP 取 XFF 末段（反代追加语义，不可伪造）。""" def test_末段入键_伪造首段轮换不绕过(self, app_client): # 伪造首段轮换 + 固定真末段 → 同键计数，第 5 次失败后第 6 次 429 for i in range(5): app_client.post(LOGIN, headers={"X-Forwarded-For": f"6.6.6.{i}, 1.2.3.4"}, json={"username": "admin", "password": "wrong"}) r = app_client.post(LOGIN, headers={"X-Forwarded-For": "9.9.9.9, 1.2.3.4"}, json={"username": "admin", "password": "wrong"}) assert r.status_code == 429 def test_末段不同即独立键_判别用例(self, app_client): # RED 判别用例：现状（无 XFF 解析）所有请求同键（client.host 恒 # testclient）→ 也会 429；新行为下 8.8.8.8 是独立键 → 401 for i in range(5): app_client.post(LOGIN, headers={"X-Forwarded-For": f"6.6.6.{i}, 1.2.3.4"}, json={"username": "admin", "password": "wrong"}) r = app_client.post(LOGIN, headers={"X-Forwarded-For": "7.7.7.7, 8.8.8.8"}, json={"username": "admin", "password": "wrong"}) assert r.status_code == 401 # 独立键：未锁，密码错 → 401（现状此断言为 429 → RED） def test_无XFF回退client_host(self, app_client): r = app_client.post(LOGIN, json={"username": "admin", "password": "wrong"}) assert r.status_code == 401 # 正常走通（键为 client.host） def test_空串XFF回退(self, app_client): r = app_client.post(LOGIN, headers={"X-Forwarded-For": " "}, json={"username": "admin", "password": "wrong"}) assert r.status_code == 401 # 空白视同无 XFF，回退 client.host def test_单段直连形态(self, app_client): # XFF "1.2.3.4"（无反代追加）→ 末段即唯一段，入键 for _ in range(5): app_client.post(LOGIN, headers={"X-Forwarded-For": "1.2.3.4"}, json={"username": "admin", "password": "wrong"}) r = app_client.post(LOGIN, headers={"X-Forwarded-For": "1.2.3.4"}, json={"username": "admin", "password": "wrong"}) assert r.status_code == 429 # ---- token 生命周期 ------------------------------------------------------------ def test_token_expiry(app_client, monkeypatch): """TTL 可注入: 拨成负值 → 签发即过期 → 受保护端点 401。""" from app.platform import auth monkeypatch.setattr(auth.token_store, "ttl_seconds", -1) r = _login(app_client) assert r.status_code == 200 token = r.json()["token"] r2 = app_client.post( CHANGE_PW, headers=_auth(token), json={"old_password": "admin123", "new_password": "n3w-pass!"}) assert r2.status_code == 401 def test_protected_requires_auth(app_client): """无 Authorization → 401(依赖链 get_current_user 先于 enforce)。""" assert app_client.post(LOGOUT).status_code == 401 def test_logout_invalidates_token(app_client): token = _login(app_client).json()["token"] # must_change=True → logout 409,先改密(enforce 语义连锁,见下) app_client.post( CHANGE_PW, headers=_auth(token), json={"old_password": "admin123", "new_password": "n3w-pass-2"}) assert app_client.post(LOGOUT, headers=_auth(token)).status_code == 200 # 同 token 再用 → 401(已撤销,立即失效) assert app_client.post(LOGOUT, headers=_auth(token)).status_code == 401 # ---- 改密 + must_change 409 语义 ------------------------------------------------- def test_change_password_flow(app_client): token = _login(app_client).json()["token"] # 旧密改新密成功 r = app_client.post( CHANGE_PW, headers=_auth(token), json={"old_password": "admin123", "new_password": "n3w-pass-1"}) assert r.status_code == 200 # 选定语义: 改密后当前 token 不失效、must_change 就地清零 # (logout 受 enforce 保护,改密后可 200 即证 token 存活且标志已清) assert app_client.post(LOGOUT, headers=_auth(token)).status_code == 200 # 旧密失效 assert _login(app_client).status_code == 401 # 新密可登录,引导标志已清 data = _login(app_client, password="n3w-pass-1").json() assert data["must_change_password"] is False def test_must_change_password_409_semantics(app_client): """引导密码强制语义: must_change=True 时受保护端点(logout)409 (detail 为固定文案——前端 client.ts 的 MUST_CHANGE_DETAIL 依赖此 契约,改动文案即红);change-password 本身豁免(200);改完 409 消除。""" token = _login(app_client).json()["token"] r = app_client.post(LOGOUT, headers=_auth(token)) assert r.status_code == 409 assert r.json()["detail"] == "必须先修改初始密码" r = app_client.post( CHANGE_PW, headers=_auth(token), json={"old_password": "admin123", "new_password": "n3w-pass-3"}) assert r.status_code == 200 assert app_client.post(LOGOUT, headers=_auth(token)).status_code == 200 # ---- 角色依赖(工厂直测;端点级覆盖在 Task 5)--------------------------------------- def test_tenant_admin_role_scoped(): tenant_admin = UserContext( user_id=2, role="tenant", tenant_slug="jiancai", must_change_password=False) with pytest.raises(HTTPException) as ei: require_super(tenant_admin) assert ei.value.status_code == 403 dep = require_tenant("jiancai") assert dep(tenant_admin) is tenant_admin # 本租户放行 super_user = UserContext( user_id=1, role="super", tenant_slug=None, must_change_password=False) assert dep(super_user) is super_user # super 全通 other = UserContext( user_id=3, role="tenant", tenant_slug="auto", must_change_password=False) with pytest.raises(HTTPException) as ei2: dep(other) assert ei2.value.status_code == 403 # 它租户 403 def test_role_dependencies_chain_must_change_409(app_client): """审查修复 Issue 1: require_super/require_tenant 参数走 platform_user (get_current_user + enforce 连锁)——must_change=True 的请求先撞 409 门, Task 5 路由挂角色依赖时不可能静默漏挂。 连锁只在 FastAPI 依赖解析时生效(直调函数体只跑角色判断),故用 一次性探针 app 挂真实 DI 验证;token 经主 app 的真实 login/change-password 路由取放(同一进程级 TokenStore)。端点级覆盖仍在 Task 5。""" from fastapi import Depends, FastAPI from fastapi.testclient import TestClient probe = FastAPI() @probe.post("/probe-super") async def _super(user=Depends(require_super)): return {"user_id": user.user_id} @probe.post("/probe-tenant") async def _tenant(user=Depends(require_tenant("jiancai"))): return {"user_id": user.user_id} token = _login(app_client).json()["token"] # 超管, must_change=True with TestClient(probe) as c: # 409 门先于角色判断 assert c.post("/probe-super", headers=_auth(token)).status_code == 409 assert c.post("/probe-tenant", headers=_auth(token)).status_code == 409 # 改密(标志就地清零)后放行到角色层: super 两个依赖都过 app_client.post( CHANGE_PW, headers=_auth(token), json={"old_password": "admin123", "new_password": "n3w-pass-9"}) assert c.post("/probe-super", headers=_auth(token)).status_code == 200 assert c.post("/probe-tenant", headers=_auth(token)).status_code == 200