[OWASP ASI02] Block SSRF in recon-tool URL arguments and gateway base URL

backend/agent/tools.py

@@ -12,6 +12,7 @@
 
 from agent.tool_hooks import run_tool_with_hooks
 from app.services.http_client import recon_client
+from app.url_guard import is_safe_public_target
 
 # Filled by the operation runner so tool outputs feed the live findings panel.
 _findings_bucket: ContextVar[list[dict] | None] = ContextVar("vecna_findings", default=None)
@@ -138,6 +139,9 @@ def resolve_dns_and_subdomains(target_url: str) -> str:
 
 
 def _analyze_headers_impl(target_url: str) -> str:
+    ok, reason = is_safe_public_target(target_url)
+    if not ok:
+        return json.dumps({"error": f"URL blocked by SSRF guard: {reason}", "findings": []})
     base = _normalize_base(target_url)
     findings: list[dict[str, str]] = []
     with recon_client(read_timeout=25.0) as client:
@@ -214,6 +218,9 @@ def analyze_http_security_headers(target_url: str) -> str:
 
 
 def _probe_paths_impl(target_url: str) -> str:
+    ok, reason = is_safe_public_target(target_url)
+    if not ok:
+        return json.dumps({"error": f"URL blocked by SSRF guard: {reason}", "paths": [], "findings": []})
     base = _normalize_base(target_url).rstrip("/")
     results: list[dict] = []
     findings: list[dict[str, str]] = []

backend/agent/vecna_agent.py

@@ -5,6 +5,28 @@
 import litellm
 from strands import Agent
 from app.config import Settings
+from app.url_guard import is_safe_public_target
+
+_OPENROUTER_API_BASE_DEFAULT = "https://openrouter.ai/api/v1"
+
+
+def _validated_gateway_base() -> str:
+    """Return the LLM gateway base URL after SSRF validation.
+
+    Reads OPENROUTER_API_BASE from the environment (falling back to the
+    official OpenRouter endpoint) and rejects any value that resolves to a
+    private/internal address — preventing server-side request forgery via
+    environment-variable injection.
+    """
+    url = (os.environ.get("OPENROUTER_API_BASE") or "").strip() or _OPENROUTER_API_BASE_DEFAULT
+    ok, reason = is_safe_public_target(url)
+    if not ok:
+        raise RuntimeError(
+            f"Gateway URL OPENROUTER_API_BASE={url!r} blocked by SSRF guard: {reason}. "
+            "Only public HTTPS endpoints are permitted for the LLM gateway."
+        )
+    return url
+
 
 SYSTEM_PROMPT = """You are Vecna Ops, a passive security reconnaissance agent.
 
@@ -45,7 +67,7 @@ def _client_args_for_model(s: Settings) -> dict | None:
             return None
         # Force OpenRouter routing: LiteLLM 1.8x can mis-detect openrouter/... and then use the
         # OpenAI client + OPENAI_API_KEY. See get_llm_provider openrouter branch.
-        base = os.environ.get("OPENROUTER_API_BASE", "https://openrouter.ai/api/v1")
+        base = _validated_gateway_base()
         return {
             "api_key": key,
             "api_base": base,

backend/app/url_guard.py

@@ -8,9 +8,28 @@
         "0.0.0.0",
         "metadata.google.internal",
         "metadata",
+        # AWS / GCP / Azure instance metadata endpoints
+        "169.254.169.254",
+        "fd00:ec2::254",
     }
 )
 
+# Matches hostnames that *begin* with a private/reserved IPv4 prefix, e.g.
+# "10.internal.host" or "172.20.proxy.local".  Covers all RFC 1918 ranges:
+#   10.0.0.0/8      → 10.
+#   172.16.0.0/12   → 172.16–172.31 (second octet 16–31)
+#   192.168.0.0/16  → 192.168.
+# Plus loopback (127.0.0.0/8) and link-local (169.254.0.0/16).
+_PRIVATE_PREFIX_RE = re.compile(
+    r"^(?:"
+    r"10\."
+    r"|127\."
+    r"|169\.254\."
+    r"|192\.168\."
+    r"|172\.(?:1[6-9]|2\d|3[01])\."
+    r")"
+)
+
 
 def is_safe_public_target(url: str) -> tuple[bool, str]:
     """
@@ -52,9 +71,10 @@ def is_safe_public_target(url: str) -> tuple[bool, str]:
         except ValueError:
             pass
 
-    # Block common private IPv4 prefixes by label (hostname won't match, but catch dotted names)
-    for prefix in ("10.", "192.168.", "127.", "169.254.", "172.16.", "172.17.", "172.18.", "172.19."):
-        if host.startswith(prefix):
-            return False, "Host not allowed"
+    # Block hostnames whose prefix matches a private/reserved IPv4 range
+    # (e.g. "172.20.proxy.local" or "10.internal.svc").  _PRIVATE_PREFIX_RE
+    # covers the full RFC 1918 space including 172.16–172.31.
+    if _PRIVATE_PREFIX_RE.match(host):
+        return False, "Host not allowed"
 
     return True, ""

backend/tests/test_url_guard.py

@@ -0,0 +1,137 @@
+"""Tests for SSRF guard — url_guard.is_safe_public_target.
+
+Verifies that all RFC 1918 private ranges, loopback, link-local, metadata
+endpoints, and the previously-missing 172.20–172.31 block are blocked.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from app.url_guard import is_safe_public_target
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _blocked(url: str) -> bool:
+    ok, _ = is_safe_public_target(url)
+    return not ok
+
+
+def _allowed(url: str) -> bool:
+    ok, _ = is_safe_public_target(url)
+    return ok
+
+
+# ---------------------------------------------------------------------------
+# Should be BLOCKED
+# ---------------------------------------------------------------------------
+
+class TestBlocked:
+    def test_localhost_name(self):
+        assert _blocked("http://localhost/")
+
+    def test_localhost_name_https(self):
+        assert _blocked("https://localhost/path")
+
+    def test_loopback_ipv4(self):
+        assert _blocked("http://127.0.0.1/")
+
+    def test_loopback_ipv4_other(self):
+        assert _blocked("http://127.100.200.1/")
+
+    def test_link_local_aws_metadata(self):
+        # Classic AWS / many clouds instance-metadata endpoint
+        assert _blocked("http://169.254.169.254/latest/meta-data/")
+
+    def test_link_local_arbitrary(self):
+        assert _blocked("http://169.254.10.5/internal")
+
+    def test_rfc1918_10_block(self):
+        assert _blocked("http://10.0.0.1/")
+
+    def test_rfc1918_10_block_deep(self):
+        assert _blocked("http://10.255.255.255/")
+
+    def test_rfc1918_192_168(self):
+        assert _blocked("http://192.168.1.1/admin")
+
+    def test_rfc1918_172_16(self):
+        assert _blocked("http://172.16.0.1/")
+
+    def test_rfc1918_172_19(self):
+        # Was covered before the fix; must still be blocked
+        assert _blocked("http://172.19.255.1/")
+
+    # --- Previously-missing 172.20–172.31 range (the core fix) ---
+
+    def test_rfc1918_172_20(self):
+        assert _blocked("http://172.20.0.1/")
+
+    def test_rfc1918_172_25(self):
+        assert _blocked("http://172.25.10.5/api")
+
+    def test_rfc1918_172_31(self):
+        assert _blocked("http://172.31.255.255/")
+
+    def test_rfc1918_172_20_hostname_prefix(self):
+        # Hostname whose dotted prefix looks like the private range
+        assert _blocked("http://172.20.proxy.local/")
+
+    def test_rfc1918_172_28_hostname_prefix(self):
+        assert _blocked("http://172.28.internal.corp/service")
+
+    def test_zero_zero(self):
+        assert _blocked("http://0.0.0.0/")
+
+    def test_metadata_google(self):
+        assert _blocked("http://metadata.google.internal/computeMetadata/v1/")
+
+    def test_metadata_aws_literal(self):
+        # 169.254.169.254 is in _BLOCKED_HOSTNAMES explicitly
+        assert _blocked("http://169.254.169.254/")
+
+    def test_dotlocalhost_subdomain(self):
+        assert _blocked("http://evil.localhost/")
+
+    def test_no_scheme_bare_ip(self):
+        # Scheme is added automatically; still blocked
+        assert _blocked("127.0.0.1")
+
+    def test_missing_host(self):
+        ok, reason = is_safe_public_target("http:///path")
+        assert not ok
+        assert reason
+
+
+# ---------------------------------------------------------------------------
+# Should be ALLOWED
+# ---------------------------------------------------------------------------
+
+class TestAllowed:
+    def test_public_https(self):
+        assert _allowed("https://example.com/")
+
+    def test_public_http(self):
+        assert _allowed("http://example.com/path?q=1")
+
+    def test_public_ip_cloudflare(self):
+        # 1.1.1.1 is a genuine public IP (Cloudflare DNS)
+        assert _allowed("https://1.1.1.1/")
+
+    def test_public_ip_8_8_8_8(self):
+        assert _allowed("https://8.8.8.8/")
+
+    def test_172_15_is_public(self):
+        # 172.15.x.x is NOT in the RFC 1918 range (172.16.0.0/12 starts at 172.16)
+        assert _allowed("https://172.15.0.1/")
+
+    def test_172_32_is_public(self):
+        # 172.32.x.x is NOT in the RFC 1918 range (172.16.0.0/12 ends at 172.31)
+        assert _allowed("https://172.32.0.1/")
+
+    def test_no_scheme_public(self):
+        # Scheme is added automatically; public host passes
+        assert _allowed("example.com")