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

fix(video-gen): trust the operator-configured origin on the first download hop

Browse Source 
The OpenRouter video content URL is built from OPENROUTER_BASE_URL, not from
a provider response, yet save_url ran the full SSRF check on it. An operator
pointing base_url at a LAN/loopback relay could submit and poll (raw
requests) but the final download was refused as SSRF unless they set
security.allow_private_urls — contradicting the issue scope ("operator-
configured endpoints out of scope") and the security.md sentence that the
operator's own base_url is unaffected.

save_url grows a `trusted_origin` flag (plumbed through save_url_video and
set only by OpenRouterVideoGenProvider._save_completed_video): the first hop
skips the private-address class check and uses a plain client, but the
cloud-metadata floor (is_always_blocked_url) still applies, auth headers
stay on hop 1 only, and every redirect target is re-validated in full so a
relay cannot bounce us to another internal address. Provider-returned result
URLs (fal, xai, image providers) keep the full guard — default is False.

security.md now states the precise scope: only the direct base_url hop is
exempt; result URLs from a LAN-hosted provider still need allow_private_urls.
This commit is contained in:
teknium1
2026-09-18 03:15:22 -07:00
committed by Teknium
parent ea05583ad5
commit b44a481334
6 changed files with 85 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
agent/provider_media.py
View File

@@ -50,7 +50,7 @@ def save_url(
kind: str, url: str, *, prefix: str, timeout: float, max_bytes: int, chunk_size: int,
content_types: Dict[str, str], url_extensions: Tuple[str, ...], default_extension: str,
label: str, empty_error: str, headers: Optional[Dict[str, str]] = None,
require_known_content_type: bool = False,
require_known_content_type: bool = False, trusted_origin: bool = False,
) -> Path:
"""Stream-download *url* into the cache with a size cap.
@@ -65,20 +65,34 @@ def save_url(
guarded transport, closing DNS-rebinding TOCTOU). Caller-supplied *headers*
(e.g. provider auth) go to the first hop only — a redirect target never
receives them.
"""
from tools.url_safety import create_ssrf_safe_client, is_safe_url
current_url, hop_headers = url, headers
*trusted_origin* is for callers that built *url* from the operator's own
provider ``base_url`` (not from a provider response): the first hop skips the
private-address class check so a LAN/loopback relay works without
``security.allow_private_urls``, but the cloud-metadata floor still applies and
every redirect target is re-validated in full.
"""
import httpx
from tools.url_safety import create_ssrf_safe_client, is_always_blocked_url, is_safe_url
current_url, hop_headers, trusted_hop = url, headers, trusted_origin
for _ in range(_MAX_SAVE_URL_REDIRECTS + 1):
if not is_safe_url(current_url):
raise ValueError(f"{label} URL failed the SSRF safety check: {current_url}")
with create_ssrf_safe_client(timeout=timeout, follow_redirects=False) as client:
if trusted_hop:
if is_always_blocked_url(current_url):
raise ValueError(f"{label} URL targets an always-blocked address: {current_url}")
client = httpx.Client(timeout=timeout, follow_redirects=False)
else:
if not is_safe_url(current_url):
raise ValueError(f"{label} URL failed the SSRF safety check: {current_url}")
client = create_ssrf_safe_client(timeout=timeout, follow_redirects=False)
with client:
with client.stream("GET", current_url, headers=hop_headers) as response:
if response.status_code in _REDIRECT_STATUS_CODES:
location = response.headers.get("location")
if not location:
raise ValueError(f"{label} download redirected without a Location: {current_url}")
current_url, hop_headers = urljoin(current_url, location), None
current_url, hop_headers, trusted_hop = urljoin(current_url, location), None, False
continue
if not response.is_success:
response.read()

6
agent/video_gen_provider.py
View File

@@ -92,15 +92,19 @@ def save_url_video(
max_bytes: int = 200 * 1024 * 1024,
headers: Optional[Dict[str, str]] = None,
require_video_content_type: bool = False,
trusted_origin: bool = False,
) -> Path:
"""Download an (often ephemeral) video URL into ``$HERMES_HOME/cache/videos/``;
raises on network / HTTP / oversize / empty errors so callers can fall back to the URL."""
raises on network / HTTP / oversize / empty errors so callers can fall back to the URL.
``trusted_origin`` is only for URLs built from the operator's configured provider
``base_url`` (see ``provider_media.save_url``)."""
return provider_media.save_url(
"videos", url, prefix=prefix, timeout=timeout, max_bytes=max_bytes,
chunk_size=256 * 1024, content_types=_URL_VIDEO_CONTENT_TYPES,
url_extensions=("mp4", "webm", "mov", "mkv"), default_extension="mp4",
label="Video", empty_error="Video at {url} was empty (0 bytes).",
headers=headers, require_known_content_type=require_video_content_type,
trusted_origin=trusted_origin,
)

6
plugins/video_gen/openrouter/__init__.py
View File

@@ -265,9 +265,11 @@ class OpenRouterVideoGenProvider(VideoGenProvider):
def _save_completed_video(self, job_id: str) -> str:
# The content endpoint is derived from our configured origin, never from ``unsigned_urls``: the
# bearer key must only ever be sent to the host the operator selected.
# bearer key must only ever be sent to the host the operator selected. That origin is operator
# chosen (a LAN relay is legitimate), so the first hop is trusted for the private-address check.
return str(save_url_video(f"{self._base_url()}/videos/{job_id}/content", prefix="openrouter",
headers=self._headers(), require_video_content_type=True))
headers=self._headers(), require_video_content_type=True,
trusted_origin=True))
def generate(
self, prompt: str, *, model: Optional[str] = None, image_url: Optional[str] = None,

51
tests/agent/test_provider_media.py
View File

@@ -111,3 +111,54 @@ def test_save_url_redirect_scopes_caller_headers_to_first_hop_and_fails_closed(m
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]
def test_save_url_trusted_origin_skips_private_check_on_first_hop_only(monkeypatch, tmp_path):
"""``trusted_origin=True`` (the caller built the URL from the operator's own
provider ``base_url``) must let a LAN/loopback relay serve the first hop, but the
cloud-metadata floor still applies and every redirect target is re-validated in
full — a relay cannot bounce us to another internal address."""
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
monkeypatch.delenv("HERMES_ALLOW_PRIVATE_URLS", raising=False)
hits = []
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
hits.append((self.path, self.headers.get("Authorization")))
if self.path == "/bounce":
self.send_response(302)
self.send_header("Location", "/content")
self.end_headers()
return
self.send_response(200)
self.send_header("Content-Type", "video/mp4")
self.end_headers()
self.wfile.write(b"clip")
def log_message(self, *_):
pass
server = HTTPServer(("127.0.0.1", 0), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()
try:
base = f"http://127.0.0.1:{server.server_port}"
path = _save_video(f"{base}/content", headers={"Authorization": "Bearer test"},
require_known_content_type=True, trusted_origin=True)
assert path.read_bytes() == b"clip"
assert hits == [("/content", "Bearer test")]
with pytest.raises(ValueError, match="SSRF safety check"):
_save_video(f"{base}/bounce", headers={"Authorization": "Bearer test"},
require_known_content_type=True, trusted_origin=True)
assert hits[1:] == [("/bounce", "Bearer test")] # hop 2 (loopback) refused before connecting
with pytest.raises(ValueError, match="always-blocked"):
_save_video("http://169.254.169.254/latest/meta-data", trusted_origin=True)
assert len(hits) == 2
finally:
server.shutdown()
server.server_close()

2
tests/plugins/video_gen/test_openrouter.py
View File

@@ -117,6 +117,8 @@ def test_generate_submits_polls_and_downloads_from_configured_origin(monkeypatch
assert [g[0] for g in session.gets] == ["https://openrouter.ai/api/v1/videos/job-1"] * 2
assert saved[0][0] == "https://openrouter.ai/api/v1/videos/job-1/content"
assert saved[0][1]["headers"]["Authorization"] == "Bearer sk-or-test" and saved[0][1]["require_video_content_type"]
# Operator-configured origin: a LAN relay must not be refused as SSRF on the first hop.
assert saved[0][1]["trusted_origin"] is True
def test_generate_rejects_local_image_paths_before_spending(monkeypatch):

2
website/docs/user-guide/security.md
View File

@@ -739,7 +739,7 @@ All URL-capable tools (web search, web extract, vision, browser) validate URLs b
SSRF protection is always active for internet-facing use and DNS failures are treated as blocked (fail-closed). Redirect chains are re-validated at each hop to prevent redirect-based bypasses.
The same guard covers fetches whose URL comes from a remote party rather than from you: image/video URLs returned by a generation provider, reference-image URLs a model supplies for edits, pet spritesheets and the petdex manifest, and skills.sh sitemap entries. A provider or index that points one of those at a private or metadata address is refused before any connection opens; the operator's own provider `base_url` is not affected, and an image-generation provider hosted on your LAN needs `security.allow_private_urls: true` (below) for its result URLs to be cached locally.
The same guard covers fetches whose URL comes from a remote party rather than from you: image/video URLs returned by a generation provider, reference-image URLs a model supplies for edits, pet spritesheets and the petdex manifest, and skills.sh sitemap entries. A provider or index that points one of those at a private or metadata address is refused before any connection opens; the operator's own provider `base_url` is not affected — a download fetched directly from your configured `base_url` (the OpenRouter video content endpoint) skips only the private-address class check on that first hop, while the cloud-metadata floor still applies and any redirect it issues is re-validated in full — and an image-generation provider hosted on your LAN needs `security.allow_private_urls: true` (below) for the *result* URLs it returns to be cached locally.
#### Intentionally allowing private URLs