140 lines
4.5 KiB
Python
140 lines
4.5 KiB
Python
"""
|
||
会话验证路由 - 实现类似 osu! 的邮件验证流程 (API v2)
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Annotated
|
||
|
||
from app.database import User
|
||
from app.dependencies import get_current_user
|
||
from app.dependencies.database import Database, get_redis
|
||
from app.service.email_verification_service import (
|
||
EmailVerificationService,
|
||
)
|
||
from app.service.login_log_service import LoginLogService
|
||
|
||
from .router import router
|
||
|
||
from fastapi import Depends, Form, HTTPException, Request, Security, status
|
||
from fastapi.responses import Response
|
||
from pydantic import BaseModel
|
||
from redis.asyncio import Redis
|
||
|
||
|
||
class SessionReissueResponse(BaseModel):
|
||
"""重新发送验证码响应"""
|
||
|
||
success: bool
|
||
message: str
|
||
|
||
|
||
@router.post(
|
||
"/session/verify", name="验证会话", description="验证邮件验证码并完成会话认证", status_code=204, tags=["验证"]
|
||
)
|
||
async def verify_session(
|
||
request: Request,
|
||
db: Database,
|
||
redis: Annotated[Redis, Depends(get_redis)],
|
||
verification_key: str = Form(..., description="8位邮件验证码"),
|
||
current_user: User = Security(get_current_user),
|
||
) -> Response:
|
||
"""
|
||
验证邮件验证码并完成会话认证
|
||
|
||
对应 osu! 的 session/verify 接口
|
||
成功时返回 204 No Content，失败时返回 401 Unauthorized
|
||
"""
|
||
try:
|
||
from app.dependencies.geoip import get_client_ip
|
||
|
||
ip_address = get_client_ip(request) # noqa: F841
|
||
user_agent = request.headers.get("User-Agent", "Unknown") # noqa: F841
|
||
|
||
# 从当前认证用户获取信息
|
||
user_id = current_user.id
|
||
if not user_id:
|
||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="用户未认证")
|
||
|
||
# 验证邮件验证码
|
||
success, message = await EmailVerificationService.verify_code(db, redis, user_id, verification_key)
|
||
|
||
if success:
|
||
# 记录成功的邮件验证
|
||
await LoginLogService.record_login(
|
||
db=db,
|
||
user_id=user_id,
|
||
request=request,
|
||
login_method="email_verification",
|
||
login_success=True,
|
||
notes="邮件验证成功",
|
||
)
|
||
|
||
# 返回 204 No Content 表示验证成功
|
||
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
||
else:
|
||
# 记录失败的邮件验证尝试
|
||
await LoginLogService.record_failed_login(
|
||
db=db,
|
||
request=request,
|
||
attempted_username=current_user.username,
|
||
login_method="email_verification",
|
||
notes=f"邮件验证失败: {message}",
|
||
)
|
||
|
||
# 返回 401 Unauthorized 表示验证失败
|
||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=message)
|
||
|
||
except ValueError:
|
||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="无效的用户会话")
|
||
except Exception:
|
||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="验证过程中发生错误")
|
||
|
||
|
||
@router.post(
|
||
"/session/verify/reissue",
|
||
name="重新发送验证码",
|
||
description="重新发送邮件验证码",
|
||
response_model=SessionReissueResponse,
|
||
tags=["验证"],
|
||
)
|
||
async def reissue_verification_code(
|
||
request: Request,
|
||
db: Database,
|
||
redis: Annotated[Redis, Depends(get_redis)],
|
||
current_user: User = Security(get_current_user),
|
||
) -> SessionReissueResponse:
|
||
"""
|
||
重新发送邮件验证码
|
||
|
||
对应 osu! 的 session/verify/reissue 接口
|
||
"""
|
||
try:
|
||
from app.dependencies.geoip import get_client_ip
|
||
|
||
ip_address = get_client_ip(request)
|
||
user_agent = request.headers.get("User-Agent", "Unknown")
|
||
|
||
# 从当前认证用户获取信息
|
||
user_id = current_user.id
|
||
if not user_id:
|
||
return SessionReissueResponse(success=False, message="用户未认证")
|
||
|
||
# 重新发送验证码
|
||
success, message = await EmailVerificationService.resend_verification_code(
|
||
db,
|
||
redis,
|
||
user_id,
|
||
current_user.username,
|
||
current_user.email,
|
||
ip_address,
|
||
user_agent,
|
||
)
|
||
|
||
return SessionReissueResponse(success=success, message=message)
|
||
|
||
except ValueError:
|
||
return SessionReissueResponse(success=False, message="无效的用户会话")
|
||
except Exception:
|
||
return SessionReissueResponse(success=False, message="重新发送过程中发生错误")
|