diff --git a/CLAUDE.md b/CLAUDE.md index 59f5ce8..c48e270 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -24,7 +24,9 @@ QMDSearch/ │ ├── config.py # 配置管理 │ ├── api/ # API 路由层 │ │ ├── response.py # 统一响应格式 (ok/error/ApiError) +│ │ ├── deps.py # 会话鉴权依赖 (Bearer → session → 用户记录) │ │ └── v1/ # API v1 版本 +│ │ ├── auth.py # 认证与用户管理接口 │ │ ├── search.py # 检索接口 │ │ ├── document.py # 文档入库/管理接口 │ │ └── knowledge.py # 知识库接口 @@ -38,6 +40,8 @@ QMDSearch/ │ │ ├── headings.py # 原生标题树解析 │ │ ├── classifier.py # 文档分类 (主类+标签+置信度) │ │ ├── chunker.py # 标题树感知 chunk 切分 +│ │ ├── users.py # 用户存储与密码哈希 (PBKDF2,Redis/内存降级) +│ │ ├── sessions.py # 会话签发与校验 (Redis TTL 12h/内存降级) │ │ └── ingestion.py # 文档入库 (总结→分类→写入) │ ├── models/ # 数据模型 │ │ ├── search.py # 检索请求/响应模型 @@ -80,25 +84,44 @@ QMDSearch/ ## API 清单 -| 方法 | 路径 | 说明 | -|------|------|------| -| GET | `/api/v1/health` | 健康检查 | -| POST | `/api/v1/search` | 分层检索 | -| POST | `/api/v1/documents` | 文档入库(202 异步入库,返回 task_id) | -| POST | `/api/v1/documents/upload` | multipart 文件上传入库(202 异步,支持 .txt/.md/.html/.htm/.pdf/.docx) | -| GET | `/api/v1/documents/tasks/{task_id}` | 入库任务状态查询(done 附 result,failed 附 error) | -| GET | `/api/v1/knowledge/categories` | 知识分类类目集 | -| GET | `/api/v1/knowledge/stats` | 统计(四层点数 + 类目分布 + uncategorized 数) | -| GET | `/api/v1/documents` | 文档列表(limit/offset 分页) | -| GET | `/api/v1/documents/{doc_id}` | 文档详情 | -| DELETE | `/api/v1/documents/{doc_id}` | 删除文档(幂等) | -| GET | `/admin` | 管理页面 | +| 方法 | 路径 | 说明 | 鉴权 | +|------|------|------|------| +| GET | `/api/v1/health` | 健康检查 | 免登录 | +| POST | `/api/v1/search` | 分层检索 | 免登录 | +| POST | `/api/v1/documents` | 文档入库(202 异步入库,返回 task_id) | Bearer | +| POST | `/api/v1/documents/upload` | multipart 文件上传入库(202 异步,支持 .txt/.md/.html/.htm/.pdf/.docx) | Bearer | +| GET | `/api/v1/documents/tasks/{task_id}` | 入库任务状态查询(done 附 result,failed 附 error) | 免登录 | +| GET | `/api/v1/knowledge/categories` | 知识分类类目集 | 免登录 | +| GET | `/api/v1/knowledge/stats` | 统计(四层点数 + 类目分布 + uncategorized 数) | 免登录 | +| GET | `/api/v1/documents` | 文档列表(limit/offset 分页) | 免登录 | +| GET | `/api/v1/documents/{doc_id}` | 文档详情 | 免登录 | +| DELETE | `/api/v1/documents/{doc_id}` | 删除文档(幂等) | Bearer | +| POST | `/api/v1/auth/login` | 用户名密码登录,签发 session token(TTL 12h) | 免登录 | +| POST | `/api/v1/auth/logout` | 退出登录(删除当前 session) | Bearer | +| POST | `/api/v1/auth/password` | 修改自己的密码(must_change_password 用户唯一可用接口) | Bearer | +| GET | `/api/v1/auth/me` | 当前登录用户信息(脱敏) | Bearer | +| GET | `/api/v1/auth/users` | 用户列表(脱敏) | Bearer + admin | +| POST | `/api/v1/auth/users` | 创建用户(重名/非法用户名/弱密码 1001) | Bearer + admin | +| POST | `/api/v1/auth/users/{username}/password` | 重置指定用户密码(成功后清除其全部 session) | Bearer + admin | +| DELETE | `/api/v1/auth/users/{username}` | 删除用户(清除其 session;禁删自己/最后一个 admin) | Bearer + admin | +| GET | `/admin` | 管理页面 | 页面登录门禁 | + +「Bearer」指请求头 `Authorization: Bearer `,token 经 `/api/v1/auth/login` 获取;变更类文档端点(POST /documents、POST /documents/upload、DELETE /documents/{id})需 Bearer token(admin/user 角色均可),查询类端点免登录。 入库任务状态持久化在 Redis(key: `ingest_task:{task_id}`):进行中与 done 保留 24h,failed 保留 7 天;Redis 不可用时降级为纯内存。 ## 管理页面 -浏览器访问 `/admin`,单页面含概览、文档管理(列表/详情/删除)、文档入库、检索测试台、类目列表五个区块。 +浏览器访问 `/admin`,页面带登录门禁(未登录/token 失效自动回登录卡片;must_change_password 用户先强制改密后方可进入)。单页面含七个区块:概览、文档管理(列表/详情/删除)、文档入库(文本 + 文件上传)、检索测试台、类目列表、API 指南(端点清单 + 在线测试台)、用户管理(仅 admin 角色挂载,含创建/重置密码/删除)。 + +## 认证与用户 + +会话制认证:登录签发 session token(Redis 持久化,TTL 12h;Redis 不可用时降级为进程内存,重启失效),请求经 `Authorization: Bearer ` 携带,鉴权依赖见 `app/api/deps.py`。 + +- **角色与权限边界**: `admin` 拥有全部权限(含 /auth/users* 用户管理);`user` 可登录并调用变更类文档端点(入库/上传/删除),访问用户管理端点返回 1006 +- **初始 admin 引导**: 空库启动时 `bootstrap_admin` 自动创建 admin 账号,随机明文密码仅在启动日志中打印一次(must_change_password=true),首次登录后须先经 POST /auth/password 改密,改密前访问其他端点返回 1006 +- **免登录端点**: 查询类端点(POST /search、GET /documents*、GET /knowledge/*、GET /health)不需要 token +- **相关错误码**: 1005 未认证或凭证无效,1006 权限不足/首次登录须先改密 ## 编码规范 diff --git a/Dockerfile b/Dockerfile index 354dd7d..d8e9a9a 100644 --- a/Dockerfile +++ b/Dockerfile @@ -16,8 +16,8 @@ FROM python:3.12-slim AS base WORKDIR /app -# 安装 uv -COPY --from=ghcr.io/astral-sh/uv:latest /uv /uv/bin/uv +# 安装 uv(放到 PATH 中,供 uv sync 与 CMD 的 uv run 使用) +COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv # 依赖层 COPY pyproject.toml uv.lock ./ diff --git a/app/api/deps.py b/app/api/deps.py new file mode 100644 index 0000000..8e52e3e --- /dev/null +++ b/app/api/deps.py @@ -0,0 +1,95 @@ +"""API 鉴权依赖:Bearer token → session → 用户记录 + +UserStore/SessionStore 为模块级懒加载单例:Redis 客户端创建失败时构造传 None, +降级为进程级内存存储(仅开发/兜底,重启失效)。存储后端读写异常统一转 2001。 +测试可 monkeypatch _user_store/_session_store 单例完成注入。 +""" + +import structlog +from fastapi import Depends, Header +from redis import asyncio as redis_async + +from app.api.response import ApiError +from app.config import settings +from app.core.sessions import SessionStore, SessionStoreError +from app.core.users import UserRecord, UserStore, UserStoreError + +logger = structlog.get_logger() + +# 模块级懒加载单例 +_user_store: UserStore | None = None +_session_store: SessionStore | None = None + + +def _create_redis_client() -> redis_async.Redis | None: + """创建 redis.asyncio 客户端(decode_responses=True);失败返回 None 走内存降级""" + try: + return redis_async.from_url(settings.redis_url, decode_responses=True) + except Exception: + logger.warning("Redis 客户端创建失败,认证存储降级为内存模式", exc_info=True) + return None + + +def _get_user_store() -> UserStore: + """用户存储懒加载单例""" + global _user_store + if _user_store is None: + _user_store = UserStore(_create_redis_client()) + return _user_store + + +def _get_session_store() -> SessionStore: + """会话存储懒加载单例""" + global _session_store + if _session_store is None: + _session_store = SessionStore(_create_redis_client()) + return _session_store + + +def _parse_bearer(authorization: str | None) -> str | None: + """解析 Authorization 头中的 Bearer token;缺失或格式非法返回 None""" + if not authorization: + return None + scheme, _, token = authorization.partition(" ") + if scheme.lower() != "bearer" or not token.strip(): + return None + return token.strip() + + +async def _resolve_user(authorization: str | None) -> tuple[UserRecord, str]: + """Bearer token → session → 用户记录(不拦截 must_change_password) + + 凭证缺失/无效/用户不存在抛 ApiError(1005);存储后端异常抛 ApiError(2001)。 + 供 get_current_user 与改密/退出端点共用(后者须对 must_change_password 用户放行)。 + """ + token = _parse_bearer(authorization) + if token is None: + raise ApiError(1005, "未认证或凭证无效") + try: + session = await _get_session_store().get(token) + if session is None: + raise ApiError(1005, "未认证或凭证无效") + user = await _get_user_store().get(session["username"]) + except (SessionStoreError, UserStoreError) as exc: + raise ApiError(2001, "认证服务暂不可用") from exc + if user is None: + raise ApiError(1005, "未认证或凭证无效") + return user, token + + +async def get_current_user(authorization: str | None = Header(None)) -> UserRecord: + """鉴权依赖:校验 Bearer token 并返回当前用户记录 + + must_change_password 用户被拦截(1006),须先经 POST /auth/password 改密。 + """ + user, _ = await _resolve_user(authorization) + if user.must_change_password: + raise ApiError(1006, "首次登录须先修改密码") + return user + + +async def require_admin(user: UserRecord = Depends(get_current_user)) -> UserRecord: + """鉴权依赖:在 get_current_user 之上要求 admin 角色""" + if user.role != "admin": + raise ApiError(1006, "权限不足") + return user diff --git a/app/api/v1/auth.py b/app/api/v1/auth.py index 5129ed6..31700ca 100644 --- a/app/api/v1/auth.py +++ b/app/api/v1/auth.py @@ -1,62 +1,192 @@ -"""认证 API:POST /auth/login、POST /auth/register、GET /auth/me""" +"""认证与用户管理 API:/api/v1/auth 下的登录/退出/改密与用户管理端点 -from typing import Any +会话制认证:登录签发 session token(存储见 app/core/sessions.py,TTL 12h), +请求经 Authorization: Bearer 携带,鉴权依赖见 app/api/deps.py。 +用户管理端点(/users*)全部要求 admin 角色;/logout 与 /password 对 +must_change_password 用户放行(改密是被拦截用户唯一能用的接口)。 +""" + +from typing import Any, Literal import structlog -from fastapi import APIRouter, Depends +from fastapi import APIRouter, Depends, Header +from pydantic import BaseModel +from app.api import deps from app.api.response import ApiError, ok -from app.config import settings -from app.core.auth import ( - ERR_REGISTER_DISABLED, - UserStore, - create_access_token, - get_current_user, - get_user_store, -) -from app.models.auth import ( - AuthUser, - LoginRequest, - RegisterRequest, - TokenResponse, -) +from app.core.sessions import SessionStoreError +from app.core.users import UserExistsError, UserRecord, UserStoreError logger = structlog.get_logger() -router = APIRouter(prefix="/api/v1", tags=["auth"]) +router = APIRouter(prefix="/api/v1/auth", tags=["auth"]) -def _build_token_response(user, store: UserStore) -> dict[str, Any]: - """签发 token 并构造统一响应 data""" - token, expires_in = create_access_token(user.username, user.role) - auth_user = AuthUser(username=user.username, role=user.role, created_at=user.created_at) - resp = TokenResponse(access_token=token, expires_in=expires_in, user=auth_user) - return resp.model_dump(mode="json") +class LoginRequest(BaseModel): + """登录请求""" + + username: str + password: str -@router.post("/auth/login") +class PasswordChangeRequest(BaseModel): + """修改自己的密码""" + + old_password: str + new_password: str + + +class UserCreateRequest(BaseModel): + """管理员创建用户""" + + username: str + password: str + role: Literal["admin", "user"] = "user" + + +class PasswordResetRequest(BaseModel): + """管理员重置他人密码""" + + new_password: str + + +def _public_user(user: UserRecord) -> dict[str, Any]: + """用户记录脱敏:剔除 password_hash/salt""" + return { + "username": user.username, + "role": user.role, + "must_change_password": user.must_change_password, + "created_at": user.created_at, + } + + +@router.post("/login") async def login(req: LoginRequest) -> dict[str, Any]: - """用户名密码登录,返回 JWT access_token""" - store = get_user_store() - user = await store.authenticate(req.username, req.password) + """用户名密码登录:成功签发 session token(must_change_password 用户也可登录)""" + try: + user = await deps._get_user_store().verify_password(req.username, req.password) + except UserStoreError as exc: + raise ApiError(2001, "认证服务暂不可用") from exc + if user is None: + raise ApiError(1005, "用户名或密码错误") + try: + token = await deps._get_session_store().create(user.username, user.role) + except SessionStoreError as exc: + raise ApiError(2001, "认证服务暂不可用") from exc logger.info("用户登录成功", username=user.username) - return ok(_build_token_response(user, store)) + return ok( + { + "token": token, + "username": user.username, + "role": user.role, + "must_change_password": user.must_change_password, + } + ) -@router.post("/auth/register") -async def register(req: RegisterRequest) -> dict[str, Any]: - """注册新用户(role=user),注册后自动签发 token +@router.get("/me") +async def me(user: UserRecord = Depends(deps.get_current_user)) -> dict[str, Any]: + """返回当前登录用户信息(脱敏),供管理页面校验登录态""" + return ok(_public_user(user)) - 受 settings.auth_register_enabled 控制,关闭时返回 ERR_REGISTER_DISABLED。 + +@router.post("/logout") +async def logout(authorization: str | None = Header(None)) -> dict[str, Any]: + """退出登录:删除当前 session(自解析 Bearer,不拦截 must_change_password)""" + _, token = await deps._resolve_user(authorization) + try: + await deps._get_session_store().delete(token) + except SessionStoreError as exc: + raise ApiError(2001, "认证服务暂不可用") from exc + return ok({"logged_out": True}) + + +@router.post("/password") +async def change_password( + req: PasswordChangeRequest, authorization: str | None = Header(None) +) -> dict[str, Any]: + """修改自己的密码(自解析 Bearer,不拦截 must_change_password——这是被拦截用户唯一可用接口) + + 旧密码错误 1005;新密码不足 8 位 1001。改密不清除既有 session(Task1 语义)。 """ - if not settings.auth_register_enabled: - raise ApiError(ERR_REGISTER_DISABLED, "注册已关闭") - store = get_user_store() - user = await store.create(req.username, req.password, role="user") - return ok(_build_token_response(user, store)) + user, _ = await deps._resolve_user(authorization) + store = deps._get_user_store() + try: + verified = await store.verify_password(user.username, req.old_password) + if verified is None: + raise ApiError(1005, "旧密码错误") + await store.set_password(user.username, req.new_password) + except (UserStoreError, SessionStoreError) as exc: + raise ApiError(2001, "认证服务暂不可用") from exc + except ValueError as exc: + raise ApiError(1001, str(exc)) from exc + logger.info("用户修改密码", username=user.username) + return ok({"username": user.username}) -@router.get("/auth/me") -async def me(user: AuthUser = Depends(get_current_user)) -> dict[str, Any]: - """返回当前登录用户信息""" - return ok(user.model_dump(mode="json")) +@router.get("/users") +async def list_users(admin: UserRecord = Depends(deps.require_admin)) -> dict[str, Any]: + """列出全部用户(脱敏:不含 password_hash/salt)""" + try: + users = await deps._get_user_store().list() + except UserStoreError as exc: + raise ApiError(2001, "认证服务暂不可用") from exc + return ok([_public_user(user) for user in users]) + + +@router.post("/users") +async def create_user( + req: UserCreateRequest, admin: UserRecord = Depends(deps.require_admin) +) -> dict[str, Any]: + """创建用户:重名/非法用户名/弱密码 1001;成功返回脱敏后的创建记录""" + try: + user = await deps._get_user_store().create(req.username, req.password, role=req.role) + except UserExistsError as exc: + raise ApiError(1001, str(exc)) from exc + except ValueError as exc: + raise ApiError(1001, str(exc)) from exc + except UserStoreError as exc: + raise ApiError(2001, "认证服务暂不可用") from exc + logger.info("管理员创建用户", username=user.username, role=user.role, operator=admin.username) + return ok(_public_user(user)) + + +@router.post("/users/{username}/password") +async def reset_user_password( + username: str, req: PasswordResetRequest, admin: UserRecord = Depends(deps.require_admin) +) -> dict[str, Any]: + """重置指定用户密码:用户不存在 1004;成功后清除该用户全部 session""" + store = deps._get_user_store() + try: + if await store.get(username) is None: + raise ApiError(1004, "用户不存在") + await store.set_password(username, req.new_password) + await deps._get_session_store().delete_by_username(username) + except (UserStoreError, SessionStoreError) as exc: + raise ApiError(2001, "认证服务暂不可用") from exc + except ValueError as exc: + raise ApiError(1001, str(exc)) from exc + logger.info("管理员重置用户密码", username=username, operator=admin.username) + return ok({"username": username}) + + +@router.delete("/users/{username}") +async def delete_user( + username: str, admin: UserRecord = Depends(deps.require_admin) +) -> dict[str, Any]: + """删除用户:不存在 1004;删除自己 1001;最后一个 admin 禁止删除 1001;成功后清除其 session""" + store = deps._get_user_store() + try: + target = await store.get(username) + if target is None: + raise ApiError(1004, "用户不存在") + if target.username == admin.username: + raise ApiError(1001, "不能删除当前登录账号") + if target.role == "admin" and await store.count_admins() <= 1: + raise ApiError(1001, "禁止删除最后一个管理员") + await store.delete(username) + await deps._get_session_store().delete_by_username(username) + except (UserStoreError, SessionStoreError) as exc: + raise ApiError(2001, "认证服务暂不可用") from exc + logger.info("管理员删除用户", username=username, operator=admin.username) + return ok({"username": username}) diff --git a/app/api/v1/document.py b/app/api/v1/document.py index 0ddf3e4..602b2d4 100644 --- a/app/api/v1/document.py +++ b/app/api/v1/document.py @@ -10,12 +10,13 @@ import structlog from fastapi import APIRouter, Depends, File, Form, Query, UploadFile from fastapi.responses import JSONResponse +from app.api.deps import get_current_user from app.api.response import ApiError, ok from app.config import Settings, settings -from app.core.auth import AuthUser, get_current_user, require_admin from app.core.file_parser import parse_file from app.core.ingest_tasks import IngestTaskManager from app.core.ingestion import Ingester +from app.core.users import UserRecord from app.models.document import DocumentInput from app.services.qdrant import QdrantService from app.services.redis import RedisCache, get_cache @@ -74,7 +75,7 @@ def _allowed_extensions() -> set[str]: @router.post("/documents") async def ingest_document( - doc: DocumentInput, user: AuthUser = Depends(get_current_user) + doc: DocumentInput, user: UserRecord = Depends(get_current_user) ) -> JSONResponse: """文档入库入口:登记异步任务并返回 202 + task_id,入库结果经任务查询端点获取""" if not doc.text.strip(): @@ -95,7 +96,7 @@ async def upload_document( metadata: str = Form( default="", description='可选元数据 JSON 字符串,如 \'{"author":"x"}\'' ), - user: AuthUser = Depends(get_current_user), + user: UserRecord = Depends(get_current_user), ) -> JSONResponse: """文件上传入库入口:校验 → 提取文本 → 落盘 → 提交异步入库流水线 @@ -183,9 +184,7 @@ async def upload_document( @router.get("/documents/tasks/{task_id}") -async def get_ingest_task( - task_id: str, user: AuthUser = Depends(get_current_user) -) -> dict[str, Any]: +async def get_ingest_task(task_id: str) -> dict[str, Any]: """查询入库任务状态:含 task_id/status/created_at/updated_at,done 附 result,failed 附 error""" task = await _get_task_manager().get(task_id) if task is None: @@ -197,7 +196,6 @@ async def get_ingest_task( async def list_documents( limit: int = Query(default=20, ge=1, le=100), offset: str | None = None, - user: AuthUser = Depends(get_current_user), ) -> dict[str, Any]: """分页列出文档(L1 摘要),返回 items 与下一页游标 next_offset""" try: @@ -209,9 +207,7 @@ async def list_documents( @router.get("/documents/{doc_id}") -async def get_document( - doc_id: str, user: AuthUser = Depends(get_current_user) -) -> dict[str, Any]: +async def get_document(doc_id: str) -> dict[str, Any]: """获取文档详情:L1 记录 + L2/L3 节点 + chunks 数量""" try: detail = await _get_qdrant().get_doc_detail(doc_id) @@ -225,7 +221,7 @@ async def get_document( @router.delete("/documents/{doc_id}") async def delete_document( - doc_id: str, user: AuthUser = Depends(require_admin) + doc_id: str, user: UserRecord = Depends(get_current_user) ) -> dict[str, Any]: """删除文档:四层集合中该 doc_id 的所有点;幂等,不存在也返回成功(删除数全 0)""" try: diff --git a/app/core/sessions.py b/app/core/sessions.py new file mode 100644 index 0000000..16a6c31 --- /dev/null +++ b/app/core/sessions.py @@ -0,0 +1,100 @@ +"""Session 核心:登录会话的签发与校验 + +存储后端为 Redis(key: session:{token},JSON 序列化,TTL 12h); +构造时传入 None 降级为进程级内存 dict(带过期时间戳,惰性清理过期项,重启失效)。 +所有 Redis 读写异常统一包装为 SessionStoreError 抛出(不静默)。 +""" + +import json +import secrets +import time + +from redis import asyncio as redis_async + +# Redis 键前缀 +_SESSION_KEY_PREFIX = "session:" + + +class SessionStoreError(Exception): + """Session 存储后端读写异常""" + + +class SessionStore: + """会话存储:Redis 持久化(带 TTL),构造传 None 降级为内存 dict""" + + TTL: int = 43200 # 会话有效期(秒),12h + + def __init__(self, redis_client: redis_async.Redis | None) -> None: + # redis.asyncio 客户端(需 decode_responses=True);None → 内存降级 + self._redis = redis_client + # 内存降级:token -> (payload, 过期时间戳) + self._memory: dict[str, tuple[dict[str, str], float]] = {} + + @staticmethod + def _key(token: str) -> str: + return f"{_SESSION_KEY_PREFIX}{token}" + + async def create(self, username: str, role: str) -> str: + """签发会话:生成 64 位 hex token,存 session:{token} = {username, role}""" + token = secrets.token_hex(32) + payload = {"username": username, "role": role} + if self._redis is None: + self._purge_expired() + self._memory[token] = (payload, time.time() + self.TTL) + return token + try: + await self._redis.setex(self._key(token), self.TTL, json.dumps(payload, ensure_ascii=False)) + except Exception as e: + raise SessionStoreError("写入会话失败") from e + return token + + async def get(self, token: str) -> dict[str, str] | None: + """校验会话:无效或已过期返回 None,有效返回 {username, role}""" + if self._redis is None: + item = self._memory.get(token) + if item is None: + return None + payload, expire_at = item + if expire_at <= time.time(): + # 惰性清理过期项 + del self._memory[token] + return None + return dict(payload) + try: + raw = await self._redis.get(self._key(token)) + except Exception as e: + raise SessionStoreError("读取会话失败") from e + if raw is None: + return None + return json.loads(raw) + + async def delete(self, token: str) -> None: + """删除单个会话(幂等)""" + if self._redis is None: + self._memory.pop(token, None) + return + try: + await self._redis.delete(self._key(token)) + except Exception as e: + raise SessionStoreError("删除会话失败") from e + + async def delete_by_username(self, username: str) -> None: + """删除指定用户的全部会话(删除用户时清理其登录态)""" + if self._redis is None: + matched = [t for t, (payload, _) in self._memory.items() if payload["username"] == username] + for token in matched: + del self._memory[token] + return + try: + async for key in self._redis.scan_iter(match=f"{_SESSION_KEY_PREFIX}*"): + raw = await self._redis.get(key) + if raw is not None and json.loads(raw).get("username") == username: + await self._redis.delete(key) + except Exception as e: + raise SessionStoreError(f"按用户删除会话失败: {username}") from e + + def _purge_expired(self) -> None: + """惰性清理内存降级模式下的过期会话""" + now = time.time() + for token in [t for t, (_, expire_at) in self._memory.items() if expire_at <= now]: + del self._memory[token] diff --git a/app/core/users.py b/app/core/users.py new file mode 100644 index 0000000..4983700 --- /dev/null +++ b/app/core/users.py @@ -0,0 +1,191 @@ +"""用户核心:用户记录存储与密码哈希 + +存储后端为 Redis(key: user:{username},JSON 序列化,无 TTL 持久化); +构造时传入 None 降级为进程级内存 dict(重启丢失,仅用于测试与 Redis 故障兜底)。 +所有 Redis 读写异常统一包装为 UserStoreError 抛出(不静默),由上层转 2001。 +密码哈希使用标准库 hashlib.pbkdf2_hmac(SHA256,100_000 迭代 + 16 字节随机 salt)。 +""" + +import hashlib +import hmac +import json +import re +import secrets +from dataclasses import asdict, dataclass +from datetime import UTC, datetime + +from redis import asyncio as redis_async +from structlog.typing import FilteringBoundLogger + +# 用户名规则:字母/数字/下划线/连字符,2~32 位 +USERNAME_PATTERN = re.compile(r"^[a-zA-Z0-9_-]{2,32}$") + +# 密码最小长度 +MIN_PASSWORD_LENGTH = 8 + +# PBKDF2 迭代次数 +_HASH_ITERATIONS = 100_000 + +# salt 字节数(hex 序列化后 32 字符) +_SALT_BYTES = 16 + +# Redis 键前缀 +_USER_KEY_PREFIX = "user:" + + +class UserExistsError(Exception): + """用户名已存在""" + + +class UserStoreError(Exception): + """用户存储后端读写异常""" + + +@dataclass +class UserRecord: + """用户记录(不含明文密码)""" + + username: str + role: str # "admin" | "user" + password_hash: str # hex + salt: str # hex, 16 字节 + must_change_password: bool + created_at: str # UTC ISO8601 + + +class UserStore: + """用户存储:Redis 持久化,构造传 None 降级为内存 dict""" + + def __init__(self, redis_client: redis_async.Redis | None) -> None: + # redis.asyncio 客户端(需 decode_responses=True);None → 内存降级 + self._redis = redis_client + # 内存降级:username -> JSON 字符串(与 Redis 路径同构) + self._memory: dict[str, str] = {} + + @staticmethod + def _key(username: str) -> str: + return f"{_USER_KEY_PREFIX}{username}" + + def hash_password(self, password: str, salt: bytes) -> str: + """PBKDF2-HMAC-SHA256(100_000 迭代),返回 hex 摘要""" + return hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, _HASH_ITERATIONS).hex() + + async def create( + self, + username: str, + password: str, + role: str = "user", + must_change_password: bool = False, + ) -> UserRecord: + """创建用户 + + 重名抛 UserExistsError;用户名不符合 USERNAME_PATTERN 或密码过短抛 ValueError。 + """ + if not USERNAME_PATTERN.match(username): + raise ValueError(f"用户名非法: {username!r}(须为 2~32 位字母/数字/_/-)") + if len(password) < MIN_PASSWORD_LENGTH: + raise ValueError(f"密码长度不得少于 {MIN_PASSWORD_LENGTH} 位") + if await self.get(username) is not None: + raise UserExistsError(f"用户已存在: {username}") + salt = secrets.token_bytes(_SALT_BYTES) + record = UserRecord( + username=username, + role=role, + password_hash=self.hash_password(password, salt), + salt=salt.hex(), + must_change_password=must_change_password, + created_at=datetime.now(UTC).isoformat(), + ) + await self._write(record) + return record + + async def get(self, username: str) -> UserRecord | None: + """按用户名查询,不存在返回 None""" + raw = await self._read_raw(username) + if raw is None: + return None + return UserRecord(**json.loads(raw)) + + async def list(self) -> list[UserRecord]: + """列出全部用户(扫描 user:* 键)""" + if self._redis is None: + raws: list[str | None] = list(self._memory.values()) + else: + try: + raws = [await self._redis.get(key) async for key in self._redis.scan_iter(match=f"{_USER_KEY_PREFIX}*")] + except Exception as e: + raise UserStoreError("列出用户失败") from e + return [UserRecord(**json.loads(raw)) for raw in raws if raw is not None] + + async def delete(self, username: str) -> bool: + """删除用户,返回是否删除成功(幂等:不存在返回 False)""" + if self._redis is None: + return self._memory.pop(username, None) is not None + try: + return bool(await self._redis.delete(self._key(username))) + except Exception as e: + raise UserStoreError(f"删除用户失败: {username}") from e + + async def verify_password(self, username: str, password: str) -> UserRecord | None: + """校验密码:成功返回用户记录,用户不存在或密码错误返回 None""" + record = await self.get(username) + if record is None: + return None + candidate = self.hash_password(password, bytes.fromhex(record.salt)) + # 常量时间比较,防时序侧信道 + if not hmac.compare_digest(candidate, record.password_hash): + return None + return record + + async def set_password(self, username: str, new_password: str) -> bool: + """重置密码:重新生成 salt 并哈希,同时清除 must_change_password 标记 + + 用户不存在返回 False;新密码过短抛 ValueError。 + """ + if len(new_password) < MIN_PASSWORD_LENGTH: + raise ValueError(f"密码长度不得少于 {MIN_PASSWORD_LENGTH} 位") + record = await self.get(username) + if record is None: + return False + salt = secrets.token_bytes(_SALT_BYTES) + record.salt = salt.hex() + record.password_hash = self.hash_password(new_password, salt) + record.must_change_password = False + await self._write(record) + return True + + async def count_admins(self) -> int: + """统计 admin 角色用户数""" + return sum(1 for record in await self.list() if record.role == "admin") + + async def _read_raw(self, username: str) -> str | None: + if self._redis is None: + return self._memory.get(username) + try: + return await self._redis.get(self._key(username)) + except Exception as e: + raise UserStoreError(f"读取用户失败: {username}") from e + + async def _write(self, record: UserRecord) -> None: + raw = json.dumps(asdict(record), ensure_ascii=False) + if self._redis is None: + self._memory[record.username] = raw + return + try: + await self._redis.set(self._key(record.username), raw) + except Exception as e: + raise UserStoreError(f"写入用户失败: {record.username}") from e + + +async def bootstrap_admin(store: UserStore, logger: FilteringBoundLogger) -> str | None: + """空库引导:无任何用户时创建随机密码的 admin 账号 + + 明文密码通过 logger.warning 打印一次并返回(仅此一次机会); + 已有用户时不做任何事,返回 None。 + """ + if await store.list(): + return None + password = secrets.token_urlsafe(12) + await store.create("admin", password, role="admin", must_change_password=True) + logger.warning("已创建默认管理员,请立即登录并修改密码", username="admin", password=password) + return password diff --git a/app/main.py b/app/main.py index 5361416..e58ddce 100644 --- a/app/main.py +++ b/app/main.py @@ -6,6 +6,7 @@ import structlog from fastapi import FastAPI, Request from fastapi.exceptions import RequestValidationError from fastapi.responses import FileResponse, JSONResponse, RedirectResponse +from redis import asyncio as redis_async from app.api.response import ApiError, error from app.api.v1.auth import router as auth_router @@ -14,7 +15,7 @@ from app.api.v1.knowledge import router as knowledge_router from app.api.v1.search import router as search_router from app.api.v1.settings import router as settings_router from app.config import settings -from app.core.auth import ensure_default_admin +from app.core.users import UserStore, bootstrap_admin from app.services.qdrant import QdrantService logger = structlog.get_logger() @@ -32,7 +33,13 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]: except Exception: logger.error("Qdrant 集合初始化失败,跳过初始化继续启动") try: - await ensure_default_admin() + # Redis 客户端创建失败(如 URL 非法)时传 None,UserStore 降级为内存模式 + redis_client = redis_async.from_url(settings.redis_url, decode_responses=True) + except Exception: + redis_client = None + try: + # 空库引导默认管理员;明文密码由 bootstrap_admin 内部 warning 打印一次 + await bootstrap_admin(UserStore(redis_client), logger) except Exception: logger.warning("默认管理员初始化失败,跳过", exc_info=True) try: diff --git a/app/static/admin.html b/app/static/admin.html index 295e6b1..a322244 100644 --- a/app/static/admin.html +++ b/app/static/admin.html @@ -79,7 +79,7 @@ button.primary:disabled { background: #9ca3af; cursor: not-allowed; } form .field { margin-bottom: 12px; } form label { display: block; font-size: 13px; margin-bottom: 4px; color: #374151; } - input[type="text"], input[type="number"], textarea { + input[type="text"], input[type="number"], input[type="password"], textarea, select { width: 100%; border: 1px solid #d1d5db; border-radius: 4px; padding: 8px; font-size: 13px; font-family: inherit; } @@ -118,6 +118,12 @@ header { position: relative; } .user-area { position: absolute; top: 14px; right: 24px; display: flex; align-items: center; gap: 8px; } .user-area .action { padding: 4px 10px; } + .user-area #user-name { color: #f9fafb; font-size: 13px; } + .role-badge { + display: inline-block; border-radius: 3px; padding: 1px 8px; font-size: 12px; + background: #eff6ff; color: #1d4ed8; border: 1px solid #93c5fd; + } + .role-badge.role-admin { background: #fef3c7; color: #b45309; border-color: #fcd34d; } .login-overlay { position: fixed; inset: 0; background: rgba(0,0,0,0.45); display: flex; align-items: center; justify-content: center; z-index: 100; @@ -139,6 +145,45 @@ padding: 12px 14px; margin-bottom: 12px; font-size: 13px; white-space: pre-wrap; } .summary-box .sum-title { font-weight: 600; margin-bottom: 6px; color: #15803d; } + .method-badge { + display: inline-block; border-radius: 3px; padding: 1px 8px; + font-size: 12px; font-weight: 600; color: #fff; margin-right: 8px; + } + .method-get { background: #16a34a; } + .method-post { background: #2563eb; } + .method-delete { background: #dc2626; } + .auth-badge { + display: inline-block; border-radius: 3px; padding: 1px 8px; + font-size: 12px; margin-left: 8px; border: 1px solid transparent; + } + .auth-badge.auth-none { background: #f0fdf4; color: #15803d; border-color: #86efac; } + .auth-badge.auth-bearer { background: #eff6ff; color: #1d4ed8; border-color: #93c5fd; } + .auth-badge.auth-admin { background: #fef3c7; color: #b45309; border-color: #fcd34d; } + .api-item { border: 1px solid #e5e7eb; border-radius: 6px; padding: 12px 14px; margin-bottom: 12px; } + .api-item .api-head { display: flex; align-items: center; flex-wrap: wrap; gap: 4px; margin-bottom: 6px; } + .api-item code { background: #f3f4f6; border-radius: 3px; padding: 1px 6px; font-size: 13px; } + .api-item .api-desc { font-size: 13px; color: #374151; margin-bottom: 8px; } + .api-item h4 { font-size: 13px; margin: 10px 0 6px; color: #374151; } + .json-pre { + background: #f9fafb; border: 1px solid #e5e7eb; border-radius: 4px; + padding: 8px; white-space: pre-wrap; word-break: break-word; + font-size: 12px; max-height: 260px; overflow: auto; + } + .curl-box { position: relative; } + .curl-box .curl-copy { position: absolute; top: 6px; right: 6px; } + .curl-pre { + background: #1f2937; color: #e5e7eb; border-radius: 4px; + padding: 8px; white-space: pre-wrap; word-break: break-all; font-size: 12px; + } + .try-panel { margin-top: 10px; border-top: 1px dashed #e5e7eb; padding-top: 10px; } + .try-panel .field { margin-bottom: 10px; } + .try-panel label { display: block; font-size: 13px; margin-bottom: 4px; color: #374151; } + .try-panel textarea { min-height: 100px; font-family: ui-monospace, monospace; } + .try-result pre { + background: #f9fafb; border: 1px solid #e5e7eb; border-radius: 4px; + padding: 8px; white-space: pre-wrap; word-break: break-word; + font-size: 12px; max-height: 320px; overflow: auto; + } @@ -159,11 +204,36 @@ +

知识库管理后台

@@ -268,8 +339,49 @@ + +
+ +