scriptoriumadmin/strix
1
0
Fork 0
mirror of https://github.com/usestrix/strix.git synced 2026-10-10 03:28:11 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge d3968c95eb into 55bc07991a

Browse source
This commit is contained in:
@siundu 2026-10-05 16:05:20 +00:00 • committed by GitHub
commit 3b61b67e5f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 348 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
pyproject.toml
View file

@ -43,6 +43,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",

93
strix/interface/utils.py
View file

@ -5,6 +5,7 @@ import os
import re
import secrets
import shutil
import socket
import subprocess
import sys
import tempfile
@ -13,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
@ -1119,14 +1121,89 @@ def resolve_diff_scope_context(
)
def _is_http_git_repo(url: str) -> bool:
check_url = f"{url.rstrip('/')}/info/refs?service=git-upload-pack"
def _is_disallowed_probe_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
return (
ip.is_private
or ip.is_loopback
or ip.is_link_local
or ip.is_multicast
or ip.is_reserved
or ip.is_unspecified
)
def _resolve_pinned_probe_ip(hostname: str, port: int) -> str | None:
"""Resolve `hostname` once and return an allowed literal IP to connect to.
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.
"""
hostname = hostname.strip("[]")
try:
with requests.get(check_url, headers={"User-Agent": "git/2.43.0"}, timeout=10) as resp:
if resp.status_code >= 400:
return resp.status_code == 401
return "x-git-upload-pack-advertisement" in resp.headers.get("Content-Type", "")
except (requests.RequestException, ValueError):
ip = ipaddress.ip_address(hostname)
except ValueError:
pass
else:
return None if _is_disallowed_probe_ip(ip) else str(ip)
try:
addr_infos = socket.getaddrinfo(hostname, port, type=socket.SOCK_STREAM)
except (OSError, UnicodeError):
return None
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 None
def _is_http_git_repo(url: str) -> bool:
parsed = urlparse(url)
hostname = parsed.hostname
if not hostname or parsed.scheme not in ("http", "https"):
return False
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:
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

257
tests/test_ssrf_guard.py Normal file
View file

@ -0,0 +1,257 @@
"""Tests for the SSRF guard on the host-side git-repo probe in interface.utils.
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
import socket
from typing import TYPE_CHECKING, Any
import pytest
from strix.interface import utils
if TYPE_CHECKING:
from typing import Self
class _FakeResponse:
def __init__(self, status: int, headers: dict[str, str] | None = None) -> None:
self.status = status
self.headers = headers or {}
self.released = False
def release_conn(self) -> None:
self.released = True
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(
"host",
[
"127.0.0.1",
"127.53.0.1",
"::1",
"10.0.0.5",
"192.168.1.1",
"172.16.0.1",
"169.254.169.254", # cloud metadata endpoint
"0.0.0.0", # nosec B104
"224.0.0.1", # multicast
],
)
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_resolve_pinned_probe_ip_accepts_public_ip_literals(host: str) -> None:
assert utils._resolve_pinned_probe_ip(host, 443) == host
def test_resolve_pinned_probe_ip_rejects_hostname_resolving_only_to_private_ips(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(
socket,
"getaddrinfo",
lambda *_a, **_kw: [(None, None, None, "", ("10.1.2.3", 0))],
)
assert utils._resolve_pinned_probe_ip("internal.corp.example", 443) is None
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, "", ("169.254.169.254", 0)),
(None, None, None, "", ("93.184.216.34", 0)),
],
)
assert utils._resolve_pinned_probe_ip("example.com", 443) == "93.184.216.34"
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._resolve_pinned_probe_ip("nonexistent.invalid", 443) is None
# --- _is_http_git_repo ---------------------------------------------------------
def test_is_http_git_repo_does_not_connect_to_disallowed_hosts(
monkeypatch: pytest.MonkeyPatch,
) -> None:
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))
assert utils._is_http_git_repo("http://metadata.internal/latest/meta-data/") is False
assert instances == []
assert calls == []
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(
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(_FakeResponse(401), calls, instances))
assert utils._is_http_git_repo("https://internal.example/service") is False
def test_is_http_git_repo_does_not_follow_redirects(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(_FakeResponse(200), calls, instances))
utils._is_http_git_repo("https://example.com/some/repo")
assert calls[0]["redirect"] is False
def test_is_http_git_repo_accepts_genuine_200_git_response(
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] = []
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

@ -2391,6 +2391,7 @@ source = { editable = "." }
dependencies = [
{ name = "boto3" },
{ name = "caido-sdk-client" },
{ name = "certifi" },
{ name = "cryptography" },
{ name = "cvss" },
{ name = "docker" },
@ -2406,6 +2407,7 @@ dependencies = [
{ name = "reportlab" },
{ name = "requests" },
{ name = "rich" },
{ name = "urllib3" },
]
[package.dev-dependencies]
@ -2425,6 +2427,7 @@ dev = [
requires-dist = [
{ name = "boto3", 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" },
@ -2440,6 +2443,7 @@ requires-dist = [
{ name = "reportlab", specifier = ">=4.0" },
{ name = "requests", specifier = ">=2.32.0" },
{ name = "rich" },
{ name = "urllib3", specifier = ">=2.0.0" },
]
[package.metadata.requires-dev]