b44a481334
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.
165 lines
5.9 KiB
Python
165 lines
5.9 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from agent import provider_media
|
|
|
|
|
|
def _save_video(url: str, **kwargs):
|
|
return provider_media.save_url(
|
|
"videos",
|
|
url,
|
|
prefix="provider-test",
|
|
timeout=30,
|
|
max_bytes=1024,
|
|
chunk_size=64,
|
|
content_types={"video/mp4": "mp4"},
|
|
url_extensions=("mp4",),
|
|
default_extension="mp4",
|
|
label="Video",
|
|
empty_error="empty: {url}",
|
|
**kwargs,
|
|
)
|
|
|
|
|
|
def _stub_fetch(monkeypatch, handler):
|
|
"""Route save_url's fetch through an httpx MockTransport and stub the URL
|
|
policy check — SSRF behavior has dedicated tests in test_save_url_image.py;
|
|
these pin the streaming/content-type/headers contract."""
|
|
import httpx
|
|
|
|
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(handler), **kw),
|
|
)
|
|
|
|
|
|
def test_save_url_forwards_headers_and_streams_to_cache(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
|
|
calls = []
|
|
|
|
import httpx
|
|
|
|
def handler(request):
|
|
calls.append(request)
|
|
return httpx.Response(200, headers={"Content-Type": "video/mp4"}, content=b"clip")
|
|
|
|
_stub_fetch(monkeypatch, handler)
|
|
|
|
path = _save_video(
|
|
"https://api.example/videos/job/content",
|
|
headers={"Authorization": "Bearer test"},
|
|
require_known_content_type=True,
|
|
)
|
|
|
|
assert path.read_bytes() == b"clip"
|
|
assert path.suffix == ".mp4"
|
|
assert len(calls) == 1
|
|
assert str(calls[0].url) == "https://api.example/videos/job/content"
|
|
assert calls[0].headers["Authorization"] == "Bearer test"
|
|
|
|
|
|
def test_save_url_strict_content_type_rejects_non_video(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
|
|
import httpx
|
|
|
|
_stub_fetch(
|
|
monkeypatch,
|
|
lambda request: httpx.Response(200, headers={"Content-Type": "text/html"}, content=b"not a video"),
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="unexpected Content-Type text/html"):
|
|
_save_video(
|
|
"https://api.example/videos/job/content",
|
|
require_known_content_type=True,
|
|
)
|
|
|
|
assert not list(provider_media.cache_dir("videos").iterdir())
|
|
|
|
|
|
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 = []
|
|
|
|
import httpx
|
|
|
|
def handler(request):
|
|
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)
|
|
|
|
path = _save_video(
|
|
"https://api.example/start",
|
|
headers={"Authorization": "Bearer test"},
|
|
require_known_content_type=True,
|
|
)
|
|
|
|
assert path.read_bytes() == b"clip"
|
|
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]
|
|
|
|
|
|
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()
|