scriptoriumadmin/strix
1
0
Fork 0
mirror of https://github.com/usestrix/strix.git synced 2026-09-23 00:41:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix(interface): pin git-repo probe connection to resolve DNS-rebinding bypass

Browse source 
The previous SSRF guard validated the target hostname and then let
requests.get() resolve and connect to it separately. That leaves a
DNS-rebinding window: a malicious DNS server can answer the validation
lookup with a public address and the connection's own lookup, moments
later, with a private one, since nothing pins the two lookups together.

Resolve the hostname once via _resolve_pinned_probe_ip, validate that
address, and connect directly to it with urllib3's connection pools
(passing server_hostname/assert_hostname for TLS SNI and certificate
hostname verification, and an explicit Host header) instead of handing
the hostname back to an HTTP client that would re-resolve it.

Adds a regression test that simulates a rebinding DNS server (public
address on the first lookup, private on any later one) and asserts the
probe still connects to the address it already validated.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
siundu254 2026-09-16 14:10:00 +03:00
parent bf621579dd
commit d3968c95eb
No known key found for this signature in database
GPG key ID: A97023C381E1DA5C
4 changed files with 210 additions and 72 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
pyproject.toml
View file

@ -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",

86
strix/interface/utils.py
View file

@ -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

190
tests/test_ssrf_guard.py
View file

@ -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

4
uv.lock generated
View file

@ -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"]