mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-21 15:55:43 +00:00
fix(gateway): deliver MEDIA: tags for every file type via validated egress
MEDIA: tags whose path had an unknown extension (.py, .log, .toml, .weirdext, ...) fell between both extraction passes: the anchored extension allowlist (MEDIA_TAG_CLEANUP_RE) did not match them, and the extension-less pass explicitly skipped any path that HAD a suffix. The file was never delivered even though the intended design (universal ingress/egress) says any non-credential file should ship. Widen _path_lacks_deliverable_extension() so the validated delivery pass (MEDIA_EXTENSIONLESS_TAG_RE + validate_media_delivery_path) covers every path the extension allowlist does not — unknown extensions and extension-less files alike. Security posture is unchanged: unknown-extension paths only deliver after full validation (exists, symlinks resolved, credential/system denylist, strict-mode allowlist+recency), and unvalidated tags stay visible in the text instead of being silently dropped. Known extensions keep their unconditional pre-existing behavior. Because extract_media, _strip_media_tag_directives (non-streaming dispatch), and strip_media_directives_for_display (streaming) all share the same two regexes + predicate, all delivery paths pick up the widened behavior with no per-site changes. Dispatch partition in gateway/run.py already routes non-image/video extensions through send_document. Closes the gap reported in PR #36060; supersedes the allowlist-append approach there (an extension allowlist can never enumerate every file type a user asks the agent to produce). Co-authored-by: Randimt <randimt@users.noreply.github.com>
This commit is contained in:
parent
eb6aa03609
commit
2fd36b17c5
2 changed files with 124 additions and 11 deletions
|
|
@ -1504,12 +1504,16 @@ MEDIA_TAG_CLEANUP_RE = re.compile(
|
|||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
# Extension-less absolute paths (e.g. Caddyfile, Dockerfile, Makefile) are
|
||||
# intentionally excluded from MEDIA_TAG_CLEANUP_RE — they are validated and
|
||||
# delivered via MEDIA_EXTENSIONLESS_TAG_RE so prompt-injection paths that do
|
||||
# not exist on disk are left visible instead of silently dropped. Paths with
|
||||
# an unknown but present extension (e.g. .weirdext) stay on the #34517
|
||||
# bare-path fallback and are not handled here.
|
||||
# Paths NOT covered by MEDIA_TAG_CLEANUP_RE's extension alternation — both
|
||||
# extension-less files (Caddyfile, Dockerfile, Makefile) and files with an
|
||||
# unknown extension (.py, .log, .weirdext, ...) — are validated and delivered
|
||||
# via MEDIA_EXTENSIONLESS_TAG_RE. Every ``MEDIA:`` path is therefore
|
||||
# deliverable regardless of file type (#36060): known extensions extract
|
||||
# unconditionally via the anchored pattern above, everything else extracts
|
||||
# only after ``validate_media_delivery_path`` accepts it (exists on disk, not
|
||||
# under the credential/system denylist, strict-mode rules honored), so
|
||||
# prompt-injection paths that do not validate are left visible instead of
|
||||
# silently dropped.
|
||||
MEDIA_EXTENSIONLESS_TAG_RE = re.compile(
|
||||
r'''[`"']?MEDIA:\s*'''
|
||||
r'''(?P<path>`[^`\n]+`|"[^"\n]+"|'[^'\n]+'|'''
|
||||
|
|
@ -1527,8 +1531,16 @@ def _normalize_media_tag_path(raw: str) -> str:
|
|||
|
||||
|
||||
def _path_lacks_deliverable_extension(path: str) -> bool:
|
||||
"""True only when the basename has no extension (Caddyfile, Makefile, …)."""
|
||||
return not Path(path).suffix
|
||||
"""True when MEDIA_TAG_CLEANUP_RE's extension alternation does not cover
|
||||
``path`` — either the basename has no extension at all (Caddyfile,
|
||||
Makefile, …) or the extension is not in MEDIA_DELIVERY_EXTS (.py, .log,
|
||||
.weirdext, …). Such paths route through the validated delivery pass
|
||||
(``validate_media_delivery_path``) instead of the unconditional one, so
|
||||
every file type is deliverable (#36060) while nonexistent / denylisted
|
||||
paths stay visible in the text.
|
||||
"""
|
||||
suffix = Path(path).suffix.lower()
|
||||
return not suffix or suffix not in MEDIA_DELIVERY_EXTS
|
||||
|
||||
|
||||
def _strip_media_tag_directives(text: str) -> str:
|
||||
|
|
|
|||
|
|
@ -631,12 +631,15 @@ class TestMediaExtensionAllowlistParity:
|
|||
|
||||
def test_unknown_extension_not_black_holed_by_cleanup(self):
|
||||
"""A MEDIA: tag with an unknown extension is NOT stripped from the
|
||||
body — it survives so extract_local_files can still see the bare path,
|
||||
rather than vanishing entirely (the core of issue #34517)."""
|
||||
body by the extension-anchored cleanup — and when the path does not
|
||||
validate (nonexistent file here), it is not delivered either, so the
|
||||
tag survives visibly instead of vanishing (the core of issue #34517).
|
||||
Validated unknown-extension paths DO deliver — see
|
||||
TestUniversalMediaEgress (#36060)."""
|
||||
from gateway.platforms.base import MEDIA_TAG_CLEANUP_RE
|
||||
text = "Saved to MEDIA:/tmp/data.weirdext done"
|
||||
media, _ = BasePlatformAdapter.extract_media(text)
|
||||
assert media == [] # unknown extension is not a deliverable MEDIA tag
|
||||
assert media == [] # nonexistent path fails validation, not delivered
|
||||
stripped = MEDIA_TAG_CLEANUP_RE.sub("", text)
|
||||
assert "/tmp/data.weirdext" in stripped # path preserved, not dropped
|
||||
|
||||
|
|
@ -712,6 +715,104 @@ class TestExtensionlessMediaDelivery:
|
|||
)
|
||||
|
||||
|
||||
class TestUniversalMediaEgress:
|
||||
"""#36060: every MEDIA: path is deliverable regardless of file type.
|
||||
|
||||
Known extensions extract unconditionally (MEDIA_TAG_CLEANUP_RE); unknown
|
||||
extensions and extension-less files extract via the validated pass —
|
||||
delivered when validate_media_delivery_path accepts them, left visible
|
||||
when it does not (nonexistent, denylisted).
|
||||
"""
|
||||
|
||||
def _patch_allow_root(self, monkeypatch, root):
|
||||
monkeypatch.setattr(
|
||||
"gateway.platforms.base.MEDIA_DELIVERY_SAFE_ROOTS",
|
||||
(str(root),),
|
||||
)
|
||||
monkeypatch.delenv("HERMES_MEDIA_DELIVERY_STRICT", raising=False)
|
||||
|
||||
@pytest.mark.parametrize("name", [
|
||||
"script.py", "server.log", "notes.weirdext", "app.ts", "run.sh",
|
||||
"config.toml", "styles.css", "contract.sol",
|
||||
])
|
||||
def test_unknown_extension_delivered_when_file_validates(
|
||||
self, tmp_path, monkeypatch, name,
|
||||
):
|
||||
root = tmp_path / "output"
|
||||
root.mkdir()
|
||||
f = root / name
|
||||
f.write_text("content", encoding="utf-8")
|
||||
self._patch_allow_root(monkeypatch, root)
|
||||
|
||||
content = f"Here you go:\nMEDIA:{f}\nDone."
|
||||
media, cleaned = BasePlatformAdapter.extract_media(content)
|
||||
assert len(media) == 1
|
||||
assert media[0][0] == str(f.resolve())
|
||||
assert "MEDIA:" not in cleaned
|
||||
assert "Done." in cleaned
|
||||
|
||||
def test_unknown_extension_left_visible_when_not_on_disk(
|
||||
self, tmp_path, monkeypatch,
|
||||
):
|
||||
root = tmp_path / "output"
|
||||
root.mkdir()
|
||||
self._patch_allow_root(monkeypatch, root)
|
||||
|
||||
content = "MEDIA:/nonexistent/script.py"
|
||||
media, cleaned = BasePlatformAdapter.extract_media(content)
|
||||
assert media == []
|
||||
assert "MEDIA:/nonexistent/script.py" in cleaned
|
||||
|
||||
def test_denylisted_paths_still_rejected_regardless_of_extension(
|
||||
self, tmp_path, monkeypatch,
|
||||
):
|
||||
# A denylisted path must not deliver even though .py/.log/etc now
|
||||
# route through the validated pass. _media_delivery_denied_paths()
|
||||
# reads _MEDIA_DELIVERY_DENIED_PREFIXES at call time, so patching the
|
||||
# tuple exercises the real denylist logic.
|
||||
secret_dir = tmp_path / "secrets"
|
||||
secret_dir.mkdir()
|
||||
f = secret_dir / "creds.py"
|
||||
f.write_text("TOKEN = 'x'", encoding="utf-8")
|
||||
monkeypatch.setattr(
|
||||
"gateway.platforms.base._MEDIA_DELIVERY_DENIED_PREFIXES",
|
||||
(str(secret_dir),),
|
||||
)
|
||||
monkeypatch.delenv("HERMES_MEDIA_DELIVERY_STRICT", raising=False)
|
||||
|
||||
content = f"MEDIA:{f}"
|
||||
media, cleaned = BasePlatformAdapter.extract_media(content)
|
||||
assert media == []
|
||||
assert "MEDIA:" in cleaned # rejected tag stays visible
|
||||
|
||||
def test_strip_for_display_strips_validated_unknown_extension(
|
||||
self, tmp_path, monkeypatch,
|
||||
):
|
||||
root = tmp_path / "output"
|
||||
root.mkdir()
|
||||
f = root / "server.log"
|
||||
f.write_text("x", encoding="utf-8")
|
||||
self._patch_allow_root(monkeypatch, root)
|
||||
|
||||
text = f"MEDIA:{f}"
|
||||
stripped = BasePlatformAdapter.strip_media_directives_for_display(text)
|
||||
assert "MEDIA:" not in stripped
|
||||
|
||||
def test_strip_for_display_keeps_unvalidated_unknown_extension(self):
|
||||
text = "MEDIA:/nonexistent/server.log"
|
||||
stripped = BasePlatformAdapter.strip_media_directives_for_display(text)
|
||||
assert "MEDIA:/nonexistent/server.log" in stripped
|
||||
|
||||
def test_known_extension_still_unconditional(self):
|
||||
# Known extensions keep the pre-#36060 behavior: extracted (and the
|
||||
# tag stripped) even when the file does not exist — downstream
|
||||
# delivery surfaces the failure.
|
||||
content = "MEDIA:/nonexistent/report.pdf"
|
||||
media, cleaned = BasePlatformAdapter.extract_media(content)
|
||||
assert media == [("/nonexistent/report.pdf", False)]
|
||||
assert "MEDIA:" not in cleaned
|
||||
|
||||
|
||||
class TestMediaDeliveryPathValidation:
|
||||
def _patch_roots(self, monkeypatch, *roots):
|
||||
monkeypatch.setattr(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue