mirror of
https://github.com/fastapiadmin/FastapiAdmin.git
synced 2026-09-22 13:05:18 +00:00
zhangtao
ccaa94e1f2
- 演示模式中增加演示白名单IP "117.10.167.220" - 中间件中优先从X-Forwarded-For请求头获取客户端真实IP，优化IP判断逻辑 - 从日志中打印用户名称，方便追踪演示环境操作用户 - 补充IP白名单和路径白名单的判断条件，非白名单用户禁止操作 - 初始化插件中的数据库连接和初始化逻辑调整，改为单独会话完成，提升代码清晰度 - 初始化脚本中更新PostgreSQL序列时仅处理含id字段的模型，避免关联表模型错误 - 修改开发环境配置，切换数据库类型为MySQL，更新相关连接配置，匹配MySQL默认端口及用户信息 - 删除MySQL数据库的完整导出脚本，确保代码库无冗余数据库备份文件
111 lines
3.9 KiB
Python
111 lines
3.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import time
|
|
from typing import Dict, List, Union
|
|
from starlette.middleware.cors import CORSMiddleware
|
|
from starlette.types import ASGIApp
|
|
from starlette.requests import Request
|
|
from starlette.middleware.gzip import GZipMiddleware
|
|
from starlette.middleware.base import Response, BaseHTTPMiddleware, RequestResponseEndpoint
|
|
|
|
from app.common.response import ErrorResponse
|
|
from app.config.setting import settings
|
|
from app.core.logger import logger
|
|
from app.core.exceptions import CustomException
|
|
|
|
|
|
class CustomCORSMiddleware(CORSMiddleware):
|
|
"""CORS跨域中间件"""
|
|
def __init__(self, app: ASGIApp) -> None:
|
|
CORSMiddlewareConfig: Dict[str, Union[List[str], bool]] = {
|
|
"allow_origins": settings.ALLOW_ORIGINS,
|
|
"allow_methods": settings.ALLOW_METHODS,
|
|
"allow_headers": settings.ALLOW_HEADERS,
|
|
"allow_credentials": settings.ALLOW_CREDENTIALS
|
|
}
|
|
super().__init__(app, **CORSMiddlewareConfig)
|
|
|
|
|
|
class RequestLogMiddleware(BaseHTTPMiddleware):
|
|
"""
|
|
记录请求日志中间件: 提供一个基础的中间件类，允许你自定义请求和响应处理逻辑。
|
|
"""
|
|
def __init__(self, app: ASGIApp) -> None:
|
|
super().__init__(app)
|
|
|
|
async def dispatch(
|
|
self, request: Request, call_next: RequestResponseEndpoint
|
|
) -> Response:
|
|
start_time = time.time()
|
|
|
|
logger.info(
|
|
f"请求来源: {request.client.host}, "
|
|
f"请求方法: {request.method}, "
|
|
f"请求路径: {request.url.path}, "
|
|
f"客户端IP: {request.client.host}"
|
|
)
|
|
|
|
try:
|
|
response = await call_next(request)
|
|
process_time = round(time.time() - start_time, 5)
|
|
response.headers["X-Process-Time"] = str(process_time)
|
|
|
|
logger.info(
|
|
f"会话ID: {request.scope.get('session_id')}, "
|
|
f"响应状态: {response.status_code}, "
|
|
f"响应内容长度: {response.headers.get('content-length', '0')}, "
|
|
f"处理时间: {process_time}s"
|
|
)
|
|
|
|
return response
|
|
|
|
except CustomException as e:
|
|
logger.error(f"系统异常: {str(e)}")
|
|
return ErrorResponse(msg=f"系统异常，请联系管理员: {str(e)}")
|
|
|
|
|
|
class DemoEnvMiddleware(BaseHTTPMiddleware):
|
|
"""演示环境中间件"""
|
|
def __init__(self, app: ASGIApp) -> None:
|
|
super().__init__(app)
|
|
|
|
async def dispatch(
|
|
self, request: Request, call_next: RequestResponseEndpoint
|
|
) -> Response:
|
|
|
|
if settings.DEMO_ENABLE and request.method != "GET":
|
|
path = request.scope.get("path")
|
|
|
|
request_ip = None
|
|
x_forwarded_for = request.headers.get('X-Forwarded-For')
|
|
if x_forwarded_for:
|
|
# 取第一个 IP 地址，通常为客户端真实 IP
|
|
request_ip = x_forwarded_for.split(',')[0].strip()
|
|
else:
|
|
# 若没有 X-Forwarded-For 头，则使用 request.client.host
|
|
request_ip = request.client.host
|
|
|
|
user_username = request.scope.get("user_username")
|
|
logger.error(f"用户名称: {user_username}")
|
|
|
|
# 检查IP是否在白名单，或路径是否在白名单，或用户是否在白名单
|
|
if (request_ip in settings.DEMO_IP_WHITE_LIST) or (path in settings.DEMO_WHITE_LIST_PATH):
|
|
return await call_next(request)
|
|
|
|
else:
|
|
# 非白名单用户，禁止操作
|
|
return ErrorResponse(msg="演示环境，禁止操作")
|
|
|
|
return await call_next(request)
|
|
|
|
|
|
class CustomGZipMiddleware(GZipMiddleware):
|
|
"""GZip压缩中间件"""
|
|
def __init__(self, app: ASGIApp) -> None:
|
|
super().__init__(
|
|
app,
|
|
minimum_size=settings.GZIP_MIN_SIZE,
|
|
compresslevel=settings.GZIP_COMPRESS_LEVEL
|
|
)
|
|
|