v2.1: 新增通知提醒功能和安全增强
功能: - 通知提醒系统: 还款日提醒、逾期提醒、系统消息横幅、登录时通知 - 通知中心UI: 铃铛图标 + 下拉列表 + 未读徽章 - 定时任务: APScheduler 每小时检查即将到期的还款 - 邮件通知: SMTP 发送还款提醒和安全警告 - 异常登录通知: 3/5/10次失败触发网页和邮件通知 安全: - 登录安全告警: 异常登录次数达到阈值时发送通知 - 密码重置链接: 一次性使用, 5分钟有效期 - 登录频率限制: IP级和账户级限制 技术: - 后端: FastAPI + SQLAlchemy + PostgreSQL - 前端: 纯 HTML/CSS/JS - 部署: Docker Compose (新增 build 配置)
This commit is contained in:
68
backend/app/core/rate_limit.py
Normal file
68
backend/app/core/rate_limit.py
Normal file
@@ -0,0 +1,68 @@
|
||||
import time
|
||||
from collections import defaultdict
|
||||
from fastapi import Request, HTTPException
|
||||
|
||||
MAX_FAILURES = 10
|
||||
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)
|
||||
|
||||
|
||||
def record_login_failure(ip: str, account: str = ""):
|
||||
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)
|
||||
|
||||
|
||||
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)
|
||||
|
||||
|
||||
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_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
|
||||
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user