diff --git a/pyproject.toml b/pyproject.toml index d63d8452..abcf9e18 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -41,6 +41,8 @@ dependencies = [ "rich", "docker>=7.1.0", "requests>=2.32.0", + "urllib3>=2.0.0", + "certifi>=2024.2.2", "cvss>=3.2", "caido-sdk-client>=0.2.0", "markdown-it-py>=3.0.0", diff --git a/strix/interface/utils.py b/strix/interface/utils.py index 3681ae5e..a2c5da73 100644 --- a/strix/interface/utils.py +++ b/strix/interface/utils.py @@ -14,7 +14,8 @@ from pathlib import Path from typing import Any from urllib.parse import parse_qs, urlparse -import requests +import certifi +import urllib3 from rich.console import Console from rich.panel import Panel from rich.text import Text @@ -1131,51 +1132,78 @@ def _is_disallowed_probe_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) - ) -def _is_ssrf_safe_host(host: str) -> bool: - """Reject hosts that resolve to loopback/private/link-local/reserved addresses. +def _resolve_pinned_probe_ip(hostname: str, port: int) -> str | None: + """Resolve `hostname` once and return an allowed literal IP to connect to. - `_is_http_git_repo` probes from the Strix host itself, so a target string an - attacker influences (e.g. a scan target read from a poisoned file, or a URL - forwarded by an automated pipeline) must not be able to make that probe reach - internal infrastructure. + The caller must open its connection to this exact IP rather than letting + the HTTP client re-resolve `hostname` itself. Checking the hostname and + then connecting to it separately (as a plain `requests.get(url)` call + would) leaves a DNS-rebinding gap: a malicious DNS server can answer the + validation lookup with a public IP and the connection's own lookup, + moments later, with a private one, since nothing pins the two lookups to + the same answer. """ - host = host.strip("[]") + hostname = hostname.strip("[]") try: - ip = ipaddress.ip_address(host) + ip = ipaddress.ip_address(hostname) except ValueError: pass else: - return not _is_disallowed_probe_ip(ip) + return None if _is_disallowed_probe_ip(ip) else str(ip) try: - addr_infos = socket.getaddrinfo(host, None) + addr_infos = socket.getaddrinfo(hostname, port, type=socket.SOCK_STREAM) except (OSError, UnicodeError): - return False + return None - resolved_ips = {info[4][0] for info in addr_infos} - if not resolved_ips: - return False + for _family, _socktype, _proto, _canon, sockaddr in addr_infos: + candidate = ipaddress.ip_address(sockaddr[0]) + if not _is_disallowed_probe_ip(candidate): + return str(candidate) - return all(not _is_disallowed_probe_ip(ipaddress.ip_address(addr)) for addr in resolved_ips) + return None def _is_http_git_repo(url: str) -> bool: - hostname = urlparse(url).hostname - if not hostname or not _is_ssrf_safe_host(hostname): + parsed = urlparse(url) + hostname = parsed.hostname + if not hostname or parsed.scheme not in ("http", "https"): return False - check_url = f"{url.rstrip('/')}/info/refs?service=git-upload-pack" + port = parsed.port or (443 if parsed.scheme == "https" else 80) + pinned_ip = _resolve_pinned_probe_ip(hostname, port) + if pinned_ip is None: + return False + + request_path = f"{parsed.path.rstrip('/')}/info/refs?service=git-upload-pack" + headers = {"User-Agent": "git/2.43.0", "Host": hostname} + try: - with requests.get( - check_url, - headers={"User-Agent": "git/2.43.0"}, - timeout=10, - allow_redirects=False, - ) as resp: - if resp.status_code != 200: - return False - return "x-git-upload-pack-advertisement" in resp.headers.get("Content-Type", "") - except (requests.RequestException, ValueError): + pool: urllib3.HTTPConnectionPool + if parsed.scheme == "https": + pool = urllib3.HTTPSConnectionPool( + pinned_ip, + port, + server_hostname=hostname, + assert_hostname=hostname, + ca_certs=certifi.where(), + timeout=10, + retries=False, + ) + else: + pool = urllib3.HTTPConnectionPool(pinned_ip, port, timeout=10, retries=False) + + with pool: + resp = pool.request( + "GET", request_path, headers=headers, redirect=False, preload_content=False + ) + try: + if resp.status != 200: + return False + return "x-git-upload-pack-advertisement" in resp.headers.get("Content-Type", "") + finally: + resp.release_conn() + except (urllib3.exceptions.HTTPError, OSError, ValueError): return False diff --git a/tests/test_ssrf_guard.py b/tests/test_ssrf_guard.py index 2e8b2fb1..2b9af73d 100644 --- a/tests/test_ssrf_guard.py +++ b/tests/test_ssrf_guard.py @@ -3,6 +3,12 @@ Covers https://github.com/usestrix/strix/issues/1132: `_is_http_git_repo` runs on the Strix host (not the sandbox), so an attacker-influenced target string must not be able to use it to probe internal/private network services. + +The guard must also resist DNS rebinding: the address that gets validated has +to be the exact address the probe connects to. A naive "resolve, check, then +`requests.get(url)`" implementation re-resolves the hostname to actually +connect, giving a malicious DNS server a second lookup — answered with a +private address — to bypass the check made against the first one. """ from __future__ import annotations @@ -11,7 +17,6 @@ import socket from typing import TYPE_CHECKING, Any import pytest -import requests from strix.interface import utils @@ -21,15 +26,50 @@ if TYPE_CHECKING: class _FakeResponse: - def __init__(self, status_code: int, headers: dict[str, str] | None = None) -> None: - self.status_code = status_code + def __init__(self, status: int, headers: dict[str, str] | None = None) -> None: + self.status = status self.headers = headers or {} + self.released = False - def __enter__(self) -> Self: - return self + def release_conn(self) -> None: + self.released = True - def __exit__(self, *_exc: object) -> None: - return None + +def _make_fake_pool_class( + response: _FakeResponse | Exception, + calls: list[dict[str, Any]], + instances: list[Any], +) -> type: + class _FakePool: + def __init__(self, host: str, port: int, **kwargs: Any) -> None: + self.host = host + self.port = port + self.kwargs = kwargs + instances.append(self) + + def __enter__(self) -> Self: + return self + + def __exit__(self, *_exc: object) -> None: + return None + + def request( + self, method: str, path: str, headers: dict[str, str] | None = None, **kwargs: Any + ) -> _FakeResponse: + calls.append({"method": method, "path": path, "headers": headers, **kwargs}) + if isinstance(response, Exception): + raise response + return response + + return _FakePool + + +def _patch_pools(monkeypatch: pytest.MonkeyPatch, pool_class: type) -> None: + monkeypatch.setattr(utils.urllib3, "HTTPSConnectionPool", pool_class) + monkeypatch.setattr(utils.urllib3, "HTTPConnectionPool", pool_class) + + +# --- _resolve_pinned_probe_ip ------------------------------------------------- @pytest.mark.parametrize( @@ -46,16 +86,16 @@ class _FakeResponse: "224.0.0.1", # multicast ], ) -def test_is_ssrf_safe_host_rejects_disallowed_ip_literals(host: str) -> None: - assert utils._is_ssrf_safe_host(host) is False +def test_resolve_pinned_probe_ip_rejects_disallowed_ip_literals(host: str) -> None: + assert utils._resolve_pinned_probe_ip(host, 443) is None @pytest.mark.parametrize("host", ["8.8.8.8", "1.1.1.1", "93.184.216.34"]) -def test_is_ssrf_safe_host_accepts_public_ip_literals(host: str) -> None: - assert utils._is_ssrf_safe_host(host) is True +def test_resolve_pinned_probe_ip_accepts_public_ip_literals(host: str) -> None: + assert utils._resolve_pinned_probe_ip(host, 443) == host -def test_is_ssrf_safe_host_rejects_hostname_resolving_to_private_ip( +def test_resolve_pinned_probe_ip_rejects_hostname_resolving_only_to_private_ips( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( @@ -63,44 +103,95 @@ def test_is_ssrf_safe_host_rejects_hostname_resolving_to_private_ip( "getaddrinfo", lambda *_a, **_kw: [(None, None, None, "", ("10.1.2.3", 0))], ) - assert utils._is_ssrf_safe_host("internal.corp.example") is False + assert utils._resolve_pinned_probe_ip("internal.corp.example", 443) is None -def test_is_ssrf_safe_host_accepts_hostname_resolving_to_public_ip( +def test_resolve_pinned_probe_ip_returns_the_first_allowed_resolved_address( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( socket, "getaddrinfo", - lambda *_a, **_kw: [(None, None, None, "", ("93.184.216.34", 0))], + lambda *_a, **_kw: [ + (None, None, None, "", ("169.254.169.254", 0)), + (None, None, None, "", ("93.184.216.34", 0)), + ], ) - assert utils._is_ssrf_safe_host("example.com") is True + assert utils._resolve_pinned_probe_ip("example.com", 443) == "93.184.216.34" -def test_is_ssrf_safe_host_rejects_unresolvable_hostname( +def test_resolve_pinned_probe_ip_rejects_unresolvable_hostname( monkeypatch: pytest.MonkeyPatch, ) -> None: def _boom(*_a: object, **_kw: object) -> Any: raise OSError("name resolution failed") monkeypatch.setattr(socket, "getaddrinfo", _boom) - assert utils._is_ssrf_safe_host("nonexistent.invalid") is False + assert utils._resolve_pinned_probe_ip("nonexistent.invalid", 443) is None -def test_is_http_git_repo_does_not_probe_disallowed_hosts( +# --- _is_http_git_repo --------------------------------------------------------- + + +def test_is_http_git_repo_does_not_connect_to_disallowed_hosts( monkeypatch: pytest.MonkeyPatch, ) -> None: - called = False + monkeypatch.setattr( + socket, + "getaddrinfo", + lambda *_a, **_kw: [(None, None, None, "", ("169.254.169.254", 0))], + ) + calls: list[dict[str, Any]] = [] + instances: list[Any] = [] + _patch_pools(monkeypatch, _make_fake_pool_class(_FakeResponse(200), calls, instances)) - def _fake_get(*_a: object, **_kw: object) -> _FakeResponse: - nonlocal called - called = True - return _FakeResponse(200) + assert utils._is_http_git_repo("http://metadata.internal/latest/meta-data/") is False + assert instances == [] + assert calls == [] - monkeypatch.setattr(requests, "get", _fake_get) - assert utils._is_http_git_repo("http://169.254.169.254/latest/meta-data/") is False - assert called is False +def test_is_http_git_repo_rejects_non_http_schemes_without_resolving( + monkeypatch: pytest.MonkeyPatch, +) -> None: + def _boom(*_a: object, **_kw: object) -> Any: + raise AssertionError("must not resolve a non-http(s) scheme") + + monkeypatch.setattr(socket, "getaddrinfo", _boom) + assert utils._is_http_git_repo("ftp://example.com/repo") is False + + +def test_is_http_git_repo_pins_the_connection_to_the_validated_address( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The core regression test for the DNS-rebinding bypass. + + A rebinding DNS server answers the first (validation) lookup with a + public address and would answer any later lookup with a private one. The + probe must connect to the address it already validated instead of + re-resolving the hostname, so it must end up talking to the public + address here even though a second lookup would be unsafe. + """ + lookups = 0 + + def _rebinding_getaddrinfo(*_a: object, **_kw: object) -> list[Any]: + nonlocal lookups + lookups += 1 + ip = "93.184.216.34" if lookups == 1 else "10.0.0.1" + return [(None, None, None, "", (ip, 0))] + + monkeypatch.setattr(socket, "getaddrinfo", _rebinding_getaddrinfo) + + calls: list[dict[str, Any]] = [] + instances: list[Any] = [] + response = _FakeResponse(200, {"Content-Type": "application/x-git-upload-pack-advertisement"}) + _patch_pools(monkeypatch, _make_fake_pool_class(response, calls, instances)) + + assert utils._is_http_git_repo("https://example.com/org/repo") is True + assert len(instances) == 1 + assert instances[0].host == "93.184.216.34" + assert instances[0].kwargs.get("server_hostname") == "example.com" + assert instances[0].kwargs.get("assert_hostname") == "example.com" + assert calls[0]["headers"]["Host"] == "example.com" def test_is_http_git_repo_no_longer_treats_401_as_a_repo_signal( @@ -111,7 +202,9 @@ def test_is_http_git_repo_no_longer_treats_401_as_a_repo_signal( "getaddrinfo", lambda *_a, **_kw: [(None, None, None, "", ("93.184.216.34", 0))], ) - monkeypatch.setattr(requests, "get", lambda *_a, **_kw: _FakeResponse(401)) + calls: list[dict[str, Any]] = [] + instances: list[Any] = [] + _patch_pools(monkeypatch, _make_fake_pool_class(_FakeResponse(401), calls, instances)) assert utils._is_http_git_repo("https://internal.example/service") is False @@ -122,16 +215,12 @@ def test_is_http_git_repo_does_not_follow_redirects(monkeypatch: pytest.MonkeyPa "getaddrinfo", lambda *_a, **_kw: [(None, None, None, "", ("93.184.216.34", 0))], ) - captured_kwargs: dict[str, Any] = {} - - def _fake_get(*_a: object, **kwargs: object) -> _FakeResponse: - captured_kwargs.update(kwargs) - return _FakeResponse(200, {"Content-Type": "application/x-git-upload-pack-advertisement"}) - - monkeypatch.setattr(requests, "get", _fake_get) + calls: list[dict[str, Any]] = [] + instances: list[Any] = [] + _patch_pools(monkeypatch, _make_fake_pool_class(_FakeResponse(200), calls, instances)) utils._is_http_git_repo("https://example.com/some/repo") - assert captured_kwargs.get("allow_redirects") is False + assert calls[0]["redirect"] is False def test_is_http_git_repo_accepts_genuine_200_git_response( @@ -142,12 +231,27 @@ def test_is_http_git_repo_accepts_genuine_200_git_response( "getaddrinfo", lambda *_a, **_kw: [(None, None, None, "", ("93.184.216.34", 0))], ) - monkeypatch.setattr( - requests, - "get", - lambda *_a, **_kw: _FakeResponse( - 200, {"Content-Type": "application/x-git-upload-pack-advertisement"} - ), - ) + calls: list[dict[str, Any]] = [] + instances: list[Any] = [] + response = _FakeResponse(200, {"Content-Type": "application/x-git-upload-pack-advertisement"}) + _patch_pools(monkeypatch, _make_fake_pool_class(response, calls, instances)) assert utils._is_http_git_repo("https://example.com/some/repo") is True + + +def test_is_http_git_repo_returns_false_on_connection_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + socket, + "getaddrinfo", + lambda *_a, **_kw: [(None, None, None, "", ("93.184.216.34", 0))], + ) + calls: list[dict[str, Any]] = [] + instances: list[Any] = [] + _patch_pools( + monkeypatch, + _make_fake_pool_class(utils.urllib3.exceptions.HTTPError("boom"), calls, instances), + ) + + assert utils._is_http_git_repo("https://example.com/some/repo") is False diff --git a/uv.lock b/uv.lock index 3c3a6b41..5acd1ddf 100644 --- a/uv.lock +++ b/uv.lock @@ -2390,6 +2390,7 @@ version = "1.6.2" source = { editable = "." } dependencies = [ { name = "caido-sdk-client" }, + { name = "certifi" }, { name = "cryptography" }, { name = "cvss" }, { name = "docker" }, @@ -2404,6 +2405,7 @@ dependencies = [ { name = "reportlab" }, { name = "requests" }, { name = "rich" }, + { name = "urllib3" }, ] [package.optional-dependencies] @@ -2431,6 +2433,7 @@ dev = [ requires-dist = [ { name = "boto3", marker = "extra == 'bedrock'", specifier = ">=1.28.0" }, { name = "caido-sdk-client", specifier = ">=0.2.0" }, + { name = "certifi", specifier = ">=2024.2.2" }, { name = "cryptography", specifier = ">=48.0.1,<49" }, { name = "cvss", specifier = ">=3.2" }, { name = "docker", specifier = ">=7.1.0" }, @@ -2446,6 +2449,7 @@ requires-dist = [ { name = "reportlab", specifier = ">=4.0" }, { name = "requests", specifier = ">=2.32.0" }, { name = "rich" }, + { name = "urllib3", specifier = ">=2.0.0" }, ] provides-extras = ["vertex", "bedrock"]