mirror of
https://github.com/usestrix/strix.git
synced 2026-09-23 00:41:50 +00:00
fix(interface): prevent SSRF in host-side git-repo probe
_is_http_git_repo runs from the Strix host itself (not the sandbox) to check whether a target URL looks like a git repo. It followed redirects and treated HTTP 401 as a positive "is a git repo" signal, so an attacker-influenced target string could make Strix's own infrastructure probe internal/private services (a classic SSRF pattern), including via a 401-gated redirect chain. Add _is_ssrf_safe_host to reject targets that resolve to loopback/private/link-local/reserved/multicast addresses before probing, disable redirect-following on the probe request, and only treat a genuine 200 response with the expected git content-type as a positive signal. Fixes usestrix/strix#1132 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
parent
2dadbb748a
commit
bf621579dd
2 changed files with 205 additions and 3 deletions
|
|
@ -5,6 +5,7 @@ import os
|
|||
import re
|
||||
import secrets
|
||||
import shutil
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
|
|
@ -1119,12 +1120,60 @@ def resolve_diff_scope_context(
|
|||
)
|
||||
|
||||
|
||||
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 _is_ssrf_safe_host(host: str) -> bool:
|
||||
"""Reject hosts that resolve to loopback/private/link-local/reserved addresses.
|
||||
|
||||
`_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.
|
||||
"""
|
||||
host = host.strip("[]")
|
||||
try:
|
||||
ip = ipaddress.ip_address(host)
|
||||
except ValueError:
|
||||
pass
|
||||
else:
|
||||
return not _is_disallowed_probe_ip(ip)
|
||||
|
||||
try:
|
||||
addr_infos = socket.getaddrinfo(host, None)
|
||||
except (OSError, UnicodeError):
|
||||
return False
|
||||
|
||||
resolved_ips = {info[4][0] for info in addr_infos}
|
||||
if not resolved_ips:
|
||||
return False
|
||||
|
||||
return all(not _is_disallowed_probe_ip(ipaddress.ip_address(addr)) for addr in resolved_ips)
|
||||
|
||||
|
||||
def _is_http_git_repo(url: str) -> bool:
|
||||
hostname = urlparse(url).hostname
|
||||
if not hostname or not _is_ssrf_safe_host(hostname):
|
||||
return False
|
||||
|
||||
check_url = f"{url.rstrip('/')}/info/refs?service=git-upload-pack"
|
||||
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
|
||||
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):
|
||||
return False
|
||||
|
|
|
|||
153
tests/test_ssrf_guard.py
Normal file
153
tests/test_ssrf_guard.py
Normal file
|
|
@ -0,0 +1,153 @@
|
|||
"""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.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import socket
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from strix.interface import utils
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from typing import Self
|
||||
|
||||
|
||||
class _FakeResponse:
|
||||
def __init__(self, status_code: int, headers: dict[str, str] | None = None) -> None:
|
||||
self.status_code = status_code
|
||||
self.headers = headers or {}
|
||||
|
||||
def __enter__(self) -> Self:
|
||||
return self
|
||||
|
||||
def __exit__(self, *_exc: object) -> None:
|
||||
return None
|
||||
|
||||
|
||||
@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_is_ssrf_safe_host_rejects_disallowed_ip_literals(host: str) -> None:
|
||||
assert utils._is_ssrf_safe_host(host) is False
|
||||
|
||||
|
||||
@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_is_ssrf_safe_host_rejects_hostname_resolving_to_private_ip(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.setattr(
|
||||
socket,
|
||||
"getaddrinfo",
|
||||
lambda *_a, **_kw: [(None, None, None, "", ("10.1.2.3", 0))],
|
||||
)
|
||||
assert utils._is_ssrf_safe_host("internal.corp.example") is False
|
||||
|
||||
|
||||
def test_is_ssrf_safe_host_accepts_hostname_resolving_to_public_ip(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.setattr(
|
||||
socket,
|
||||
"getaddrinfo",
|
||||
lambda *_a, **_kw: [(None, None, None, "", ("93.184.216.34", 0))],
|
||||
)
|
||||
assert utils._is_ssrf_safe_host("example.com") is True
|
||||
|
||||
|
||||
def test_is_ssrf_safe_host_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
|
||||
|
||||
|
||||
def test_is_http_git_repo_does_not_probe_disallowed_hosts(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
called = False
|
||||
|
||||
def _fake_get(*_a: object, **_kw: object) -> _FakeResponse:
|
||||
nonlocal called
|
||||
called = True
|
||||
return _FakeResponse(200)
|
||||
|
||||
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_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))],
|
||||
)
|
||||
monkeypatch.setattr(requests, "get", lambda *_a, **_kw: _FakeResponse(401))
|
||||
|
||||
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))],
|
||||
)
|
||||
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)
|
||||
|
||||
utils._is_http_git_repo("https://example.com/some/repo")
|
||||
assert captured_kwargs.get("allow_redirects") 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))],
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
requests,
|
||||
"get",
|
||||
lambda *_a, **_kw: _FakeResponse(
|
||||
200, {"Content-Type": "application/x-git-upload-pack-advertisement"}
|
||||
),
|
||||
)
|
||||
|
||||
assert utils._is_http_git_repo("https://example.com/some/repo") is True
|
||||
Loading…
Add table
Reference in a new issue