From 104232979d6ec24e82a42dfbf14ac98f3df3c827 Mon Sep 17 00:00:00 2001
From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com>
Date: Fri, 3 Jul 2026 18:42:04 +0530
Subject: [PATCH] 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).
---
 plugins/video_gen/xai/__init__.py          | 20 ++++++++++++
 tests/plugins/video_gen/test_xai_plugin.py | 38 ++++++++++++++++++++++
 2 files changed, 58 insertions(+)

diff --git a/plugins/video_gen/xai/__init__.py b/plugins/video_gen/xai/__init__.py
index edc981c78..90dfa57bf 100644
--- a/plugins/video_gen/xai/__init__.py
+++ b/plugins/video_gen/xai/__init__.py
@@ -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
diff --git a/tests/plugins/video_gen/test_xai_plugin.py b/tests/plugins/video_gen/test_xai_plugin.py
index eb495b969..e1a2a5ec9 100644
--- a/tests/plugins/video_gen/test_xai_plugin.py
+++ b/tests/plugins/video_gen/test_xai_plugin.py
@@ -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))