安全加固：CORS白名单、JWT密钥自动生成、SSRF防护、路径遍历与错误信息泄露修复

backend/utils/auth.py

@@ -107,7 +107,7 @@ def admin_required(f):
                 return redirect(url_for('auth.login'))
         except Exception as exc:
             if _is_ajax_request():
-                return jsonify({'success': False, 'error': str(exc)}), 500
+                return jsonify({'success': False, 'error': '服务器内部错误，请稍后重试'}), 500
             return redirect(url_for('auth.login'))
         return f(*args, **kwargs)
     return decorated

backend/utils/db.py

@@ -2,6 +2,7 @@
 数据库连接与初始化
 """
 import os
+import re
 import pymysql
 from werkzeug.security import generate_password_hash
 
@@ -31,6 +32,13 @@ mysql_user_source = config_mysql_user_source
 mysql_database_source = config_mysql_database_source
 
 
+def _safe_identifier(name):
+    """仅允许字母、数字、下划线的数据库/表名片段，防止注入 DDL 片段。"""
+    if not re.fullmatch(r'[A-Za-z0-9_]+', name or ''):
+        raise ValueError(f'非法标识符: {name!r}')
+    return name
+
+
 def describe_db_target():
     return (
         f"{mysql_user}@{mysql_host}/{mysql_database} "
@@ -80,9 +88,10 @@ def init_db():
         charset='utf8mb4'
     )
     try:
+        db_name = _safe_identifier(mysql_database)
         with conn.cursor() as cur:
-            cur.execute(f"CREATE DATABASE IF NOT EXISTS `{mysql_database}` DEFAULT CHARSET utf8mb4")
-            cur.execute(f"USE `{mysql_database}`")
+            cur.execute(f"CREATE DATABASE IF NOT EXISTS `{db_name}` DEFAULT CHARSET utf8mb4")
+            cur.execute(f"USE `{db_name}`")
             cur.execute("""
                 CREATE TABLE IF NOT EXISTS users (
                     id INT AUTO_INCREMENT PRIMARY KEY,

backend/utils/ssrf.py

@@ -0,0 +1,58 @@
+"""
+SSRF 防护：检测 URL 是否指向内网/本机地址，禁止服务端请求。
+"""
+import ipaddress
+import socket
+from urllib.parse import urlparse
+
+_PRIVATE_NETWORKS = [
+    ipaddress.ip_network('0.0.0.0/8'),
+    ipaddress.ip_network('10.0.0.0/8'),
+    ipaddress.ip_network('100.64.0.0/10'),
+    ipaddress.ip_network('127.0.0.0/8'),
+    ipaddress.ip_network('169.254.0.0/16'),
+    ipaddress.ip_network('172.16.0.0/12'),
+    ipaddress.ip_network('192.0.0.0/24'),
+    ipaddress.ip_network('192.168.0.0/16'),
+    ipaddress.ip_network('198.18.0.0/15'),
+    ipaddress.ip_network('224.0.0.0/4'),
+    ipaddress.ip_network('240.0.0.0/4'),
+    ipaddress.ip_network('::1/128'),
+    ipaddress.ip_network('fc00::/7'),
+    ipaddress.ip_network('fe80::/10'),
+]
+
+_LOCAL_HOSTNAMES = {
+    'localhost',
+    'localhost.localdomain',
+    'metadata.google.internal',
+    'metadata.azure.internal',
+    '169.254.169.254',
+}
+
+
+def is_internal_url(url):
+    """判断 URL 是否解析到内网/本机/保留地址。解析失败视为不可信返回 True。"""
+    if not url or not isinstance(url, str):
+        return True
+    parsed = urlparse(url)
+    if parsed.scheme not in ('http', 'https'):
+        return True
+    host = parsed.hostname
+    if not host:
+        return True
+    host_lower = host.lower().rstrip('.')
+    if host_lower in _LOCAL_HOSTNAMES:
+        return True
+    try:
+        infos = socket.getaddrinfo(host, parsed.port or 80)
+    except socket.gaierror:
+        return True
+    for info in infos:
+        try:
+            ip = ipaddress.ip_address(info[4][0])
+        except ValueError:
+            continue
+        if any(ip in network for network in _PRIVATE_NETWORKS):
+            return True
+    return False