From 6f7aa95067fb88012db59b0a808710067625251e Mon Sep 17 00:00:00 2001 From: Naman Singh Date: Sun, 2 Aug 2026 12:03:32 +0530 Subject: [PATCH 1/4] fix(crawler): re-validate redirect hops and strip credentials cross-origin Fixes #2369 crawl_target() followed redirects via httpx's automatic handling with no re-validation of any hop, and forwarded operator credentials (including the decrypted vault Authorization header injected by the executor) and cookies to every redirect destination. Two defenses are added in backend/secuscan/crawler.py: - Redirect hops are re-validated against the network policy engine before being fetched, so a hostile or compromised seed cannot pivot the crawler into cloud-metadata, loopback, private/CGNAT, or IPv6 link-local/ULA ranges (SSRF). - Credentials are only sent to the seed origin; they are stripped on any cross-origin redirect, so vault credentials cannot be exfiltrated to an attacker-controlled host (browser-equivalent behavior). Redirects are now followed manually (follow_redirects=False) to keep the existing max_redirects and max_size constraints intact. --- backend/secuscan/crawler.py | 221 ++++++++++++--- .../unit/test_crawler_redirect_security.py | 267 ++++++++++++++++++ 2 files changed, 446 insertions(+), 42 deletions(-) create mode 100644 testing/backend/unit/test_crawler_redirect_security.py diff --git a/backend/secuscan/crawler.py b/backend/secuscan/crawler.py index f7a21dcf8..66cfd198f 100644 --- a/backend/secuscan/crawler.py +++ b/backend/secuscan/crawler.py @@ -3,14 +3,23 @@ from __future__ import annotations from html.parser import HTMLParser +import asyncio +import logging import re -from typing import Any, Dict, List +from typing import Any, Dict, List, Tuple from urllib.parse import parse_qsl, urljoin, urlparse import httpx from .config import settings +logger = logging.getLogger(__name__) + +# HTTP statuses that trigger redirect following. httpx also follows 303/307/308 +# alongside 301/302; we enumerate them explicitly because redirects are handled +# manually so every hop can be re-validated. +_REDIRECT_STATUSES = {301, 302, 303, 307, 308} + class _SurfaceParser(HTMLParser): def __init__(self) -> None: @@ -74,6 +83,60 @@ def _build_headers(extra_headers: Dict[str, Any] | None = None) -> Dict[str, str return headers +def _is_same_origin(a: Any, b: Any) -> bool: + """Return True when two parsed URLs share scheme, host, and effective port. + + Mirrors the same-origin definition browsers use for credential handling: a + redirect to a different scheme, host, or port is a new origin and must not + inherit the seed's credentials. + """ + try: + scheme_a = (a.scheme or "").lower() + scheme_b = (b.scheme or "").lower() + host_a = (a.hostname or "").lower() + host_b = (b.hostname or "").lower() + port_a = a.port if a.port is not None else (443 if scheme_a == "https" else 80) + port_b = b.port if b.port is not None else (443 if scheme_b == "https" else 80) + except ValueError: + # Malformed port on either side: treat as cross-origin so credentials + # are never forwarded. + return False + return scheme_a == scheme_b and host_a == host_b and port_a == port_b + + +def _validate_redirect_target(url: str) -> Tuple[bool, str]: + """Re-validate a redirect destination against the network policy. + + The seed target is validated by the executor before the scanner runs, but + httpx-followed redirects are not. Without this check a hostile or + compromised seed can pivot the crawler into cloud-metadata, loopback, + private/CGNAT, or IPv6 link-local/ULA ranges that were never authorized. + """ + parsed = urlparse(url) + if parsed.scheme not in {"http", "https"}: + return False, f"redirect to unsupported scheme '{parsed.scheme}'" + hostname = parsed.hostname + if not hostname: + return False, "redirect target has no hostname" + + from .network_policy import get_policy_engine + + try: + engine = get_policy_engine() + allowed, reason, _ = engine.check_access( + dest_ip=hostname, + dest_hostname=hostname, + plugin_id="crawler", + task_id="crawler-redirect", + ) + except Exception as exc: + # Never fail open: if the policy engine cannot evaluate the target, + # block the redirect instead of silently fetching an internal host. + logger.warning("Redirect target validation failed for %s: %s", url, exc) + return False, "redirect target could not be validated" + return allowed, reason + + async def crawl_target( url: str, *, @@ -83,42 +146,125 @@ async def crawl_target( max_redirects: int = 10, max_size: int = 5 * 1024 * 1024, ) -> Dict[str, Any]: - """Fetch a target and normalize discovered links/forms/scripts/API hints.""" - headers = _build_headers(extra_headers) + """Fetch a target and normalize discovered links/forms/scripts/API hints. + + Redirects are followed manually rather than by httpx's automatic handling + so every hop is re-validated: + + - Each redirect destination is checked against the network policy before + it is fetched, closing the SSRF gap where a hostile seed pivots the + crawler into internal or metadata-only networks that were never + authorized (active when ``enforce_network_policy`` is enabled, matching + how the executor validates the seed target). + - Credentials supplied via ``extra_headers``/``cookies`` are only sent to + the seed origin; they are stripped on any cross-origin redirect so vault + credentials cannot be exfiltrated to an attacker-controlled host. + """ + base_headers = _build_headers(None) + seed_origin = urlparse(url) + redirect_chain: List[Dict[str, Any]] = [] + current_url = url + redirects_followed = 0 + final_response: httpx.Response | None = None + body_bytes = b"" + response_headers: Dict[str, str] = {} + set_cookie_headers: List[str] = [] + forward_credentials = True + async with httpx.AsyncClient( - follow_redirects=True, - max_redirects=max_redirects, + follow_redirects=False, timeout=timeout, - headers=headers, - cookies=cookies or {}, + headers=base_headers, verify=settings.verify_ssl, ) as client: - async with client.stream("GET", url) as response: - # Check Content-Length header if present - content_length = response.headers.get("content-length") - if content_length: - try: - cl_val = int(content_length) - except ValueError: - cl_val = 0 - if cl_val > max_size: - raise ValueError(f"Response size exceeds limit of {max_size} bytes") - - # Read response in chunks to enforce size limit - body_chunks = [] - bytes_read = 0 - async for chunk in response.aiter_bytes(): - bytes_read += len(chunk) - if bytes_read > max_size: - raise ValueError(f"Response size exceeds limit of {max_size} bytes") - body_chunks.append(chunk) - - body_bytes = b"".join(body_chunks) - body = body_bytes.decode("utf-8", errors="replace") + for _ in range(max_redirects + 1): + if not _is_same_origin(seed_origin, urlparse(current_url)): + forward_credentials = False + + hop_headers = dict(base_headers) + if forward_credentials and extra_headers: + for key, value in extra_headers.items(): + if key and value is not None: + hop_headers[str(key)] = str(value) + hop_cookies = dict(cookies or {}) if forward_credentials else {} + + if redirects_followed > 0 and settings.enforce_network_policy: + allowed, reason = await asyncio.to_thread( + _validate_redirect_target, current_url + ) + if not allowed: + raise ValueError( + f"Redirect to {current_url} rejected by network policy: {reason}" + ) + + async with client.stream( + "GET", current_url, headers=hop_headers, cookies=hop_cookies + ) as response: + # Check Content-Length header if present + content_length = response.headers.get("content-length") + if content_length: + try: + cl_val = int(content_length) + except ValueError: + cl_val = 0 + if cl_val > max_size: + raise ValueError(f"Response size exceeds limit of {max_size} bytes") + + # Read response in chunks to enforce size limit + body_chunks: List[bytes] = [] + bytes_read = 0 + async for chunk in response.aiter_bytes(): + bytes_read += len(chunk) + if bytes_read > max_size: + raise ValueError(f"Response size exceeds limit of {max_size} bytes") + body_chunks.append(chunk) + hop_body = b"".join(body_chunks) + + status = response.status_code + location = response.headers.get("location") + if status in _REDIRECT_STATUSES and location: + redirect_chain.append( + { + "url": str(response.url), + "status_code": status, + "location": location, + } + ) + if redirects_followed >= max_redirects: + raise httpx.TooManyRedirects( + f"Exceeded maximum redirects ({max_redirects})", + request=response.request, + ) + redirects_followed += 1 + current_url = urljoin(str(response.url), location) + continue + + final_response = response + body_bytes = hop_body + response_headers = dict(response.headers) + set_cookie_headers = ( + list(response.headers.get_list("set-cookie")) + if hasattr(response.headers, "get_list") + else [] + ) + break + else: + raise httpx.TooManyRedirects( + f"Exceeded maximum redirects ({max_redirects})", + request=None, + ) + + if final_response is None: + raise httpx.TooManyRedirects( + f"Exceeded maximum redirects ({max_redirects})", + request=None, + ) + + body = body_bytes.decode("utf-8", errors="replace") parser = _SurfaceParser() parser.feed(body) - base_url = str(response.url) + base_url = str(final_response.url) final_parsed = urlparse(base_url) normalized_links = sorted({urljoin(base_url, link) for link in parser.links if link}) normalized_scripts = sorted({urljoin(base_url, script) for script in parser.scripts if script}) @@ -139,30 +285,21 @@ async def crawl_target( path_hints.append({"url": candidate, "kind": path_tag}) forms = [_normalize_form(base_url, form) for form in parser.forms[:50]] - headers_snapshot = dict(response.headers) - set_cookie_headers = list(response.headers.get_list("set-cookie")) if hasattr(response.headers, "get_list") else [] + headers_snapshot = response_headers tech_hints = _extract_tech_hints(headers_snapshot, parser.meta_generators, normalized_scripts, body) cms_hints = _extract_cms_hints(parser.meta_generators, body, normalized_scripts) - redirect_chain = [ - { - "url": str(item.url), - "status_code": item.status_code, - "location": item.headers.get("location"), - } - for item in response.history - ] return { "seed_url": url, "final_url": base_url, - "status_code": response.status_code, + "status_code": final_response.status_code, "scheme": final_parsed.scheme, "headers": headers_snapshot, "set_cookie_headers": set_cookie_headers[:20], "redirect_chain": redirect_chain[:10], "tech_hints": tech_hints[:20], "cms_hints": cms_hints[:10], - "pages": [{"url": base_url, "title": _extract_title(body), "content_type": response.headers.get("content-type", "")}] + [ + "pages": [{"url": base_url, "title": _extract_title(body), "content_type": response_headers.get("content-type", "")}] + [ {"url": link, "title": "", "content_type": ""} for link in normalized_links[:100] ], "forms": forms, diff --git a/testing/backend/unit/test_crawler_redirect_security.py b/testing/backend/unit/test_crawler_redirect_security.py new file mode 100644 index 000000000..42f4bdd49 --- /dev/null +++ b/testing/backend/unit/test_crawler_redirect_security.py @@ -0,0 +1,267 @@ +"""Unit tests for crawler redirect hardening. + +Covers the two halves of the redirect SSRF / credential-exfiltration fix in +``backend/secuscan/crawler.py``: + +1. Redirect hops are re-validated against the network policy before they are + fetched, so a hostile seed cannot pivot the crawler into cloud-metadata, + loopback, or private ranges (SSRF). +2. Credentials (``extra_headers`` / ``cookies``) are only sent to the seed + origin and are stripped on any cross-origin redirect, so vault credentials + cannot be leaked to an attacker-controlled host. + +Related issue: https://github.com/utksh1/SecuScan/issues/2369 +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock, patch +from urllib.parse import urlparse + +import httpx +import pytest + +from backend.secuscan.crawler import ( + _is_same_origin, + _validate_redirect_target, + crawl_target, +) + + +class _FakePolicyEngine: + """Minimal stand-in for NetworkPolicyEngine. + + Denies the cloud-metadata / link-local range (169.254.0.0/16) like the + real mandatory denylist, allows everything else, and records calls so the + tests can assert the crawler passes the redirect host for validation. + """ + + def __init__(self, *, raise_on_check: bool = False): + self.calls = [] + self.raise_on_check = raise_on_check + + def check_access(self, dest_ip, dest_port=0, plugin_id="unknown", task_id="unknown", dest_hostname=None): + self.calls.append({"dest_ip": dest_ip, "dest_hostname": dest_hostname}) + if self.raise_on_check: + raise RuntimeError("policy engine exploded") + try: + import ipaddress + ip = ipaddress.ip_address(dest_ip) + if ip in ipaddress.ip_network("169.254.0.0/16"): + return False, "Blocked by mandatory denylist (matched: 169.254.0.0/16)", None + except ValueError: + pass + return True, "Allowed", None + + +def _make_response(status_code, url, *, headers=None, body=b""): + resp = MagicMock() + resp.status_code = status_code + resp.url = url + resp.headers = dict(headers or {}) + resp.history = [] + + async def _aiter_bytes(): + yield body + + resp.aiter_bytes = _aiter_bytes + resp.request = httpx.Request("GET", url) + return resp + + +def _stream_context(response): + ctx = MagicMock() + ctx.__aenter__ = AsyncMock(return_value=response) + ctx.__aexit__ = AsyncMock(return_value=None) + return ctx + + +def _make_client(*responses): + """Build a mocked AsyncClient that yields one streamed response per call.""" + client = MagicMock() + client.__aenter__ = AsyncMock(return_value=client) + client.__aexit__ = AsyncMock(return_value=None) + client.stream.side_effect = [_stream_context(resp) for resp in responses] + return client + + +# --------------------------------------------------------------------------- +# _is_same_origin +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "origin,other,expected", + [ + ("https://example.com/a", "https://example.com/b", True), + ("http://example.com:80/x", "http://example.com/y", True), + ("https://example.com:443/x", "https://example.com/y", True), + ("https://example.com/x", "https://other.com/y", False), + ("http://example.com/x", "https://example.com/y", False), + ("https://example.com:8080/x", "https://example.com/y", False), + ("https://example.com/x", "https://EXAMPLE.com/y", True), + ], +) +def test_is_same_origin(origin, other, expected): + assert _is_same_origin(urlparse(origin), urlparse(other)) is expected + + +def test_is_same_origin_malformed_port_is_cross_origin(): + # A malformed port on the redirect target must never be treated as the + # same origin, so credentials are dropped. + assert _is_same_origin( + urlparse("https://example.com/"), urlparse("https://example.com:bad/") + ) is False + + +# --------------------------------------------------------------------------- +# _validate_redirect_target +# --------------------------------------------------------------------------- + + +def test_validate_redirect_target_rejects_unsupported_scheme(): + ok, reason = _validate_redirect_target("ftp://example.com/x") + assert ok is False + assert "unsupported scheme" in reason + + +def test_validate_redirect_target_rejects_missing_hostname(): + ok, reason = _validate_redirect_target("http://") + assert ok is False + assert "no hostname" in reason + + +def test_validate_redirect_target_consults_policy_engine(): + engine = _FakePolicyEngine() + with patch("backend.secuscan.network_policy.get_policy_engine", return_value=engine): + ok, reason = _validate_redirect_target("http://169.254.169.254/latest/meta-data/") + assert ok is False + assert "denylist" in reason + assert engine.calls[-1]["dest_ip"] == "169.254.169.254" + assert engine.calls[-1]["dest_hostname"] == "169.254.169.254" + + +def test_validate_redirect_target_fails_closed_on_engine_error(): + engine = _FakePolicyEngine(raise_on_check=True) + with patch("backend.secuscan.network_policy.get_policy_engine", return_value=engine): + ok, reason = _validate_redirect_target("http://169.254.169.254/latest/meta-data/") + assert ok is False + assert "could not be validated" in reason + + +# --------------------------------------------------------------------------- +# crawl_target redirect behavior +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_crawl_target_blocks_redirect_to_cloud_metadata(): + """A redirect into the metadata/link-local range must abort the crawl. + + The redirect destination is never fetched (stream is called once for the + seed), proving the SSRF pivot is closed rather than merely logged. + """ + seed = "http://example.com/" + redirect_url = "http://169.254.169.254/latest/meta-data/" + client = _make_client( + _make_response(302, seed, headers={"location": redirect_url}), + _make_response(200, redirect_url, body=b"iam metadata"), + ) + engine = _FakePolicyEngine() + + with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=client): + with patch("backend.secuscan.crawler.settings") as mock_settings: + mock_settings.verify_ssl = True + mock_settings.enforce_network_policy = True + with patch("backend.secuscan.network_policy.get_policy_engine", return_value=engine): + with pytest.raises(ValueError, match="rejected by network policy"): + await crawl_target(seed, extra_headers={"Authorization": "Basic dXNlcjpwYXNz"}) + + # The metadata endpoint must never have been requested. + assert client.stream.call_count == 1 + assert engine.calls[-1]["dest_ip"] == "169.254.169.254" + + +@pytest.mark.asyncio +async def test_crawl_target_strips_credentials_on_cross_origin_redirect(): + """Cross-origin redirects must not inherit the seed's Authorization or cookies.""" + seed = "https://example.com/" + attacker = "https://attacker.example/collect" + client = _make_client( + _make_response(302, seed, headers={"location": attacker}), + _make_response(200, attacker, body=b"pwned"), + ) + extra_headers = {"Authorization": "Basic dXNlcjpwYXNz", "X-Custom": "keep"} + cookies = {"session": "secret"} + + with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=client): + with patch("backend.secuscan.crawler.settings") as mock_settings: + mock_settings.verify_ssl = True + mock_settings.enforce_network_policy = False + result = await crawl_target( + seed, extra_headers=extra_headers, cookies=cookies + ) + + calls = client.stream.call_args_list + assert len(calls) == 2 + + first_headers = calls[0].kwargs["headers"] + assert first_headers["Authorization"] == "Basic dXNlcjpwYXNz" + assert first_headers["X-Custom"] == "keep" + assert calls[0].kwargs["cookies"] == {"session": "secret"} + + second_headers = calls[1].kwargs["headers"] + assert "Authorization" not in second_headers + assert "X-Custom" not in second_headers + assert calls[1].kwargs["cookies"] == {} + + assert result["final_url"] == attacker + assert result["redirect_chain"] == [ + {"url": seed, "status_code": 302, "location": attacker} + ] + + +@pytest.mark.asyncio +async def test_crawl_target_keeps_credentials_on_same_origin_redirect(): + """Same-origin redirects legitimately keep the configured credentials.""" + seed = "https://example.com/" + login = "https://example.com/login" + client = _make_client( + _make_response(302, seed, headers={"location": login}), + _make_response(200, login, body=b"login"), + ) + extra_headers = {"Authorization": "Basic dXNlcjpwYXNz"} + cookies = {"session": "secret"} + + with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=client): + with patch("backend.secuscan.crawler.settings") as mock_settings: + mock_settings.verify_ssl = True + mock_settings.enforce_network_policy = False + result = await crawl_target( + seed, extra_headers=extra_headers, cookies=cookies + ) + + calls = client.stream.call_args_list + assert len(calls) == 2 + assert calls[1].kwargs["headers"]["Authorization"] == "Basic dXNlcjpwYXNz" + assert calls[1].kwargs["cookies"] == {"session": "secret"} + assert result["final_url"] == login + + +@pytest.mark.asyncio +async def test_crawl_target_enforces_manual_redirect_limit(): + """The manual redirect loop must still cap the chain at max_redirects.""" + client = _make_client( + _make_response(302, "http://example.com/", headers={"location": "http://example.com/b"}), + _make_response(302, "http://example.com/b", headers={"location": "http://example.com/c"}), + _make_response(302, "http://example.com/c", headers={"location": "http://example.com/d"}), + ) + + with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=client): + with patch("backend.secuscan.crawler.settings") as mock_settings: + mock_settings.verify_ssl = True + mock_settings.enforce_network_policy = False + with pytest.raises(httpx.TooManyRedirects): + await crawl_target("http://example.com/", max_redirects=2) + + assert client.stream.call_count == 3 From 4731318587086a400dce4ee2430c46da74870889 Mon Sep 17 00:00:00 2001 From: Naman Singh Date: Tue, 4 Aug 2026 20:14:40 +0530 Subject: [PATCH 2/4] fix(crawler): resolve DNS once and pin IP for redirect hops _validate_redirect_target now resolves DNS up front and validates every returned IP against the network policy, returning the validated address. crawl_target rewrites redirect-hop URLs to use the pinned IP with the original hostname in the Host header, closing the DNS-rebinding window between policy validation and the actual HTTP fetch. Add tests for DNS resolution, validated IP return, resolution failure, and IP-pinning behavior on redirect hops. --- backend/secuscan/crawler.py | 81 ++++++++--- .../unit/test_crawler_redirect_security.py | 133 +++++++++++++----- 2 files changed, 165 insertions(+), 49 deletions(-) diff --git a/backend/secuscan/crawler.py b/backend/secuscan/crawler.py index 66cfd198f..0801dcc6a 100644 --- a/backend/secuscan/crawler.py +++ b/backend/secuscan/crawler.py @@ -7,7 +7,7 @@ import logging import re from typing import Any, Dict, List, Tuple -from urllib.parse import parse_qsl, urljoin, urlparse +from urllib.parse import parse_qsl, urljoin, urlparse, urlunparse import httpx @@ -104,37 +104,63 @@ def _is_same_origin(a: Any, b: Any) -> bool: return scheme_a == scheme_b and host_a == host_b and port_a == port_b -def _validate_redirect_target(url: str) -> Tuple[bool, str]: +def _validate_redirect_target(url: str) -> Tuple[bool, str, str | None]: """Re-validate a redirect destination against the network policy. The seed target is validated by the executor before the scanner runs, but httpx-followed redirects are not. Without this check a hostile or compromised seed can pivot the crawler into cloud-metadata, loopback, private/CGNAT, or IPv6 link-local/ULA ranges that were never authorized. + + Resolves DNS up front and validates every returned IP so that the caller + can pin the connection to the validated address, closing any + DNS-rebinding window between validation and fetch. """ parsed = urlparse(url) if parsed.scheme not in {"http", "https"}: - return False, f"redirect to unsupported scheme '{parsed.scheme}'" + return False, f"redirect to unsupported scheme '{parsed.scheme}'", None hostname = parsed.hostname if not hostname: - return False, "redirect target has no hostname" + return False, "redirect target has no hostname", None + + import socket as _socket + try: + addr_infos = _socket.getaddrinfo( + hostname, parsed.port or (443 if parsed.scheme == "https" else 80), + proto=_socket.IPPROTO_TCP, + ) + except OSError: + return False, "redirect hostname could not be resolved", None from .network_policy import get_policy_engine try: engine = get_policy_engine() - allowed, reason, _ = engine.check_access( - dest_ip=hostname, - dest_hostname=hostname, - plugin_id="crawler", - task_id="crawler-redirect", - ) except Exception as exc: - # Never fail open: if the policy engine cannot evaluate the target, - # block the redirect instead of silently fetching an internal host. logger.warning("Redirect target validation failed for %s: %s", url, exc) - return False, "redirect target could not be validated" - return allowed, reason + return False, "redirect target could not be validated", None + + validated_ip: str | None = None + for _family, _stype, _proto, _cname, sockaddr in addr_infos: + ip_str = sockaddr[0] + try: + allowed, reason, _ = engine.check_access( + dest_ip=ip_str, + dest_hostname=hostname, + plugin_id="crawler", + task_id="crawler-redirect", + ) + except Exception as exc: + logger.warning("Redirect target validation failed for %s: %s", url, exc) + return False, "redirect target could not be validated", None + if not allowed: + return False, reason, None + if validated_ip is None: + validated_ip = ip_str + + if validated_ip is None: + return False, "redirect target did not resolve to any address", None + return True, "Allowed", validated_ip async def crawl_target( @@ -188,17 +214,40 @@ async def crawl_target( hop_headers[str(key)] = str(value) hop_cookies = dict(cookies or {}) if forward_credentials else {} + # Pin the connection address for redirect hops to prevent + # DNS-rebinding: the hostname is resolved once for policy + # validation and the same IP is used for the actual fetch. + pinned_url = current_url + pinned_headers = dict(hop_headers) + if redirects_followed > 0 and settings.enforce_network_policy: - allowed, reason = await asyncio.to_thread( + allowed, reason, validated_ip = await asyncio.to_thread( _validate_redirect_target, current_url ) if not allowed: raise ValueError( f"Redirect to {current_url} rejected by network policy: {reason}" ) + if validated_ip: + parsed_hop = urlparse(current_url) + hop_host = parsed_hop.hostname + new_netloc = ( + f"[{validated_ip}]" if ":" in validated_ip else validated_ip + ) + if parsed_hop.port: + new_netloc = f"{new_netloc}:{parsed_hop.port}" + pinned_url = urlunparse(( + parsed_hop.scheme, + new_netloc, + parsed_hop.path, + parsed_hop.params, + parsed_hop.query, + parsed_hop.fragment, + )) + pinned_headers["Host"] = hop_host async with client.stream( - "GET", current_url, headers=hop_headers, cookies=hop_cookies + "GET", pinned_url, headers=pinned_headers, cookies=hop_cookies ) as response: # Check Content-Length header if present content_length = response.headers.get("content-length") diff --git a/testing/backend/unit/test_crawler_redirect_security.py b/testing/backend/unit/test_crawler_redirect_security.py index 42f4bdd49..d2f731f56 100644 --- a/testing/backend/unit/test_crawler_redirect_security.py +++ b/testing/backend/unit/test_crawler_redirect_security.py @@ -9,6 +9,8 @@ 2. Credentials (``extra_headers`` / ``cookies``) are only sent to the seed origin and are stripped on any cross-origin redirect, so vault credentials cannot be leaked to an attacker-controlled host. +3. DNS is resolved once for validation and the same IP is pinned for the + actual fetch, closing any DNS-rebinding window. Related issue: https://github.com/utksh1/SecuScan/issues/2369 """ @@ -85,6 +87,11 @@ def _make_client(*responses): return client +def _fake_addrinfo(ip): + """Return a fake socket.getaddrinfo result for the given IP.""" + return [(2, 1, 6, "", (ip, 0, 0, 0))] + + # --------------------------------------------------------------------------- # _is_same_origin # --------------------------------------------------------------------------- @@ -120,33 +127,54 @@ def test_is_same_origin_malformed_port_is_cross_origin(): def test_validate_redirect_target_rejects_unsupported_scheme(): - ok, reason = _validate_redirect_target("ftp://example.com/x") + ok, reason, _ip = _validate_redirect_target("ftp://example.com/x") assert ok is False assert "unsupported scheme" in reason def test_validate_redirect_target_rejects_missing_hostname(): - ok, reason = _validate_redirect_target("http://") + ok, reason, _ip = _validate_redirect_target("http://") assert ok is False assert "no hostname" in reason def test_validate_redirect_target_consults_policy_engine(): engine = _FakePolicyEngine() - with patch("backend.secuscan.network_policy.get_policy_engine", return_value=engine): - ok, reason = _validate_redirect_target("http://169.254.169.254/latest/meta-data/") + with patch("backend.secuscan.network_policy.get_policy_engine", return_value=engine), \ + patch("socket.getaddrinfo", return_value=_fake_addrinfo("169.254.169.254")): + ok, reason, ip = _validate_redirect_target("http://169.254.169.254/latest/meta-data/") assert ok is False assert "denylist" in reason + assert ip is None assert engine.calls[-1]["dest_ip"] == "169.254.169.254" assert engine.calls[-1]["dest_hostname"] == "169.254.169.254" def test_validate_redirect_target_fails_closed_on_engine_error(): engine = _FakePolicyEngine(raise_on_check=True) - with patch("backend.secuscan.network_policy.get_policy_engine", return_value=engine): - ok, reason = _validate_redirect_target("http://169.254.169.254/latest/meta-data/") + with patch("backend.secuscan.network_policy.get_policy_engine", return_value=engine), \ + patch("socket.getaddrinfo", return_value=_fake_addrinfo("169.254.169.254")): + ok, reason, ip = _validate_redirect_target("http://169.254.169.254/latest/meta-data/") assert ok is False assert "could not be validated" in reason + assert ip is None + + +def test_validate_redirect_target_resolves_dns_and_returns_validated_ip(): + engine = _FakePolicyEngine() + with patch("backend.secuscan.network_policy.get_policy_engine", return_value=engine), \ + patch("socket.getaddrinfo", return_value=_fake_addrinfo("93.184.216.34")): + ok, reason, ip = _validate_redirect_target("http://example.com/ok") + assert ok is True + assert ip == "93.184.216.34" + + +def test_validate_redirect_target_rejects_dns_resolution_failure(): + with patch("socket.getaddrinfo", side_effect=OSError("no such host")): + ok, reason, ip = _validate_redirect_target("http://nonexistent.invalid/") + assert ok is False + assert "could not be resolved" in reason + assert ip is None # --------------------------------------------------------------------------- @@ -169,13 +197,14 @@ async def test_crawl_target_blocks_redirect_to_cloud_metadata(): ) engine = _FakePolicyEngine() - with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=client): - with patch("backend.secuscan.crawler.settings") as mock_settings: - mock_settings.verify_ssl = True - mock_settings.enforce_network_policy = True - with patch("backend.secuscan.network_policy.get_policy_engine", return_value=engine): - with pytest.raises(ValueError, match="rejected by network policy"): - await crawl_target(seed, extra_headers={"Authorization": "Basic dXNlcjpwYXNz"}) + with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=client), \ + patch("backend.secuscan.crawler.settings") as mock_settings, \ + patch("backend.secuscan.network_policy.get_policy_engine", return_value=engine), \ + patch("socket.getaddrinfo", return_value=_fake_addrinfo("169.254.169.254")): + mock_settings.verify_ssl = True + mock_settings.enforce_network_policy = True + with pytest.raises(ValueError, match="rejected by network policy"): + await crawl_target(seed, extra_headers={"Authorization": "Basic dXNlcjpwYXNz"}) # The metadata endpoint must never have been requested. assert client.stream.call_count == 1 @@ -194,13 +223,13 @@ async def test_crawl_target_strips_credentials_on_cross_origin_redirect(): extra_headers = {"Authorization": "Basic dXNlcjpwYXNz", "X-Custom": "keep"} cookies = {"session": "secret"} - with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=client): - with patch("backend.secuscan.crawler.settings") as mock_settings: - mock_settings.verify_ssl = True - mock_settings.enforce_network_policy = False - result = await crawl_target( - seed, extra_headers=extra_headers, cookies=cookies - ) + with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=client), \ + patch("backend.secuscan.crawler.settings") as mock_settings: + mock_settings.verify_ssl = True + mock_settings.enforce_network_policy = False + result = await crawl_target( + seed, extra_headers=extra_headers, cookies=cookies + ) calls = client.stream.call_args_list assert len(calls) == 2 @@ -233,13 +262,13 @@ async def test_crawl_target_keeps_credentials_on_same_origin_redirect(): extra_headers = {"Authorization": "Basic dXNlcjpwYXNz"} cookies = {"session": "secret"} - with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=client): - with patch("backend.secuscan.crawler.settings") as mock_settings: - mock_settings.verify_ssl = True - mock_settings.enforce_network_policy = False - result = await crawl_target( - seed, extra_headers=extra_headers, cookies=cookies - ) + with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=client), \ + patch("backend.secuscan.crawler.settings") as mock_settings: + mock_settings.verify_ssl = True + mock_settings.enforce_network_policy = False + result = await crawl_target( + seed, extra_headers=extra_headers, cookies=cookies + ) calls = client.stream.call_args_list assert len(calls) == 2 @@ -257,11 +286,49 @@ async def test_crawl_target_enforces_manual_redirect_limit(): _make_response(302, "http://example.com/c", headers={"location": "http://example.com/d"}), ) - with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=client): - with patch("backend.secuscan.crawler.settings") as mock_settings: - mock_settings.verify_ssl = True - mock_settings.enforce_network_policy = False - with pytest.raises(httpx.TooManyRedirects): - await crawl_target("http://example.com/", max_redirects=2) + with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=client), \ + patch("backend.secuscan.crawler.settings") as mock_settings: + mock_settings.verify_ssl = True + mock_settings.enforce_network_policy = False + with pytest.raises(httpx.TooManyRedirects): + await crawl_target("http://example.com/", max_redirects=2) assert client.stream.call_count == 3 + + +@pytest.mark.asyncio +async def test_crawl_target_pins_ip_for_redirect_hop(): + """Redirect hops connect to the validated (pinned) IP, not a fresh DNS lookup. + + This prevents DNS-rebinding: the hostname is resolved once for policy + validation and the resulting IP is used for the actual HTTP connection, + with the original hostname set in the Host header. + """ + seed = "http://example.com/" + redirect_url = "http://example.com/secret" + client = _make_client( + _make_response(302, seed, headers={"location": redirect_url}), + _make_response(200, redirect_url, body=b"ok"), + ) + engine = _FakePolicyEngine() + + with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=client), \ + patch("backend.secuscan.crawler.settings") as mock_settings, \ + patch("backend.secuscan.network_policy.get_policy_engine", return_value=engine), \ + patch("socket.getaddrinfo", return_value=_fake_addrinfo("93.184.216.34")): + mock_settings.verify_ssl = True + mock_settings.enforce_network_policy = True + result = await crawl_target(seed) + + assert client.stream.call_count == 2 + calls = client.stream.call_args_list + + # The redirect hop must use the pinned IP in the URL. + second_url = calls[1].args[1] + assert "93.184.216.34" in second_url + + # The Host header must preserve the original hostname. + second_headers = calls[1].kwargs["headers"] + assert second_headers.get("Host") == "example.com" + + assert result["final_url"] == redirect_url From 35d2508924efd6798bde4cbec9adc1311f11f57e Mon Sep 17 00:00:00 2001 From: Naman Singh Date: Wed, 5 Aug 2026 20:46:20 +0530 Subject: [PATCH 3/4] fix(crawler): honor log-only network policy mode for redirect denials --- backend/secuscan/crawler.py | 16 +++++++--- .../unit/test_crawler_redirect_security.py | 32 +++++++++++++++++++ 2 files changed, 44 insertions(+), 4 deletions(-) diff --git a/backend/secuscan/crawler.py b/backend/secuscan/crawler.py index 0801dcc6a..e56eff5c2 100644 --- a/backend/secuscan/crawler.py +++ b/backend/secuscan/crawler.py @@ -225,10 +225,18 @@ async def crawl_target( _validate_redirect_target, current_url ) if not allowed: - raise ValueError( - f"Redirect to {current_url} rejected by network policy: {reason}" - ) - if validated_ip: + if settings.network_policy_failure_mode == "log_only": + logger.warning( + "[Log Only] Redirect to %s denied by network policy but " + "allowed in log-only mode: %s", + current_url, + reason, + ) + else: + raise ValueError( + f"Redirect to {current_url} rejected by network policy: {reason}" + ) + if allowed and validated_ip: parsed_hop = urlparse(current_url) hop_host = parsed_hop.hostname new_netloc = ( diff --git a/testing/backend/unit/test_crawler_redirect_security.py b/testing/backend/unit/test_crawler_redirect_security.py index d2f731f56..9b163750f 100644 --- a/testing/backend/unit/test_crawler_redirect_security.py +++ b/testing/backend/unit/test_crawler_redirect_security.py @@ -211,6 +211,38 @@ async def test_crawl_target_blocks_redirect_to_cloud_metadata(): assert engine.calls[-1]["dest_ip"] == "169.254.169.254" +@pytest.mark.asyncio +async def test_crawl_target_log_only_allows_denied_redirect(): + """In log-only failure mode a denied redirect hop is allowed with a warning. + + This mirrors the executor's fail-open behavior: policy violations are + logged but the crawl continues, so a dry-run/log-only network-policy + deployment does not abort crawler scans on denied redirect targets. + """ + seed = "http://example.com/" + redirect_url = "http://169.254.169.254/latest/meta-data/" + client = _make_client( + _make_response(302, seed, headers={"location": redirect_url}), + _make_response(200, redirect_url, body=b"iam metadata"), + ) + engine = _FakePolicyEngine() + + with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=client), \ + patch("backend.secuscan.crawler.settings") as mock_settings, \ + patch("backend.secuscan.network_policy.get_policy_engine", return_value=engine), \ + patch("socket.getaddrinfo", return_value=_fake_addrinfo("169.254.169.254")): + mock_settings.verify_ssl = True + mock_settings.enforce_network_policy = True + mock_settings.network_policy_failure_mode = "log_only" + result = await crawl_target(seed) + + # The denied hop is still fetched (no pinning available for a denied target). + assert client.stream.call_count == 2 + calls = client.stream.call_args_list + assert "169.254.169.254" in calls[1].args[1] + assert result["final_url"] == redirect_url + + @pytest.mark.asyncio async def test_crawl_target_strips_credentials_on_cross_origin_redirect(): """Cross-origin redirects must not inherit the seed's Authorization or cookies.""" From 6173580102922c80f51b19a171b37766ab247033 Mon Sep 17 00:00:00 2001 From: Naman Singh Date: Wed, 5 Aug 2026 21:38:42 +0530 Subject: [PATCH 4/4] ci: retrigger frontend checks (flaky failure on Node 20 shard)