fix(xai): route video-gen local inputs through the shared read guard

Fold the xAI video credential-read guard into the same shared
agent.file_safety.raise_if_read_blocked chokepoint this PR introduces for
the image providers, so the whole image+video bug class is covered by one
enforced boundary. Consolidates the parallel salvage of #57695 (xAI
image+video) into this PR; #57727 is now redundant and will be closed.

- video_gen/xai: guard _image_ref_to_xai_url and _video_ref_to_xai_url
  (the video image + video byte-read chokepoints) via the shared helper.
- Regression tests: symlinked auth.json with .png/.mp4 names are blocked
  across both video read paths (mutation-checked).
This commit is contained in:
kshitijk4poor 2026-07-03 18:42:04 +05:30 committed by kshitij
parent c1826e2690
commit 104232979d
2 changed files with 58 additions and 0 deletions

View file

@ -127,6 +127,22 @@ def _xai_headers(api_key: str) -> Dict[str, str]:
}
def _raise_if_blocked_local_input(ref: str) -> None:
"""Refuse to read a local media path that Hermes' read deny-list blocks.
Thin wrapper over the shared ``agent.file_safety.raise_if_read_blocked``
chokepoint so xAI video inputs enforce the same credential-store guard as
the image providers. Fails open if the guard machinery is unavailable
(defense-in-depth, per the denylist's own framing).
"""
try:
from agent.file_safety import raise_if_read_blocked
except Exception as exc: # noqa: BLE001 - guard must never break loading
logger.debug("xAI media input read guard unavailable: %s", exc)
return
raise_if_read_blocked(ref)
def _image_ref_to_xai_url(value: str) -> str:
"""Return a URL/data URI accepted by xAI for image inputs."""
ref = (value or "").strip()
@ -140,6 +156,8 @@ def _image_ref_to_xai_url(value: str) -> str:
if not path.is_file():
return ref
_raise_if_blocked_local_input(ref)
mime = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
if not mime.startswith("image/"):
return ref
@ -195,6 +213,8 @@ def _video_ref_to_xai_url(value: str) -> str:
if not path.is_file():
return ref
_raise_if_blocked_local_input(ref)
mime = mimetypes.guess_type(path.name)[0] or "video/mp4"
if not mime.startswith("video/"):
return ref

View file

@ -186,3 +186,41 @@ def test_video_input_from_public_url_rejects_bare_file_id():
)
)
assert result is None
def test_xai_video_image_input_blocks_credential_store_symlink(tmp_path, monkeypatch):
from plugins.video_gen.xai import _image_ref_to_xai_input
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
auth_json = hermes_home / "auth.json"
auth_json.write_text('{"api_key":"sk-secret"}', encoding="utf-8")
image_link = hermes_home / "leak.png"
try:
image_link.symlink_to(auth_json)
except OSError as exc:
pytest.skip(f"symlink unavailable on this platform: {exc}")
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
with pytest.raises(ValueError, match="credential store"):
_image_ref_to_xai_input(str(image_link))
def test_xai_video_file_input_blocks_credential_store_symlink(tmp_path, monkeypatch):
from plugins.video_gen.xai import _video_ref_to_xai_url
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
auth_json = hermes_home / "auth.json"
auth_json.write_text('{"api_key":"sk-secret"}', encoding="utf-8")
video_link = hermes_home / "leak.mp4"
try:
video_link.symlink_to(auth_json)
except OSError as exc:
pytest.skip(f"symlink unavailable on this platform: {exc}")
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
with pytest.raises(ValueError, match="credential store"):
_video_ref_to_xai_url(str(video_link))