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))