pj0235-eai_agentplatform/docs/04_Backend/BE01_Auth_Module.md

BE01 — 认证模块设计

版本：V1.1 | 技术：JWT（python-jose）+ bcrypt | 参考：pj006-zhilianyuan2 auth/security.py + dependencies.py

---

1. 模块职责

• 用户注册（仅管理员可创建账号）
• 密码登录 → JWT 签发
• Token 校验 + 用户状态检查
• 角色守卫（普通用户 / 管理员）

2. 核心流程

POST /api/auth/login
  → 校验 username + password
  → bcrypt verify
  → 签发 JWT（含 sub=username, role, exp）
  → 返回 { token, expires_in, user }

GET /api/auth/me
  → Authorization: Bearer <token>
  → 解码 JWT → 校验用户状态
  → 返回用户信息

3. 密码哈希（参考 zhilianyuan2 模式）

# core/security.py
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(password: str, hashed: str) -> bool:
    return pwd_context.verify(password, hashed)

4. JWT 签发与校验

# core/security.py
from jose import jwt, JWTError
from datetime import datetime, timedelta, timezone

def create_access_token(
    *, subject: str, role: str, settings: Settings
) -> str:
    issued_at = datetime.now(timezone.utc)
    payload = {
        "sub": subject,
        "role": role,
        "iat": issued_at,
        "exp": issued_at + timedelta(minutes=settings.jwt_expire_minutes),
    }
    return jwt.encode(payload, settings.jwt_secret, algorithm="HS256")

def decode_access_token(token: str, settings: Settings) -> dict:
    try:
        return jwt.decode(token, settings.jwt_secret, algorithms=["HS256"])
    except JWTError as e:
        raise AuthError(f"令牌无效或已过期：{e}")

5. 依赖注入（参考 zhilianyuan2 dependencies.py）

# core/deps.py
from fastapi import Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

async def get_current_user(
    credentials: HTTPAuthorizationCredentials | None = Depends(
        HTTPBearer(auto_error=False)
    ),
    db: Session = Depends(get_db),
    settings: Settings = Depends(get_settings),
) -> User:
    """解析 JWT + 校验用户状态"""
    if credentials is None:
        raise AuthError("缺少 Authorization Bearer 令牌")
    payload = decode_access_token(credentials.credentials, settings)
    username = payload.get("sub")
    user = db.query(User).filter(User.username == username).first()
    if user is None:
        raise AuthError("用户不存在")
    if user.status != "active":
        raise AuthError("账号已禁用")
    return user

def require_admin(current_user: User = Depends(get_current_user)) -> User:
    """管理员角色守卫"""
    if current_user.role != "admin":
        raise ForbiddenError("需要管理员权限")
    return current_user

6. API 路由

# api/auth.py
from fastapi import APIRouter, Depends
from pydantic import BaseModel

router = APIRouter(prefix="/api/auth", tags=["auth"])

class LoginRequest(BaseModel):
    username: str
    password: str

class TokenResponse(BaseModel):
    token: str
    expires_in: int
    user: UserPublic

@router.post("/login")
def login(request: LoginRequest, settings: SettingsDep):
    user = authenticate(request.username, request.password)
    token = create_access_token(subject=user.username, role=user.role, settings=settings)
    return {"data": {
        "token": token,
        "expires_in": settings.jwt_expire_minutes * 60,
        "user": UserPublic.from_orm(user),
    }}

@router.get("/me")
def me(current_user: CurrentUser):
    return {"data": UserPublic.from_orm(current_user)}

7. 数据表

CREATE TABLE user (
    id              BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
    username        VARCHAR(64) NOT NULL UNIQUE,
    password_hash   VARCHAR(256) NOT NULL,
    full_name       VARCHAR(64) NOT NULL,
    role            ENUM('employee','admin') NOT NULL DEFAULT 'employee',
    status          ENUM('active','disabled') NOT NULL DEFAULT 'active',
    created_at      DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
    updated_at      DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6)
);

8. 安全约束

• 禁用账号即时失效：token 校验时检查 status
• 前端路由守卫仅作 UX 隐藏，不以之为安全边界
• 所有权限以后端鉴权为准