121 lines
3.3 KiB
Python
121 lines
3.3 KiB
Python
# ===========================================
|
|
# 班级操行分管理系统 - 后端服务
|
|
#
|
|
# 开发者: Canglan
|
|
# 联系方式: admin@sea-studio.top
|
|
# 版权归属: Sea Network Technology Studio
|
|
# 许可证: MIT License
|
|
#
|
|
# 版权所有 © Sea Network Technology Studio
|
|
# ===========================================
|
|
|
|
from fastapi import Request
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
from typing import Dict, Any
|
|
import re
|
|
|
|
|
|
class SanitizeMiddleware(BaseHTTPMiddleware):
|
|
"""输入过滤中间件"""
|
|
|
|
async def dispatch(self, request: Request, call_next):
|
|
# 只处理POST、PUT、PATCH请求
|
|
if request.method in ["POST", "PUT", "PATCH"]:
|
|
# 获取请求体
|
|
body = await request.body()
|
|
if body:
|
|
import json
|
|
try:
|
|
data = json.loads(body)
|
|
# 清理数据
|
|
cleaned_data = self._sanitize_data(data)
|
|
# 替换请求体
|
|
request._body = json.dumps(cleaned_data).encode()
|
|
except:
|
|
pass
|
|
|
|
response = await call_next(request)
|
|
return response
|
|
|
|
def _sanitize_data(self, data: Any) -> Any:
|
|
"""递归清理数据"""
|
|
if isinstance(data, dict):
|
|
return {k: self._sanitize_data(v) for k, v in data.items()}
|
|
elif isinstance(data, list):
|
|
return [self._sanitize_data(item) for item in data]
|
|
elif isinstance(data, str):
|
|
return self._sanitize_string(data)
|
|
else:
|
|
return data
|
|
|
|
def _sanitize_string(self, value: str) -> str:
|
|
"""清理字符串"""
|
|
if not value:
|
|
return ""
|
|
|
|
# 去除首尾空格
|
|
value = value.strip()
|
|
|
|
# 限制长度
|
|
if len(value) > 1000:
|
|
value = value[:1000]
|
|
|
|
# 转义HTML特殊字符
|
|
html_chars = {
|
|
'&': '&',
|
|
'<': '<',
|
|
'>': '>',
|
|
'"': '"',
|
|
"'": ''',
|
|
'/': '/'
|
|
}
|
|
for char, escape in html_chars.items():
|
|
value = value.replace(char, escape)
|
|
|
|
return value
|
|
|
|
|
|
def sanitize_input(value: str, max_length: int = 255) -> str:
|
|
"""清理单个输入值"""
|
|
if not value:
|
|
return ""
|
|
|
|
value = value.strip()
|
|
if len(value) > max_length:
|
|
value = value[:max_length]
|
|
|
|
return value
|
|
|
|
|
|
def validate_points(points: int, min_val: int = -100, max_val: int = 100) -> tuple:
|
|
"""
|
|
验证分值
|
|
返回: (是否有效, 错误信息)
|
|
"""
|
|
if points == 0:
|
|
return False, "分值不能为0"
|
|
if points < min_val or points > max_val:
|
|
return False, f"分值必须在{min_val}到{max_val}之间"
|
|
return True, ""
|
|
|
|
|
|
def validate_reason(reason: str) -> tuple:
|
|
"""
|
|
验证原因
|
|
返回: (是否有效, 错误信息)
|
|
"""
|
|
if not reason or not reason.strip():
|
|
return False, "原因不能为空"
|
|
if len(reason) > 255:
|
|
return False, "原因长度不能超过255个字符"
|
|
return True, ""
|
|
|
|
|
|
def validate_date(date_str: str) -> bool:
|
|
"""验证日期格式 YYYY-MM-DD"""
|
|
if not date_str:
|
|
return False
|
|
pattern = r'^\d{4}-\d{2}-\d{2}$'
|
|
if not re.match(pattern, date_str):
|
|
return False
|
|
return True |