From 16332af60b5a5262111f1171c5d85f07463f929d Mon Sep 17 00:00:00 2001 From: srojk34 <286497132+srojk34@users.noreply.github.com> Date: Fri, 3 Jul 2026 05:45:35 +0300 Subject: [PATCH] security(gateway): anchor api_server MEDIA tag resolution to safe paths _resolve_media_to_data_urls's ad-hoc _MEDIA_TAG_RE matched any bare token after MEDIA: (no absolute-path anchor) and read the resolved path directly with no denylist. A relative/traversal path like MEDIA:../../../../etc/passwd.png slipped through, and any image- suffixed file the process could read (including under ~/.ssh, ~/.aws, etc.) was base64-inlined into the API response if its path merely appeared in the model's own final reply text. Every other platform adapter's MEDIA: handling already goes through two shared primitives in gateway/platforms/base.py: - MEDIA_TAG_CLEANUP_RE, which anchors the path to ~/, /, or a Windows drive letter plus a known deliverable extension. - validate_media_delivery_path, which resolves symlinks and rejects paths under the credential/system-path denylist. Reuse both here instead of the local unanchored pattern and naive Path().expanduser() resolution. --- gateway/platforms/api_server.py | 30 ++++++++++++---- .../test_api_server_media_data_urls.py | 34 +++++++++++++++++++ 2 files changed, 57 insertions(+), 7 deletions(-) diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 3cf11f3359d..a029596849d 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -54,9 +54,11 @@ except ImportError: from gateway.config import Platform, PlatformConfig from gateway.platforms.base import ( + MEDIA_TAG_CLEANUP_RE, BasePlatformAdapter, SendResult, is_network_accessible, + validate_media_delivery_path, ) from agent.redact import redact_sensitive_text @@ -581,9 +583,6 @@ _MEDIA_MIME = { ".webp": "image/webp", ".bmp": "image/bmp", } -_MEDIA_TAG_RE = re.compile( - r"[`\"']?MEDIA:\s*(`[^`\n]+`|\"[^\"\n]+\"|'[^'\n]+'|\S+)[`\"']?" -) _MEDIA_DATA_URL_MAX_BYTES = 5 * 1024 * 1024 # skip images larger than 5MB @@ -594,18 +593,35 @@ def _resolve_media_to_data_urls(text: str) -> str: ``MEDIA:`` tags referencing images on the server are useless to them. Inline small local images as markdown data URLs; non-image or unreadable paths are left untouched. + + Uses the same anchored ``MEDIA_TAG_CLEANUP_RE`` matcher and + ``validate_media_delivery_path`` safety check every other platform + adapter's media delivery already goes through (gateway/platforms/base.py) + — an absolute-path anchor plus a known-extension requirement, and a + resolved-path check against the credential/system-path denylist. The + prior pattern here matched any bare token after ``MEDIA:`` (including a + relative/traversal path like ``../../etc/passwd.png``) and read the file + directly with no denylist, so any image-suffixed, readable file the + process could see was base64-exfiltrated to the API caller if its path + merely appeared in the model's own final reply text. """ if not text or "MEDIA:" not in text: return text import base64 def _to_data_url(path_str: str) -> Optional[str]: - p = Path(path_str.strip().strip("`\"'")).expanduser() + # validate_media_delivery_path() strips wrapping quotes/backticks + # and trailing punctuation internally, same as MEDIA_TAG_CLEANUP_RE's + # other callers (extract_media / _strip_media_tag_directives) rely on. + safe_path = validate_media_delivery_path(path_str) + if not safe_path: + return None + p = Path(safe_path) suffix = p.suffix.lower() if suffix not in _MEDIA_IMG_EXT: return None try: - if not p.is_file() or p.stat().st_size > _MEDIA_DATA_URL_MAX_BYTES: + if p.stat().st_size > _MEDIA_DATA_URL_MAX_BYTES: return None b64 = base64.b64encode(p.read_bytes()).decode() except OSError: @@ -613,10 +629,10 @@ def _resolve_media_to_data_urls(text: str) -> str: return f"![image](data:{_MEDIA_MIME[suffix]};base64,{b64})" def _repl(m: "re.Match[str]") -> str: - return _to_data_url(m.group(1)) or m.group(0) + return _to_data_url(m.group("path")) or m.group(0) try: - return _MEDIA_TAG_RE.sub(_repl, text) + return MEDIA_TAG_CLEANUP_RE.sub(_repl, text) except Exception: return text diff --git a/tests/gateway/test_api_server_media_data_urls.py b/tests/gateway/test_api_server_media_data_urls.py index 960f4b194f8..bf0036b3248 100644 --- a/tests/gateway/test_api_server_media_data_urls.py +++ b/tests/gateway/test_api_server_media_data_urls.py @@ -72,6 +72,40 @@ class TestResolveMediaToDataUrls(unittest.TestCase): out = _resolve_media_to_data_urls(f"MEDIA:{p1}\nand MEDIA:{p2}") self.assertEqual(out.count("data:image/png;base64,"), 2) + def test_relative_traversal_path_not_inlined(self): + """A relative/traversal path must never be inlined — the anchored + MEDIA_TAG_CLEANUP_RE matcher requires an absolute-path prefix + (~/, /, or a Windows drive letter), so a bare relative token after + MEDIA: is left as literal text rather than resolved against cwd.""" + text = "MEDIA:../../../../etc/passwd.png" + self.assertEqual(_resolve_media_to_data_urls(text), text) + + def test_credential_path_not_inlined_even_with_image_extension(self): + """An absolute path under the credential/system-path denylist + (validate_media_delivery_path) must not be inlined even though it + has an allowed image extension and the tag matcher's shape.""" + text = "MEDIA:~/.ssh/id_rsa.png" + self.assertEqual(_resolve_media_to_data_urls(text), text) + + def test_symlink_escaping_to_denylisted_target_not_inlined(self): + """A symlink whose resolved target lands under a denylisted system + prefix (/etc) must not be inlined — validate_media_delivery_path + resolves symlinks before the containment/denylist check runs, so + the traversal can't be laundered through an innocuous-looking + image-suffixed symlink name.""" + import os + import tempfile + from pathlib import Path + + d = Path(tempfile.mkdtemp(prefix="hermes_media_test_symlink")) + link = d / "shot.png" + try: + os.symlink("/etc/hosts", link) + except OSError: + self.skipTest("symlink creation not supported in this environment") + text = f"MEDIA:{link}" + self.assertEqual(_resolve_media_to_data_urls(text), text) + if __name__ == "__main__": unittest.main()