diff --git a/.env.example b/.env.example index 4487faf..a894ec1 100644 --- a/.env.example +++ b/.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(用于自动搜集目标资产;不填则只能手动录入目标) diff --git a/app/api/settings.py b/app/api/settings.py index f14ff5a..c8954b9 100644 --- a/app/api/settings.py +++ b/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) diff --git a/app/settings_service.py b/app/settings_service.py index 9dc323e..c861502 100644 --- a/app/settings_service.py +++ b/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}"} diff --git a/app/tools/netguard.py b/app/tools/netguard.py index 05ccbfc..a9b68fd 100644 --- a/app/tools/netguard.py +++ b/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,6 +106,8 @@ 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] @@ -82,6 +115,8 @@ def assert_safe_outbound_url(url: str, *, allow_extra_hosts: set[str] | None = N 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