fix(security): P1/P2 审计修复 + JWT HttpOnly Cookie 双模认证 + 限流

- P0/P1 审计修复: 滑块验证码不再下发 x_position/成败即销毁 key、 user-login 补失败计数+滑块门控、限流标识改 X-Real-IP、百度翻译 appkey 环境化、ChangeEmail/ChangePhone 补调 avalidate、 logs/tasks.py Count(filter=Q) 修复、chat 收藏 SSRF 内网黑名单 - P1 #6/7: token_blacklist + ROTATE_REFRESH_TOKENS 开启, /user/token/refresh/ 挂载 - #2 JWT HttpOnly Cookie 双模认证: user/cookie_auth.py 种/清 Cookie, user/authentication.py CookieOrHeaderJWTAuthentication(Bearer 优先/_COOKIE 兜底), user/views/token.py CookieTokenRefreshView + UserLogoutAPIView(/user/logout/), create_standardized_response 自动对含 token 的响应种 Cookie, 异步视图内 RefreshToken.for_user 全部 sync_to_async 包裹(修 SynchronousOnlyOperation 500), WS ChatConsumer 优先读 Cookie token - P2 #11 限流: utils/rate_limit.py 固定窗口频控, shorturl 生成匿名10次/分+登录60次/分, 邮箱验证码同邮箱60s1次+同IP10次/10min, 登录/注册验证码错5次作废+成功即销毁防重放, 换绑邮箱/手机同步落地, urls.py 补挂 shorturl 路由(此前 404)
2026-09-08 11:28:00 +08:00
parent 2944b19e6f
commit 3618323192
22 changed files with 711 additions and 37 deletions
@@ -0,0 +1,102 @@
+# utils/rate_limit.py — 基于 Redis 缓存的频控与失败计数（同步/异步兼容）
+from django.core.cache import caches
+
+default_cache = caches['default']
+
+
+def get_client_ip(request) -> str:
+    """
+    安全提取客户端真实 IP：优先取反向代理注入的 X-Real-IP，杜绝伪造 X-Forwarded-For 绕过限流
+    """
+    if not request:
+        return ''
+    real_ip = request.META.get('HTTP_X_REAL_IP')
+    if real_ip:
+        return real_ip.strip()
+    fwd = request.META.get('HTTP_X_FORWARDED_FOR')
+    if fwd:
+        return fwd.split(',')[0].strip()
+    return request.META.get('REMOTE_ADDR', '').strip()
+
+
+def _client_key(scope: str, identifier: str) -> str:
+    return f'rl_{scope}_{identifier}'
+
+
+def check_rate_limit(scope: str, identifier: str, limit: int, window_seconds: int) -> bool:
+    """
+    固定窗口频控：同 identifier 在 window_seconds 秒内允许最多 limit 次请求。
+    返回 True 表示放行，False 表示已被限流。
+    """
+    if not identifier:
+        return True
+    key = _client_key(scope, identifier)
+    try:
+        count = default_cache.get(key)
+    except Exception:
+        # Redis 异常时保持放行，避免误伤业务
+        return True
+
+    if count is None:
+        try:
+            default_cache.set(key, 1, timeout=window_seconds)
+        except Exception:
+            return True
+        return True
+
+    try:
+        count = int(count)
+    except (TypeError, ValueError):
+        count = 0
+
+    if count >= limit:
+        return False
+
+    try:
+        default_cache.set(key, count + 1, timeout=window_seconds)
+    except Exception:
+        return True
+    return True
+
+
+def record_failure(scope: str, identifier: str, max_failures: int = 5, window_seconds: int = 300) -> int:
+    """
+    记录一次失败并返回当前累计失败次数。
+    """
+    if not identifier:
+        return 0
+    key = f'fail_{scope}_{identifier}'
+    try:
+        count = default_cache.get(key)
+        new_count = (int(count) + 1) if count is not None else 1
+        default_cache.set(key, new_count, timeout=window_seconds)
+        return new_count
+    except Exception:
+        return 0
+
+
+def get_failure_count(scope: str, identifier: str) -> int:
+    """
+    获取当前失败次数
+    """
+    if not identifier:
+        return 0
+    key = f'fail_{scope}_{identifier}'
+    try:
+        count = default_cache.get(key)
+        return int(count) if count is not None else 0
+    except Exception:
+        return 0
+
+
+def reset_failures(scope: str, identifier: str):
+    """
+    重置失败计数
+    """
+    if not identifier:
+        return
+    key = f'fail_{scope}_{identifier}'
+    try:
+        default_cache.delete(key)
+    except Exception:
+        pass
@@ -104,7 +104,18 @@ def create_standardized_response(data=None, code=None, status_code=200, message=
    if code is not None:
        response_data['code'] = int(code)
        response_data['message'] = message or get_response_message(code)
-    return Response(response_data, status=status_code)
+    response = Response(response_data, status=status_code)
+    # 安全加固：若返回数据含 access 或 refresh，自动植入 HttpOnly Cookie 杜绝 XSS 窃取
+    if isinstance(data, dict):
+        access = data.get('access')
+        refresh = data.get('refresh')
+        if access or refresh:
+            try:
+                from user.cookie_auth import set_auth_cookies
+                set_auth_cookies(response, access_token=access, refresh_token=refresh)
+            except Exception:
+                pass
+    return response


 def create_standardized_error_response(data=None, code=None, status_code=400, message=None):
@@ -85,7 +85,8 @@ def _image_to_base64(image: Image.Image) -> str:
 def generate_slider_captcha() -> Dict:
    captcha_key = str(uuid.uuid4())

-    x_position = secrets.randbelow(200) + 50
+    # 上限 239：保证缺口在客户端可拖动范围内可达（滑块行程 = 图宽300 - 块宽60 = 240，含 ±5 容差）
+    x_position = secrets.randbelow(190) + 50
    y_position = secrets.randbelow(50) + 30

    bg_image = _generate_background_image()
@@ -108,12 +109,13 @@ def generate_slider_captcha() -> Dict:
    }
    default_cache.set(f'slider_captcha_{captcha_key}', cache_data, timeout=CAPTCHA_TIMEOUT)

+    # 安全修复：坐标答案（x）只存服务端缓存，绝不下发客户端（否则脚本可直接读答案绕过）；
+    # y_position 非答案，仅用于客户端垂直摆放拼图块
    return {
        'captcha_key': captcha_key,
        'bg_image': bg_base64,
        'slider_image': slider_base64,
-        'x_position': x_position,
-        'y_position': y_position
+        'y_position': y_position,
    }


@@ -126,11 +128,9 @@ def verify_slider_captcha(captcha_key: str, x_position: int) -> bool:

    stored_x = cached_data['x_position']

-    if abs(stored_x - x_position) <= TOLERANCE:
-        default_cache.delete(cache_key)
-        return True
-
-    return False
+    # 安全修复：无论成败都销毁 key，杜绝同一 key 的穷举重试
+    default_cache.delete(cache_key)
+    return abs(stored_x - x_position) <= TOLERANCE


 def verify_slider_trajectory(
@@ -146,16 +146,21 @@ def verify_slider_trajectory(

    stored_x = cached_data['x_position']

+    # 安全修复：x 错误或轨迹校验失败均立即销毁 key，客户端需重新获取验证码
    if abs(stored_x - x_position) > TOLERANCE:
+        default_cache.delete(cache_key)
        return False

    if len(trajectory) < TRAJECTORY_MIN_POINTS:
+        default_cache.delete(cache_key)
        return False

    if trajectory[-1]['t'] - trajectory[0]['t'] > TRAJECTORY_MAX_DURATION:
+        default_cache.delete(cache_key)
        return False

    if not _analyze_trajectory(trajectory):
+        default_cache.delete(cache_key)
        return False

    default_cache.delete(cache_key)
@@ -0,0 +1,60 @@
+from .slider_captcha import (
+    generate_slider_captcha as _base_generate,
+    verify_slider_captcha as _base_verify,
+    verify_slider_trajectory as _base_verify_traj,
+    SliderCaptchaError,
+)
+
+from .rate_limit import check_rate_limit
+
+# 频控阈值：单 IP 10 秒内最多 6 次生成（普通用户基本用不到这个量）
+SLIDER_GENERATE_LIMIT = 6
+SLIDER_GENERATE_WINDOW = 10  # 秒
+
+# 验证失败计数：单 IP 5 分钟内累计 10 次失败即拒绝继续验证，
+# 防止对单 captcha_key 失败重试耗尽缓存/拖慢接口
+SLIDER_VERIFY_FAIL_LIMIT = 10
+SLIDER_VERIFY_FAIL_WINDOW = 300  # 秒
+
+
+def _client_ip(request) -> str:
+    fwd = request.META.get('HTTP_X_FORWARDED_FOR')
+    if fwd:
+        return fwd.split(',')[0].strip()
+    return request.META.get('REMOTE_ADDR', '')
+
+
+def generate_slider_captcha(request=None):
+    if request is not None:
+        ip = _client_ip(request)
+        if ip and not check_rate_limit('slider_gen', ip, SLIDER_GENERATE_LIMIT, SLIDER_GENERATE_WINDOW):
+            from utils.response_codes import ResponseCode, create_standardized_error_response
+            from rest_framework import status
+            return create_standardized_error_response(
+                code=ResponseCode.RATE_LIMITED,
+                message='请求过于频繁，请稍后再试',
+                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+            )
+    return _base_generate()
+
+
+def verify_slider_captcha(captcha_key: str, x_position: int, request=None) -> bool:
+    ok = _base_verify(captcha_key, x_position)
+    if request is not None and not ok:
+        ip = _client_ip(request)
+        if ip:
+            check_rate_limit('slider_fail', ip, SLIDER_VERIFY_FAIL_LIMIT, SLIDER_VERIFY_FAIL_WINDOW)
+    return ok
+
+
+def verify_slider_trajectory(captcha_key: str, x_position: int, trajectory, request=None) -> bool:
+    ok = _base_verify_traj(captcha_key, x_position, trajectory)
+    if request is not None and not ok:
+        ip = _client_ip(request)
+        if ip:
+            check_rate_limit('slider_fail', ip, SLIDER_VERIFY_FAIL_LIMIT, SLIDER_VERIFY_FAIL_WINDOW)
+    return ok
+
+
+def is_captcha_error(err) -> bool:
+    return isinstance(err, SliderCaptchaError)