""" 用户认证与资料路由。 实现接口（统一响应格式 {success, data, message}）： - POST /api/register 注册：邮箱 + 用户名 + 密码（≥6 位），密码 bcrypt 加密存储 - POST /api/login 登录：校验密码，成功返回 JWT 令牌与用户基础信息（带失败限流） - GET /api/user/level 查询当前登录用户的权限级别（角色） - GET /api/user/me 查询当前登录用户的完整资料（头像、简介） - GET /api/user/blogger 查询博主公开资料（我的简介页面，无需登录） - PUT /api/user/profile 更新当前用户资料（头像 / 简介） """ import os from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Request, status from sqlalchemy.orm import Session from ..auth import create_access_token, get_current_user, hash_password, verify_password from ..database import get_db from ..models import PURPOSE_REGISTER, ROLE_BLOGGER, ROLE_VISITOR, User from ..schemas import ( LoginRequest, ProfileOut, ProfileUpdate, TokenOut, UnifiedResponse, UserCreate, UserLevelOut, VerifyCodeRequest, ) from ..security import ( get_client_ip, is_valid_email, login_failure_limiter, login_ip_limiter, register_ip_limiter, ) from .email import verify_code router = APIRouter(prefix="/api", tags=["auth"]) # 密码长度限制（可通过 .env 的 PASSWORD_MIN_LENGTH 调整，默认 6；前端同步校验） PASSWORD_MIN_LENGTH = int(os.getenv("PASSWORD_MIN_LENGTH") or "6") PASSWORD_MAX_LENGTH = 128 @router.post("/register", response_model=UnifiedResponse) def register(payload: UserCreate, request: Request, db: Session = Depends(get_db)) -> UnifiedResponse: """注册新用户：邮箱、用户名唯一，密码以 bcrypt 哈希存储，角色固定为 visitor。 安全要求：必须携带邮箱验证码（防垃圾注册与邮箱盗用），并按 IP 限流。 """ email = payload.email.strip().lower() username = payload.username.strip() password = payload.password # 注册限流（IP 维度）：同一 IP 每小时最多 3 次，防止批量注册 client_ip = get_client_ip(request) if register_ip_limiter.is_blocked(client_ip): raise HTTPException(status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="注册过于频繁，请稍后再试") # 格式校验（与前端规则保持一致） if not is_valid_email(email): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="邮箱格式不正确") if len(username) < 2: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户名至少 2 个字符") if len(username) > 50: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户名不能超过 50 个字符") if len(password) < PASSWORD_MIN_LENGTH: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="密码至少 6 位") if len(password) > PASSWORD_MAX_LENGTH: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="密码不能超过 128 位") # 邮箱验证码校验：未提供或错误时拒绝注册（校验通过后验证码一次性作废） if not payload.code.strip(): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请先获取并填写邮箱验证码") verify_code(VerifyCodeRequest(email=email, code=payload.code, purpose=PURPOSE_REGISTER), db) # 唯一性校验：邮箱或用户名任一冲突即返回统一提示，不区分具体是哪一项， # 防止攻击者通过注册接口枚举已注册的邮箱 / 用户名 email_taken = db.query(User).filter(User.email == email).first() is not None username_taken = db.query(User).filter(User.username == username).first() is not None if email_taken or username_taken: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="该邮箱或用户名已被使用，请直接登录或更换后重试", ) # 注册用户固定为 visitor 角色，防止通过注册接口越权提升权限 user = User( email=email, username=username, password_hash=hash_password(password), role=ROLE_VISITOR, ) db.add(user) db.commit() db.refresh(user) register_ip_limiter.hit(client_ip) data = { "id": user.id, "email": user.email, "username": user.username, "role": user.role, } return UnifiedResponse(success=True, data=data, message="注册成功") @router.post("/login", response_model=UnifiedResponse) def login(payload: LoginRequest, request: Request, db: Session = Depends(get_db)) -> UnifiedResponse: """登录：校验邮箱与密码，成功后返回 JWT 令牌；失败过多时按邮箱与 IP 双重限流。""" email = payload.email.strip().lower() client_ip = get_client_ip(request) # IP 维度限流：同一 IP 在窗口内尝试过多直接拒绝（防分布式爆破） if login_ip_limiter.is_blocked(client_ip): raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="尝试次数过多，请稍后再试", ) # 邮箱维度限流：同一邮箱 15 分钟内失败 5 次后直接拒绝，防止暴力破解 if login_failure_limiter.is_blocked(email): raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="尝试次数过多，请 15 分钟后再试", ) user = db.query(User).filter(User.email == email).first() # 统一提示，避免泄露用户是否存在；失败同时记录邮箱与 IP 维度计数 if user is None or not verify_password(payload.password, user.password_hash): login_failure_limiter.hit(email) login_ip_limiter.hit(client_ip) raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="邮箱或密码错误") # 登录成功：重置两类失败计数并返回令牌与用户信息 login_failure_limiter.reset(email) login_ip_limiter.reset(client_ip) token = create_access_token(user) data = TokenOut( token=token, user_id=user.id, username=user.username, role=user.role, ).model_dump() return UnifiedResponse(success=True, data=data, message="登录成功") @router.get("/user/level", response_model=UnifiedResponse) def get_user_level(current_user: User = Depends(get_current_user)) -> UnifiedResponse: """查询当前登录用户的权限级别（角色）。""" level = UserLevelOut(user_id=current_user.id, role=current_user.role) return UnifiedResponse(success=True, data=level.model_dump(), message="获取成功") @router.get("/user/me", response_model=UnifiedResponse) def get_my_profile(current_user: User = Depends(get_current_user)) -> UnifiedResponse: """查询当前登录用户的完整资料（含头像、简介），供个人设置使用。""" profile = ProfileOut.model_validate(current_user) return UnifiedResponse(success=True, data=profile.model_dump(), message="获取成功") @router.get("/user/blogger", response_model=UnifiedResponse) def get_blogger_profile(db: Session = Depends(get_db)) -> UnifiedResponse: """查询博主公开资料（我的简介页面，无需登录）。""" blogger = db.query(User).filter(User.role == ROLE_BLOGGER).order_by(User.id.asc()).first() if blogger is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="博主资料不存在") data = { "username": blogger.username, "avatar": blogger.avatar, "bio": blogger.bio, "role": blogger.role, } return UnifiedResponse(success=True, data=data, message="获取成功") @router.put("/user/profile", response_model=UnifiedResponse) def update_profile( payload: ProfileUpdate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ) -> UnifiedResponse: """更新当前用户资料：头像路径与个人简介（博主简介展示在“我的简介”页面）。""" if payload.avatar is not None: avatar = payload.avatar.strip() or None # 头像仅允许站内上传路径（禁 http/https 外链，防止追踪与钓鱼图片） if avatar and not avatar.startswith("/uploads/"): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="头像地址不合法，请使用站内上传的头像") current_user.avatar = avatar if payload.bio is not None: bio = payload.bio.strip() current_user.bio = bio or None db.commit() db.refresh(current_user) profile = ProfileOut.model_validate(current_user) return UnifiedResponse(success=True, data=profile.model_dump(), message="资料已更新")