498d29dfa9
严重修复: - C1: JWT密钥生成64字节随机字符串替代硬编码占位符 - C2: 数据库密码改为随机强密码，移除宿主机端口映射 - C3: 管理员默认密码改为强密码 高危修复: - H1: 移除数据库54326端口映射，仅内部网络访问 - H3: 添加X-Frame-Options/CSP/XSS-Protection等安全头 - H4: 前端所有innerHTML拼接处添加escapeHtml转义防XSS - H5: SMTP密码在API响应中脱敏显示 - H6: 债务列表接口添加用户隔离，普通用户只能看自己的数据 - H7: 债务详情/计划/记录接口添加归属校验 - H8: 账户级暴力破解锁定(5次失败锁定5分钟) - H9: 速率限制数据增加过期清理机制防内存泄漏 中危修复: - M1: 注册/重置密码接口添加后端密码强度校验(最少8位) - M4: 添加全局异常处理器，500错误不再暴露堆栈 - M5: Docker容器改为非root用户运行 - M6: 添加.dockerignore排除pyc和敏感文件 - M7: 管理员操作写入审计日志 - M8: CORS限制为实际使用的HTTP方法和头 其他: - Nginx隐藏版本号(server_tokens off) - 管理员角色修改添加枚举校验 - 密码重置验证码接口返回统一错误信息
104 lines
3.6 KiB
Python
104 lines
3.6 KiB
Python
import time
|
|
from collections import defaultdict
|
|
from fastapi import Request, HTTPException
|
|
|
|
MAX_FAILURES = 10
|
|
ACCOUNT_MAX_FAILURES = 5
|
|
WINDOW_SECONDS = 300
|
|
ALERT_THRESHOLDS = [3, 5, 10]
|
|
|
|
_login_attempts: dict[str, list[float]] = defaultdict(list)
|
|
_account_failures: dict[str, list[float]] = defaultdict(list)
|
|
_account_notified: dict[str, set[int]] = defaultdict(set)
|
|
_account_blocked: dict[str, float] = {}
|
|
|
|
|
|
def _cleanup():
|
|
now = time.time()
|
|
expired_ips = [ip for ip, times in _login_attempts.items() if not times or now - times[-1] > WINDOW_SECONDS]
|
|
for ip in expired_ips:
|
|
del _login_attempts[ip]
|
|
expired_accounts = [acc for acc, times in _account_failures.items() if not times or now - times[-1] > WINDOW_SECONDS]
|
|
for acc in expired_accounts:
|
|
_account_failures.pop(acc, None)
|
|
_account_notified.pop(acc, None)
|
|
expired_blocked = [acc for acc, until in _account_blocked.items() if now > until]
|
|
for acc in expired_blocked:
|
|
del _account_blocked[acc]
|
|
|
|
|
|
def record_login_failure(ip: str, account: str = ""):
|
|
_cleanup()
|
|
now = time.time()
|
|
_login_attempts[ip] = [t for t in _login_attempts[ip] if now - t < WINDOW_SECONDS]
|
|
_login_attempts[ip].append(now)
|
|
if account:
|
|
_account_failures[account] = [t for t in _account_failures[account] if now - t < WINDOW_SECONDS]
|
|
_account_failures[account].append(now)
|
|
count = len(_account_failures[account])
|
|
if count >= ACCOUNT_MAX_FAILURES:
|
|
_account_blocked[account] = now + WINDOW_SECONDS
|
|
|
|
|
|
def clear_login_failures(ip: str, account: str = ""):
|
|
_login_attempts.pop(ip, None)
|
|
if account:
|
|
_account_failures.pop(account, None)
|
|
_account_notified.pop(account, None)
|
|
_account_blocked.pop(account, None)
|
|
|
|
|
|
def get_remaining_seconds(ip: str) -> int:
|
|
now = time.time()
|
|
_login_attempts[ip] = [t for t in _login_attempts[ip] if now - t < WINDOW_SECONDS]
|
|
if not _login_attempts[ip]:
|
|
return 0
|
|
oldest = _login_attempts[ip][0]
|
|
remaining = int(WINDOW_SECONDS - (now - oldest))
|
|
return max(remaining, 0)
|
|
|
|
|
|
def get_account_remaining_seconds(account: str) -> int:
|
|
now = time.time()
|
|
if account in _account_blocked:
|
|
remaining = int(_account_blocked[account] - now)
|
|
if remaining > 0:
|
|
return remaining
|
|
del _account_blocked[account]
|
|
return 0
|
|
|
|
|
|
def get_account_failure_count(account: str) -> int:
|
|
now = time.time()
|
|
_account_failures[account] = [t for t in _account_failures[account] if now - t < WINDOW_SECONDS]
|
|
return len(_account_failures[account])
|
|
|
|
|
|
def get_unnotified_threshold(account: str) -> int | None:
|
|
now = time.time()
|
|
_account_failures[account] = [t for t in _account_failures[account] if now - t < WINDOW_SECONDS]
|
|
count = len(_account_failures[account])
|
|
notified = _account_notified[account]
|
|
for threshold in ALERT_THRESHOLDS:
|
|
if count >= threshold and threshold not in notified:
|
|
notified.add(threshold)
|
|
return threshold
|
|
return None
|
|
|
|
|
|
def is_ip_blocked(ip: str) -> bool:
|
|
return get_remaining_seconds(ip) > 0 and len(_login_attempts[ip]) >= MAX_FAILURES
|
|
|
|
|
|
def is_account_blocked(account: str) -> bool:
|
|
return get_account_remaining_seconds(account) > 0
|
|
|
|
|
|
async def ip_rate_limit_middleware(request: Request, call_next):
|
|
ip = request.client.host if request.client else "unknown"
|
|
if is_ip_blocked(ip):
|
|
remaining = get_remaining_seconds(ip)
|
|
raise HTTPException(status_code=429, detail=f"登录失败次数过多，请 {remaining} 秒后再试")
|
|
response = await call_next(request)
|
|
return response
|