security(vision): route local-file inputs through the shared credential-read guard

video_analyze_tool's local-path branch read raw bytes via
_detect_video_mime_type (extension-only, no magic-byte check) with no
call to agent.file_safety.raise_if_read_blocked, unlike the image-gen
and video-gen provider plugins that already route local inputs through
that shared chokepoint (#57698). A model could point video_url at a
credential store (e.g. .env, auth.json) renamed or symlinked to a
video-like extension and have its raw bytes base64-encoded and sent to
the vision provider.

vision_analyze_tool and its native fast path (_vision_analyze_native)
had the same gap in their local-file branches; they were only
incidentally protected by the image magic-byte sniff rejecting
non-image content, not by the intended read guard.

Add raise_if_read_blocked() to all three local-file branches, mirroring
the existing plugins/image_gen and plugins/video_gen call sites.
fix/verification-admin-route-recovery
srojk34 2026-07-03 19:17:10 +03:00 committed by Teknium
parent b3c7b34885
commit 9ae17b8ac5
4 changed files with 73 additions and 0 deletions

View File

@ -193,6 +193,29 @@ class TestVideoAnalyzeTool:
assert data["success"] is True
assert "demo" in data["analysis"].lower()
def test_local_file_read_guard_blocks_env_via_video_extension(self, tmp_path):
"""A .env file symlinked with a video extension must still be blocked.
_detect_video_mime_type only checks the file extension, not file
content, so without a read guard a model could point video_url at
any credential-store file (renamed/symlinked to look like a video)
and have its raw bytes base64-encoded and sent to the vision
provider. Regression for the shared agent.file_safety chokepoint
added to video_analyze_tool's local-file branch.
"""
secret = tmp_path / ".env"
secret.write_text("OPENAI_API_KEY=sk-super-secret\n", encoding="utf-8")
disguised = tmp_path / "video.mp4"
disguised.symlink_to(secret)
with patch("tools.vision_tools.async_call_llm", new_callable=AsyncMock) as mock_llm:
result = self._run(video_analyze_tool(str(disguised), "What is this?"))
data = json.loads(result)
assert data["success"] is False
assert "secret-bearing environment file" in data["error"]
mock_llm.assert_not_awaited()
def test_local_file_not_found(self, tmp_path):
"""Non-existent file raises appropriate error."""
result = self._run(video_analyze_tool("/nonexistent/video.mp4", "What?"))

View File

@ -495,6 +495,38 @@ class TestVisionSafetyGuards:
assert "not a recognized image" in result["error"]
mock_llm.assert_not_awaited()
@pytest.mark.asyncio
async def test_local_env_file_blocked_via_read_guard(self, tmp_path):
"""A .env file must be blocked even if given an image-like path.
Mirrors the video_analyze_tool regression: the local-file branch
must route through agent.file_safety.raise_if_read_blocked before
vision_analyze_tool ever opens the file, not rely solely on the
magic-byte mime check as an accidental side effect.
"""
secret = tmp_path / ".env"
secret.write_text("OPENAI_API_KEY=sk-super-secret\n", encoding="utf-8")
with patch("tools.vision_tools.async_call_llm", new_callable=AsyncMock) as mock_llm:
result = json.loads(await vision_analyze_tool(str(secret), "extract text"))
assert result["success"] is False
assert "secret-bearing environment file" in result["error"]
mock_llm.assert_not_awaited()
@pytest.mark.asyncio
async def test_native_fast_path_local_env_file_blocked_via_read_guard(self, tmp_path):
"""Same read guard must apply to the native vision fast path."""
from tools.vision_tools import _vision_analyze_native
secret = tmp_path / ".env"
secret.write_text("OPENAI_API_KEY=sk-super-secret\n", encoding="utf-8")
result = json.loads(await _vision_analyze_native(str(secret), "extract text"))
assert result["success"] is False
assert "secret-bearing environment file" in result["error"]
@pytest.mark.asyncio
async def test_blocked_remote_url_short_circuits_before_download(self):
blocked = {

View File

@ -117,6 +117,22 @@ async def resolve_image_source(src: str, ctx: ResolveContext) -> ResolvedImage:
# path outside the caches never yields the host's bytes.
host_target = _permitted_host_read_target(p, ctx)
if host_target is not None and host_target.is_file():
# Shared credential-read guard (agent.file_safety, #57698): refuse
# secret-bearing files (.env, auth.json, ...) with an intentional,
# specific error instead of relying on the magic-byte sniff to
# reject them incidentally. Same chokepoint the image-gen/video-gen
# provider plugins enforce on model-supplied local paths. Import is
# best-effort (guard unavailability must not break image loading);
# a real block always propagates.
try:
from agent.file_safety import raise_if_read_blocked
except Exception: # noqa: BLE001 — guard unavailable: proceed
raise_if_read_blocked = None
if raise_if_read_blocked is not None:
try:
raise_if_read_blocked(str(host_target))
except ValueError as exc:
raise SourceUnsafe(str(exc), src=s, origin="file")
data = await asyncio.to_thread(host_target.read_bytes)
return _finalize(data, "", "file", s)
if _is_local_terminal_backend():

View File

@ -1666,6 +1666,8 @@ async def video_analyze_tool(
local_path = Path(os.path.expanduser(resolved_url))
if local_path.is_file():
from agent.file_safety import raise_if_read_blocked
raise_if_read_blocked(str(local_path))
logger.info("Using local video file: %s", video_url)
temp_video_path = local_path
should_cleanup = False