在 /home/enzo/projects/hive-manage-v3/backend/ 下实现完整的 JWT 鉴权 + 用户管理 + 审计日志系统。全部使用 backend/.venv/bin/python 和 backend/.venv/bin/pip。 已有: User + AuditLog ORM 模型, FastAPI 入口, database.py, config.py ## 第一部分: auth_service.py 创建 backend/app/services/__init__.py (空) 创建 backend/app/services/auth_service.py 核心函数: - hash_password(password: str) -> str — bcrypt (passlib) - verify_password(plain: str, hashed: str) -> bool - create_access_token(user_id: str, role: str) -> str — JWT(HS256, config.jwt_secret, expires=config.jwt_expire_minutes) - decode_access_token(token: str) -> dict — jose.jwt.decode, 捕获 ExpiredSignatureError/JWTError 抛 HTTPException(401) - get_user_by_id(db, user_id) -> User | None - get_user_by_username(db, username) -> User | None - authenticate_user(db, username, password) -> User — 验证密码, 失败检查/递增 login_attempts, >=5 次锁定 15 分钟, 成功后重置 login_attempts - create_user(db, username, email, password) -> User — 首个用户自动 Owner + is_approved=True, 后续用户 Viewer + is_approved=False ## 第二部分: auth middleware 创建 backend/app/middleware/__init__.py (空) 创建 backend/app/middleware/auth.py ```python from fastapi import Depends, HTTPException, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db from app.services.auth_service import decode_access_token, get_user_by_id security = HTTPBearer() async def require_auth( credentials: HTTPAuthorizationCredentials = Depends(security), db: AsyncSession = Depends(get_db), ): payload = decode_access_token(credentials.credentials) user = await get_user_by_id(db, payload["sub"]) if not user or not user.is_active: raise HTTPException(status_code=401, detail="Invalid or inactive user") return user def require_role(*roles: str): async def role_check(user = Depends(require_auth)): if user.role.value not in roles: raise HTTPException(status_code=403, detail="Insufficient permissions") return user return role_check ``` ## 第三部分: Auth Schemas 创建 backend/app/schemas/__init__.py (空) 创建 backend/app/schemas/auth.py ```python from pydantic import BaseModel, EmailStr class LoginRequest(BaseModel): username: str password: str class RegisterRequest(BaseModel): username: str email: str # 先用 str,后续升级 EmailStr password: str class TokenResponse(BaseModel): access_token: str token_type: str = "bearer" expires_in: int class ChangePasswordRequest(BaseModel): old_password: str new_password: str ``` 创建 backend/app/schemas/user.py ```python from pydantic import BaseModel from datetime import datetime class UserResponse(BaseModel): id: str username: str email: str role: str is_active: bool is_approved: bool created_at: datetime class Config: from_attributes = True class UserUpdateRequest(BaseModel): role: str | None = None is_active: bool | None = None ``` ## 第四部分: Auth API 创建 backend/app/api/__init__.py (空) 创建 backend/app/api/auth.py POST /api/v1/auth/register — 注册, 返回 201 POST /api/v1/auth/login — 登录, 返回 TokenResponse POST /api/v1/auth/refresh — 刷新 token, 需 require_auth POST /api/v1/auth/change-password — 改密码, 需 require_auth ## 第五部分: User Management API 创建 backend/app/api/users.py GET /api/v1/users — 用户列表 (需 owner), 支持分页 GET /api/v1/users/{id} — 用户详情 (需 owner) PATCH /api/v1/users/{id} — 改角色/状态 (需 owner) POST /api/v1/users/{id}/approve — 审批新用户 (需 owner) ## 第六部分: Audit Middleware + API 创建 backend/app/middleware/audit.py ```python # 审计中间件: 拦截所有 POST/PUT/PATCH/DELETE 请求 # 从 request.state 获取当前用户 (需 auth middleware 先执行) # 记录: user_id, username, action, target, detail, ip_address # 写入 AuditLog 表 (异步) ``` 创建 backend/app/api/audit.py GET /api/v1/audit — 审计日志列表 (需 owner), 支持分页 + 按 action 筛选 ## 第七部分: 注册路由 修改 backend/app/main.py, 注册所有 router: ```python from app.api import auth, users, audit app.include_router(auth.router) app.include_router(users.router) app.include_router(audit.router) ``` ## 第八部分: 测试 创建 backend/tests/test_auth.py - test_register_success (201) - test_register_duplicate (409) - test_login_success (200 + JWT) - test_login_wrong_password (401) - test_login_locked (423 after 5 failures) - test_require_auth_no_token (401) - test_viewer_cannot_manage_users (403) 创建 backend/tests/conftest.py - async test fixtures: async_client (httpx.AsyncClient + FastAPI app) - test database (SQLite in-memory 或者用现有的 PG) 注意: 测试使用 pytest + pytest-asyncio + httpx.AsyncClient ## 第九部分: 提交 ```bash cd /home/enzo/projects/hive-manage-v3 && \ git add backend/app/services/ backend/app/middleware/ backend/app/schemas/ backend/app/api/ backend/tests/ backend/app/main.py && \ git commit -m "feat: JWT auth, user management, audit logging system" ``` 请按顺序执行以上 9 个部分,每个部分完成后再进入下一部分。pip install 需要时使用 `cd backend && .venv/bin/pip install `。