Harden auth, CSRF, and email log UX

app_security.py

@@ -10,9 +10,12 @@ import re
 import time
 import hashlib
 import secrets
+import ipaddress
+import socket
 from pathlib import Path
 from typing import Optional
 from functools import wraps
+from urllib.parse import urlparse
 from flask import request, jsonify, session
 from collections import defaultdict
 import threading
@@ -199,7 +202,7 @@ def require_ip_not_locked(f):
     """装饰器：检查IP是否被锁定"""
     @wraps(f)
     def decorated_function(*args, **kwargs):
-        ip_address = request.remote_addr
+        ip_address = get_rate_limit_ip()
 
         # P0 / O-01：统一使用 services.state 的线程安全限流状态
         try:
@@ -450,6 +453,65 @@ def get_client_ip(trust_proxy=False):
     return request.remote_addr
 
 
+def get_rate_limit_ip() -> str:
+    """在可信代理场景下取真实IP，用于限流/风控。"""
+    remote_addr = request.remote_addr or ""
+    try:
+        remote_ip = ipaddress.ip_address(remote_addr)
+    except ValueError:
+        remote_ip = None
+
+    if remote_ip and (remote_ip.is_private or remote_ip.is_loopback or remote_ip.is_link_local):
+        forwarded = request.headers.get("X-Forwarded-For", "")
+        if forwarded:
+            candidate = forwarded.split(",")[0].strip()
+            try:
+                ipaddress.ip_address(candidate)
+                return candidate
+            except ValueError:
+                pass
+        real_ip = request.headers.get("X-Real-IP", "").strip()
+        if real_ip:
+            try:
+                ipaddress.ip_address(real_ip)
+                return real_ip
+            except ValueError:
+                pass
+
+    return remote_addr
+
+
+def is_safe_outbound_url(url: str) -> bool:
+    """限制向内网/保留地址发起请求，降低SSRF风险。"""
+    try:
+        parsed = urlparse(str(url or "").strip())
+    except Exception:
+        return False
+
+    if parsed.scheme not in ("http", "https"):
+        return False
+
+    host = parsed.hostname
+    if not host:
+        return False
+
+    ips = []
+    try:
+        ips = [ipaddress.ip_address(host)]
+    except ValueError:
+        try:
+            infos = socket.getaddrinfo(host, None)
+            ips = [ipaddress.ip_address(info[4][0]) for info in infos]
+        except Exception:
+            return False
+
+    for ip in ips:
+        if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved:
+            return False
+
+    return True
+
+
 if __name__ == '__main__':
     # 测试文件路径安全
     print("文件路径安全测试:")