kplam
92b062c048
- 新增 app/api/deps.py、app/core/users.py、app/core/sessions.py：会话鉴权依赖、 用户存储（PBKDF2-HMAC-SHA256 + 随机 salt，Redis/内存降级）、会话签发与校验（TTL 12h） - auth.py 新增用户管理端点（列表/创建/重置密码/删除）与 admin/user 角色权限边界， user 访问用户管理返回 1006，禁删自己与最后一个 admin - admin.html 新增用户管理面板（仅 admin 挂载）与 API 指南在线测试台 - Dockerfile 将 uv 放入 PATH；docker-compose 调整 qdrant 依赖为 service_started 并移除依赖 curl 的 healthcheck（官方镜像不含 curl） - 新增用户管理测试（users/sessions/auth_api/auth_integration），全量 461 项测试通过 Co-Authored-By: WorkBuddy <workbuddy@tencent.com>
220 lines
9.2 KiB
Python
220 lines
9.2 KiB
Python
"""用户体系集成验证：内存环境全链路
|
||
|
||
内存 UserStore/SessionStore 注入 app.api.deps 单例（不跑 lifespan，无需真实 Redis）；
|
||
文档闭环复用 test_e2e_integration 的内存 Qdrant + FakeOllama 环境（真实入库→删除）。
|
||
|
||
覆盖：
|
||
- 空库 bootstrap_admin 引导 → 登录 → must_change_password 拦截(1006) → 改密 → 恢复(200)
|
||
- 多角色权限：admin 创建 user，user 真实入库/删除文档成功，用户管理端点 1006
|
||
- 未认证与免登录边界：变更类文档端点无 token 1005；查询类端点/health 免登录
|
||
- admin 用户管理闭环：重置密码（旧失效/新可登）、删除用户（session 清除 → me 1005）
|
||
- 引导幂等：已有用户时 bootstrap_admin 返回 None 不重复创建
|
||
"""
|
||
|
||
from typing import Any
|
||
|
||
import pytest
|
||
import structlog
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.api import deps
|
||
from app.api.v1 import document as document_module
|
||
from app.api.v1 import search as search_module
|
||
from app.config import Settings
|
||
from app.core.ingest_tasks import IngestTaskManager, IngestTaskStatus
|
||
from app.core.sessions import SessionStore
|
||
from app.core.users import UserStore, bootstrap_admin
|
||
from app.main import app
|
||
from tests.test_e2e_integration import FakeCache, _make_env, _structured_doc
|
||
|
||
|
||
@pytest.fixture
|
||
def empty_auth_stores(monkeypatch: pytest.MonkeyPatch) -> tuple[UserStore, SessionStore]:
|
||
"""注入空的内存 UserStore/SessionStore 到 deps 单例（不预置账号，用于引导场景）"""
|
||
user_store = UserStore(None)
|
||
session_store = SessionStore(None)
|
||
monkeypatch.setattr(deps, "_user_store", user_store)
|
||
monkeypatch.setattr(deps, "_session_store", session_store)
|
||
return user_store, session_store
|
||
|
||
|
||
def _login(client: TestClient, username: str, password: str) -> dict[str, Any]:
|
||
"""调登录接口并返回响应体"""
|
||
return client.post("/api/v1/auth/login", json={"username": username, "password": password}).json()
|
||
|
||
|
||
def _bearer(token: str) -> dict[str, str]:
|
||
"""构造 Authorization Bearer 请求头"""
|
||
return {"Authorization": f"Bearer {token}"}
|
||
|
||
|
||
class TestBootstrapAndForcedChange:
|
||
"""空库引导 + must_change_password 强制改密全链路"""
|
||
|
||
async def test_bootstrap_login_forced_change(self, empty_auth_stores: tuple[UserStore, SessionStore]) -> None:
|
||
user_store, _ = empty_auth_stores
|
||
|
||
# 空库引导：创建 must_change_password=true 的 admin，明文密码仅本次返回
|
||
password = await bootstrap_admin(user_store, structlog.get_logger())
|
||
assert password is not None
|
||
record = await user_store.get("admin")
|
||
assert record is not None
|
||
assert record.role == "admin"
|
||
assert record.must_change_password is True
|
||
|
||
# 不进上下文：跳过 lifespan，无需真实 Qdrant/Redis
|
||
client = TestClient(app)
|
||
|
||
# 引导密码可登录，响应携带 must_change_password 标记
|
||
login = _login(client, "admin", password)
|
||
assert login["code"] == 0
|
||
assert login["data"]["must_change_password"] is True
|
||
headers = _bearer(login["data"]["token"])
|
||
|
||
# 改密前：业务端点（用户列表）被 1006 拦截
|
||
assert client.get("/api/v1/auth/users", headers=headers).json()["code"] == 1006
|
||
|
||
# 改密（被拦截用户唯一可用接口）
|
||
change = client.post(
|
||
"/api/v1/auth/password",
|
||
json={"old_password": password, "new_password": "admin-new-pass-1"},
|
||
headers=headers,
|
||
).json()
|
||
assert change["code"] == 0
|
||
|
||
# 改密后：同一 token 恢复访问
|
||
resp = client.get("/api/v1/auth/users", headers=headers)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["code"] == 0
|
||
|
||
async def test_bootstrap_idempotent(self, empty_auth_stores: tuple[UserStore, SessionStore]) -> None:
|
||
user_store, _ = empty_auth_stores
|
||
logger = structlog.get_logger()
|
||
|
||
first = await bootstrap_admin(user_store, logger)
|
||
second = await bootstrap_admin(user_store, logger)
|
||
|
||
assert first is not None
|
||
assert second is None # 已有用户时不重复创建
|
||
assert len(await user_store.list()) == 1
|
||
|
||
|
||
class TestRolesAndDocumentChain:
|
||
"""多角色权限 + 文档入库/删除真实内存链路"""
|
||
|
||
async def test_user_role_document_chain(self, auth_stores, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
# 文档端点模块级单例替换为内存链路实例（内存 Qdrant + FakeOllama）
|
||
env = await _make_env()
|
||
manager = IngestTaskManager(env.ingester, None, Settings())
|
||
monkeypatch.setattr(document_module, "_task_manager", manager)
|
||
monkeypatch.setattr(document_module, "_qdrant", env.qdrant)
|
||
|
||
client = TestClient(app)
|
||
|
||
# admin 登录并创建 user 角色账号
|
||
admin_login = _login(client, "admin", "admin-pass-123")
|
||
assert admin_login["code"] == 0
|
||
admin_headers = _bearer(admin_login["data"]["token"])
|
||
created = client.post(
|
||
"/api/v1/auth/users",
|
||
json={"username": "alice", "password": "alice-pass-123", "role": "user"},
|
||
headers=admin_headers,
|
||
).json()
|
||
assert created["code"] == 0
|
||
assert created["data"]["role"] == "user"
|
||
|
||
# user 登录
|
||
user_login = _login(client, "alice", "alice-pass-123")
|
||
assert user_login["code"] == 0
|
||
user_headers = _bearer(user_login["data"]["token"])
|
||
|
||
# user 提交文档入库（202），等待任务跑完写入四层集合
|
||
doc = _structured_doc()
|
||
resp = client.post(
|
||
"/api/v1/documents", json={"text": doc.text, "title": doc.title}, headers=user_headers
|
||
)
|
||
assert resp.status_code == 202
|
||
task_id = resp.json()["data"]["task_id"]
|
||
final = await manager.wait_done(task_id)
|
||
assert final["status"] == IngestTaskStatus.DONE
|
||
doc_id = final["result"]["document_id"]
|
||
|
||
# user 删除自己入库的文档（四层集合真实清除）
|
||
deleted = client.delete(f"/api/v1/documents/{doc_id}", headers=user_headers).json()
|
||
assert deleted["code"] == 0
|
||
assert deleted["data"]["doc_id"] == doc_id
|
||
assert deleted["data"]["deleted_total"] > 0
|
||
|
||
# user 角色访问用户管理端点 → 1006
|
||
assert client.get("/api/v1/auth/users", headers=user_headers).json()["code"] == 1006
|
||
forbidden = client.post(
|
||
"/api/v1/auth/users",
|
||
json={"username": "mallory", "password": "mallory-pass-123"},
|
||
headers=user_headers,
|
||
).json()
|
||
assert forbidden["code"] == 1006
|
||
|
||
|
||
class TestUnauthenticatedBoundaries:
|
||
"""未认证拦截与免登录端点回归"""
|
||
|
||
async def test_boundaries(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
# 查询类端点挂内存链路，避免触达真实 Qdrant/Redis
|
||
env = await _make_env()
|
||
monkeypatch.setattr(document_module, "_task_manager", IngestTaskManager(env.ingester, None, Settings()))
|
||
monkeypatch.setattr(document_module, "_qdrant", env.qdrant)
|
||
monkeypatch.setattr(search_module, "_retriever", env.retriever)
|
||
monkeypatch.setattr(search_module, "get_cache", lambda: FakeCache())
|
||
|
||
client = TestClient(app)
|
||
|
||
# 变更类文档端点：无 token → 1005
|
||
assert client.post("/api/v1/documents", json={"text": "正文"}).json()["code"] == 1005
|
||
assert client.delete("/api/v1/documents/some-doc-id").json()["code"] == 1005
|
||
|
||
# 查询类端点免登录（检索鉴权由 conftest 覆盖旧 JWT 依赖放行）
|
||
search = client.post("/api/v1/search", json={"query": "安装步骤有哪些注意事项？"}).json()
|
||
assert search["code"] == 0
|
||
listing = client.get("/api/v1/documents").json()
|
||
assert listing["code"] == 0
|
||
health = client.get("/api/v1/health")
|
||
assert health.status_code == 200
|
||
assert health.json()["status"] == "ok"
|
||
|
||
|
||
class TestAdminUserManagement:
|
||
"""admin 重置密码 / 删除用户闭环"""
|
||
|
||
def test_reset_and_delete_user(self, auth_stores) -> None:
|
||
client = TestClient(app)
|
||
|
||
admin_login = _login(client, "admin", "admin-pass-123")
|
||
assert admin_login["code"] == 0
|
||
admin_headers = _bearer(admin_login["data"]["token"])
|
||
|
||
# admin 创建 user
|
||
created = client.post(
|
||
"/api/v1/auth/users",
|
||
json={"username": "bob", "password": "bob-pass-123"},
|
||
headers=admin_headers,
|
||
).json()
|
||
assert created["code"] == 0
|
||
assert created["data"]["role"] == "user"
|
||
|
||
# 重置 bob 密码：旧密码登录 1005，新密码登录成功
|
||
reset = client.post(
|
||
"/api/v1/auth/users/bob/password",
|
||
json={"new_password": "bob-new-pass-456"},
|
||
headers=admin_headers,
|
||
).json()
|
||
assert reset["code"] == 0
|
||
assert _login(client, "bob", "bob-pass-123")["code"] == 1005
|
||
bob_login = _login(client, "bob", "bob-new-pass-456")
|
||
assert bob_login["code"] == 0
|
||
bob_headers = _bearer(bob_login["data"]["token"])
|
||
|
||
# admin 删除 bob：其 session 一并清除，me → 1005
|
||
deleted = client.delete("/api/v1/auth/users/bob", headers=admin_headers).json()
|
||
assert deleted["code"] == 0
|
||
assert client.get("/api/v1/auth/me", headers=bob_headers).json()["code"] == 1005
|