hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 29 Packages Projects Releases Wiki Activity

test: one class invariant across every remote-fetch site

Browse Source 
Replace the per-site refusal tests with a single parametrized invariant
driven through a real loopback listener: each of the nine call sites
refuses a loopback target with zero connections recorded, and a
safe-looking first hop that 302s to the metadata endpoint is stopped at
the hop with nothing cached. save_url's redirect contract (caller headers
first-hop only, missing-Location 3xx fails closed) stays in
test_provider_media. Existing save_url tests keep their httpx/allow-
private adaptations; the codex edit test keeps its guarded-client stub.
This commit is contained in:
teknium1
2026-09-18 00:56:39 -07:00
committed by Teknium
parent 5236a58005
commit ea05583ad5
8 changed files with 151 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
tests/agent/test_pet_engine.py
View File

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

13
tests/agent/test_provider_media.py
View File

@@ -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"<html>3xx body</html>")
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]

138
tests/agent/test_remote_fetch_ssrf_guard.py Normal file
View File

@@ -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 ``<loc>`` 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"<sitemapindex><sitemap><loc>http://{self.headers['Host']}/sitemap-skills-1.xml"
"</loc></sitemap></sitemapindex>").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 <loc> 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 # <loc> 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"

45
tests/agent/test_save_url_image.py
View File

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

8
tests/plugins/image_gen/test_openai_codex_provider.py
View File

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

9
tests/plugins/image_gen/test_openai_provider.py
View File

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

36
tests/tools/test_skills_hub.py
View File

@@ -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 <loc> must not be fetched."""
index_xml = (
"<urlset>"
"<url><loc>https://www.skills.sh/sitemap-skills-1.xml</loc></url>"
"<url><loc>http://169.254.169.254/sitemap-skills-2.xml</loc></url>"
"</urlset>"
)
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 = "<urlset><url><loc>https://skills.sh/acme/tools/my-skill</loc></url></urlset>"
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

99
tests/tui_gateway/test_methods_images.py
View File

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