fix(url-crawler): SSRF — the scanner fetched attacker-chosen addresses

src/malwar/detectors/url_crawler/detector.py

@@ -4,6 +4,7 @@
 
 import logging
 
+from malwar.core.config import get_settings
 from malwar.core.constants import DetectorLayer, Severity, ThreatCategory
 from malwar.detectors.url_crawler.analyzer import analyze_fetch_result
 from malwar.detectors.url_crawler.extractor import extract_urls
@@ -23,7 +24,21 @@ class UrlCrawlerDetector(BaseDetector):
     """Layer 2: Fetch and analyze URLs found in skill files."""
 
     def __init__(self, fetcher: SafeFetcher | None = None) -> None:
-        self._fetcher = fetcher or SafeFetcher()
+        # The crawler_* settings existed but nothing read them, so tuning
+        # crawler_timeout or crawler_max_redirects silently did nothing while
+        # SafeFetcher used its own constructor defaults. The values happen to
+        # match, so wiring them changes no behaviour today -- it makes the
+        # documented knobs real.
+        if fetcher is None:
+            settings = get_settings()
+            fetcher = SafeFetcher(
+                max_urls=settings.crawler_max_urls,
+                timeout=settings.crawler_timeout,
+                max_redirects=settings.crawler_max_redirects,
+                max_bytes=settings.crawler_max_response_bytes,
+                concurrency=settings.crawler_concurrency,
+            )
+        self._fetcher = fetcher
 
     @property
     def layer_name(self) -> str:

src/malwar/detectors/url_crawler/fetcher.py

@@ -5,11 +5,18 @@
 import asyncio
 import logging
 from dataclasses import dataclass, field
+from urllib.parse import urljoin
 
 import httpx
 
+from malwar.detectors.url_crawler.safety import Resolver, check_url, resolve_host
+
 logger = logging.getLogger("malwar.detectors.url_crawler.fetcher")
 
+
+class UnsafeURLError(Exception):
+    """Raised when a URL (or a redirect target) must not be fetched."""
+
 # Content types we consider textual (worth fetching the body for)
 _TEXT_CONTENT_TYPES = frozenset(
     {
@@ -83,12 +90,16 @@ def __init__(
         max_redirects: int = 3,
         max_bytes: int = 1_048_576,
         concurrency: int = 5,
+        resolver: Resolver = resolve_host,
     ) -> None:
         self.max_urls = max_urls
         self.timeout = timeout
         self.max_redirects = max_redirects
         self.max_bytes = max_bytes
         self.concurrency = concurrency
+        # Injectable so tests pin DNS answers instead of depending on the
+        # network; see safety.check_url.
+        self.resolver = resolver
 
     async def fetch_urls(self, urls: list[str]) -> list[FetchResult]:
         """Fetch multiple URLs concurrently with safety bounds."""
@@ -97,9 +108,12 @@ async def fetch_urls(self, urls: list[str]) -> list[FetchResult]:
 
         semaphore = asyncio.Semaphore(self.concurrency)
 
+        # Redirects are followed by hand in _request, not by httpx, so every
+        # hop can be validated before it is fetched. Letting the transport
+        # chase them means the first unsafe destination is already requested
+        # by the time we could look at it.
         async with httpx.AsyncClient(
-            follow_redirects=True,
-            max_redirects=self.max_redirects,
+            follow_redirects=False,
             timeout=httpx.Timeout(self.timeout),
         ) as client:
             tasks = [
@@ -137,6 +151,19 @@ async def _fetch_one(
                     content="",
                     error=f"Request timed out ({self.timeout}s)",
                 )
+            except UnsafeURLError as exc:
+                # Refused before any request was sent. Surfaced as a result
+                # rather than dropped, so a blocked fetch is visible in the
+                # report instead of looking like a URL that simply had nothing
+                # interesting at the other end.
+                return FetchResult(
+                    url=url,
+                    final_url=url,
+                    status_code=0,
+                    content_type="",
+                    content="",
+                    error=f"Refused to fetch: {exc}",
+                )
             except Exception as exc:
                 logger.debug("Fetch failed for %s: %s", url, exc)
                 return FetchResult(
@@ -148,25 +175,46 @@ async def _fetch_one(
                     error=str(exc),
                 )
 
+    async def _request(
+        self,
+        client: httpx.AsyncClient,
+        method: str,
+        url: str,
+    ) -> tuple[httpx.Response, list[str], str]:
+        """Issue ``method`` against ``url``, following redirects by hand.
+
+        Every hop is checked before it is requested, which is the whole point:
+        a URL that passes on the first request can redirect to internal
+        infrastructure on the second, so validating only the entry point is
+        equivalent to validating nothing.
+
+        Returns ``(response, redirect_chain, final_url)``.
+        """
+        chain: list[str] = []
+        current = url
+        for _ in range(self.max_redirects + 1):
+            safe, reason = check_url(current, resolver=self.resolver)
+            if not safe:
+                raise UnsafeURLError(reason)
+            resp = await client.request(method, current)
+            location = resp.headers.get("location")
+            if not (resp.is_redirect and location):
+                return resp, chain, current
+            chain.append(current)
+            # Relative Location headers are legal and common.
+            current = urljoin(current, location)
+        raise httpx.TooManyRedirects(f"exceeded {self.max_redirects} redirects")
+
     async def _do_fetch(
         self,
         client: httpx.AsyncClient,
         url: str,
     ) -> FetchResult:
         """Perform the actual fetch with HEAD pre-check."""
-        redirect_chain: list[str] = []
-
-        # Attempt HEAD first to inspect content-type / size
-        head_resp = await client.head(url)
+        head_resp, redirect_chain, final_url = await self._request(client, "HEAD", url)
         content_type_raw = head_resp.headers.get("content-type", "")
         content_type = content_type_raw.split(";")[0].strip().lower()
 
-        # Build redirect chain from the response history
-        for resp in head_resp.history:
-            redirect_chain.append(str(resp.url))
-
-        final_url = str(head_resp.url)
-
         # Check content-length to avoid huge downloads
         content_length = head_resp.headers.get("content-length")
         if content_length and int(content_length) > self.max_bytes:
@@ -204,10 +252,8 @@ async def _do_fetch(
                 redirect_chain=redirect_chain,
             )
 
-        # GET request with body-size limit
-        get_resp = await client.get(url)
-        redirect_chain = [str(r.url) for r in get_resp.history]
-        final_url = str(get_resp.url)
+        # GET request with body-size limit, redirects validated the same way.
+        get_resp, redirect_chain, final_url = await self._request(client, "GET", url)
 
         body = get_resp.text[: self.max_bytes]
 

src/malwar/detectors/url_crawler/safety.py

@@ -0,0 +1,133 @@
+"""Decide whether a URL is safe for the scanner itself to fetch.
+
+The URL crawler exists to follow links found in skill content, and skill
+content is hostile input by definition -- it is the thing being analysed. URLs
+come from the body, from ``source_url`` and ``author_url``, and from arbitrary
+frontmatter keys, so the destination is fully attacker-controlled.
+
+Without a guard, scanning a skill containing::
+
+    http://169.254.169.254/latest/meta-data/iam/security-credentials/
+
+makes the scanner fetch cloud instance credentials, from CI or from whatever
+machine a user ran it on. That is server-side request forgery reached through
+the tool's normal, default-on code path (``use_urls=True``).
+
+Three rules, and the third is the one that is easy to get wrong:
+
+1. Only ``http`` and ``https``. Blocks ``file://``, ``gopher://`` and friends.
+2. Resolve the hostname and reject if **any** returned address is private,
+   loopback, link-local, reserved or otherwise not a public unicast address.
+   Checking the literal string is not enough: an attacker controls DNS for
+   their own domain, so ``evil.example`` with an A record of ``127.0.0.1``
+   walks straight past a textual check.
+3. Re-check every redirect hop. A URL that passes on the first request can
+   redirect to metadata on the second, so validating only the initial target
+   is equivalent to not validating at all.
+
+Residual risk, stated rather than papered over: this resolves the name and
+then hands the URL to httpx, which resolves it again to connect. A DNS entry
+that changes between those two lookups (rebinding) defeats the check. Closing
+that needs a transport that pins the validated address, which is a larger
+change; it is tracked as a known limit, not silently ignored.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import socket
+from collections.abc import Callable
+from urllib.parse import urlsplit
+
+ALLOWED_SCHEMES: frozenset[str] = frozenset({"http", "https"})
+
+IPAddress = ipaddress.IPv4Address | ipaddress.IPv6Address
+Resolver = Callable[[str], list[IPAddress]]
+
+
+def _address_is_public(ip: IPAddress) -> bool:
+    """True only for ordinary routable unicast addresses.
+
+    ``is_global`` alone is not sufficient on every Python version for every
+    family, so the specific categories that matter for SSRF are also named.
+    """
+    if (
+        ip.is_private            # 10/8, 172.16/12, 192.168/16, fc00::/7, ...
+        or ip.is_loopback        # 127/8, ::1
+        or ip.is_link_local      # 169.254/16 (cloud metadata), fe80::/10
+        or ip.is_multicast
+        or ip.is_reserved
+        or ip.is_unspecified     # 0.0.0.0, ::
+    ):
+        return False
+    # IPv4-mapped and 6to4 addresses can smuggle a private v4 address inside a
+    # v6 literal, so unwrap and re-check rather than trusting the outer form.
+    mapped = getattr(ip, "ipv4_mapped", None)
+    if mapped is not None:
+        return _address_is_public(mapped)
+    sixtofour = getattr(ip, "sixtofour", None)
+    if sixtofour is not None:
+        return _address_is_public(sixtofour)
+    return True
+
+
+def resolve_host(host: str) -> list[IPAddress]:
+    """Return every address ``host`` resolves to, or [] if it does not resolve.
+
+    An IP literal resolves to itself without a DNS lookup.
+    """
+    try:
+        return [ipaddress.ip_address(host)]
+    except ValueError:
+        pass
+    try:
+        infos = socket.getaddrinfo(host, None, proto=socket.IPPROTO_TCP)
+    except (OSError, UnicodeError):
+        return []
+    out: list[IPAddress] = []
+    for info in infos:
+        try:
+            out.append(ipaddress.ip_address(info[4][0]))
+        except ValueError:
+            continue
+    return out
+
+
+def check_url(url: str, resolver: Resolver = resolve_host) -> tuple[bool, str]:
+    """Return ``(safe, reason)`` for one URL.
+
+    ``reason`` is empty when safe and names the specific failure otherwise, so
+    a blocked fetch is explainable rather than a silent drop.
+
+    ``resolver`` is injectable so tests can exercise the address-classification
+    logic against fixed answers. Making the whole check skippable instead would
+    mean the tests that mock HTTP stop covering the guard entirely, which is
+    how a control ends up passing its own suite while doing nothing.
+    """
+    try:
+        parts = urlsplit(url)
+    except ValueError as exc:
+        return False, f"unparseable URL: {exc}"
+
+    if parts.scheme.lower() not in ALLOWED_SCHEMES:
+        return False, f"scheme {parts.scheme!r} not allowed"
+
+    host = parts.hostname
+    if not host:
+        return False, "no host in URL"
+
+    addresses = resolver(host)
+    if not addresses:
+        # Unresolvable is not unsafe, but there is nothing to fetch and letting
+        # it through would mean the connect-time resolution is the only one
+        # that ever happens -- i.e. no check at all.
+        return False, f"host {host!r} does not resolve"
+
+    # Every address must be public. One private answer among several is enough
+    # to reach an internal service, because which one gets connected to is not
+    # ours to choose.
+    for ip in addresses:
+        if not _address_is_public(ip):
+            return False, f"host {host!r} resolves to non-public address {ip}"
+
+    return True, ""