hive-manage-v3/docs/superpowers/plans/phase-1-backend-spec.md

在 /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

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

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

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

# 审计中间件: 拦截所有 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:

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

第九部分: 提交

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 <package>。