"""认证路由""" import os from datetime import datetime from fastapi import APIRouter, Depends, HTTPException, Response from sqlalchemy.orm import Session from ..database import get_db from ..models import User, AuditLog from ..security import verify_password, create_access_token, ACCESS_TOKEN_EXPIRE_HOURS from ..permissions import get_user_permissions, ROLE_LABELS from ..schemas import LoginRequest, TokenResponse from ..deps import get_current_user router = APIRouter(prefix="/api/auth", tags=["认证"]) # Cookie 安全配置(生产环境 HTTPS 部署时设置 COOKIE_SECURE=true) _COOKIE_SECURE = os.getenv("COOKIE_SECURE", "false").lower() == "true" @router.post("/login", response_model=TokenResponse) def login(req: LoginRequest, response: Response, db: Session = Depends(get_db)): user = db.query(User).filter(User.username == req.username).first() if not user or not verify_password(req.password, user.password_hash): raise HTTPException(status_code=401, detail="用户名或密码错误") if not user.is_active: raise HTTPException(status_code=403, detail="账号已禁用,请联系管理员") token = create_access_token({"sub": str(user.id), "role": user.role}) perms = get_user_permissions(user) # 设置 httpOnly cookie(前端无需 JS 读取 token,防 XSS 窃取) response.set_cookie( key="access_token", value=token, httponly=True, secure=_COOKIE_SECURE, samesite="lax", max_age=ACCESS_TOKEN_EXPIRE_HOURS * 3600, path="/", ) # 审计 db.add(AuditLog(user_id=user.id, username=user.username, action="login", target="auth")) db.commit() return TokenResponse( access_token=token, role=user.role, username=user.username, permissions=perms, ) @router.get("/me") def me(current_user: User = Depends(get_current_user)): return { "id": current_user.id, "username": current_user.username, "role": current_user.role, "role_label": ROLE_LABELS.get(current_user.role, current_user.role), "is_active": current_user.is_active, "remark": current_user.remark, "permissions": get_user_permissions(current_user), } @router.post("/logout") def logout(response: Response, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)): response.delete_cookie(key="access_token", path="/") db.add(AuditLog(user_id=current_user.id, username=current_user.username, action="logout", target="auth")) db.commit() return {"message": "已登出"}