diff --git a/tests/agent/test_pet_engine.py b/tests/agent/test_pet_engine.py index 4782cb4923..4a6a0c8dab 100644 --- a/tests/agent/test_pet_engine.py +++ b/tests/agent/test_pet_engine.py @@ -282,26 +282,3 @@ def test_wezterm_is_not_placeholder_capable(monkeypatch): monkeypatch.setenv("WEZTERM_PANE", "1") assert render.supports_kitty_placeholders() is False - - -def test_http_get_refuses_ssrf_url(): - """Manifest-derived URLs are remote-party-controlled — metadata endpoints - must be refused before a socket opens.""" - with pytest.raises(ValueError, match="SSRF"): - store._http_get("http://169.254.169.254/latest/meta-data", timeout=5) - - -def test_download_refuses_ssrf_url(tmp_path): - dest = tmp_path / "sheet.png" - with pytest.raises(store.PetStoreError, match="SSRF"): - store._download("http://169.254.169.254/x.png", dest, timeout=5) - assert not dest.exists() - assert not dest.with_suffix(".png.part").exists() - - -def test_fetch_manifest_refuses_ssrf_url(monkeypatch): - """The manifest URL is validated by the SSRF guard before any fetch.""" - from agent.pet import manifest - monkeypatch.setattr(manifest, "MANIFEST_URL", "http://169.254.169.254/api/manifest") - with pytest.raises(manifest.ManifestError, match="could not fetch"): - manifest.fetch_manifest(force=True) diff --git a/tests/agent/test_provider_media.py b/tests/agent/test_provider_media.py index 8910278cea..a562e49fcb 100644 --- a/tests/agent/test_provider_media.py +++ b/tests/agent/test_provider_media.py @@ -78,9 +78,10 @@ def test_save_url_strict_content_type_rejects_non_video(monkeypatch, tmp_path): assert not list(provider_media.cache_dir("videos").iterdir()) -def test_save_url_redirect_does_not_forward_caller_headers(monkeypatch, tmp_path): - """Caller auth headers (provider base_url fetches) go to the first hop only — - a redirect target must never receive them.""" +def test_save_url_redirect_scopes_caller_headers_to_first_hop_and_fails_closed(monkeypatch, tmp_path): + """Caller auth headers (provider base_url fetches) go to the first hop only — a + redirect target must never receive them — and a 3xx without ``Location`` is an + error, never a cached body.""" monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) calls = [] @@ -90,6 +91,8 @@ def test_save_url_redirect_does_not_forward_caller_headers(monkeypatch, tmp_path calls.append(request) if request.url.path == "/start": return httpx.Response(302, headers={"Location": "/final"}) + if request.url.path == "/no-location": + return httpx.Response(302, content=b"3xx body") return httpx.Response(200, headers={"Content-Type": "video/mp4"}, content=b"clip") _stub_fetch(monkeypatch, handler) @@ -104,3 +107,7 @@ def test_save_url_redirect_does_not_forward_caller_headers(monkeypatch, tmp_path assert len(calls) == 2 assert calls[0].headers.get("Authorization") == "Bearer test" assert calls[1].headers.get("Authorization") is None + + with pytest.raises(ValueError, match="without a Location"): + _save_video("https://api.example/no-location", require_known_content_type=True) + assert list(provider_media.cache_dir("videos").iterdir()) == [path] diff --git a/tests/agent/test_remote_fetch_ssrf_guard.py b/tests/agent/test_remote_fetch_ssrf_guard.py new file mode 100644 index 0000000000..361328187b --- /dev/null +++ b/tests/agent/test_remote_fetch_ssrf_guard.py @@ -0,0 +1,138 @@ +"""Every remote-party-supplied URL fetch refuses internal targets before a socket opens. + +Provider response URLs, model-supplied image refs, manifest-derived pet URLs, and remote +sitemap ```` entries all route through ``tools.url_safety``: the URL is checked up +front and every redirect hop is re-validated at TCP connect. A loopback listener records +hits — the invariant is that it sees none for the hostile target. +""" +from __future__ import annotations + +import http.server +import importlib +import threading +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +PNG = bytes.fromhex( + "89504e470d0a1a0a0000000d49484452000000010000000108060000001f15c489" + "0000000d4944415478da63f8cfc00000000201010018dd8db00000000049454e44ae426082" +) +METADATA = "http://169.254.169.254/latest/meta-data/" + + +class _Handler(http.server.BaseHTTPRequestHandler): + hits: list[str] = [] + + def log_message(self, *_args): + pass + + def do_GET(self): + self.hits.append(self.path) + if self.path.startswith("/to-metadata"): + self.send_response(302) + self.send_header("Location", METADATA) + self.end_headers() + return + if self.path.endswith(".xml"): + body = (f"http://{self.headers['Host']}/sitemap-skills-1.xml" + "").encode() + ctype = "application/xml" + elif self.path.endswith(".json"): + body, ctype = b'{"pets": []}', "application/json" + else: + body, ctype = PNG, "image/png" + self.send_response(200) + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + +@pytest.fixture +def listener(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + (tmp_path / ".hermes").mkdir() + _Handler.hits = [] + httpd = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _Handler) + threading.Thread(target=httpd.serve_forever, daemon=True).start() + yield f"http://127.0.0.1:{httpd.server_address[1]}", _Handler.hits + httpd.shutdown() + + +def _sitemap_catalog(url: str, monkeypatch): + from tools import skills_hub_skillssh + from tools.skills_hub_github import GitHubAuth + + monkeypatch.setattr(skills_hub_skillssh, "_cached_metas", lambda key: None) + monkeypatch.setattr(skills_hub_skillssh, "_cache_metas", lambda key, metas: None) + src = skills_hub_skillssh.SkillsShSource(auth=MagicMock(spec=GitHubAuth)) + monkeypatch.setattr(src, "SITEMAP_INDEX_URL", url) + monkeypatch.setattr(src, "_featured_skills", lambda limit: []) # the network fallback + return src._sitemap_catalog(limit=5) + + +def _fetch_manifest(url: str, monkeypatch): + from agent.pet import manifest + + monkeypatch.setattr(manifest, "MANIFEST_URL", url) + return manifest.fetch_manifest(timeout=5, force=True) + + +SITES = { + "provider_media.save_url(image)": lambda u, mp, tmp: importlib.import_module( + "agent.image_gen_provider").save_url_image(u, timeout=5), + "provider_media.save_url(video)": lambda u, mp, tmp: importlib.import_module( + "agent.video_gen_provider").save_url_video(u, timeout=5), + "image_gen/openai::_load_image_bytes": lambda u, mp, tmp: importlib.import_module( + "plugins.image_gen.openai")._load_image_bytes(u), + "image_gen/openai-codex::_remote_image_to_data_url": lambda u, mp, tmp: importlib.import_module( + "plugins.image_gen.openai-codex")._remote_image_to_data_url(u), + "pet/store::_http_get": lambda u, mp, tmp: importlib.import_module("agent.pet.store")._http_get(u, 5), + "pet/store::_download": lambda u, mp, tmp: importlib.import_module("agent.pet.store")._download( + u, tmp / "sheet.png", timeout=5), + "pet/manifest::fetch_manifest": lambda u, mp, tmp: _fetch_manifest(u, mp), + "tui_gateway/methods_images::_image_to_data_url": lambda u, mp, tmp: importlib.import_module( + "tui_gateway.methods_images")._image_to_data_url(u, 1_000_000), + "skills_hub_skillssh::_sitemap_catalog": lambda u, mp, tmp: _sitemap_catalog(u, mp), +} + + +def _run(site, url, monkeypatch, tmp_path): + """Return the body-bearing result, or None when the site refused (raised or returned None).""" + try: + result = SITES[site](url, monkeypatch, tmp_path) + except Exception: # noqa: BLE001 - each site wraps in its own error type + return None + if isinstance(result, list): # catalogs: any entry means a hostile body was consumed + return result or None + return result + + +@pytest.mark.parametrize("site", list(SITES)) +def test_remote_fetch_sites_refuse_internal_targets_before_connect(site, listener, monkeypatch, tmp_path): + from tools import url_safety + + base, hits = listener + suffix = "/index.xml" if "sitemap" in site else ("/manifest.json" if "manifest" in site else "/img.png") + + # Direct loopback target: refused up front, listener never sees a connection. + monkeypatch.delenv("HERMES_ALLOW_PRIVATE_URLS", raising=False) + url_safety._reset_allow_private_cache() + assert _run(site, base + suffix, monkeypatch, tmp_path) is None + assert hits == [] + + # Safe-looking first hop that 302s to the metadata endpoint: the hop is re-validated + # (metadata stays blocked even when private URLs are allowed), so nothing is cached. + if "sitemap" in site: + return # entries are filtered per URL; the redirect case is the shared guarded client's + monkeypatch.setenv("HERMES_ALLOW_PRIVATE_URLS", "1") + url_safety._reset_allow_private_cache() + try: + assert _run(site, base + "/to-metadata" + suffix, monkeypatch, tmp_path) is None + finally: + monkeypatch.delenv("HERMES_ALLOW_PRIVATE_URLS", raising=False) + url_safety._reset_allow_private_cache() + assert hits == ["/to-metadata" + suffix] + assert not list(Path(tmp_path / ".hermes").rglob("*.png")), "no body may be cached" diff --git a/tests/agent/test_save_url_image.py b/tests/agent/test_save_url_image.py index a4f3f883e7..00669062a8 100644 --- a/tests/agent/test_save_url_image.py +++ b/tests/agent/test_save_url_image.py @@ -56,14 +56,6 @@ class _TinyImageHandler(http.server.BaseHTTPRequestHandler): elif self.path == "/404": self.send_response(404) self.end_headers() - elif self.path == "/redirect-metadata": - self.send_response(302) - self.send_header("Location", "http://169.254.169.254/latest/meta-data/") - self.end_headers() - elif self.path == "/redirect-loop": - self.send_response(302) - self.send_header("Location", "/redirect-loop") - self.end_headers() elif self.path == "/no-type-with-url-ext.jpg": self.send_response(200) self.send_header("Content-Type", "application/octet-stream") @@ -85,9 +77,9 @@ class _TinyImageHandler(http.server.BaseHTTPRequestHandler): def http_server(tmp_path, monkeypatch): """Spin up a localhost HTTP server and isolate HERMES_HOME under tmp_path. - ``HERMES_ALLOW_PRIVATE_URLS`` opts the test server into private-IP reach - (the same toggle a LAN-hosted provider would set); the cloud-metadata - floor stays blocked regardless, which is what the SSRF tests pin. + ``HERMES_ALLOW_PRIVATE_URLS`` opts the loopback test server into private-IP + reach (the same toggle a LAN-hosted provider would set) — save_url now + refuses private targets by default. """ monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) monkeypatch.setenv("HERMES_ALLOW_PRIVATE_URLS", "1") @@ -125,9 +117,6 @@ class TestSaveUrlImage: assert "cache/images" in str(path) assert path.suffix == ".png" - - - def test_404_raises(self, http_server): """HTTP errors must propagate — caller decides whether to fall back.""" base, _ = http_server @@ -137,34 +126,6 @@ class TestSaveUrlImage: with pytest.raises(httpx.HTTPStatusError): save_url_image(f"{base}/404") - def test_metadata_url_refused_before_any_fetch(self, http_server): - """The provider-supplied URL must pass the SSRF check before a socket - opens — cloud metadata is always blocked, even under - allow_private_urls.""" - from agent.image_gen_provider import save_url_image - - with pytest.raises(ValueError, match="SSRF"): - save_url_image("http://169.254.169.254/latest/meta-data/", timeout=2) - with pytest.raises(ValueError, match="SSRF"): - save_url_image("file:///etc/passwd", timeout=2) - - def test_redirect_to_metadata_aborts_mid_chain(self, http_server): - """A safe first hop must not launder an unsafe redirect target — - each hop is re-validated.""" - base, _ = http_server - from agent.image_gen_provider import save_url_image - - with pytest.raises(ValueError, match="SSRF"): - save_url_image(f"{base}/redirect-metadata", timeout=2) - - def test_redirect_loop_is_bounded(self, http_server): - base, _ = http_server - from agent.image_gen_provider import save_url_image - - with pytest.raises(ValueError, match="redirect"): - save_url_image(f"{base}/redirect-loop") - - def test_oversize_raises_and_cleans_up(self, http_server, tmp_path): """Oversize downloads must NOT leak a partial file into the cache.""" base, _ = http_server diff --git a/tests/plugins/image_gen/test_openai_codex_provider.py b/tests/plugins/image_gen/test_openai_codex_provider.py index 2c3cab50b4..b20e86f8d8 100644 --- a/tests/plugins/image_gen/test_openai_codex_provider.py +++ b/tests/plugins/image_gen/test_openai_codex_provider.py @@ -191,14 +191,6 @@ class TestGenerate: body = json.loads(codex_backend["requests"][0].content) assert body["images"] == [{"image_url": "data:image/png;base64," + _b64_png()}] - def test_remote_source_url_refused_by_ssrf_guard(self, provider, codex_backend): - """A model-supplied image_url pointing at a metadata endpoint must be - refused before any fetch.""" - result = provider.generate("edit", image_url="http://169.254.169.254/latest/meta-data") - - assert result["success"] is False - assert codex_backend["requests"] == [] - def test_capabilities_advertise_image_inputs(self, provider): caps = provider.capabilities() assert caps["modalities"] == ["text", "image"] diff --git a/tests/plugins/image_gen/test_openai_provider.py b/tests/plugins/image_gen/test_openai_provider.py index caba14a6eb..93ad401321 100644 --- a/tests/plugins/image_gen/test_openai_provider.py +++ b/tests/plugins/image_gen/test_openai_provider.py @@ -119,15 +119,6 @@ class TestSourceImageLoading: openai_plugin._load_image_bytes(str(auth_json)) - def test_load_image_bytes_refuses_ssrf_url(self, tmp_path, monkeypatch): - """A model-supplied remote image ref pointing at a metadata endpoint must be - refused before any fetch.""" - monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) - (tmp_path / ".hermes").mkdir() - - with pytest.raises(ValueError, match="SSRF"): - openai_plugin._load_image_bytes("http://169.254.169.254/latest/meta-data") - def test_load_image_bytes_allows_legit_local_image(self, tmp_path, monkeypatch): """Negative control: a legitimate local image path is NOT blocked and loads normally — proves the guard doesn't over-fire on everything.""" diff --git a/tests/tools/test_skills_hub.py b/tests/tools/test_skills_hub.py index 8244ad7fa8..a7d0e14b42 100644 --- a/tests/tools/test_skills_hub.py +++ b/tests/tools/test_skills_hub.py @@ -228,41 +228,6 @@ class TestSkillsShSource: assert results[0].path == "vercel-react-best-practices" assert results[0].extra["installs"] == 207679 - def test_sitemap_skips_metadata_loc_entries(self, monkeypatch): - """A hostile sitemap index advertising an internal must not be fetched.""" - index_xml = ( - "" - "https://www.skills.sh/sitemap-skills-1.xml" - "http://169.254.169.254/sitemap-skills-2.xml" - "" - ) - monkeypatch.setattr("tools.skills_hub_skillssh._cached_metas", lambda key: None) - monkeypatch.setattr("tools.skills_hub_skillssh._cache_metas", lambda key, m: None) - fetched = [] - - def _fake_get_text(url, **kw): - fetched.append(url) - return index_xml if "sitemap.xml" in url else None - - monkeypatch.setattr("tools.skills_hub_skillssh._get_text", _fake_get_text) - - class _Client: - def __enter__(self): return self - def __exit__(self, *a): return False - def get(self, url, **kw): - fetched.append(url) - body = "https://skills.sh/acme/tools/my-skill" - return httpx.Response(200, text=body, request=httpx.Request("GET", url)) - - monkeypatch.setattr("tools.url_safety.create_ssrf_safe_client", lambda **kw: _Client()) - - self._source()._sitemap_catalog(limit=5) - - assert "http://169.254.169.254/sitemap-skills-2.xml" not in fetched - assert fetched == [ - "https://www.skills.sh/sitemap.xml", - "https://www.skills.sh/sitemap-skills-1.xml", - ] @patch("tools.skills_hub._write_index_cache") @patch("tools.skills_hub._read_index_cache", return_value=None) @@ -2077,4 +2042,3 @@ class TestUrlSourceFetchMissingReferencedFile: assert bundle is not None assert bundle.name == "demo" assert "references/missing.md" not in bundle.files - diff --git a/tests/tui_gateway/test_methods_images.py b/tests/tui_gateway/test_methods_images.py deleted file mode 100644 index c6dbc4d9f3..0000000000 --- a/tests/tui_gateway/test_methods_images.py +++ /dev/null @@ -1,99 +0,0 @@ -"""``image.generate`` RPC: the provider-result ``image`` field may be a bare remote -URL (cache_url_best_effort falls back to it), so ``_image_to_data_url`` fetches it -server-side — it must route through the SSRF guard like provider_media.save_url. -""" - -import http.server -import socketserver -import threading - -import httpx -import pytest - -from tui_gateway import methods_images - -_PNG = bytes.fromhex( - "89504e470d0a1a0a0000000d49484452000000010000000108060000001f15c4" - "890000000d49444154789c6300010000000500010d0a2db40000000049454e44" - "ae426082" -) - - -def test_image_to_data_url_refuses_metadata_url(): - assert methods_images._image_to_data_url("http://169.254.169.254/latest/meta-data", 1024) is None - - -def test_image_to_data_url_refuses_file_scheme(): - assert methods_images._image_to_data_url("file:///etc/passwd", 1024) is None - - -def test_image_to_data_url_fetches_safe_url(monkeypatch): - monkeypatch.setattr("tools.url_safety.is_safe_url", lambda url: True) - monkeypatch.setattr( - "tools.url_safety.create_ssrf_safe_client", - lambda **kw: httpx.Client( - transport=httpx.MockTransport( - lambda request: httpx.Response( - 200, content=_PNG, headers={"content-type": "image/png"}, request=request)), - **kw)) - - data_url = methods_images._image_to_data_url("https://example.com/i.png", 1024) - - assert data_url is not None - assert data_url.startswith("data:image/png;base64,") - - -def test_image_to_data_url_caps_remote_body(monkeypatch): - monkeypatch.setattr("tools.url_safety.is_safe_url", lambda url: True) - monkeypatch.setattr( - "tools.url_safety.create_ssrf_safe_client", - lambda **kw: httpx.Client( - transport=httpx.MockTransport( - lambda request: httpx.Response( - 200, content=_PNG * 8, headers={"content-type": "image/png"}, request=request)), - **kw)) - - assert methods_images._image_to_data_url("https://example.com/i.png", 64) is None - - -class _ImageHandler(http.server.BaseHTTPRequestHandler): - def do_GET(self): # noqa: N802 - if self.path == "/ok.png": - self.send_response(200) - self.send_header("Content-Type", "image/png") - self.end_headers() - self.wfile.write(_PNG) - elif self.path == "/redirect-metadata": - self.send_response(302) - self.send_header("Location", "http://169.254.169.254/latest/meta-data/") - self.end_headers() - else: - self.send_response(404) - self.end_headers() - - def log_message(self, *args, **kw): - return - - -@pytest.fixture -def image_server(monkeypatch): - """Real local HTTP server — same e2e shape as tests/agent/test_save_url_image.""" - monkeypatch.setenv("HERMES_ALLOW_PRIVATE_URLS", "1") - from tools import url_safety - url_safety._reset_allow_private_cache() - httpd = socketserver.TCPServer(("127.0.0.1", 0), _ImageHandler) - threading.Thread(target=httpd.serve_forever, daemon=True).start() - yield f"http://127.0.0.1:{httpd.server_address[1]}" - httpd.shutdown() - monkeypatch.delenv("HERMES_ALLOW_PRIVATE_URLS", raising=False) - url_safety._reset_allow_private_cache() - - -def test_image_to_data_url_e2e_real_fetch(image_server): - """End-to-end: real guarded client, real socket, real cache path.""" - data_url = methods_images._image_to_data_url(f"{image_server}/ok.png", 1 << 20) - assert data_url is not None and data_url.startswith("data:image/png;base64,") - - -def test_image_to_data_url_e2e_redirect_to_metadata_blocked(image_server): - assert methods_images._image_to_data_url(f"{image_server}/redirect-metadata", 1 << 20) is None