diff --git a/backend/agent/tools.py b/backend/agent/tools.py index 5e94f32..5b3cd55 100644 --- a/backend/agent/tools.py +++ b/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]] = [] diff --git a/backend/agent/vecna_agent.py b/backend/agent/vecna_agent.py index b61f3d9..0d345a4 100644 --- a/backend/agent/vecna_agent.py +++ b/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, diff --git a/backend/app/url_guard.py b/backend/app/url_guard.py index a8e6ace..a1509bf 100644 --- a/backend/app/url_guard.py +++ b/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, "" diff --git a/backend/tests/test_url_guard.py b/backend/tests/test_url_guard.py new file mode 100644 index 0000000..62076c7 --- /dev/null +++ b/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")