Don't follow HTTP redirects off the scan target

src/header_audit.py

@@ -96,13 +96,16 @@ def _flatten_headers(msg) -> dict:
 def fetch_headers(url: str, timeout: float = 8.0) -> tuple[dict, int]:
     """
     Fetch HTTP headers from a URL. Returns (headers_dict, status_code).
-    Follows redirects (urllib default), ignores cert errors. Never raises:
-    on any failure returns ({}, 0) so the caller can produce a sane
-    "could not audit" result instead of a misleading all-missing report.
+    Does not follow redirects (a 3xx to metadata / RFC1918 would be SSRF).
+    Ignores cert errors. Never raises: on any failure returns ({}, 0) so the
+    caller can produce a sane "could not audit" result instead of a misleading
+    all-missing report.
 
     Also stashes a short body snippet on ``headers['_body_snippet']`` so bot-
     challenge pages can be detected when vendor headers are intermittent.
     """
+    from src.http_fetch import urlopen_no_redirect  # noqa: PLC0415
+
     ctx = ssl.create_default_context()
     ctx.check_hostname = False
     ctx.verify_mode = ssl.CERT_NONE
@@ -121,7 +124,7 @@ def _pack(msg, status: int, body: bytes = b"") -> tuple[dict, int]:
         return headers, status
 
     try:
-        with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
+        with urlopen_no_redirect(req, timeout=timeout, context=ctx) as resp:
             body = resp.read(2048)
             return _pack(resp.headers, resp.status, body)
     except urllib.error.HTTPError as e:

src/http_fetch.py

@@ -0,0 +1,37 @@
+"""HTTP fetch helpers that refuse to follow redirects.
+
+urllib.request.urlopen follows 3xx by default, including cross-host hops.
+A scan of an authorized public target that returns
+``Location: http://169.254.169.254/...`` (or RFC1918) would otherwise make
+the agent connect to metadata or internal neighbors — the same SSRF class
+already blocked for LLM URLs, SIEM, vuln_prober, takeover, and active
+validation. Target-facing auditors must fail the same way: return the 3xx
+as HTTPError and never fetch the redirect destination.
+"""
+from __future__ import annotations
+
+import urllib.error
+import urllib.request
+from typing import Optional
+
+
+class NoRedirectHandler(urllib.request.HTTPRedirectHandler):
+    """Replace the default redirect handler so 3xx is surfaced, not followed."""
+
+    def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: ANN001
+        raise urllib.error.HTTPError(
+            newurl, code, "HTTP redirects are not followed", headers, fp,
+        )
+
+
+def urlopen_no_redirect(
+    req: urllib.request.Request,
+    timeout: float,
+    context: Optional[object] = None,
+):
+    """urlopen that never follows 3xx. Optional SSL context for CERT_NONE probes."""
+    handlers: list = [NoRedirectHandler]
+    if context is not None:
+        handlers.append(urllib.request.HTTPSHandler(context=context))
+    opener = urllib.request.build_opener(*handlers)
+    return opener.open(req, timeout=timeout)

src/service_enum.py

@@ -384,7 +384,8 @@ def probe_http_auth_state(host: str, port: int, scheme: str = "http",
         req = urllib.request.Request(url, headers={"User-Agent": "NetLogic/2.0"})
         status, hdrs, body_head = None, {}, ""
         try:
-            with urllib.request.urlopen(req, timeout=timeout) as resp:
+            from src.http_fetch import urlopen_no_redirect  # noqa: PLC0415
+            with urlopen_no_redirect(req, timeout=timeout) as resp:
                 status = resp.status
                 hdrs = {k.lower(): v for k, v in resp.headers.items()}
                 body_head = resp.read(512).decode("utf-8", errors="replace").lower()

src/service_prober.py

@@ -59,7 +59,8 @@ def _http_get(host: str, port: int, path: str = "/", scheme: str = "http",
         ctx.verify_mode = ssl.CERT_NONE
         url = f"{scheme}://{host}:{port}{path}"
         req = urllib.request.Request(url, headers=headers or {})
-        with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
+        from src.http_fetch import urlopen_no_redirect  # noqa: PLC0415
+        with urlopen_no_redirect(req, timeout=timeout, context=ctx) as resp:
             body = resp.read(16384).decode("utf-8", errors="replace")
             return resp.status, dict(resp.headers), body
     except urllib.error.HTTPError as e:

src/stack_fingerprint.py

@@ -284,7 +284,8 @@ def _fetch(url: str, payload: str = None, timeout: float = 8.0) -> tuple[dict, s
         "Accept-Language": "en-US,en;q=0.5",
     })
     try:
-        with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
+        from src.http_fetch import urlopen_no_redirect  # noqa: PLC0415
+        with urlopen_no_redirect(req, timeout=timeout, context=ctx) as resp:
             body = resp.read(65536).decode("utf-8", errors="replace")
             headers = _flatten_headers(resp.headers)
             return headers, body, resp.status

src/web_fingerprint.py

@@ -120,7 +120,8 @@ def _get(url: str, timeout: float, binary: bool = False, max_bytes: int = 262144
         "Accept-Language": "en-US,en;q=0.9",
     })
     try:
-        with urllib.request.urlopen(req, timeout=timeout) as resp:
+        from src.http_fetch import urlopen_no_redirect  # noqa: PLC0415
+        with urlopen_no_redirect(req, timeout=timeout) as resp:
             raw = resp.read(max_bytes)
             return resp.status, (raw if binary else raw.decode("utf-8", errors="replace"))
     except urllib.error.HTTPError as e:

test_http_ssrf_redirect.py

@@ -0,0 +1,155 @@
+"""Off-origin HTTP redirects must not be followed by target-facing fetchers.
+
+urllib.request.urlopen follows 3xx by default. A public scan target that
+returns ``Location: http://169.254.169.254/...`` (or another host) would
+make the agent connect there — scanner SSRF. These tests use two local
+HTTP servers: origin 302s to a "metadata" host that records hits.
+"""
+from __future__ import annotations
+
+import http.server
+import threading
+
+import pytest
+
+from src.header_audit import fetch_headers
+from src.http_fetch import urlopen_no_redirect
+from src.service_enum import probe_http_auth_state
+from src.service_prober import _http_get
+from src.stack_fingerprint import _fetch
+from src.web_fingerprint import _get
+
+
+def _two_servers():
+    """Origin 302s to off-origin /secret. Off-origin 200s and counts hits."""
+    hits = {"secret": 0}
+
+    class Off(http.server.BaseHTTPRequestHandler):
+        def do_GET(self):
+            hits["secret"] += 1
+            self.send_response(200)
+            self.send_header("Server", "imds-fake")
+            self.send_header("Content-Type", "text/plain")
+            self.end_headers()
+            self.wfile.write(b"instance-id-i-0secret")
+
+        def log_message(self, *_a):
+            return
+
+    off = http.server.HTTPServer(("127.0.0.1", 0), Off)
+    off_port = off.server_address[1]
+
+    class Origin(http.server.BaseHTTPRequestHandler):
+        def do_GET(self):
+            self.send_response(302)
+            self.send_header("Location", f"http://127.0.0.1:{off_port}/secret")
+            self.send_header("Server", "in-scope")
+            self.end_headers()
+
+        def log_message(self, *_a):
+            return
+
+    origin = http.server.HTTPServer(("127.0.0.1", 0), Origin)
+    threads = [
+        threading.Thread(target=origin.serve_forever, daemon=True),
+        threading.Thread(target=off.serve_forever, daemon=True),
+    ]
+    for t in threads:
+        t.start()
+    return origin, off, hits
+
+
+@pytest.fixture
+def redirect_pair():
+    origin, off, hits = _two_servers()
+    try:
+        yield origin.server_address[1], hits
+    finally:
+        origin.shutdown()
+        off.shutdown()
+        origin.server_close()
+        off.server_close()
+
+
+def test_urlopen_no_redirect_does_not_fetch_destination(redirect_pair):
+    import urllib.error
+    import urllib.request
+
+    origin_port, hits = redirect_pair
+    req = urllib.request.Request(f"http://127.0.0.1:{origin_port}/")
+    with pytest.raises(urllib.error.HTTPError) as ctx:
+        urlopen_no_redirect(req, timeout=2.0)
+    assert ctx.value.code == 302
+    assert hits["secret"] == 0
+
+
+def test_fetch_headers_does_not_follow_off_origin_redirect(redirect_pair):
+    origin_port, hits = redirect_pair
+    headers, status = fetch_headers(f"http://127.0.0.1:{origin_port}/", timeout=2.0)
+    assert status == 302
+    assert "in-scope" in (headers.get("server") or "")
+    assert "instance-id" not in headers.get("_body_snippet", "")
+    assert hits["secret"] == 0
+
+
+def test_stack_fingerprint_does_not_follow_off_origin_redirect(redirect_pair):
+    origin_port, hits = redirect_pair
+    headers, body, status = _fetch(f"http://127.0.0.1:{origin_port}/", timeout=2.0)
+    assert status == 302
+    assert "imds-fake" not in (headers.get("server") or "")
+    assert "instance-id" not in (body or "")
+    assert hits["secret"] == 0
+
+
+def test_service_prober_http_get_does_not_follow_off_origin_redirect(redirect_pair):
+    origin_port, hits = redirect_pair
+    result = _http_get("127.0.0.1", origin_port, "/", timeout=2.0)
+    assert result is not None
+    status, hdrs, body = result
+    assert status == 302
+    assert "instance-id" not in (body or "")
+    assert hits["secret"] == 0
+
+
+def test_web_fingerprint_get_does_not_follow_off_origin_redirect(redirect_pair):
+    origin_port, hits = redirect_pair
+    status, body = _get(f"http://127.0.0.1:{origin_port}/", timeout=2.0)
+    assert status == 302
+    assert "instance-id" not in (body or "")
+    assert hits["secret"] == 0
+
+
+def test_service_enum_does_not_classify_redirect_destination_as_open(redirect_pair):
+    """302 to another host must not be fetched; must not look 'open' from the 200."""
+    origin_port, hits = redirect_pair
+    attrs = probe_http_auth_state("127.0.0.1", origin_port, scheme="http", timeout=2.0)
+    assert hits["secret"] == 0
+    values = {a.value for a in attrs}
+    assert "open" not in values
+
+
+def test_fetch_headers_still_returns_200_body():
+    """A normal 200 must still be fetched (redirects are the only thing blocked)."""
+
+    class Ok(http.server.BaseHTTPRequestHandler):
+        def do_GET(self):
+            self.send_response(200)
+            self.send_header("Server", "ok")
+            self.send_header("Content-Type", "text/plain")
+            self.end_headers()
+            self.wfile.write(b"hello-target")
+
+        def log_message(self, *_a):
+            return
+
+    srv = http.server.HTTPServer(("127.0.0.1", 0), Ok)
+    threading.Thread(target=srv.serve_forever, daemon=True).start()
+    try:
+        port = srv.server_address[1]
+        headers, status = fetch_headers(f"http://127.0.0.1:{port}/", timeout=2.0)
+        assert status == 200
+        assert "ok" in (headers.get("server") or "")
+        assert "hello-target" in headers.get("_body_snippet", "")
+    finally:
+        srv.shutdown()
+        srv.server_close()