Bennett/g0v0-server
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore(linter): make linter happy

Browse Source
This commit is contained in:
MingxuanGame
2025-09-30 07:57:08 +00:00
parent 0f637446df
commit 017b058e63
15 changed files with 99 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
app/middleware/__init__.py
View File

@@ -1,9 +1,5 @@
"""
中间件模块
from __future__ import annotations
提供会话验证和其他中间件功能
"""
from .verify_session import SessionState, VerifySessionMiddleware
from .verify_session import VerifySessionMiddleware, SessionState
__all__ = ["VerifySessionMiddleware", "SessionState"]
__all__ = ["SessionState", "VerifySessionMiddleware"]

61
app/middleware/session_verification.py
View File

@@ -6,20 +6,21 @@
from __future__ import annotations
import asyncio
from typing import Awaitable, Callable, Optional
from collections.abc import Awaitable, Callable
from typing import ClassVar, Literal, cast
from app.database.lazer_user import User
from app.database.verification import LoginSession
from app.dependencies.database import get_redis, with_db
from app.log import logger
from app.service.verification_service import LoginSessionService
from app.utils import bg_tasks
from fastapi import HTTPException, Request, Response, status
from fastapi.responses import JSONResponse
from redis.asyncio import Redis
from sqlmodel.ext.asyncio.session import AsyncSession
from app.database.lazer_user import User
from app.database.verification import LoginSession
from app.dependencies.database import with_db, get_redis
from app.log import logger
from app.service.verification_service import LoginSessionService
class SessionVerificationState:
"""会话验证状态管理类
@@ -39,7 +40,7 @@ class SessionVerificationState:
db: AsyncSession,
redis: Redis,
user: User,
) -> Optional[SessionVerificationState]:
) -> SessionVerificationState | None:
"""获取当前会话验证状态"""
try:
# 从请求头或token中获取会话信息
@@ -58,7 +59,7 @@ class SessionVerificationState:
return None
@staticmethod
def _extract_session_token(request: Request) -> Optional[str]:
def _extract_session_token(request: Request) -> str | None:
"""从请求中提取会话token"""
# 尝试从Authorization header提取
auth_header = request.headers.get("Authorization", "")
@@ -79,11 +80,11 @@ class SessionVerificationState:
# 智能选择验证方法
# 参考osu-web: API版本 < 20250913 或用户没有TOTP时使用邮件验证
# 这里简化为检查用户是否有TOTP
totp_key = getattr(self.user, 'totp_key', None)
current_method = 'totp' if totp_key else 'mail'
totp_key = getattr(self.user, "totp_key", None)
current_method = "totp" if totp_key else "mail"
# 设置验证方法
asyncio.create_task(self._set_verification_method(current_method))
bg_tasks.add_task(self._set_verification_method, current_method)
return current_method
@@ -91,11 +92,14 @@ class SessionVerificationState:
"""内部方法：设置验证方法"""
try:
token_id = self.session.token_id
if token_id is not None and method in ['totp', 'mail']:
if token_id is not None and method in ["totp", "mail"]:
# 类型检查确保method是正确的字面量类型
verification_method = method if method in ['totp', 'mail'] else 'totp'
verification_method = method if method in ["totp", "mail"] else "totp"
await LoginSessionService.set_login_method(
self.user.id, token_id, verification_method, self.redis # type: ignore
self.user.id,
token_id,
cast(Literal["totp", "mail"], verification_method),
self.redis,
)
except Exception as e:
logger.error(f"[Session Verification] Error setting verification method: {e}")
@@ -112,9 +116,7 @@ class SessionVerificationState:
try:
token_id = self.session.token_id
if token_id is not None:
await LoginSessionService.mark_session_verified(
db, self.redis, self.user.id, token_id
)
await LoginSessionService.mark_session_verified(db, self.redis, self.user.id, token_id)
finally:
await db.close()
except Exception as e:
@@ -142,8 +144,7 @@ class SessionVerificationState:
db = with_db()
try:
await EmailVerificationService.send_verification_email(
db, self.redis, self.user.id, self.user.username,
self.user.email, None, None
db, self.redis, self.user.id, self.user.username, self.user.email, None, None
)
finally:
await db.close()
@@ -158,7 +159,7 @@ class SessionVerificationController:
"""
# 需要跳过验证的路由（参考osu-web的SKIP_VERIFICATION_ROUTES）
SKIP_VERIFICATION_ROUTES = {
SKIP_VERIFICATION_ROUTES: ClassVar[set[str]] = {
"/api/v2/session/verify",
"/api/v2/session/verify/reissue",
"/api/v2/me",
@@ -190,26 +191,18 @@ class SessionVerificationController:
# API请求返回JSON响应
if request.url.path.startswith("/api/"):
return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED,
content={"method": method}
)
return JSONResponse(status_code=status.HTTP_401_UNAUTHORIZED, content={"method": method})
# 其他情况可以扩展支持HTML响应
return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED,
content={
"authentication": "verify",
"method": method,
"message": "Session verification required"
}
content={"authentication": "verify", "method": method, "message": "Session verification required"},
)
except Exception as e:
logger.error(f"[Session Verification] Error initiating verification: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Verification initiation failed"
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Verification initiation failed"
)
@@ -262,7 +255,7 @@ class SessionVerificationMiddleware:
# 出错时允许请求继续，避免阻塞正常流程
return await call_next(request)
async def _get_user(self, request: Request) -> Optional[User]:
async def _get_user(self, request: Request) -> User | None:
"""获取当前用户"""
try:
# 这里需要手动获取用户，因为在中间件中无法直接使用依赖注入

13
app/middleware/setup.py
View File

@@ -1,14 +1,10 @@
"""
中间件设置和配置
展示如何将会话验证中间件集成到FastAPI应用中
"""
from fastapi import FastAPI
from __future__ import annotations
from app.config import settings
from app.middleware.verify_session import VerifySessionMiddleware
from fastapi import FastAPI
def setup_session_verification_middleware(app: FastAPI) -> None:
"""设置会话验证中间件
@@ -22,9 +18,11 @@ def setup_session_verification_middleware(app: FastAPI) -> None:
# 可以在这里添加中间件配置日志
from app.log import logger
logger.info("[Middleware] Session verification middleware enabled")
else:
from app.log import logger
logger.info("[Middleware] Session verification middleware disabled")
@@ -41,4 +39,5 @@ def setup_all_middlewares(app: FastAPI) -> None:
# app.add_middleware(OtherMiddleware)
from app.log import logger
logger.info("[Middleware] All middlewares configured")

54
app/middleware/verify_session.py
View File

@@ -6,22 +6,23 @@ FastAPI会话验证中间件
from __future__ import annotations
from typing import Callable, Optional
from collections.abc import Callable
from typing import ClassVar
from app.auth import get_token_by_access_token
from app.database.lazer_user import User
from app.database.verification import LoginSession
from app.dependencies.database import get_redis, with_db
from app.log import logger
from app.service.verification_service import LoginSessionService
from fastapi import Request, Response, status
from fastapi.responses import JSONResponse
from redis.asyncio import Redis
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from starlette.middleware.base import BaseHTTPMiddleware
from app.database.lazer_user import User
from app.database.verification import LoginSession
from app.dependencies.database import with_db, get_redis
from app.auth import get_token_by_access_token
from app.log import logger
from app.service.verification_service import LoginSessionService
from sqlmodel import select
class VerifySessionMiddleware(BaseHTTPMiddleware):
"""会话验证中间件
@@ -30,7 +31,7 @@ class VerifySessionMiddleware(BaseHTTPMiddleware):
"""
# 需要跳过验证的路由
SKIP_VERIFICATION_ROUTES = {
SKIP_VERIFICATION_ROUTES: ClassVar[set[str]] = {
"/api/v2/session/verify",
"/api/v2/session/verify/reissue",
"/api/v2/me",
@@ -44,7 +45,7 @@ class VerifySessionMiddleware(BaseHTTPMiddleware):
}
# 需要强制验证的路由模式（敏感操作）
ALWAYS_VERIFY_PATTERNS = {
ALWAYS_VERIFY_PATTERNS: ClassVar[set[str]] = {
"/api/v2/account/",
"/api/v2/settings/",
"/api/private/admin/",
@@ -110,9 +111,9 @@ class VerifySessionMiddleware(BaseHTTPMiddleware):
return True
# 特权用户或非活跃用户需要验证
if hasattr(user, 'is_privileged') and user.is_privileged():
return True
if hasattr(user, 'is_inactive') and user.is_inactive():
# if hasattr(user, 'is_privileged') and user.is_privileged():
# return True
if not user.is_active:
return True
# 安全方法（GET/HEAD/OPTIONS）一般不需要验证
@@ -123,7 +124,7 @@ class VerifySessionMiddleware(BaseHTTPMiddleware):
# 修改操作（POST/PUT/DELETE/PATCH）需要验证
return method in {"POST", "PUT", "DELETE", "PATCH"}
async def _get_current_user(self, request: Request) -> Optional[User]:
async def _get_current_user(self, request: Request) -> User | None:
"""获取当前用户"""
try:
# 从Authorization header提取token
@@ -151,7 +152,7 @@ class VerifySessionMiddleware(BaseHTTPMiddleware):
logger.debug(f"[Verify Session Middleware] Error getting user: {e}")
return None
async def _get_session_state(self, request: Request, user: User) -> Optional[SessionState]:
async def _get_session_state(self, request: Request, user: User) -> SessionState | None:
"""获取会话状态"""
try:
# 提取会话token（这里简化为使用相同的auth token）
@@ -191,17 +192,13 @@ class VerifySessionMiddleware(BaseHTTPMiddleware):
# 返回验证要求响应
return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED,
content={
"method": method,
"message": "Session verification required"
}
content={"method": method, "message": "Session verification required"},
)
except Exception as e:
logger.error(f"[Verify Session Middleware] Error initiating verification: {e}")
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content={"error": "Verification initiation failed"}
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"error": "Verification initiation failed"}
)
@@ -216,7 +213,7 @@ class SessionState:
self.user = user
self.redis = redis
self.db = db
self._verification_method: Optional[str] = None
self._verification_method: str | None = None
def is_verified(self) -> bool:
"""检查会话是否已验证"""
@@ -236,8 +233,8 @@ class SessionState:
if self._verification_method is None:
# 检查用户是否有TOTP密钥
await self.user.awaitable_attrs.totp_key # 预加载
totp_key = getattr(self.user, 'totp_key', None)
self._verification_method = 'totp' if totp_key else 'mail'
totp_key = getattr(self.user, "totp_key", None)
self._verification_method = "totp" if totp_key else "mail"
# 保存选择的方法
token_id = self.session.token_id
@@ -253,9 +250,7 @@ class SessionState:
try:
token_id = self.session.token_id
if token_id is not None:
await LoginSessionService.mark_session_verified(
self.db, self.redis, self.user.id, token_id
)
await LoginSessionService.mark_session_verified(self.db, self.redis, self.user.id, token_id)
self.session.is_verified = True # 更新本地状态
except Exception as e:
logger.error(f"[Session State] Error marking verified: {e}")
@@ -268,8 +263,7 @@ class SessionState:
# 这里可以触发邮件发送
await EmailVerificationService.send_verification_email(
self.db, self.redis, self.user.id, self.user.username,
self.user.email, None, None
self.db, self.redis, self.user.id, self.user.username, self.user.email, None, None
)
except Exception as e:
logger.error(f"[Session State] Error issuing mail: {e}")