fix(ssrf): restore local LLM loopback egress and unify pre/post outbound validation

README.md

@@ -493,6 +493,12 @@ Notes:
   - allow additional exact hosts via `OPENCLAW_LLM_ALLOWED_HOSTS=host1,host2`
   - or opt in to any public host via `OPENCLAW_ALLOW_ANY_PUBLIC_LLM_HOST=1`
   - `OPENCLAW_ALLOW_INSECURE_BASE_URL=1` disables SSRF blocking (not recommended)
+- Local providers (`ollama`, `lmstudio`) are loopback-only by design:
+  - valid targets: `localhost` / `127.0.0.1` / `::1`
+  - do **not** enable `OPENCLAW_ALLOW_INSECURE_BASE_URL` just to use local LLM
+  - recommended examples:
+    - Ollama: `http://127.0.0.1:11434`
+    - LM Studio: `http://localhost:1234/v1`
 
 ### Webhooks
 
@@ -768,6 +774,24 @@ python scripts/operator_doctor.py --json
 
 Set webhook auth env vars (see Quick Start) and restart ComfyUI.
 
+### LLM model list shows `HTTP 403 ... Private/reserved IP blocked: 127.0.0.1`
+
+This error usually means your OpenClaw version is older than the local-loopback SSRF fix.
+For local providers, `127.0.0.1` and `localhost` are valid targets and do not require insecure SSRF flags.
+
+Checklist:
+
+1. Update OpenClaw to the latest release.
+2. For Ollama:
+   - run `ollama serve`
+   - verify `http://127.0.0.1:11434/api/tags` is reachable on the same machine
+3. In OpenClaw Settings:
+   - Provider: `Ollama (Local)` or `LM Studio (Local)`
+   - Base URL: leave empty (use provider default) or set loopback URL explicitly
+4. Keep these flags disabled:
+   - `OPENCLAW_ALLOW_ANY_PUBLIC_LLM_HOST=0`
+   - `OPENCLAW_ALLOW_INSECURE_BASE_URL=0`
+
 ### Admin Token: server-side vs UI
 
 `OPENCLAW_ADMIN_TOKEN` is a **server-side environment variable**.

api/config.py

@@ -80,6 +80,7 @@ if __package__ and "." in __package__:
         get_admin_token,
         get_apply_semantics,
         get_effective_config,
+        get_llm_egress_controls,
         get_settings_schema,
         is_loopback_client,
         update_config,
@@ -110,6 +111,7 @@ else:  # pragma: no cover (test-only import mode)
         get_admin_token,
         get_apply_semantics,
         get_effective_config,
+        get_llm_egress_controls,
         get_settings_schema,
         is_loopback_client,
         update_config,
@@ -399,10 +401,10 @@ async def llm_models_handler(request: web.Request) -> web.Response:
 
     try:
         from ..services.providers.catalog import ProviderType, get_provider_info
-        from ..services.providers.keys import get_api_key_for_provider
+        from ..services.providers.keys import get_api_key_for_provider, requires_api_key
     except ImportError:
         from services.providers.catalog import ProviderType, get_provider_info
-        from services.providers.keys import get_api_key_for_provider
+        from services.providers.keys import get_api_key_for_provider, requires_api_key
 
     info = get_provider_info(provider)
     if not info:
@@ -443,7 +445,11 @@ async def llm_models_handler(request: web.Request) -> web.Response:
             )
 
     api_key = get_api_key_for_provider(provider)
-    if not api_key:
+    # CRITICAL:
+    # Local providers (e.g. ollama/lmstudio) intentionally work without API keys.
+    # Do not change this gate back to `if not api_key`, or local model-list loading
+    # will regress with false 400 errors.
+    if requires_api_key(provider) and not api_key:
         return web.json_response(
             {"ok": False, "error": f"No API key configured for provider '{provider}'."},
             status=400,
@@ -459,16 +465,12 @@ async def llm_models_handler(request: web.Request) -> web.Response:
         except ImportError:
             from services.safe_io import STANDARD_OUTBOUND_POLICY, validate_outbound_url
 
-        allowed_hosts = _get_llm_allowed_hosts()
-        allow_any = _env_flag(
-            "OPENCLAW_ALLOW_ANY_PUBLIC_LLM_HOST",
-            "MOLTBOT_ALLOW_ANY_PUBLIC_LLM_HOST",
-            default=False,
-        )
+        controls = get_llm_egress_controls(provider, base_url)
         validate_outbound_url(
             base_url,
-            allow_hosts=None if allow_any else allowed_hosts,
-            allow_any_public_host=allow_any,
+            allow_hosts=controls.get("allow_hosts"),
+            allow_any_public_host=bool(controls.get("allow_any_public_host")),
+            allow_loopback_hosts=controls.get("allow_loopback_hosts"),
             policy=STANDARD_OUTBOUND_POLICY,
         )
     except Exception as e:
@@ -492,19 +494,24 @@ async def llm_models_handler(request: web.Request) -> web.Response:
             )
 
         url = f"{base_url.rstrip('/')}/models"
+        request_headers = {
+            "User-Agent": f"ComfyUI-OpenClaw/{PACK_VERSION}",
+            "Accept": "application/json",
+        }
+        if api_key:
+            request_headers["Authorization"] = f"Bearer {api_key}"
 
         # S65: Enforce outbound policy via safe_io
         payload = safe_request_json(
             method="GET",
             url=url,
-            headers={
-                "User-Agent": f"ComfyUI-OpenClaw/{PACK_VERSION}",
-                "Authorization": f"Bearer {api_key}",
-                "Accept": "application/json",
-            },
+            json_body=None,
+            headers=request_headers,
             timeout_sec=10,
             policy=STANDARD_OUTBOUND_POLICY,
-            allow_hosts=None if allow_any else allowed_hosts,
+            allow_hosts=controls.get("allow_hosts"),
+            allow_any_public_host=bool(controls.get("allow_any_public_host")),
+            allow_loopback_hosts=controls.get("allow_loopback_hosts"),
         )
 
         models = _extract_models_from_payload(payload)

docs/deploy/local-only.md

@@ -34,6 +34,12 @@ No special configuration is required.
 
 - **Admin Token**: Not required for loopback-only operations (unless `OPENCLAW_CONNECTOR_ADMIN_TOKEN` is explicitly set).
 - **Webhooks**: Disabled by default.
+- **Local LLM (optional)**:
+  - Ollama: `http://127.0.0.1:11434`
+  - LM Studio: `http://localhost:1234/v1`
+  - Keep SSRF relax flags disabled:
+    - `OPENCLAW_ALLOW_ANY_PUBLIC_LLM_HOST=0`
+    - `OPENCLAW_ALLOW_INSECURE_BASE_URL=0`
 
 ### 3. "Red Lines" (What NOT to do)
 
@@ -45,4 +51,6 @@ No special configuration is required.
 1. Open `http://127.0.0.1:8188` in your browser.
 2. Open the OpenClaw tab in the sidebar.
 3. Go to **Settings** -> **Health**.
-4. All checks should be green.
+4. If using Ollama, verify `http://127.0.0.1:11434/api/tags` responds.
+5. In **Settings -> LLM**, set provider to `Ollama (Local)` and click **Load Models**.
+6. All checks should be green.

docs/release/config_secrets_contract.md

@@ -39,6 +39,10 @@ Controls the core LLM client used by nodes (Planner, Refiner, etc.).
 | `OPENCLAW_ALLOW_ANY_PUBLIC_LLM_HOST` | `0` | Set `1` to bypass host allowlist and allow any public IP. |
 | `OPENCLAW_ALLOW_INSECURE_BASE_URL` | `0` | Set `1` to allow HTTP or private IP targets (Dangerous!). |
 
+Notes:
+- Local providers (`ollama`, `lmstudio`) are loopback-only by design and should use `localhost` / `127.0.0.1` / `::1`.
+- Local loopback provider targets do not require enabling insecure SSRF flags.
+
 ### 2.2 Security & Authentication
 
 Controls access to APIs and administrative features.
@@ -50,7 +54,7 @@ Controls access to APIs and administrative features.
 | `OPENCLAW_WEBHOOK_SECRET` | **High** | Secret for signing/verifying inbound webhook payloads. |
 | `OPENCLAW_REQUIRE_APPROVAL_FOR_TRIGGERS` | Low | Set `1` to require admin approval for all external triggers (default: `0`). |
 | `OPENCLAW_PRESETS_PUBLIC_READ` | Low | Set `0` to require Admin Token for listing presets (default: `1`). |
-| `OPENCLAW_STRICT_LOCALHOST_AUTH` | Low | Set `1` to enforce auth even on localhost (default: `0`). |
+| `OPENCLAW_STRICT_LOCALHOST_AUTH` | Low | Set `1` to enforce auth even on localhost (default: `1`). |
 
 ### 2.3 Connector & Delivery (Chat Apps)
 

docs/security_deployment_guide.md

@@ -79,9 +79,10 @@ OPENCLAW_ADMIN_TOKEN=change-this-local-admin-token
 1. Keep ComfyUI bound to localhost only.
 2. Keep remote admin disabled.
 3. Keep external tools/registry sync/transforms disabled unless explicitly needed.
-4. Run:
+4. For local LLM providers (Ollama/LM Studio), use loopback URLs only (`localhost`/`127.0.0.1`/`::1`); keep `OPENCLAW_ALLOW_ANY_PUBLIC_LLM_HOST=0` and `OPENCLAW_ALLOW_INSECURE_BASE_URL=0`.
+5. Run:
    - `python scripts/check_deployment_profile.py --profile local`
-5. If you enable optional high-risk features, document why and time-box the change.
+6. If you enable optional high-risk features, document why and time-box the change.
 
 ## 4. LAN (Trusted Subnet)
 

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "comfyui-openclaw"
 description = "Your own personal AIGC Factory. Any picture. Any reel. The Comfy way.©️"
-version = "0.5.0"
+version = "0.5.2"
 license = {text = "MIT"}
 readme = "README.md"
 requires-python = ">=3.10"

scripts/preflight_check.py

@@ -12,10 +12,10 @@ Usage:
     python scripts/preflight_check.py [--strict]
 """
 
-import sys
-import subprocess
-import re
 import argparse
+import re
+import subprocess
+import sys
 
 # Minimum requirements
 MIN_PYTHON_VERSION = (3, 10)
@@ -31,39 +31,52 @@ RED = "\033[91m"
 YELLOW = "\033[93m"
 RESET = "\033[0m"
 
+
 def log_ok(msg: str):
     print(f"{GREEN}[OK]{RESET} {msg}")
 
+
 def log_fail(msg: str):
     print(f"{RED}[FAIL]{RESET} {msg}")
 
+
 def log_warn(msg: str):
     print(f"{YELLOW}[WARN]{RESET} {msg}")
 
+
 def check_python_version() -> bool:
     """Validate Python interpreter version."""
     current = sys.version_info[:3]
     if current < MIN_PYTHON_VERSION:
-        log_fail(f"Python version {sys.version} is too old. Required: >={'.'.join(map(str, MIN_PYTHON_VERSION))}")
+        log_fail(
+            f"Python version {sys.version} is too old. Required: >={'.'.join(map(str, MIN_PYTHON_VERSION))}"
+        )
         return False
     log_ok(f"Python version: {'.'.join(map(str, current))}")
     return True
 
+
 def check_node_version() -> bool:
     """Validate Node.js version."""
     try:
-        output = subprocess.check_output(["node", "--version"], stderr=subprocess.STDOUT).decode("utf-8").strip()
+        output = (
+            subprocess.check_output(["node", "--version"], stderr=subprocess.STDOUT)
+            .decode("utf-8")
+            .strip()
+        )
         # Output is usually vX.Y.Z
         match = re.search(r"v(\d+)\.(\d+)\.(\d+)", output)
         if not match:
             log_warn(f"Could not parse Node version from '{output}'.")
             return False
-            
+
         major, minor, patch = map(int, match.groups())
         if (major, minor, patch) < MIN_NODE_VERSION:
-            log_fail(f"Node version {output} is too old. Required: >={'.'.join(map(str, MIN_NODE_VERSION))}")
+            log_fail(
+                f"Node version {output} is too old. Required: >={'.'.join(map(str, MIN_NODE_VERSION))}"
+            )
             return False
-            
+
         log_ok(f"Node version: {output}")
         return True
     except FileNotFoundError:
@@ -73,23 +86,24 @@ def check_node_version() -> bool:
         log_warn(f"Failed to check Node version: {e}")
         return False
 
+
 def check_python_packages() -> bool:
     """Validate installed Python packages."""
     all_ok = True
     try:
-        from importlib.metadata import version, PackageNotFoundError
-        
+        from importlib.metadata import PackageNotFoundError, version
+
         for pkg, min_ver in REQUIRED_PYTHON_PACKAGES:
             try:
                 installed_ver = version(pkg)
                 # Simple version compare (not fully semver compliant but enough for preflight)
                 # Using pkg_resources or packaging.version is better but might add deps.
                 # We'll just split by dot.
-                
+
                 # normalize versions
                 def parse_ver(v_str):
                     return tuple(map(int, v_str.split(".")[:3]))
-                
+
                 if parse_ver(installed_ver) < parse_ver(min_ver):
                     log_fail(f"Package '{pkg}' version {installed_ver} < {min_ver}")
                     all_ok = False
@@ -100,30 +114,37 @@ def check_python_packages() -> bool:
                 all_ok = False
             except ValueError:
                 # Fallback for complex version strings
-                log_warn(f"Package '{pkg}' version {installed_ver} checked (complex format). Assuming OK.")
-                
+                log_warn(
+                    f"Package '{pkg}' version {installed_ver} checked (complex format). Assuming OK."
+                )
+
     except ImportError:
-        log_warn("importlib.metadata not available (Python < 3.8?). Skipping package checks.")
-        
+        log_warn(
+            "importlib.metadata not available (Python < 3.8?). Skipping package checks."
+        )
+
     return all_ok
 
+
 def main():
     parser = argparse.ArgumentParser(description="Run dependency preflight checks.")
-    parser.add_argument("--strict", action="store_true", help="Fail with exit code 1 on any error")
+    parser.add_argument(
+        "--strict", action="store_true", help="Fail with exit code 1 on any error"
+    )
     args = parser.parse_args()
-    
+
     print("Running R120 Dependency Preflight...")
     print("-" * 40)
-    
+
     checks = [
         check_python_version(),
         check_node_version(),
         check_python_packages(),
     ]
-    
+
     print("-" * 40)
     success = all(checks)
-    
+
     if success:
         print(f"{GREEN}Preflight PASSED.{RESET}")
         sys.exit(0)
@@ -134,5 +155,6 @@ def main():
         # Non-strict mode (e.g. dev) might exit 0 or just warn
         sys.exit(1)
 
+
 if __name__ == "__main__":
     main()

services/llm_client.py

@@ -254,6 +254,20 @@ class LLMClient:
 
         return candidates_3d
 
+    def _get_egress_controls(
+        self, provider: str, base_url: Optional[str]
+    ) -> Dict[str, Any]:
+        """Build canonical SSRF controls for provider egress."""
+        try:
+            from ..services.runtime_config import get_llm_egress_controls
+        except ImportError:
+            from services.runtime_config import get_llm_egress_controls
+
+        # IMPORTANT:
+        # Keep provider egress controls centralized. Falling back to policy-only or
+        # ad-hoc allowlists causes path drift and can reintroduce S65 regressions.
+        return get_llm_egress_controls(provider, base_url or "")
+
     def _validate_candidate_url(self, provider: str, base_url: Optional[str]) -> bool:
         """
         Validate base_url against S16/S16.1 SSRF policy.
@@ -277,51 +291,25 @@ class LLMClient:
         except ImportError:
             from services.safe_io import STANDARD_OUTBOUND_POLICY, validate_outbound_url
 
-        def _env_flag(primary: str, legacy: str, default: bool = False) -> bool:
-            val = os.environ.get(primary)
-            if val is None:
-                val = os.environ.get(legacy)
-            if val is None:
-                return default
-            return str(val).strip().lower() in ("1", "true", "yes", "y", "on")
-
         try:
-            # S16.1: Strict host allowlist (exact match) OR explicit opt-in for any public host.
-            allowed_hosts_str = os.environ.get(
-                "OPENCLAW_LLM_ALLOWED_HOSTS"
-            ) or os.environ.get("MOLTBOT_LLM_ALLOWED_HOSTS", "")
-            allowed_hosts_env = set(
-                h.lower().strip() for h in allowed_hosts_str.split(",") if h.strip()
-            )
-            try:
-                from ..services.providers.catalog import get_default_public_llm_hosts
-            except ImportError:
-                from services.providers.catalog import (
-                    get_default_public_llm_hosts,  # type: ignore
-                )
-
-            allowed_hosts = set(get_default_public_llm_hosts()) | allowed_hosts_env
-            allow_any = _env_flag(
-                "OPENCLAW_ALLOW_ANY_PUBLIC_LLM_HOST",
-                "MOLTBOT_ALLOW_ANY_PUBLIC_LLM_HOST",
-                default=False,
-            )
+            controls = self._get_egress_controls(provider, base_url)
 
             # S16/S16.1/S51: Validate URL (raises on block).
             validate_outbound_url(
                 base_url,
-                allow_hosts=allowed_hosts if not allow_any else None,
-                allow_any_public_host=allow_any,
+                allow_hosts=controls.get("allow_hosts"),
+                allow_any_public_host=bool(controls.get("allow_any_public_host")),
+                allow_loopback_hosts=controls.get("allow_loopback_hosts"),
                 policy=STANDARD_OUTBOUND_POLICY,
             )
             return True
         except Exception as e:
             # Allow override via explicit risk-acceptance flag (keeps behavior consistent with runtime_config validation).
-            if _env_flag(
-                "OPENCLAW_ALLOW_INSECURE_BASE_URL",
-                "MOLTBOT_ALLOW_INSECURE_BASE_URL",
-                default=False,
-            ):
+            if (
+                os.environ.get("OPENCLAW_ALLOW_INSECURE_BASE_URL")
+                or os.environ.get("MOLTBOT_ALLOW_INSECURE_BASE_URL")
+                or ""
+            ).strip().lower() in ("1", "true", "yes", "y", "on"):
                 logger.warning(
                     f"Failover candidate {provider} with base_url={base_url} allowed by "
                     f"OPENCLAW_ALLOW_INSECURE_BASE_URL despite SSRF policy: {e}"
@@ -765,6 +753,8 @@ class LLMClient:
         max_tokens: int,
     ) -> Dict[str, Any]:
         """Complete using Anthropic Messages API."""
+        egress_controls = self._get_egress_controls(self.provider, self.base_url)
+
         if image_base64:
             message = anthropic.build_vision_message(
                 user_message, image_base64, image_media_type
@@ -781,6 +771,9 @@ class LLMClient:
             temperature=temperature,
             max_tokens=max_tokens,
             timeout=self.timeout,
+            allow_hosts=egress_controls.get("allow_hosts"),
+            allow_any_public_host=bool(egress_controls.get("allow_any_public_host")),
+            allow_loopback_hosts=egress_controls.get("allow_loopback_hosts"),
         )
 
     def _complete_openai_compat(
@@ -796,6 +789,7 @@ class LLMClient:
         tool_choice: Optional[str] = None,
     ) -> Dict[str, Any]:
         """Complete using OpenAI-compatible API."""
+        egress_controls = self._get_egress_controls(self.provider, self.base_url)
         messages = [{"role": "system", "content": system}]
 
         if image_base64:
@@ -817,6 +811,9 @@ class LLMClient:
             timeout=self.timeout,
             tools=tools,
             tool_choice=tool_choice,
+            allow_hosts=egress_controls.get("allow_hosts"),
+            allow_any_public_host=bool(egress_controls.get("allow_any_public_host")),
+            allow_loopback_hosts=egress_controls.get("allow_loopback_hosts"),
         )
 
     def get_provider_summary(self) -> Dict[str, Any]:

services/providers/anthropic.py

@@ -51,6 +51,9 @@ def make_request(
     temperature: float = 0.7,
     max_tokens: int = 4096,
     timeout: float = 120.0,
+    allow_hosts: Optional[set[str]] = None,
+    allow_any_public_host: bool = False,
+    allow_loopback_hosts: Optional[set[str]] = None,
 ) -> Dict[str, Any]:
     """
     Make a request to Anthropic /v1/messages endpoint.
@@ -80,7 +83,9 @@ def make_request(
             headers=headers,
             timeout_sec=int(timeout),
             policy=STANDARD_OUTBOUND_POLICY,
-            allow_hosts=None,  # Use policy + strict DNS check
+            allow_hosts=allow_hosts,
+            allow_any_public_host=allow_any_public_host,
+            allow_loopback_hosts=allow_loopback_hosts,
         )
 
         # Extract text from response

services/providers/catalog.py

@@ -318,6 +318,50 @@ def list_providers() -> list:
     return list(PROVIDER_CATALOG.keys())
 
 
+def _normalize_host(host: str) -> str:
+    return host.lower().strip().rstrip(".")
+
+
+def is_loopback_host(host: str) -> bool:
+    """Return True if host is one of the canonical loopback names."""
+    return _normalize_host(host) in {"localhost", "127.0.0.1", "::1"}
+
+
+def get_loopback_host_aliases(host: str) -> set[str]:
+    """
+    Return canonical loopback aliases when host is loopback.
+
+    This intentionally returns all canonical aliases so validation remains stable
+    regardless of whether callers use localhost, IPv4 loopback, or IPv6 loopback.
+    """
+    if not is_loopback_host(host):
+        return set()
+    return {"localhost", "127.0.0.1", "::1"}
+
+
+def is_local_provider(provider: str) -> bool:
+    """
+    Return True for catalog providers intended for local-loopback use.
+
+    Local providers are identified by:
+    - no API key requirement, and
+    - loopback default endpoint or explicit "(Local)" naming.
+    """
+    info = get_provider_info(provider)
+    if not info:
+        return False
+    if info.env_key_name is not None:
+        return False
+
+    try:
+        parsed = urlparse(info.base_url or "")
+        host = parsed.hostname or ""
+    except Exception:
+        host = ""
+
+    return is_loopback_host(host) or info.name.lower().endswith("(local)")
+
+
 def get_default_public_llm_hosts() -> set[str]:
     """
     Return the default *public* LLM hosts that are safe to allow by default.
@@ -326,8 +370,8 @@ def get_default_public_llm_hosts() -> set[str]:
     - We want built-in providers to work out-of-the-box without requiring users to
       configure an SSRF allowlist.
     - Custom Base URLs must still pass SSRF validation (host allowlist + public IP).
-    - Local providers are intentionally excluded here because SSRF validation blocks
-      loopback/private IPs by design.
+    - Local providers are intentionally excluded from this *public* allowlist.
+      Their loopback behavior is handled by explicit provider-aware controls.
     """
     hosts: set[str] = set()
 

services/providers/openai_compat.py

@@ -70,6 +70,9 @@ def make_request(
     timeout: float = 120.0,
     tools: Optional[List[Dict[str, Any]]] = None,  # R39: Optional tools
     tool_choice: Optional[str] = None,  # R39: Optional tool_choice
+    allow_hosts: Optional[set[str]] = None,
+    allow_any_public_host: bool = False,
+    allow_loopback_hosts: Optional[set[str]] = None,
 ) -> Dict[str, Any]:
     """
     Make a request to an OpenAI-compatible /chat/completions endpoint.
@@ -101,7 +104,9 @@ def make_request(
             headers=headers,
             timeout_sec=int(timeout),
             policy=STANDARD_OUTBOUND_POLICY,
-            allow_hosts=None,  # Use policy + strict DNS check
+            allow_hosts=allow_hosts,
+            allow_any_public_host=allow_any_public_host,
+            allow_loopback_hosts=allow_loopback_hosts,
         )
 
         # Extract text from response

services/runtime_config.py

@@ -8,6 +8,7 @@ import json
 import logging
 import os
 from typing import Any, Dict, List, Optional, Tuple
+from urllib.parse import urlparse
 
 logger = logging.getLogger("ComfyUI-OpenClaw.services.runtime_config")
 
@@ -41,7 +42,12 @@ try:
     from .state_dir import get_state_dir
 
     CONFIG_FILE = os.path.join(get_state_dir(), "config.json")
-    from .providers.catalog import PROVIDER_CATALOG
+    from .providers.catalog import (
+        PROVIDER_CATALOG,
+        get_default_public_llm_hosts,
+        get_loopback_host_aliases,
+        is_local_provider,
+    )
     from .safe_io import SSRFError, is_private_ip, validate_outbound_url
 except ImportError:
     try:
@@ -49,7 +55,12 @@ except ImportError:
         from services.state_dir import get_state_dir  # type: ignore
 
         CONFIG_FILE = os.path.join(get_state_dir(), "config.json")
-        from services.providers.catalog import PROVIDER_CATALOG  # type: ignore
+        from services.providers.catalog import (  # type: ignore
+            PROVIDER_CATALOG,
+            get_default_public_llm_hosts,
+            get_loopback_host_aliases,
+            is_local_provider,
+        )
         from services.safe_io import is_private_ip  # type: ignore
         from services.safe_io import SSRFError, validate_outbound_url
     except ImportError:
@@ -58,6 +69,9 @@ except ImportError:
         )
         # Fallback to empty if missing
         PROVIDER_CATALOG = {}
+        get_default_public_llm_hosts = lambda: set()  # type: ignore
+        get_loopback_host_aliases = lambda _host: set()  # type: ignore
+        is_local_provider = lambda _provider: False  # type: ignore
 
         # Mock for validation if missing (Fail Closed)
         class SSRFError(ValueError):
@@ -225,6 +239,49 @@ def _env_flag(primary: str, legacy: str, default: bool = False) -> bool:
     return str(v).strip().lower() in ("1", "true", "yes", "on")
 
 
+def get_llm_egress_controls(provider: str, base_url: str) -> Dict[str, Any]:
+    """
+    Build canonical outbound SSRF controls for LLM egress paths.
+
+    IMPORTANT:
+    Callers must reuse this same control set for both pre-validation and request-time
+    validation. Diverging parameters caused the S65 loopback regression (pre-check
+    passed while request-time check failed with HTTP 403).
+    """
+    allowed_hosts_str = os.environ.get("OPENCLAW_LLM_ALLOWED_HOSTS") or os.environ.get(
+        "MOLTBOT_LLM_ALLOWED_HOSTS", ""
+    )
+    env_hosts = {h.lower().strip() for h in allowed_hosts_str.split(",") if h.strip()}
+    allowed_hosts = set(get_default_public_llm_hosts()) | env_hosts
+
+    allow_any = _env_flag(
+        "OPENCLAW_ALLOW_ANY_PUBLIC_LLM_HOST",
+        "MOLTBOT_ALLOW_ANY_PUBLIC_LLM_HOST",
+        default=False,
+    )
+
+    allow_loopback_hosts: Optional[set[str]] = None
+    try:
+        host = (urlparse(base_url).hostname or "").lower().rstrip(".")
+    except Exception:
+        host = ""
+
+    # CRITICAL:
+    # Local providers can use loopback only. Never widen this to blanket private IPs;
+    # doing so would reopen SSRF paths into internal networks.
+    if host and is_local_provider(provider):
+        loopback_aliases = get_loopback_host_aliases(host)
+        if loopback_aliases:
+            allow_loopback_hosts = loopback_aliases
+            allowed_hosts |= loopback_aliases
+
+    return {
+        "allow_hosts": None if allow_any else allowed_hosts,
+        "allow_any_public_host": allow_any,
+        "allow_loopback_hosts": allow_loopback_hosts,
+    }
+
+
 def get_scheduler_config() -> Dict[str, Any]:
     """
     Get effective Scheduler config (Env > Defaults).
@@ -414,19 +471,8 @@ def validate_config_update(updates: Dict[str, Any]) -> Tuple[Dict[str, Any], lis
                 # Matches known good default
                 pass
 
-            elif known_provider and known_provider.name.lower().endswith("(local)"):
-                # Loopback only for local providers
-                if not (
-                    val.startswith("http://localhost")
-                    or val.startswith("http://127.0.0.1")
-                ):
-                    errors.append(
-                        f"Local provider {provider_key} must use localhost URL"
-                    )
-                    continue
-
             else:
-                # Custom URL (either custom provider OR overriding default URL)
+                # Custom URL (either custom provider OR overriding default URL).
 
                 # Check opt-in for custom URLs
                 if provider_key == "custom" and not _env_flag(
@@ -439,37 +485,27 @@ def validate_config_update(updates: Dict[str, Any]) -> Tuple[Dict[str, Any], lis
                     )
                     continue
 
-                # S16.1: Strict Host Allowlist (Exact Match)
-                # Deny by default unless host is explicitly allowed.
-                # NOTE: built-in provider public hosts are allowlisted by default.
-                allowed_hosts_str = os.environ.get(
-                    "OPENCLAW_LLM_ALLOWED_HOSTS"
-                ) or os.environ.get("MOLTBOT_LLM_ALLOWED_HOSTS", "")
-                allowed_hosts_env = set(
-                    h.lower().strip() for h in allowed_hosts_str.split(",") if h.strip()
-                )
+                controls = get_llm_egress_controls(provider_key, val)
+
+                # Keep local providers strict: only loopback endpoints are acceptable.
+                if is_local_provider(provider_key) and not controls.get(
+                    "allow_loopback_hosts"
+                ):
+                    errors.append(
+                        f"Local provider {provider_key} must use localhost URL"
+                    )
+                    continue
+
                 try:
-                    from .providers.catalog import get_default_public_llm_hosts
                     from .safe_io import STANDARD_OUTBOUND_POLICY
 
-                    allowed_hosts = (
-                        set(get_default_public_llm_hosts()) | allowed_hosts_env
-                    )
-                except Exception:
-                    allowed_hosts = allowed_hosts_env
-
-                # Check opt-in for "Any Public Host" (risky, for flexibility)
-                allow_any = _env_flag(
-                    "OPENCLAW_ALLOW_ANY_PUBLIC_LLM_HOST",
-                    "MOLTBOT_ALLOW_ANY_PUBLIC_LLM_HOST",
-                    default=False,
-                )
-
-                try:
                     validate_outbound_url(
                         val,
-                        allow_hosts=allowed_hosts if not allow_any else None,
-                        allow_any_public_host=allow_any,
+                        allow_hosts=controls.get("allow_hosts"),
+                        allow_any_public_host=bool(
+                            controls.get("allow_any_public_host")
+                        ),
+                        allow_loopback_hosts=controls.get("allow_loopback_hosts"),
                         policy=STANDARD_OUTBOUND_POLICY,
                     )
                 except SSRFError as e:

services/safe_io.py

@@ -298,6 +298,7 @@ def validate_outbound_url(
     *,
     allow_hosts: Optional[Set[str]] = None,
     allow_any_public_host: bool = False,
+    allow_loopback_hosts: Optional[Set[str]] = None,
     policy: Optional[OutboundPolicy] = None,
 ) -> Tuple[str, str, int, list[str]]:
     """
@@ -307,6 +308,8 @@ def validate_outbound_url(
         url: URL to validate.
         allow_hosts: If provided, only these hosts are allowed.
         allow_any_public_host: If True, allow any host that resolves to a public IP.
+        allow_loopback_hosts: Optional host allowlist for controlled loopback-only
+            exceptions. This does not allow general private networks.
         policy: S51 OutboundPolicy for scheme+port enforcement.
 
     Returns:
@@ -346,6 +349,9 @@ def validate_outbound_url(
 
     # Normalize host
     normalized_host = _normalize_host(host)
+    normalized_loopback_allowlist = {
+        _normalize_host(h) for h in (allow_loopback_hosts or set())
+    }
 
     # Check allowlist if provided or enforced
     if not allow_any_public_host:
@@ -365,6 +371,21 @@ def validate_outbound_url(
         for _, _, _, _, sockaddr in addr_infos:
             ip = sockaddr[0]
             if is_private_ip(ip):
+                # CRITICAL:
+                # Only allow loopback IPs when the target host is explicitly listed in
+                # allow_loopback_hosts. Never relax this into blanket private-IP allow.
+                try:
+                    ip_obj = ipaddress.ip_address(ip)
+                except ValueError:
+                    ip_obj = None
+                if (
+                    ip_obj is not None
+                    and ip_obj.is_loopback
+                    and normalized_host in normalized_loopback_allowlist
+                ):
+                    if ip not in resolved_ips:
+                        resolved_ips.append(ip)
+                    continue
                 raise SSRFError(f"Private/reserved IP blocked: {ip}")
             if ip not in resolved_ips:
                 resolved_ips.append(ip)
@@ -518,9 +539,11 @@ def safe_fetch(
 def safe_request_json(
     method: str,
     url: str,
-    json_body: Any,
+    json_body: Any = None,
     *,
     allow_hosts: Optional[Set[str]] = None,
+    allow_any_public_host: bool = False,
+    allow_loopback_hosts: Optional[Set[str]] = None,
     headers: Optional[dict] = None,
     timeout_sec: int = 10,
     max_response_bytes: int = 1_000_000,
@@ -541,8 +564,15 @@ def safe_request_json(
 
     while True:
         # Validate URL + Pin IPs
+        # IMPORTANT:
+        # Keep these controls aligned with any caller pre-validation. Divergence
+        # between pre-check and request-time check caused S65 regressions.
         scheme, host, port, pinned_ips = validate_outbound_url(
-            current_url, allow_hosts=allow_hosts, policy=policy
+            current_url,
+            allow_hosts=allow_hosts,
+            allow_any_public_host=allow_any_public_host,
+            allow_loopback_hosts=allow_loopback_hosts,
+            policy=policy,
         )
 
         # Build request

tests/test_api_model_list.py

@@ -63,9 +63,47 @@ class TestModelListAPI(unittest.IsolatedAsyncioTestCase):
 
         # Verify safe_request_json was called (S65 compliance)
         mock_safe_request.assert_called_once()
-        call_kwargs = mock_safe_request.call_args
+        call_kwargs = mock_safe_request.call_args.kwargs
         # Should have passed allow_hosts (not allow_any)
-        self.assertIn("allow_hosts", call_kwargs.kwargs)
+        self.assertIn("allow_hosts", call_kwargs)
+        self.assertFalse(call_kwargs.get("allow_any_public_host", False))
+        self.assertIsNone(call_kwargs.get("json_body"))
+
+    @patch("api.config.get_effective_config")
+    @patch("services.providers.keys.get_api_key_for_provider")
+    @patch("services.providers.keys.requires_api_key")
+    @patch("api.config.check_rate_limit")
+    @patch("api.config.require_admin_token")
+    @patch("services.safe_io.safe_request_json")
+    async def test_handler_local_provider_no_api_key_allowed(
+        self,
+        mock_safe_request,
+        mock_require_admin,
+        mock_rate_limit,
+        mock_requires_key,
+        mock_get_key,
+        mock_get_config,
+    ):
+        """Local providers should not be blocked by API-key gate."""
+        mock_rate_limit.return_value = True
+        mock_require_admin.return_value = (True, None)
+        mock_get_config.return_value = (
+            {"provider": "ollama", "base_url": "http://127.0.0.1:11434"},
+            {},
+        )
+        mock_requires_key.return_value = False
+        mock_get_key.return_value = None
+        mock_safe_request.return_value = {"models": [{"id": "gemma3:4b"}]}
+
+        request = MagicMock()
+        request.query = {}
+        request.remote = "127.0.0.1"
+
+        resp = await llm_models_handler(request)
+        self.assertEqual(resp.status, 200)
+        data = json.loads(resp.body)
+        self.assertTrue(data["ok"])
+        self.assertIn("gemma3:4b", data["models"])
 
     @patch("api.config.get_effective_config")
     @patch("services.providers.keys.get_api_key_for_provider")
@@ -191,6 +229,48 @@ class TestModelListAPI(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(resp.status, 200)
         # Check cache key uses custom url
         self.assertIn(("custom", "http://custom-host:8080/v1"), _MODEL_LIST_CACHE)
+        kwargs = mock_safe_request.call_args.kwargs
+        self.assertFalse(kwargs.get("allow_any_public_host", False))
+        self.assertIsNotNone(kwargs.get("allow_hosts"))
+
+    @patch("api.config.get_effective_config")
+    @patch("services.providers.keys.get_api_key_for_provider")
+    @patch("api.config.check_rate_limit")
+    @patch("api.config.require_admin_token")
+    @patch("services.safe_io.safe_request_json")
+    @patch("services.safe_io.validate_outbound_url")
+    async def test_handler_allow_any_public_host_controls_are_consistent(
+        self,
+        mock_validate_url,
+        mock_safe_request,
+        mock_require_admin,
+        mock_rate_limit,
+        mock_get_key,
+        mock_get_config,
+    ):
+        mock_rate_limit.return_value = True
+        mock_require_admin.return_value = (True, None)
+        mock_get_config.return_value = (
+            {"provider": "custom", "base_url": "https://example.com/v1"},
+            {},
+        )
+        mock_get_key.return_value = "sk-custom"
+        mock_safe_request.return_value = {"models": [{"id": "x"}]}
+
+        request = MagicMock()
+        request.query = {}
+        request.remote = "127.0.0.1"
+
+        with patch.dict("os.environ", {"OPENCLAW_ALLOW_ANY_PUBLIC_LLM_HOST": "1"}):
+            resp = await llm_models_handler(request)
+
+        self.assertEqual(resp.status, 200)
+        validate_kwargs = mock_validate_url.call_args.kwargs
+        request_kwargs = mock_safe_request.call_args.kwargs
+        self.assertTrue(validate_kwargs["allow_any_public_host"])
+        self.assertTrue(request_kwargs["allow_any_public_host"])
+        self.assertIsNone(validate_kwargs["allow_hosts"])
+        self.assertIsNone(request_kwargs["allow_hosts"])
 
     @patch("api.config.get_effective_config")
     @patch("services.providers.keys.get_api_key_for_provider")

tests/test_providers.py

@@ -16,7 +16,10 @@ from services.providers.catalog import (
     PROVIDER_CATALOG,
     ProviderInfo,
     ProviderType,
+    get_loopback_host_aliases,
     get_provider_info,
+    is_local_provider,
+    is_loopback_host,
     list_providers,
 )
 from services.providers.keys import (
@@ -86,6 +89,21 @@ class TestProviderCatalog(unittest.TestCase):
         for provider in PROVIDER_CATALOG.keys():
             self.assertIn(provider, DEFAULT_MODEL_BY_PROVIDER)
 
+    def test_local_provider_detection(self):
+        """Local providers should be identified explicitly."""
+        self.assertTrue(is_local_provider("ollama"))
+        self.assertTrue(is_local_provider("lmstudio"))
+        self.assertFalse(is_local_provider("openai"))
+
+    def test_loopback_helpers(self):
+        self.assertTrue(is_loopback_host("localhost"))
+        self.assertTrue(is_loopback_host("127.0.0.1"))
+        self.assertFalse(is_loopback_host("api.openai.com"))
+        self.assertEqual(
+            get_loopback_host_aliases("localhost"),
+            {"localhost", "127.0.0.1", "::1"},
+        )
+
 
 class TestProviderKeys(unittest.TestCase):
 

tests/test_s65_egress_policy_parity.py

@@ -33,6 +33,10 @@ class TestS65EgressPolicyParity(unittest.TestCase):
 
             self.assertEqual(result["text"], "Hello")
             mock_safe.assert_called_once()
+            kwargs = mock_safe.call_args.kwargs
+            self.assertIn("allow_hosts", kwargs)
+            self.assertIn("allow_any_public_host", kwargs)
+            self.assertIn("allow_loopback_hosts", kwargs)
 
     def test_openai_compat_uses_safe_io(self):
         with patch("services.providers.openai_compat.safe_request_json") as mock_safe:
@@ -50,6 +54,10 @@ class TestS65EgressPolicyParity(unittest.TestCase):
 
             self.assertEqual(result["text"], "Hello")
             mock_safe.assert_called_once()
+            kwargs = mock_safe.call_args.kwargs
+            self.assertIn("allow_hosts", kwargs)
+            self.assertIn("allow_any_public_host", kwargs)
+            self.assertIn("allow_loopback_hosts", kwargs)
 
 
 class TestS65ModelListEgressConvergence(unittest.TestCase):

tests/test_safe_io.py

@@ -15,6 +15,7 @@ from services.safe_io import (
     resolve_under_root,
     safe_read_bytes,
     safe_read_text,
+    safe_request_json,
     safe_write_text,
     validate_outbound_url,
 )
@@ -179,6 +180,53 @@ class TestURLSafety(unittest.TestCase):
             validate_outbound_url("https://example.com", allow_hosts={"example.com"})
         self.assertIn("Private/reserved IP", str(ctx.exception))
 
+    @patch("socket.getaddrinfo")
+    def test_allow_loopback_private_ip_with_explicit_host_gate(self, mock_dns):
+        """Loopback may be allowed only with explicit allow_loopback_hosts host gate."""
+        mock_dns.return_value = [(2, 1, 6, "", ("127.0.0.1", 443))]
+
+        result = validate_outbound_url(
+            "https://localhost",
+            allow_hosts={"localhost"},
+            allow_loopback_hosts={"localhost"},
+        )
+        self.assertEqual(result, ("https", "localhost", 443, ["127.0.0.1"]))
+
+    @patch("socket.getaddrinfo")
+    def test_loopback_allowlist_does_not_allow_other_private_ranges(self, mock_dns):
+        """Loopback exception must not allow non-loopback private IPs."""
+        mock_dns.return_value = [(2, 1, 6, "", ("192.168.1.9", 443))]
+
+        with self.assertRaises(SSRFError) as ctx:
+            validate_outbound_url(
+                "https://localhost",
+                allow_hosts={"localhost"},
+                allow_loopback_hosts={"localhost"},
+            )
+        self.assertIn("Private/reserved IP", str(ctx.exception))
+
+    @patch("services.safe_io._build_pinned_opener")
+    @patch("services.safe_io.validate_outbound_url")
+    def test_safe_request_json_get_without_body(self, mock_validate, mock_build):
+        """GET requests should work when json_body is omitted/None."""
+        mock_validate.return_value = ("https", "example.com", 443, ["93.184.216.34"])
+
+        mock_response = MagicMock()
+        mock_response.getcode.return_value = 200
+        mock_response.read.return_value = b'{"ok": true}'
+
+        mock_opener = MagicMock()
+        mock_opener.open.return_value.__enter__.return_value = mock_response
+        mock_build.return_value = mock_opener
+
+        out = safe_request_json(
+            method="GET",
+            url="https://example.com/models",
+            json_body=None,
+            allow_hosts={"example.com"},
+        )
+        self.assertEqual(out["ok"], True)
+
     def test_host_normalization_case(self):
         """Test host normalization is case-insensitive."""
         self.assertEqual(_normalize_host("Example.COM"), "example.com")

tests/test_ssrf_policy.py

@@ -6,7 +6,7 @@ import os
 import unittest
 from unittest.mock import Mock, patch
 
-from services.runtime_config import validate_config_update
+from services.runtime_config import get_llm_egress_controls, validate_config_update
 
 
 class TestSSRFPolicy(unittest.TestCase):
@@ -85,6 +85,17 @@ class TestSSRFPolicy(unittest.TestCase):
             sanitized, errors = validate_config_update(updates)
             self.assertEqual(errors, [])
 
+    def test_egress_controls_local_provider_loopback_only(self):
+        controls = get_llm_egress_controls("ollama", "http://127.0.0.1:11434")
+        self.assertIsNotNone(controls["allow_loopback_hosts"])
+        self.assertIn("127.0.0.1", controls["allow_loopback_hosts"])
+        self.assertIn("localhost", controls["allow_hosts"])
+
+    def test_egress_controls_custom_provider_does_not_get_loopback_exception(self):
+        controls = get_llm_egress_controls("custom", "http://127.0.0.1:11434")
+        self.assertIsNone(controls["allow_loopback_hosts"])
+        self.assertNotIn("127.0.0.1", controls["allow_hosts"])
+
 
 if __name__ == "__main__":
     unittest.main()