# router_users_auth.py # ========================================== # 🔐 用户认证路由模块 # ========================================== # 作用：处理用户登录、注册、密码重置、验证码发送等认证相关接口 # 关联文件： # - 安全认证.py (密码哈希 + JWT 安全模块) 🔒 P0安全增强 # - verify_code_engine.py (验证码缓存与发送) # - 数据库连接.py (JSON数据库读写) # - models.py (Pydantic数据模型定义) # ========================================== from fastapi import APIRouter, HTTPException, BackgroundTasks, Request import time import re import random import json import 数据库连接 as db from models import UserRegister, UserLogin, SendCodeRequest from verify_code_engine import VERIFY_CODES, send_email_code, send_sms_code, cleanup_expired_codes, check_send_cooldown # 🔒 P0安全增强：导入密码哈希和 JWT 工具 from 安全认证 import hash_password, verify_password, create_token, require_password_match # 🚀 P2优化：速率限制 from slowapi import Limiter from slowapi.util import get_remote_address limiter = Limiter(key_func=get_remote_address) # 创建子路由实例 router = APIRouter() # ========================================== # 📤 发送验证码接口（异步后台任务版） # ========================================== # 作用：接收验证码发送请求，后台异步发送邮件/短信 # 关联：verify_code_engine.py 的 send_email_code / send_sms_code # 前端调用：注册表单组件.js、重置密码表单组件.js @router.post("/api/users/send-code") @limiter.limit("5/minute") # 🚀 P2优化：每分钟最多5次 async def send_verify_code(request: Request, req: SendCodeRequest, bg_tasks: BackgroundTasks): """ 发送验证码接口（异步版本） 请求参数： - contact: 邮箱或手机号 - contact_type: "email" 或 "phone" - action_type: "register" 注册 / "reset" 重置密码 - account: (仅重置密码时必填) 用户账号 """ # 如果是重置密码，需要先验证账号与联系方式匹配 if req.action_type == "reset": if not req.account: raise HTTPException(status_code=400, detail="找回密码需先填写当前账号") users_db = db.load_data("users.json", default_data={}) user = users_db.get(req.account) if not user: raise HTTPException(status_code=404, detail="该账号不存在") # 校验联系方式是否与账号绑定的一致 if req.contact_type == "email" and user.get("email") != req.contact: raise HTTPException(status_code=400, detail="填写的邮箱与该账号绑定的邮箱不一致") if req.contact_type == "phone" and user.get("phone") != req.contact: raise HTTPException(status_code=400, detail="填写的手机号与该账号绑定的手机号不一致") # 🔒 检查发送频率限制（同一联系方式60秒内只能发送1条） check_send_cooldown(req.contact) # 生成6位随机验证码 code = str(random.randint(100000, 999999)) # 🚀 P1性能优化：每次写入前触发清理过期验证码 cleanup_expired_codes() # 构建缓存键（联系方式_动作类型） cache_key = f"{req.contact}_{req.action_type}" # 将验证码存入内存缓存，有效期5分钟 VERIFY_CODES[cache_key] = { "code": code, "expires_at": int(time.time()) + 300 # 当前时间 + 300秒(5分钟) } # 根据联系方式类型，添加后台发送任务 if req.contact_type == "email": bg_tasks.add_task(send_email_code, req.contact, code, req.action_type) elif req.contact_type == "phone": bg_tasks.add_task(send_sms_code, req.contact, code, req.action_type) else: raise HTTPException(status_code=400, detail="不支持的验证方式") return {"status": "success", "message": "验证码发送请求已提交"} # ========================================== # 📤 发送验证码接口（同步版本，备用） # ========================================== # 作用：同步方式发送验证码，部分场景需要立即知道发送结果 # 关联：verify_code_engine.py 的 send_email_code @router.post("/api/users/send_code") async def send_code_api(req: SendCodeRequest): """发送验证码接口（同步版本，直接等待发送结果）""" # 🔒 检查发送频率限制（同一联系方式60秒内只能发送1条） check_send_cooldown(req.contact) # 生成6位随机验证码 code = str(random.randint(100000, 999999)) key = f"{req.contact}_{req.action_type}" # 存入缓存，有效期5分钟 VERIFY_CODES[key] = { "code": code, "expires_at": time.time() + 300 # 300秒(5分钟) } # 同步发送（会阻塞等待结果） if req.contact_type == "email": try: send_email_code(req.contact, code, req.action_type) return {"status": "success", "message": "验证码已成功发送至邮箱"} except Exception as e: raise HTTPException(status_code=500, detail=f"邮件发送失败: {str(e)}") elif req.contact_type == "phone": return {"status": "success", "message": "验证码已成功发送至手机"} else: raise HTTPException(status_code=400, detail="不支持的验证方式") # ========================================== # 📝 用户注册接口 # ========================================== # 作用：新用户注册，需要验证邮箱/手机验证码 # 关联： # - verify_code_engine.py 的 VERIFY_CODES (校验验证码) # - 数据库连接.py (保存新用户到 users.json) # - 前端 注册表单组件.js @router.post("/api/users/register") @limiter.limit("3/minute") # 🔒 P0安全优化：注册每分钟最多3次 async def register_user(request: Request, user: UserRegister): """ 用户注册接口 请求参数：(UserRegister 模型) - account: 账号（6-20位字母数字下划线） - password: 密码（至少6位） - name: 昵称 - email: 邮箱（与 phone 二选一） - phone: 手机号（与 email 二选一） - code: 验证码 - intro: 个人介绍（可选，最多100字） """ users_db = db.load_data("users.json", default_data={}) # ========== 第一步：查重检查 ========== # 检查账号是否已存在 if user.account in users_db: raise HTTPException(status_code=400, detail="该账号已被注册，请更换一个") # 检查邮箱和手机号是否已被其他用户绑定 for existing_user in users_db.values(): if user.email and existing_user.get("email") == user.email: raise HTTPException(status_code=400, detail="此邮箱已注册，请直接登录或找回密码") if user.phone and existing_user.get("phone") == user.phone: raise HTTPException(status_code=400, detail="该手机号已被绑定") # ========== 第二步：验证码校验（原子性获取+删除） ========== # 根据注册方式构建缓存键 cache_key = f"{user.email}_register" if user.email else f"{user.phone}_register" # 🔒 P0安全修复：验证码一次性使用，原子性pop防止并发重用 cached = VERIFY_CODES.pop(cache_key, None) # 兼容新老缓存格式（expires_at 或 expires） expire_time = cached.get("expires_at", cached.get("expires", 0)) if cached else 0 # 校验验证码是否正确且未过期 if not cached or cached["code"] != user.code or time.time() > expire_time: raise HTTPException(status_code=400, detail="验证码不正确或已过期") # ========== 第三步：格式校验 ========== if len(user.account) <= 5: raise HTTPException(status_code=400, detail="账号必须大于5个字符") if not re.match(r'^[a-zA-Z0-9_]{6,20}$', user.account): raise HTTPException(status_code=400, detail="账号仅支持大小写英文字母、数字及下划线") if len(user.password) < 6: raise HTTPException(status_code=400, detail="密码必须大于等于6个字符") if not re.match(r'^[a-zA-Z0-9!@#$%^&*()_+\-=\[\]{};\':"\\|,.<>\/?]{6,}$', user.password): raise HTTPException(status_code=400, detail="密码包含不支持的特殊字符") if user.intro and len(user.intro) > 100: raise HTTPException(status_code=400, detail="个人介绍不能超过100个字符") # ========== 第四步：保存新用户 ========== # 构建用户数据对象 new_user = user.dict() new_user.pop("code", None) # 移除验证码字段，不存入数据库 # 🔒 安全防护：拒绝 base64 头像数据 if (new_user.get("avatarDataUrl") or "").startswith("data:"): new_user["avatarDataUrl"] = "" # 🔒 P0安全增强：密码哈希化存储（不再存储明文密码） new_user["password"] = hash_password(new_user["password"]) new_user.update({ "created_at": int(time.time()), # 注册时间戳 "followers": [], # 粉丝列表 "following": [], # 关注列表 "privacy": {"follows": False, "likes": False, "favorites": False, "downloads": False} # 隐私设置 }) # 保存到数据库 users_db[user.account] = new_user db.save_data("users.json", users_db) # 返回用户信息（排除密码） return {"status": "success", "message": "注册成功", "data": {k: v for k, v in new_user.items() if k != "password"}} # ========================================== # 🔑 用户登录接口 # ========================================== # 作用：验证账号密码，返回登录凭证 # 关联： # - 数据库连接.py (读取 users.json 校验密码) # - 前端 登录表单组件.js @router.post("/api/users/login") @limiter.limit("10/minute") # 🚀 P2优化：每分钟最多10次登录尝试 async def login_user(request: Request, user: UserLogin): """ 用户登录接口 请求参数：(UserLogin 模型) - account: 账号 - password: 密码 """ users_db = db.load_data("users.json", default_data={}) # 检查账号是否存在 if user.account not in users_db: raise HTTPException(status_code=404, detail="账号不存在") user_data = users_db[user.account] stored_password = user_data.get("password", "") # 🔒 P0安全增强：密码哈希验证（使用统一验证函数） if not require_password_match(stored_password, user.password): raise HTTPException(status_code=401, detail="密码错误") # 🔒 P0安全增强：登录成功后，检查是否需要迁移旧密码为bcrypt if not user_data["password"].startswith('$2b$') and not user_data["password"].startswith('$2a$'): # 旧版SHA256密码，自动迁移为bcrypt user_data["password"] = hash_password(user.password) db.save_data("users.json", users_db) # 🔒 P0安全增强：生成 JWT Token（替代 mock_token） # 获取password_version用于Token生成（如不存在则默认为0） password_version = user_data.get("password_version", 0) # 根据用户选择的"保持登录"选项设置Token有效期 token = create_token(user.account, extra_data={"pwd_ver": password_version}, remember=user.remember) return { "status": "success", "token": token, # 🔒 JWT Token "account": user.account, "name": user_data["name"], "avatar": user_data.get("avatarDataUrl", "") } # ========================================== # 🔄 重置密码接口（万能兼容版） # ========================================== # 作用：通过邮箱/手机验证码重置用户密码 # 关联： # - verify_code_engine.py 的 VERIFY_CODES (校验验证码) # - 数据库连接.py (更新 users.json 中的密码) # - 前端 重置密码表单组件.js # 特点：万能解析器，兼容各种前端数据格式 @router.post("/api/users/reset_password") @limiter.limit("3/minute") # 🔒 P0安全优化：重置密码每分钟最多3次 async def reset_password(request: Request): """ 重置密码接口（万能兼容版） 支持的请求格式： - 标准 JSON - 双重字符串化 JSON - FormData """ # ========== 第一步：万能数据解析器 ========== # 作用：兼容各种前端可能发送的数据格式 try: data = await request.json() # 处理前端可能造成的"双重字符串化"问题 if isinstance(data, str): data = json.loads(data) except Exception: # 降级尝试 FormData 格式 try: form = await request.form() data = dict(form) except Exception: raise HTTPException(status_code=400, detail="请求数据解析失败，请检查网络") if not isinstance(data, dict): raise HTTPException(status_code=400, detail=f"前端数据格式异常，收到的是: {type(data).__name__}") # ========== 第二步：万能字段提取器 ========== # 作用：兼容前端可能使用的各种字段命名 account = data.get("account") new_password = data.get("new_password") or data.get("password") verify_contact = data.get("verifyContact") or data.get("verify_contact") or data.get("email") or data.get("phone") verify_type = data.get("verifyType") or data.get("verify_type") or data.get("contact_type") code = data.get("code") # 参数完整性校验 if not all([account, new_password, verify_contact, verify_type, code]): raise HTTPException(status_code=400, detail="缺失必要参数 (账号/密码/验证码/联系方式)，请检查表单") # ========== 第三步：核心业务逻辑 ========== users_db = db.load_data("users.json", default_data={}) # 检查用户是否存在 if account not in users_db: raise HTTPException(status_code=404, detail="该用户不存在") user = users_db[account] # 校验联系方式是否与账号绑定的一致 if verify_type == "email" and user.get("email") != verify_contact: raise HTTPException(status_code=400, detail="填写的邮箱与该账号绑定的邮箱不匹配") if verify_type == "phone" and user.get("phone") != verify_contact: raise HTTPException(status_code=400, detail="填写的手机号与该账号绑定的手机号不匹配") # 校验验证码（原子性获取+删除） cache_key = f"{verify_contact}_reset" # 🔒 P0安全修复：验证码一次性使用，原子性pop防止并发重用 cached = VERIFY_CODES.pop(cache_key, None) expire_time = cached.get("expires_at", cached.get("expires", 0)) if cached else 0 if not cached or cached["code"] != code or time.time() > expire_time: raise HTTPException(status_code=400, detail="验证码不正确或已过期") # 校验新密码格式 if len(new_password) < 6: raise HTTPException(status_code=400, detail="新密码必须大于等于6个字符") if not re.match(r'^[a-zA-Z0-9!@#$%^&*()_+\-=\[\]{};\':"\\|,.<>\/?]{6,}$', new_password): raise HTTPException(status_code=400, detail="新密码包含不支持的特殊字符") # ========== 第四步：更新密码并保存 ========== # 🔒 P0安全增强：新密码哈希化存储 user["password"] = hash_password(new_password) # 🔒 P0安全增强：更新password_version使旧Token失效 import time as time_module user["password_version"] = int(time_module.time()) db.save_data("users.json", users_db) # 🔒 P0安全增强：生成新Token返回给前端替换旧Token new_token = create_token(account, extra_data={"pwd_ver": user["password_version"]}) return {"status": "success", "message": "密码修改成功", "token": new_token}