浅念
248 lines
10 KiB
Python
248 lines
10 KiB
Python
"""平台认证(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
|