feat(security): 为私有LLM端点添加LLM_ALLOWED_HOSTS白名单

.env.example

@@ -38,6 +38,11 @@ LLM_PROVIDER_BEHAVIOR_FAIL_THRESHOLD=3
 LLM_PROVIDER_FAILED_RETRY_SECONDS=60
 LLM_PROVIDER_BEHAVIOR_PROBE_SECONDS=900
 LLM_PROVIDER_COOLDOWN_SECONDS=300,900,1800,3600
+# 【可选】允许 LLM base_url 指向内网/私有地址的 host 白名单（逗号分隔，支持单 IP / CIDR / 域名）。
+# 出于防 SSRF/防 key 外泄，默认阻断把携带 LLM Key 的请求发往内网/云元数据；
+# 若你用本地/私有 LLM 部署（vllm/ollama/llama.cpp/xinference 等），把它的地址填这里显式放行。
+# 例：LLM_ALLOWED_HOSTS=192.168.141.3,10.0.0.0/8,llm.lan
+LLM_ALLOWED_HOSTS=
 
 # ---------------------------------------------------------------------------
 # 【推荐】FOFA 网络空间测绘 Key（用于自动搜集目标资产；不填则只能手动录入目标）

app/api/settings.py

@@ -15,7 +15,7 @@
 from app.config import LLMConfig
 from app.db.session import get_session
 from app.llm.client import _is_kimi_coding_endpoint, _resolve_user_agent, llm_request_url
-from app.tools.netguard import SsrfBlocked, assert_safe_outbound_url
+from app.tools.netguard import SsrfBlocked, _env_llm_allowed_hosts, assert_safe_outbound_url
 from app.workdir_cleanup import cleanup_workdir, get_workdir_stats
 from app.ui_prefs import (
     MAX_WALLPAPER_BYTES,
@@ -287,7 +287,7 @@ async def _test_llm_one(name: str, provider: LLMConfig) -> dict:
         result["error_copy"] = _llm_test_error_copy(result)
         return result
     try:
-        assert_safe_outbound_url(url)
+        assert_safe_outbound_url(url, allow_extra_hosts=_env_llm_allowed_hosts())
     except SsrfBlocked as exc:
         result["error"] = f"base_url 不被允许：{exc}"
         result["error_copy"] = _llm_test_error_copy(result)

app/settings_service.py

@@ -788,10 +788,10 @@ async def list_available_models(
         return {"ok": False, "error": "未配置 API Key，无法拉取模型列表", "models": []}
     from app.llm.client import llm_models_url
     url = llm_models_url(base)
-    from app.tools.netguard import SsrfBlocked, assert_safe_outbound_url
+    from app.tools.netguard import SsrfBlocked, _env_llm_allowed_hosts, assert_safe_outbound_url
 
     try:
-        assert_safe_outbound_url(url)
+        assert_safe_outbound_url(url, allow_extra_hosts=_env_llm_allowed_hosts())
     except SsrfBlocked as e:
         return {"ok": False, "error": f"base_url 不被允许：{e}", "models": []}
     headers = {"Authorization": f"Bearer {key}"}

app/tools/netguard.py

@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 import ipaddress
+import os
 import socket
 from urllib.parse import urlparse
 
@@ -38,10 +39,40 @@ def _ip_is_forbidden(ip: ipaddress._BaseAddress) -> bool:
     )
 
 
-def assert_safe_outbound_url(url: str, *, allow_extra_hosts: set[str] | None = None) -> str:
+def _env_llm_allowed_hosts() -> set[str]:
+    """读取 LLM_ALLOWED_HOSTS env（逗号分隔）：放行 base_url 指向的内网 host/IP/CIDR。
+
+    默认空集 = 拦截所有内网/云元数据（防 SSRF + 防 key 外泄）。
+    本地自部署 LLM（vllm/ollama/llama.cpp/xinference 等）把 host/IP/CIDR 填这里显式放行。
+    """
+    raw = os.environ.get("LLM_ALLOWED_HOSTS", "")
+    return {h.strip().lower() for h in raw.split(",") if h.strip()}
+
+
+def _ip_in_allowed(ip: ipaddress._BaseAddress, allowed: set[str]) -> bool:
+    """判断 IP 是否命中白名单（支持单 IP / CIDR 网段 / 主机名）。"""
+    ip_str = str(ip)
+    for item in allowed:
+        if "/" in item:
+            try:
+                if ip in ipaddress.ip_network(item, strict=False):
+                    return True
+            except ValueError:
+                continue
+        elif ip_str == item:
+            return True
+    return False
+
+
+def assert_safe_outbound_url(
+    url: str,
+    *,
+    allow_extra_hosts: set[str] | None = None,
+) -> str:
     """校验并返回原 URL；不安全时抛 SsrfBlocked。
 
     allow_extra_hosts：显式放行的 host（如用户在 env 里配置的私有 FOFA 代理域名）。
+    LLM 内网地址通过 env LLM_ALLOWED_HOSTS 配置（支持单 IP / CIDR / 域名）。
     """
     raw = str(url or "").strip()
     if not raw:
@@ -75,13 +106,17 @@ def assert_safe_outbound_url(url: str, *, allow_extra_hosts: set[str] | None = N
     except OSError as exc:
         raise SsrfBlocked(f"主机解析失败: {host}") from exc
 
+    llm_allowed = _env_llm_allowed_hosts()
+
     for info in infos:
         sockaddr = info[4]
         ip_str = sockaddr[0]
         try:
             ip = ipaddress.ip_address(ip_str)
         except ValueError:
             raise SsrfBlocked(f"无效 IP: {ip_str}")
+        if _ip_in_allowed(ip, llm_allowed):
+            continue
         if _ip_is_forbidden(ip):
             raise SsrfBlocked(f"目标解析到私有/保留地址({ip_str})，已拦截")
     return raw