mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-10 13:31:38 +00:00
ac94f2c8a1
vision_analyze embedded SVG (and BMP/TIFF) tool-results into conversation history with media_type image/svg+xml. Anthropic only accepts jpeg/png/ gif/webp, so the request fails with a non-retryable 400. Because the image is baked into immutable history and re-sent every turn, the session is permanently wedged on resume — retries re-send the same bad bytes. Add _normalize_to_supported_image(): SVG is rasterized to PNG (best-effort via cairosvg/svglib/rsvg-convert/inkscape), other non-supported raster formats are re-encoded to PNG via Pillow, and if conversion is impossible the tool returns an actionable error instead of a session-wedging payload. Wired into both the native-vision fast path and the auxiliary-API path so the whole bug class is covered, not just the one call site. All 99 existing vision tests pass.
225 lines
9.6 KiB
Python
225 lines
9.6 KiB
Python
"""Tests for tools/image_source.py — the unified vision image-source resolver.
|
|
|
|
Covers the delivery contract (data:/http/file/local/container source handling,
|
|
size cap, magic-byte sniff) AND the terminal-backend confinement security model
|
|
(GHSA-gpxw-6wxv-w3qq): under a non-local backend, host reads are confined to the
|
|
media caches and every other path is read inside the sandbox via exec-read.
|
|
"""
|
|
|
|
import base64
|
|
import importlib
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
|
|
PNG = b"\x89PNG\r\n\x1a\n" + b"\x00" * 64
|
|
JPEG = b"\xff\xd8\xff" + b"\x00" * 64
|
|
|
|
|
|
def _reload(monkeypatch, hermes_home: Path):
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
import hermes_constants
|
|
importlib.reload(hermes_constants)
|
|
import tools.image_source as isrc
|
|
importlib.reload(isrc)
|
|
return isrc
|
|
|
|
|
|
class TestDataUrl:
|
|
@pytest.mark.asyncio
|
|
async def test_valid_data_url_resolves_to_bytes(self, tmp_path, monkeypatch):
|
|
isrc = _reload(monkeypatch, tmp_path / "hermes")
|
|
b64 = base64.b64encode(PNG).decode()
|
|
res = await isrc.resolve_image_source(
|
|
f"data:image/png;base64,{b64}", isrc.ResolveContext())
|
|
assert res.data == PNG
|
|
assert res.mime == "image/png"
|
|
assert res.origin == "data"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_image_data_url_rejected(self, tmp_path, monkeypatch):
|
|
isrc = _reload(monkeypatch, tmp_path / "hermes")
|
|
b64 = base64.b64encode(b"not an image").decode()
|
|
with pytest.raises(isrc.NotAnImage):
|
|
await isrc.resolve_image_source(
|
|
f"data:text/plain;base64,{b64}", isrc.ResolveContext())
|
|
|
|
|
|
class TestLocalBackend:
|
|
@pytest.mark.asyncio
|
|
async def test_local_backend_reads_any_host_path(self, tmp_path, monkeypatch):
|
|
isrc = _reload(monkeypatch, tmp_path / "hermes")
|
|
monkeypatch.setenv("TERMINAL_ENV", "local")
|
|
img = tmp_path / "outside" / "pic.png"
|
|
img.parent.mkdir(parents=True)
|
|
img.write_bytes(PNG)
|
|
res = await isrc.resolve_image_source(str(img), isrc.ResolveContext())
|
|
assert res.data == PNG
|
|
assert res.origin == "file"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_uri_scheme_stripped(self, tmp_path, monkeypatch):
|
|
isrc = _reload(monkeypatch, tmp_path / "hermes")
|
|
monkeypatch.setenv("TERMINAL_ENV", "local")
|
|
img = tmp_path / "pic.jpg"
|
|
img.write_bytes(JPEG)
|
|
res = await isrc.resolve_image_source(f"file://{img}", isrc.ResolveContext())
|
|
assert res.mime == "image/jpeg"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bare_relative_path_resolves(self, tmp_path, monkeypatch):
|
|
"""A cwd-relative bare filename ('pic.png') is a valid local source —
|
|
main accepted it; the resolver must not regress it (PR review)."""
|
|
isrc = _reload(monkeypatch, tmp_path / "hermes")
|
|
monkeypatch.setenv("TERMINAL_ENV", "local")
|
|
img = tmp_path / "pic.png"
|
|
img.write_bytes(PNG)
|
|
monkeypatch.chdir(tmp_path)
|
|
res = await isrc.resolve_image_source("pic.png", isrc.ResolveContext())
|
|
assert res.data == PNG
|
|
assert res.origin == "file"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unknown_url_scheme_rejected(self, tmp_path, monkeypatch):
|
|
isrc = _reload(monkeypatch, tmp_path / "hermes")
|
|
monkeypatch.setenv("TERMINAL_ENV", "local")
|
|
with pytest.raises(isrc.UnsupportedScheme):
|
|
await isrc.resolve_image_source(
|
|
"ftp://example.com/pic.png", isrc.ResolveContext())
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_svg_passes_through_for_rasterization(self, tmp_path, monkeypatch):
|
|
"""SVG has no raster magic bytes but is passed through with mime
|
|
image/svg+xml so the vision call sites can rasterize it to PNG."""
|
|
isrc = _reload(monkeypatch, tmp_path / "hermes")
|
|
monkeypatch.setenv("TERMINAL_ENV", "local")
|
|
svg = tmp_path / "art.svg"
|
|
svg_bytes = b'<svg xmlns="http://www.w3.org/2000/svg"></svg>'
|
|
svg.write_bytes(svg_bytes)
|
|
res = await isrc.resolve_image_source(str(svg), isrc.ResolveContext())
|
|
assert res.mime == "image/svg+xml"
|
|
assert res.data == svg_bytes
|
|
|
|
|
|
class TestNonLocalBackendConfinement:
|
|
"""The security model: under a sandbox backend, host reads are confined to
|
|
the media caches; every other path is read inside the sandbox."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_media_cache_path_host_read(self, tmp_path, monkeypatch):
|
|
home = tmp_path / "hermes"
|
|
isrc = _reload(monkeypatch, home)
|
|
monkeypatch.setenv("TERMINAL_ENV", "docker")
|
|
cached = home / "cache" / "images" / "inbound.png"
|
|
cached.parent.mkdir(parents=True)
|
|
cached.write_bytes(PNG)
|
|
# No sandbox env needed — a cache path is host-read directly.
|
|
res = await isrc.resolve_image_source(str(cached), isrc.ResolveContext())
|
|
assert res.data == PNG
|
|
assert res.origin == "file"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_host_secret_outside_cache_routes_to_sandbox_not_host(self, tmp_path, monkeypatch):
|
|
"""A non-cache host path (e.g. /etc/passwd) must NOT be host-read — it
|
|
routes to the in-sandbox exec-read, which reads the CONTAINER's file."""
|
|
home = tmp_path / "hermes"
|
|
isrc = _reload(monkeypatch, home)
|
|
monkeypatch.setenv("TERMINAL_ENV", "docker")
|
|
|
|
# A real host file outside the caches, holding a "secret".
|
|
secret = tmp_path / "id_rsa"
|
|
secret.write_bytes(b"HOST-PRIVATE-KEY-DO-NOT-LEAK")
|
|
|
|
# Fake sandbox env: its exec-read returns a *different* (container) image,
|
|
# proving we read the container filesystem, not the host secret.
|
|
container_png_b64 = base64.b64encode(PNG).decode()
|
|
calls = {}
|
|
|
|
def fake_execute(cmd, **kw):
|
|
calls["cmd"] = cmd
|
|
return {"returncode": 0, "output": container_png_b64}
|
|
|
|
with patch("tools.image_source._get_active_env",
|
|
return_value=SimpleNamespace(execute=fake_execute)):
|
|
res = await isrc.resolve_image_source(str(secret), isrc.ResolveContext(task_id="t1"))
|
|
|
|
# Read came from the sandbox exec-read, returning the container image —
|
|
# the host secret bytes never appear.
|
|
assert res.origin == "container"
|
|
assert res.data == PNG
|
|
assert b"HOST-PRIVATE-KEY" not in res.data
|
|
assert "base64 --" in calls["cmd"] # option-injection-safe form
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_cache_path_fails_closed_without_sandbox(self, tmp_path, monkeypatch):
|
|
"""No active sandbox env -> refuse rather than fall back to a host read."""
|
|
home = tmp_path / "hermes"
|
|
isrc = _reload(monkeypatch, home)
|
|
monkeypatch.setenv("TERMINAL_ENV", "docker")
|
|
secret = tmp_path / "id_rsa"
|
|
secret.write_bytes(b"HOST-PRIVATE-KEY")
|
|
|
|
with patch("tools.image_source._get_active_env", return_value=None):
|
|
with pytest.raises(isrc.SourceNotFound):
|
|
await isrc.resolve_image_source(str(secret), isrc.ResolveContext(task_id="t1"))
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_symlink_in_cache_pointing_outside_is_not_host_read(self, tmp_path, monkeypatch):
|
|
"""A symlink planted inside a cache dir that points at a host secret must
|
|
not be host-read (resolve() escapes the cache) — it routes to sandbox."""
|
|
home = tmp_path / "hermes"
|
|
isrc = _reload(monkeypatch, home)
|
|
monkeypatch.setenv("TERMINAL_ENV", "docker")
|
|
secret = tmp_path / "outside" / "id_rsa"
|
|
secret.parent.mkdir(parents=True)
|
|
secret.write_bytes(b"HOST-PRIVATE-KEY")
|
|
cache_dir = home / "cache" / "images"
|
|
cache_dir.mkdir(parents=True)
|
|
link = cache_dir / "sneaky.png"
|
|
try:
|
|
link.symlink_to(secret)
|
|
except (OSError, NotImplementedError):
|
|
pytest.skip("symlinks unsupported")
|
|
|
|
# Fails closed (no sandbox) rather than host-reading the symlink target.
|
|
with patch("tools.image_source._get_active_env", return_value=None):
|
|
with pytest.raises(isrc.SourceNotFound):
|
|
await isrc.resolve_image_source(str(link), isrc.ResolveContext(task_id="t1"))
|
|
|
|
|
|
class TestExecReadSafety:
|
|
@pytest.mark.asyncio
|
|
async def test_exec_read_uses_option_terminator(self, tmp_path, monkeypatch):
|
|
"""Leading-dash paths must not be parsed as base64 options."""
|
|
home = tmp_path / "hermes"
|
|
isrc = _reload(monkeypatch, home)
|
|
monkeypatch.setenv("TERMINAL_ENV", "docker")
|
|
captured = {}
|
|
|
|
def fake_execute(cmd, **kw):
|
|
captured["cmd"] = cmd
|
|
return {"returncode": 0, "output": base64.b64encode(PNG).decode()}
|
|
|
|
with patch("tools.image_source._get_active_env",
|
|
return_value=SimpleNamespace(execute=fake_execute)):
|
|
await isrc.resolve_image_source(
|
|
"/workspace/-i-etc-shadow.png", isrc.ResolveContext(task_id="t1"))
|
|
assert "base64 -- " in captured["cmd"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_read_nonzero_returncode_raises(self, tmp_path, monkeypatch):
|
|
home = tmp_path / "hermes"
|
|
isrc = _reload(monkeypatch, home)
|
|
monkeypatch.setenv("TERMINAL_ENV", "docker")
|
|
|
|
def fake_execute(cmd, **kw):
|
|
return {"returncode": 1, "output": ""}
|
|
|
|
with patch("tools.image_source._get_active_env",
|
|
return_value=SimpleNamespace(execute=fake_execute)):
|
|
with pytest.raises(isrc.SourceNotFound):
|
|
await isrc.resolve_image_source(
|
|
"/workspace/nope.png", isrc.ResolveContext(task_id="t1"))
|