From c13281ab57fbe616457b55cd80f3e89d22cb7695 Mon Sep 17 00:00:00 2001 From: dsad Date: Sun, 5 Jul 2026 01:00:01 +0300 Subject: [PATCH] Guard native image routing with file safety --- agent/image_routing.py | 11 +++++++++ tests/agent/test_image_routing.py | 38 +++++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+) diff --git a/agent/image_routing.py b/agent/image_routing.py index ba6d8da32a2..1fe52d9565b 100644 --- a/agent/image_routing.py +++ b/agent/image_routing.py @@ -632,6 +632,17 @@ def _file_to_data_url(path: Path) -> Optional[str]: caller reports those paths in ``skipped`` and the rest of the turn proceeds. """ + try: + from agent.file_safety import raise_if_read_blocked + + raise_if_read_blocked(str(path)) + except ValueError as exc: + logger.warning("image_routing: blocked local image attachment %s -- %s", path, exc) + return None + except Exception: + # Keep attachment routing best-effort if the guard itself is unavailable. + pass + try: raw = path.read_bytes() except Exception as exc: diff --git a/tests/agent/test_image_routing.py b/tests/agent/test_image_routing.py index 675823112cf..671e6aaa18d 100644 --- a/tests/agent/test_image_routing.py +++ b/tests/agent/test_image_routing.py @@ -780,6 +780,44 @@ class TestFormatCompatibility: b64 = url.split(",", 1)[1] assert base64.b64decode(b64) == _png_bytes() + def test_file_to_data_url_blocks_read_denied_image_path(self, tmp_path: Path): + """Native image routing must honor the shared credential read guard.""" + from agent.image_routing import _file_to_data_url + + img_path = tmp_path / ".env" + img_path.write_bytes(_png_bytes()) + + assert _file_to_data_url(img_path) is None + + def test_native_content_parts_skip_read_denied_local_image(self, tmp_path: Path): + from agent.image_routing import build_native_content_parts + + img_path = tmp_path / ".env.local" + img_path.write_bytes(_png_bytes()) + + parts, skipped = build_native_content_parts("inspect this", [str(img_path)]) + + assert skipped == [str(img_path)] + assert all(part.get("type") != "image_url" for part in parts) + + def test_native_content_parts_blocks_image_symlink_to_read_denied_file(self, tmp_path: Path): + from agent.image_routing import build_native_content_parts + import os + import pytest + + secret = tmp_path / ".env" + secret.write_bytes(_png_bytes()) + img_link = tmp_path / "secret.png" + try: + os.symlink(secret, img_link) + except (OSError, NotImplementedError) as exc: + pytest.skip(f"symlinks unavailable: {exc}") + + parts, skipped = build_native_content_parts("inspect this", [str(img_link)]) + + assert skipped == [str(img_link)] + assert all(part.get("type") != "image_url" for part in parts) + def test_jpeg_passes_through_no_transcode(self, tmp_path: Path): from agent.image_routing import _file_to_data_url