From f49e8e92b1d20b99c4181cd4bc5d9bd297ea5dbc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Tue, 13 Oct 2026 08:18:15 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=B9=B3=E5=8F=B0=E8=AE=A4=E8=AF=81?= =?UTF-8?q?=EF=BC=88scrypt=E4=B8=8A=E7=A7=BB/=E5=86=85=E5=AD=98token=208h/?= =?UTF-8?q?=E7=99=BB=E5=BD=95=E9=99=90=E6=B5=81/=E6=94=B9=E5=AF=86/?= =?UTF-8?q?=E8=A7=92=E8=89=B2=E4=BE=9D=E8=B5=96/409=E8=AF=AD=E4=B9=89?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Code --- app/main.py | 6 + app/platform/auth.py | 295 +++++++++++++++++++++++++++++ app/platform/db.py | 69 +------ app/platform/routes/__init__.py | 1 + app/platform/routes/auth_routes.py | 132 +++++++++++++ tests/conftest.py | 11 ++ tests/test_auth.py | 161 ++++++++++++++++ tests/test_platform_db.py | 22 +-- 8 files changed, 626 insertions(+), 71 deletions(-) create mode 100644 app/platform/auth.py create mode 100644 app/platform/routes/__init__.py create mode 100644 app/platform/routes/auth_routes.py create mode 100644 tests/test_auth.py diff --git a/app/main.py b/app/main.py index 2c9e203..3acb72c 100644 --- a/app/main.py +++ b/app/main.py @@ -80,9 +80,15 @@ app = FastAPI(title="HESUPIPE AI 客服", lifespan=lifespan) from app.engine.routes.chat import router as chat_router from app.engine.routes.chat_rest import router as chat_rest_router from app.engine.routes.public import router as public_router +from app.platform.routes.auth_routes import router as platform_auth_router app.include_router(chat_router) app.include_router(chat_rest_router) app.include_router(public_router) +# 平台认证路由(Task 2):本任务即挂——已知租户 Host(如 jiancai)下 +# /api/platform/auth/* 经 Host 中间件直达,TDD 循环闭环。Task 3 将为 +# /api/platform/ 前缀加中间件豁免(平台后台不依赖租户域名解析); +# 豁免≠放行——认证由路由依赖(get_current_user 系)守护。 +app.include_router(platform_auth_router) app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") diff --git a/app/platform/auth.py b/app/platform/auth.py new file mode 100644 index 0000000..b604cb5 --- /dev/null +++ b/app/platform/auth.py @@ -0,0 +1,295 @@ +"""平台认证内核(Phase 2 Task 2):哈希上移 + 内存 token + 登录限流 + 依赖。 + +本模块零 app 内依赖(不 import db/main)——hash_password/verify_password +从 db.py 上移到此为单一事实源(不留双份漂移),db.py 反向 re-export 兼容 +既有导入;verify 签名解 ORM 绑定为纯参式 (password, password_hash, salt) +(Task 1 加固 M-2b)。 + +构成: +- hash_password / verify_password: 版本化 scrypt,格式与解析兼容不变 + ("scrypt$$$

$$";无前缀过渡行 = salt 列 + + 默认参,兼容照旧) +- TokenStore: 内存 token 表,secrets.token_urlsafe(32),TTL 默认 8h(可注入) +- LoginLimiter: (username, ip) 滑动窗失败计数,5 次/15 分钟(窗长可注入), + 风格照抄 engine/services/rate_limit.py +- UserContext + get_current_user / require_super / require_tenant(slug) / + enforce_password_changed / platform_user: FastAPI 依赖 + +单进程纪律: TokenStore/LoginLimiter 均进程内存态,uvicorn 必须 --workers 1 +(>1 时各 worker 各持一份:token 在 A 签发 B 不识 → 401 误伤;限流配额随 +worker 数翻倍)。与 rate_limit.py 同一取舍(单 worker 部署形态)。 + +安全语义: +- 防枚举: 用户不存在/密码错/停用同一 401 静态文案(auth_routes);用户不 + 存在时也跑一次等代价哑 scrypt,抹平时序侧信道差。 +- 密码/token 永不进日志: 本层与路由层不 log 任何凭据字段,错误 detail + 均静态串。 +- token 走 Authorization: Bearer 头,无 Cookie 免 CSRF(spec §7.2)。 +- 偏差注记: spec §7.2 写 bcrypt/argon2,实现用 hashlib.scrypt(stdlib 零 + 新依赖,抗 GPU 性能同级)——有意取舍。单次 ~50ms CPU,登录/改密路径 + 可接受(不在每请求热路径——依赖只查内存 token 表)。 +""" +import functools +import hashlib +import hmac +import secrets +import time +from collections import defaultdict +from dataclasses import dataclass +from typing import Annotated + +from fastapi import Depends, HTTPException, Request + +# ---- 密码哈希(scrypt,从 db.py 上移;偏差注记见模块 docstring)------------------ + +_SCRYPT_N = 2 ** 14 +_SCRYPT_R = 8 +_SCRYPT_P = 1 + + +def hash_password(password: str, salt: bytes | None = None) -> tuple[str, str]: + """scrypt(password, salt) → (password_hash, salt_hex)。 + + password_hash 为版本化串 "scrypt$$$

$$": + 参数与摘要同存,将来提参/换算法时旧行自解释,有迁移路径(M-1)。 + salt 返回值(列存)与串内嵌的同一盐 hex——单一写方在此同步两处, + verify 以串内为准。salt 缺省每次随机 16 字节;显式传 salt 可复现 + (测试/verify)。""" + if salt is None: + salt = secrets.token_bytes(16) + digest = hashlib.scrypt( + password.encode("utf-8"), salt=salt, + n=_SCRYPT_N, r=_SCRYPT_R, p=_SCRYPT_P) + versioned = ( + f"scrypt${_SCRYPT_N}${_SCRYPT_R}${_SCRYPT_P}" + f"${salt.hex()}${digest.hex()}") + return versioned, salt.hex() + + +def verify_password(password: str, password_hash: str, salt: str) -> bool: + """解析版本化串重算比对,常量时间比较(hmac.compare_digest)。 + + 纯参式 (password, password_hash, salt)(M-2b 解 ORM 绑定):salt 仅在 + 无前缀过渡行时使用(列存 hex),版本化行以串内为准。 + + 解析兼容(M-1): 带 "scrypt$" 前缀 → 参数/盐/摘要全取自串内;无前缀 + (过渡/手工行)→ 盐取 salt 列、参数取当前默认。任何解析失败(坏 hex/ + 坏段数/坏整数)按验败返回 False 而非炸——存储异常不该变 500。""" + stored = password_hash + try: + if stored.startswith("scrypt$"): + parts = stored.split("$") + if len(parts) != 6: + return False + _, n_s, r_s, p_s, salt_hex, hash_hex = parts + n, r, p = int(n_s), int(r_s), int(p_s) + salt_bytes = bytes.fromhex(salt_hex) + expected = bytes.fromhex(hash_hex) + else: + salt_bytes = bytes.fromhex(salt) + expected = bytes.fromhex(stored) + n, r, p = _SCRYPT_N, _SCRYPT_R, _SCRYPT_P + except ValueError: + return False + actual = hashlib.scrypt( + password.encode("utf-8"), salt=salt_bytes, n=n, r=r, p=p) + return hmac.compare_digest(expected, actual) + + +@functools.lru_cache(maxsize=1) +def _dummy_credentials() -> tuple[str, str]: + """固定哑凭据 (password_hash, salt)——登录查无用户时跑一次同代价 + verify,抹平"用户不存在秒回/密码错 50ms"的时序枚举面。""" + return hash_password("", salt=b"\x00" * 16) + + +# ---- TokenStore ----------------------------------------------------------------- + +TOKEN_TTL_SECONDS = 8 * 3600 # 默认 8h(可注入,测试拨窗用) + + +@dataclass +class _TokenRecord: + user_id: int + role: str # "super" | "tenant" + tenant_slug: str | None + must_change_password: bool + expires_at: float # time.time() 纪元秒 + + +class TokenStore: + """内存 token 表: token → 用户快照。单进程纪律见模块 docstring。 + + must_change_password 是签发时快照;change-password 就地清零当前 + token 的标志(auth_routes 语义注记),其他会话的旧 token 保持各自 + 快照(重登后消除)。""" + + def __init__(self, ttl_seconds: float = TOKEN_TTL_SECONDS) -> None: + self.ttl_seconds = ttl_seconds + self._tokens: dict[str, _TokenRecord] = {} + + def issue( + self, *, user_id: int, role: str, tenant_slug: str | None, + must_change_password: bool) -> str: + self._prune() # 懒清理(签发时) + token = secrets.token_urlsafe(32) # 256 bit,碰撞可忽略 + self._tokens[token] = _TokenRecord( + user_id=user_id, role=role, tenant_slug=tenant_slug, + must_change_password=must_change_password, + expires_at=time.time() + self.ttl_seconds) + return token + + def validate(self, token: str) -> _TokenRecord | None: + """有效返回快照;缺失/过期返回 None(过期顺手删除——懒 prune)。""" + record = self._tokens.get(token) + if record is None: + return None + if record.expires_at <= time.time(): + del self._tokens[token] + return None + return record + + def revoke(self, token: str) -> None: + """logout: 立即失效。不存在也静默(重复 logout 幂等)。""" + self._tokens.pop(token, None) + + def clear_must_change(self, token: str) -> None: + """改密成功后就地清当前 token 的引导标志(见类 docstring)。""" + record = self._tokens.get(token) + if record is not None: + record.must_change_password = False + + def _prune(self) -> None: + now = time.time() + for token in [t for t, r in self._tokens.items() + if r.expires_at <= now]: + del self._tokens[token] + + +# ---- LoginLimiter --------------------------------------------------------------- + +LOGIN_WINDOW_SECONDS = 15 * 60 +LOGIN_MAX_FAILS = 5 + + +class LoginLimiter: + """(username, ip) 键滑动窗失败计数: 窗内第 5 次失败后拒绝后续尝试, + 窗过恢复;登录成功即清该键计数(正确密码不受历史失败连累)。 + + 单进程内存态(同 TokenStore 纪律);风格照抄 rate_limit.py。只计 + 失败不计成功——正常用户重复登录不锁。""" + + def __init__( + self, + window_seconds: float = LOGIN_WINDOW_SECONDS, + max_fails: int = LOGIN_MAX_FAILS) -> None: + self.window_seconds = window_seconds + self.max_fails = max_fails + self._fails: dict[tuple[str, str], list[float]] = defaultdict(list) + + def allow(self, username: str, ip: str) -> bool: + """窗内失败数是否仍低于上限(不记录——记录走 record_fail)。""" + self._prune((username, ip), time.time()) + return len(self._fails.get((username, ip), ())) < self.max_fails + + def record_fail(self, username: str, ip: str) -> None: + key = (username, ip) + now = time.time() + self._prune(key, now) + self._fails[key].append(now) + + def reset(self, username: str, ip: str) -> None: + """登录成功清计数(窗口恢复的正路)。""" + self._fails.pop((username, ip), None) + + def _prune(self, key: tuple[str, str], now: float) -> None: + fails = self._fails.get(key) + if fails is None: + return + kept = [t for t in fails if t > now - self.window_seconds] + if kept: + self._fails[key] = kept + else: + self._fails.pop(key, None) + + +# 模块级单例: 路由与依赖共用同一份进程内存态(同 rate_limit._buckets 模式) +token_store = TokenStore() +login_limiter = LoginLimiter() + + +def _reset_for_test() -> None: + """单例复位(测试隔离): token/失败计数全清,注入的 TTL/窗长还原默认。 + conftest autouse 前后各跑一次(照抄 _reset_rate_limit 纪律)。""" + token_store._tokens.clear() + token_store.ttl_seconds = TOKEN_TTL_SECONDS + login_limiter._fails.clear() + login_limiter.window_seconds = LOGIN_WINDOW_SECONDS + + +# ---- FastAPI 依赖 ------------------------------------------------------------------ + +@dataclass(frozen=True) +class UserContext: + user_id: int + role: str # "super" | "tenant" + tenant_slug: str | None # super 为 None + must_change_password: bool + + +def _bearer_token(request: Request) -> str | None: + """Authorization: Bearer → token;非该形态返回 None。""" + header = request.headers.get("authorization", "") + if not header.startswith("Bearer "): + return None + return header[len("Bearer "):].strip() + + +async def get_current_user(request: Request) -> UserContext: + """Bearer 解析 → TokenStore 查验 → UserContext。 + + 缺头/坏格式/未知/过期一律 401(静态文案,不细分原因——token 本身 + 就是凭据,细化文案只会泄露信息)。""" + token = _bearer_token(request) + record = token_store.validate(token) if token else None + if record is None: + raise HTTPException(status_code=401, detail="未认证或凭证已失效") + return UserContext( + user_id=record.user_id, + role=record.role, + tenant_slug=record.tenant_slug, + must_change_password=record.must_change_password) + + +CurrentUser = Annotated[UserContext, Depends(get_current_user)] + + +def require_super(user: CurrentUser) -> UserContext: + """非 super → 403(端点级覆盖在 Task 5;本任务依赖直测)。""" + if user.role != "super": + raise HTTPException(status_code=403, detail="需要平台超管权限") + return user + + +def require_tenant(slug: str): + """依赖工厂: super 或指定租户的管理员放行,否则 403。""" + def _dep(user: CurrentUser) -> UserContext: + if user.role == "super" or user.tenant_slug == slug: + return user + raise HTTPException(status_code=403, detail="无权访问该租户资源") + return _dep + + +def enforce_password_changed(user: CurrentUser) -> UserContext: + """must_change_password=True → 409:除 login/change-password 外的受保护 + 端点都要过本依赖(引导/重置出的初始密码不得只当摆设——spec 强制 + 语义;auth_routes.logout 是 Task 2 的挂载位)。""" + if user.must_change_password: + raise HTTPException(status_code=409, detail="必须先修改初始密码") + return user + + +# 组合依赖: Task 5 起全部 /api/platform/* 受保护端点统一注 +# `user: platform_user`(= get_current_user + enforce_password_changed, +# 链式 Depends 自动先认证后查标志)。 +platform_user = Annotated[UserContext, Depends(enforce_password_changed)] diff --git a/app/platform/db.py b/app/platform/db.py index eb3b4a7..e98ede8 100644 --- a/app/platform/db.py +++ b/app/platform/db.py @@ -17,19 +17,16 @@ PlatformBase.metadata,engine/models.ensure_schema 只跑 Base.metadata—— - admin_users:tenant_id NULL = 超管。password_hash 存版本化串 "scrypt$$$

$$"(参数与摘要同存,将来提参/ 换算法有迁移路径,verify 解析兼容);salt 列同步存同一盐 hex(单一写方 - hash_password,verify 以串内为准)。哈希 = hashlib.scrypt(n=2**14, r=8, - p=1)——偏差注记:spec §7.2 写 bcrypt/argon2,用 stdlib scrypt 为有意 - 取舍(零新依赖,抗 GPU 性能同级);verify 走 hmac.compare_digest - 常量时间比较。 + hash_password,verify 以串内为准)。哈希实现 = hashlib.scrypt(n=2**14, + r=8, p=1)——Task 2 起上移至 platform/auth.py(单一事实源,签名纯参式), + 本模块 re-export 兼容既有导入;偏差注记:spec §7.2 写 bcrypt/argon2, + 用 stdlib scrypt 为有意取舍(零新依赖,抗 GPU 性能同级)。 测试/teardown 纪律:建过 platform engine 的测试结尾 await engine.dispose() (Windows 文件锁,同 registry)。 """ import datetime -import hashlib -import hmac import pathlib -import secrets from sqlalchemy import ( JSON, @@ -51,6 +48,10 @@ from sqlalchemy.ext.asyncio import ( ) from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column +# 密码哈希单一定义点:Task 2 上移至 platform/auth.py(纯参式签名,解 ORM +# 绑定——M-2b),此处 re-export 兼容既有 from app.platform.db import 路径 +from app.platform.auth import hash_password, verify_password # noqa: F401 + def _utcnow() -> datetime.datetime: # 跟随租户库 models.py 既有约定:datetime.now(timezone.utc)(tz-aware UTC) @@ -173,59 +174,7 @@ async def ensure_platform_schema(engine: AsyncEngine) -> None: await conn.run_sync(PlatformBase.metadata.create_all) -# ---- 密码哈希(scrypt,偏差注记见模块 docstring)--------------------------- - -_SCRYPT_N = 2 ** 14 -_SCRYPT_R = 8 -_SCRYPT_P = 1 - - -def hash_password(password: str, salt: bytes | None = None) -> tuple[str, str]: - """scrypt(password, salt) → (password_hash, salt)。 - - password_hash 为版本化串 "scrypt$$$

$$": - 参数与摘要同存,将来提参/换算法时旧行自解释,有迁移路径(M-1)。 - salt 返回值(列存)与串内嵌的同一盐 hex——单一写方在此同步两处, - verify 以串内为准。salt 缺省每次随机 16 字节;显式传 salt 可复现 - (测试/verify)。""" - if salt is None: - salt = secrets.token_bytes(16) - digest = hashlib.scrypt( - password.encode("utf-8"), salt=salt, - n=_SCRYPT_N, r=_SCRYPT_R, p=_SCRYPT_P) - versioned = ( - f"scrypt${_SCRYPT_N}${_SCRYPT_R}${_SCRYPT_P}" - f"${salt.hex()}${digest.hex()}") - return versioned, salt.hex() - - -def verify_password(password: str, user: AdminUser) -> bool: - """解析 password_hash 版本化串重算比对,常量时间比较 - (hmac.compare_digest)——防时序侧信道。 - - 解析兼容(M-1):带 "scrypt$" 前缀 → 参数/盐/摘要全取自串内; - 无前缀(过渡/手工行)→ 盐取 salt 列、参数取当前默认。任何解析失败 - (坏 hex/坏段数/坏整数)按验败返回 False 而非炸——存储异常不该变 500。""" - stored = user.password_hash - try: - if stored.startswith("scrypt$"): - parts = stored.split("$") - if len(parts) != 6: - return False - _, n_s, r_s, p_s, salt_hex, hash_hex = parts - n, r, p = int(n_s), int(r_s), int(p_s) - salt = bytes.fromhex(salt_hex) - expected = bytes.fromhex(hash_hex) - else: - salt = bytes.fromhex(user.salt) - expected = bytes.fromhex(stored) - n, r, p = _SCRYPT_N, _SCRYPT_R, _SCRYPT_P - except ValueError: - return False - actual = hashlib.scrypt( - password.encode("utf-8"), salt=salt, n=n, r=r, p=p) - return hmac.compare_digest(expected, actual) - +# ---- 密码哈希:定义在 auth.py(见文件头 re-export)--------------------------- # ---- 超管引导 --------------------------------------------------------------- diff --git a/app/platform/routes/__init__.py b/app/platform/routes/__init__.py new file mode 100644 index 0000000..2150570 --- /dev/null +++ b/app/platform/routes/__init__.py @@ -0,0 +1 @@ +"""平台层路由包(Phase 2 Task 2 起):/api/platform/* 下的后台端点。""" diff --git a/app/platform/routes/auth_routes.py b/app/platform/routes/auth_routes.py new file mode 100644 index 0000000..965b9ce --- /dev/null +++ b/app/platform/routes/auth_routes.py @@ -0,0 +1,132 @@ +"""平台认证路由(Phase 2 Task 2):/api/platform/auth/{login,logout,change-password}。 + +挂载在主 app(Task 2 即挂——jiancai 等已知租户 Host 下经 Host 中间件直达, +TDD 循环闭环;Task 3 为 /api/platform/ 前缀加中间件豁免,豁免≠放行—— +认证守护在本路由依赖)。 + +安全纪律: +- 防枚举: 用户不存在/停用/密码错同一 401 静态文案;查无用户时跑等代价 + 哑 scrypt(auth._dummy_credentials)抹平时序差。 +- 密码/token 永不进日志: 本模块不 log 任何凭据字段,HTTPException + detail 均静态串。 +- token 走 Authorization: Bearer(无 Cookie 免 CSRF,spec §7.2)。 +- 登录限流先于验证: 锁定期内正确密码也 429(不泄露"密码对了")。 +""" +from fastapi import APIRouter, HTTPException, Request +from pydantic import BaseModel +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.platform.auth import ( + CurrentUser, + _dummy_credentials, + _bearer_token, + hash_password, + login_limiter, + platform_user, + token_store, + verify_password, +) +from app.platform.db import AdminUser, Tenant + +router = APIRouter(prefix="/api/platform/auth", tags=["platform-auth"]) + +BAD_LOGIN_DETAIL = "用户名或密码错误" +LOCKED_DETAIL = "登录尝试过于频繁，请15分钟后再试" + + +class LoginRequest(BaseModel): + username: str + password: str + + +class ChangePasswordRequest(BaseModel): + old_password: str + new_password: str + + +def _client_ip(request: Request) -> str: + return request.client.host if request.client else "" + + +@router.post("/login") +async def login(body: LoginRequest, request: Request): + """限流 → 查用户(active)→ 验密 → 发 token。""" + ip = _client_ip(request) + if not login_limiter.allow(body.username, ip): + raise HTTPException(status_code=429, detail=LOCKED_DETAIL) + + engine = request.app.state.platform_engine + async with AsyncSession(engine, expire_on_commit=False) as session: + row = (await session.execute( + select(AdminUser, Tenant.slug) + .outerjoin(Tenant, AdminUser.tenant_id == Tenant.id) + .filter(AdminUser.username == body.username) + )).first() + + user = row[0] if row else None + if user is not None and user.status == "active": + ok = verify_password(body.password, user.password_hash, user.salt) + else: + # 防枚举(时序): 查无用户/停用也跑一次等代价 scrypt,响应耗时与 + # 密码错不可区分;结果仍统一 401 + dummy_hash, dummy_salt = _dummy_credentials() + verify_password(body.password, dummy_hash, dummy_salt) + ok = False + + if not ok: + login_limiter.record_fail(body.username, ip) + raise HTTPException(status_code=401, detail=BAD_LOGIN_DETAIL) + + login_limiter.reset(body.username, ip) # 成功清失败计数(窗口恢复正路) + role = "super" if user.tenant_id is None else "tenant" + tenant_slug = row[1] + token = token_store.issue( + user_id=user.id, role=role, tenant_slug=tenant_slug, + must_change_password=bool(user.must_change_password)) + return { + "token": token, + "role": role, + "tenant_slug": tenant_slug, + "must_change_password": bool(user.must_change_password), + } + + +@router.post("/logout") +async def logout(request: Request, user: platform_user): + """撤销当前 token(立即失效)。 + + Task 2 的 enforce_password_changed 挂载位:must_change=True → 409 + (Task 5 起所有 /api/platform/* 受保护端点统一走 platform_user, + login/change-password 豁免)。依赖已验过 token,此处重读头取原串 + 撤销——409/401 都在依赖层先行。""" + token_store.revoke(_bearer_token(request) or "") + return {"ok": True} + + +@router.post("/change-password") +async def change_password( + body: ChangePasswordRequest, request: Request, user: CurrentUser): + """验旧密 → 写新 hash+salt → must_change 清零。 + + 本端点豁免 enforce_password_changed(否则引导密码死锁:不改不让用、 + 要用得先改)。 + + 改密后 token 语义(本实现选定并写明):当前 token 不失效——must_change + 标志就地在 TokenStore 清零,用户改完密无需重登。旧密码即刻失效 + (hash+salt 已换)。其他会话的存量 token 不吊销(内存态实现的简化 + 取舍;将来要"改密踢全部会话"再按 user_id 索引吊销,接口不用动)。""" + engine = request.app.state.platform_engine + async with AsyncSession(engine, expire_on_commit=False) as session: + db_user = await session.get(AdminUser, user.user_id) + if db_user is None: + raise HTTPException(status_code=401, detail="账号不存在或已删除") + if not verify_password( + body.old_password, db_user.password_hash, db_user.salt): + raise HTTPException(status_code=400, detail="原密码不正确") + db_user.password_hash, db_user.salt = hash_password(body.new_password) + db_user.must_change_password = False + await session.commit() + + token_store.clear_must_change(_bearer_token(request) or "") + return {"ok": True, "must_change_password": False} diff --git a/tests/conftest.py b/tests/conftest.py index 955f859..134fb86 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -75,6 +75,17 @@ def _reset_conn_limiter(): _reset_for_test() +@pytest.fixture(autouse=True) +def _reset_platform_auth(): + """Phase 2 Task 2: 平台认证模块级单例(内存 token 表/登录限流失败 + 计数)前后复位,跨测试不泄漏——token/限流窗不隔离会串测试(前一脚的 + 锁定状态顶掉后一脚的登录);注入的 TTL/窗长一并还原默认。""" + from app.platform.auth import _reset_for_test + _reset_for_test() + yield + _reset_for_test() + + @pytest.fixture def jiancai_runtime(): """直连测试用的 jiancai TenantRuntime(Task 6: manager 的 runtime 必填)。 diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..9f5a8ae --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,161 @@ +"""平台认证(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 + + +# ---- 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; + change-password 本身豁免(200);改完 409 消除。""" + token = _login(app_client).json()["token"] + assert app_client.post(LOGOUT, headers=_auth(token)).status_code == 409 + 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 diff --git a/tests/test_platform_db.py b/tests/test_platform_db.py index 58bc5f0..4c6fd9d 100644 --- a/tests/test_platform_db.py +++ b/tests/test_platform_db.py @@ -70,8 +70,8 @@ async def test_bootstrap_creates_superuser_once(tmp_path): assert len(users) == 1 and users[0].username == "admin" assert users[0].tenant_id is None # NULL = 超管 assert users[0].must_change_password is True # 引导密码必须首登改掉 - assert verify_password("s3cret!", users[0]) - assert not verify_password("other", users[0]) + assert verify_password("s3cret!", users[0].password_hash, users[0].salt) + assert not verify_password("other", users[0].password_hash, users[0].salt) finally: await engine.dispose() @@ -85,7 +85,7 @@ async def test_bootstrap_noop_when_any_user_exists(tmp_path): await bootstrap_superuser(engine, username="root", password="second") users = await list_admin_users(engine) assert len(users) == 1 and users[0].username == "admin" - assert verify_password("first", users[0]) + assert verify_password("first", users[0].password_hash, users[0].salt) finally: await engine.dispose() @@ -128,23 +128,23 @@ def test_hash_password_versioned_format_and_fresh_salt(): def test_verify_password_bad_stored_hash_returns_false(): """存储异常按验败处理不炸(M-4):坏 hex/坏段数/坏参数 → False; - 无前缀过渡行兼容:salt 列 + 默认参数仍可验真(M-1 解析兼容)。""" + 无前缀过渡行兼容:salt 列 + 默认参数仍可验真(M-1 解析兼容)。 + Task 2 起 verify 纯参式 (password, password_hash, salt)——不再需要 + ORM 实体即可直测。""" bad_stored = [ "not-hex!", # 无前缀,坏 hex "scrypt$16384$8$1$deadbeef", # 段数不足 "scrypt$abc$8$1$00$00", # 参数非整数 ] for stored in bad_stored: - user = AdminUser(username="u", password_hash=stored, salt="zz") - assert verify_password("whatever", user) is False + assert verify_password("whatever", stored, "zz") is False salt = bytes(16) digest = hashlib.scrypt( # 裸摘要 + salt 列的过渡行形态 b"legacy", salt=salt, n=2 ** 14, r=8, p=1) - legacy = AdminUser( - username="u", password_hash=digest.hex(), salt=salt.hex()) - assert verify_password("legacy", legacy) is True - assert verify_password("other", legacy) is False + legacy_hash, legacy_salt = digest.hex(), salt.hex() + assert verify_password("legacy", legacy_hash, legacy_salt) is True + assert verify_password("other", legacy_hash, legacy_salt) is False # ---- 模型约束 ------------------------------------------------------------- @@ -323,5 +323,5 @@ def test_lifespan_platform_engine_seam_and_bootstrap(tmp_path, monkeypatch): users = asyncio.run(_probe()) assert [u.username for u in users] == ["bootadmin"] - assert verify_password("boot-pw", users[0]) + assert verify_password("boot-pw", users[0].password_hash, users[0].salt) assert users[0].must_change_password is True