diff --git a/src/malwar/detectors/url_crawler/detector.py b/src/malwar/detectors/url_crawler/detector.py index c3380b1..9f04c35 100644 --- a/src/malwar/detectors/url_crawler/detector.py +++ b/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: diff --git a/src/malwar/detectors/url_crawler/fetcher.py b/src/malwar/detectors/url_crawler/fetcher.py index 0a03301..6f02725 100644 --- a/src/malwar/detectors/url_crawler/fetcher.py +++ b/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] diff --git a/src/malwar/detectors/url_crawler/safety.py b/src/malwar/detectors/url_crawler/safety.py new file mode 100644 index 0000000..875cd25 --- /dev/null +++ b/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, "" diff --git a/tests/unit/detectors/test_url_crawler.py b/tests/unit/detectors/test_url_crawler.py index 66416bd..95aeb7d 100644 --- a/tests/unit/detectors/test_url_crawler.py +++ b/tests/unit/detectors/test_url_crawler.py @@ -2,6 +2,8 @@ from __future__ import annotations +import ipaddress + import httpx import pytest import respx @@ -16,6 +18,7 @@ SAFE_DOMAINS, check_domain_reputation, ) +from malwar.detectors.url_crawler.safety import check_url from malwar.models.skill import CodeBlock, SkillContent, SkillMetadata # --------------------------------------------------------------------------- @@ -202,12 +205,31 @@ def test_all_known_malicious_are_zero(self) -> None: # =================================================================== +def _public(_host: str) -> list[ipaddress.IPv4Address]: + """Pin DNS to one ordinary public address. + + respx intercepts at the transport layer, so mocked hosts never resolve for + real and the SSRF guard would reject every one of them. Injecting the + answer keeps the guard fully active -- scheme checks and address + classification still run -- while making these tests about fetching rather + than about whether ``short.url`` happens to exist. + """ + return [ipaddress.ip_address("93.184.216.34")] + + +def _resolves_to(addr: str): + """Pin DNS to a specific address, for the SSRF tests.""" + def _resolver(_host: str) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]: + return [ipaddress.ip_address(addr)] + return _resolver + + class TestSafeFetcher: """Tests for malwar.detectors.url_crawler.fetcher using respx.""" @pytest.mark.asyncio async def test_fetch_simple_text(self) -> None: - fetcher = SafeFetcher(max_urls=5, timeout=5.0) + fetcher = SafeFetcher(max_urls=5, timeout=5.0, resolver=_public) with respx.mock: respx.head("https://example.com/page").mock( return_value=httpx.Response( @@ -235,7 +257,7 @@ async def test_fetch_simple_text(self) -> None: @pytest.mark.asyncio async def test_fetch_binary_skips_body(self) -> None: - fetcher = SafeFetcher() + fetcher = SafeFetcher(resolver=_public) with respx.mock: respx.head("https://example.com/file.zip").mock( return_value=httpx.Response( @@ -254,7 +276,7 @@ async def test_fetch_binary_skips_body(self) -> None: @pytest.mark.asyncio async def test_fetch_respects_max_urls(self) -> None: - fetcher = SafeFetcher(max_urls=2) + fetcher = SafeFetcher(max_urls=2, resolver=_public) urls = [f"https://example.com/{i}" for i in range(5)] with respx.mock: @@ -279,7 +301,7 @@ async def test_fetch_respects_max_urls(self) -> None: @pytest.mark.asyncio async def test_fetch_handles_timeout(self) -> None: - fetcher = SafeFetcher(timeout=0.5) + fetcher = SafeFetcher(timeout=0.5, resolver=_public) with respx.mock: respx.head("https://slow.example.com/").mock( side_effect=httpx.ReadTimeout("timed out") @@ -294,7 +316,7 @@ async def test_fetch_handles_timeout(self) -> None: @pytest.mark.asyncio async def test_fetch_handles_connection_error(self) -> None: - fetcher = SafeFetcher() + fetcher = SafeFetcher(resolver=_public) with respx.mock: respx.head("https://dead.example.com/").mock( side_effect=httpx.ConnectError("Connection refused") @@ -307,7 +329,7 @@ async def test_fetch_handles_connection_error(self) -> None: @pytest.mark.asyncio async def test_fetch_too_large_content(self) -> None: - fetcher = SafeFetcher(max_bytes=1000) + fetcher = SafeFetcher(max_bytes=1000, resolver=_public) with respx.mock: respx.head("https://example.com/huge").mock( return_value=httpx.Response( @@ -327,7 +349,7 @@ async def test_fetch_too_large_content(self) -> None: @pytest.mark.asyncio async def test_fetch_redirect_chain(self) -> None: - fetcher = SafeFetcher() + fetcher = SafeFetcher(resolver=_public) with respx.mock: # HEAD follows redirect respx.head("https://short.url/x").mock( @@ -515,7 +537,7 @@ async def test_detector_with_known_malicious_url(self) -> None: body_markdown="Install from https://download.setup-service.com/tool.sh" ) context = ScanContext(skill=skill, scan_id="test-001") - detector = UrlCrawlerDetector(fetcher=SafeFetcher(max_urls=10, timeout=2.0)) + detector = UrlCrawlerDetector(fetcher=SafeFetcher(max_urls=10, timeout=2.0, resolver=_public)) # Mock the fetcher to avoid real HTTP with respx.mock: @@ -581,3 +603,120 @@ async def test_detector_properties(self) -> None: detector = UrlCrawlerDetector() assert detector.layer_name == DetectorLayer.URL_CRAWLER assert detector.order == 20 + + +# =================================================================== +# SSRF guard +# =================================================================== + + +class TestUrlSafety: + """Tests for malwar.detectors.url_crawler.safety. + + The crawler follows links out of skill content, and skill content is the + hostile input under analysis: URLs come from the body, from source_url and + author_url, and from arbitrary frontmatter keys. Before this guard existed, + scanning a skill containing the EC2 metadata address made the scanner fetch + instance credentials, on the default code path (use_urls=True). + """ + + @pytest.mark.parametrize( + "addr", + [ + "169.254.169.254", # cloud instance metadata + "127.0.0.1", # loopback + "10.0.0.5", # RFC1918 + "172.16.4.4", + "192.168.1.1", + "0.0.0.0", # unspecified # noqa: S104 + "::1", # v6 loopback + "fd00::1", # v6 unique-local + "fe80::1", # v6 link-local + ], + ) + def test_rejects_non_public_addresses(self, addr: str) -> None: + safe, reason = check_url("http://anything.example/", _resolves_to(addr)) + assert not safe, f"{addr} should be refused" + assert addr in reason or "non-public" in reason + + def test_rejects_ipv4_mapped_v6_smuggling(self) -> None: + # ::ffff:169.254.169.254 is the metadata address wearing a v6 costume. + safe, _ = check_url("http://anything.example/", _resolves_to("::ffff:169.254.169.254")) + assert not safe + + def test_rejects_non_http_schemes(self) -> None: + for url in ("file:///etc/passwd", "gopher://x/", "ftp://x/"): + safe, reason = check_url(url, _public) + assert not safe, url + assert "scheme" in reason + + def test_allows_ordinary_public_address(self) -> None: + safe, reason = check_url("https://example.com/x", _public) + assert safe, reason + + def test_dns_name_pointing_inward_is_refused(self) -> None: + # The attacker owns DNS for their own domain, so a textual check on the + # hostname is no check at all. This is why resolution happens first. + safe, _ = check_url("https://totally-normal.example/", _resolves_to("169.254.169.254")) + assert not safe + + @pytest.mark.asyncio + async def test_fetcher_refuses_metadata_address(self) -> None: + fetcher = SafeFetcher(resolver=_resolves_to("169.254.169.254")) + with respx.mock: + route = respx.head("http://169.254.169.254/latest/meta-data/").mock( + return_value=httpx.Response(200, headers={"content-type": "text/plain"}) + ) + results = await fetcher.fetch_urls(["http://169.254.169.254/latest/meta-data/"]) + + assert not route.called, "the request must never be sent" + assert results[0].error is not None + assert "Refused to fetch" in results[0].error + + @pytest.mark.asyncio + async def test_redirect_to_internal_address_is_refused(self) -> None: + # The bypass that matters: the entry point is public and passes, then + # hop two goes inward. Validating only the first URL catches nothing. + def resolver(host: str) -> list[ipaddress.IPv4Address]: + inward = host == "internal.example" + return [ipaddress.ip_address("169.254.169.254" if inward else "93.184.216.34")] + + fetcher = SafeFetcher(resolver=resolver) + with respx.mock: + respx.head("https://public.example/go").mock( + return_value=httpx.Response( + 302, headers={"location": "http://internal.example/creds"} + ) + ) + inner = respx.head("http://internal.example/creds").mock( + return_value=httpx.Response(200, headers={"content-type": "text/plain"}) + ) + results = await fetcher.fetch_urls(["https://public.example/go"]) + + assert not inner.called, "the redirect target must never be requested" + assert results[0].error is not None + assert "Refused to fetch" in results[0].error + + @pytest.mark.asyncio + async def test_public_redirect_is_still_followed(self) -> None: + # The guard must not break ordinary redirects. + fetcher = SafeFetcher(resolver=_public) + with respx.mock: + respx.head("https://a.example/1").mock( + return_value=httpx.Response(302, headers={"location": "https://b.example/2"}) + ) + respx.head("https://b.example/2").mock( + return_value=httpx.Response(200, headers={"content-type": "text/plain"}) + ) + respx.get("https://a.example/1").mock( + return_value=httpx.Response(302, headers={"location": "https://b.example/2"}) + ) + respx.get("https://b.example/2").mock( + return_value=httpx.Response(200, headers={"content-type": "text/plain"}, text="ok") + ) + results = await fetcher.fetch_urls(["https://a.example/1"]) + + assert results[0].error is None, results[0].error + assert results[0].final_url == "https://b.example/2" + assert results[0].redirect_chain == ["https://a.example/1"] + assert results[0].content == "ok"