Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 44 additions & 19 deletions gateway/platforms/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -2504,6 +2504,37 @@ def filter_local_delivery_paths(file_paths) -> List[str]:
logger.warning("Skipping unsafe local file path: %s", _log_safe_path(raw))
return safe_paths

# Shared extension list for MEDIA:<path> tag extraction (extract_media)
# and bare-path extraction (extract_local_files). Both methods must
# accept the same set of extensions or a MEDIA-tagged path can fall
# back to plain text when the bare-path path would have delivered it.
# See #34321 — commit ea49b3862 narrowed extract_media's regex to fix
# Mattermost false positives but left out .md, .doc, .odt, .rtf, .ods,
# .tsv, .json, .xml, .yaml/.yml, .ppt, .odp, .key, .tar, .gz/.tgz/.bz2/
# .xz, .html/.htm, .tiff, .svg, .bmp. Centralizing the list here keeps
# the two methods in lockstep.
_MEDIA_DELIVERY_EXTS: Tuple[str, ...] = (
# Images (embed inline)
".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tiff", ".svg",
# Video (embed inline where supported)
".mp4", ".mov", ".avi", ".mkv", ".webm",
# Audio (delivered as voice/audio where supported)
".mp3", ".wav", ".ogg", ".opus", ".m4a", ".flac",
# Documents (uploaded as file attachments)
".pdf", ".epub", ".docx", ".doc", ".odt", ".rtf", ".txt", ".md",
# Spreadsheets / data
".xlsx", ".xls", ".ods", ".csv", ".tsv", ".json", ".xml",
".yaml", ".yml",
# Presentations
".pptx", ".ppt", ".odp", ".key",
# Archives
".zip", ".tar", ".gz", ".tgz", ".bz2", ".xz", ".7z", ".rar",
# Web / rendered output
".html", ".htm",
# Mobile app packages
".apk", ".ipa",
)

@staticmethod
def extract_media(content: str) -> Tuple[List[Tuple[str, bool]], str]:
"""
Expand Down Expand Up @@ -2543,8 +2574,17 @@ def extract_media(content: str) -> Tuple[List[Tuple[str, bool]], str]:

# Extract MEDIA:<path> tags, allowing optional whitespace after the colon
# and quoted/backticked paths for LLM-formatted outputs.
# Extension list is the shared _MEDIA_DELIVERY_EXTS so MEDIA-tagged
# paths cover the same set as bare paths (see #34321).
_ext_alt = "|".join(
sorted(set(e.lstrip(".") for e in BasePlatformAdapter._MEDIA_DELIVERY_EXTS),
key=lambda s: -len(s))
)
media_pattern = re.compile(
r'''[`"']?MEDIA:\s*(?P<path>`[^`\n]+`|"[^"\n]+"|'[^'\n]+'|(?:~/|/)\S+(?:[^\S\n]+\S+)*?\.(?:png|jpe?g|gif|webp|mp4|mov|avi|mkv|webm|ogg|opus|mp3|wav|m4a|flac|epub|pdf|zip|rar|7z|docx?|xlsx?|pptx?|txt|csv|apk|ipa)(?=[\s`"',;:)\]}]|$))[`"']?'''
r'''[`"']?MEDIA:\s*(?P<path>`[^`\n]+`|"[^"\n]+"|'[^'\n]+'|(?:~/|/)\S+(?:[^\S\n]+\S+)*?\.(?:'''
+ _ext_alt
+ r''')(?=[\s`"',;:)\]}]|$))[`"']?''',
re.IGNORECASE,
)
for match in media_pattern.finditer(content):
path = match.group("path").strip()
Expand Down Expand Up @@ -2591,24 +2631,9 @@ def extract_local_files(content: str) -> Tuple[List[str], str]:
Tuple of (list of expanded file paths, cleaned text with the
raw path strings removed).
"""
_LOCAL_MEDIA_EXTS = (
# Images (embed inline)
'.png', '.jpg', '.jpeg', '.gif', '.webp', '.bmp', '.tiff', '.svg',
# Video (embed inline where supported)
'.mp4', '.mov', '.avi', '.mkv', '.webm',
# Audio (delivered as voice/audio where supported)
'.mp3', '.wav', '.ogg', '.m4a', '.flac',
# Documents (uploaded as file attachments)
'.pdf', '.docx', '.doc', '.odt', '.rtf', '.txt', '.md',
# Spreadsheets / data
'.xlsx', '.xls', '.ods', '.csv', '.tsv', '.json', '.xml', '.yaml', '.yml',
# Presentations
'.pptx', '.ppt', '.odp', '.key',
# Archives
'.zip', '.tar', '.gz', '.tgz', '.bz2', '.xz', '.7z', '.rar',
# Web / rendered output
'.html', '.htm',
)
# Use the shared _MEDIA_DELIVERY_EXTS constant so MEDIA-tagged paths
# and bare-path detection stay in lockstep. See #34321.
_LOCAL_MEDIA_EXTS = BasePlatformAdapter._MEDIA_DELIVERY_EXTS
ext_part = '|'.join(e.lstrip('.') for e in _LOCAL_MEDIA_EXTS)

# (?<![/:\w.]) prevents matching inside URLs (e.g. https://…/img.png)
Expand Down
104 changes: 104 additions & 0 deletions tests/gateway/test_platform_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -362,6 +362,110 @@ def test_both_directives_can_coexist(self):
assert "[[as_document]]" not in cleaned


class TestExtractMediaExtensionParity:
"""Regression tests for #34321.

extract_media() and extract_local_files() must accept the same set of
file extensions. Before #34321 the bare-path extractor would happily
deliver a .md file but a MEDIA:/path/foo.md tag would fall through to
plain text on every platform.
"""

# Extensions that #34321 reported missing from extract_media.
_PREVIOUSLY_MISSING = (
"md", "doc", "odt", "rtf", "ods", "tsv", "json", "xml",
"yaml", "yml", "ppt", "odp", "key", "tar", "gz", "tgz",
"bz2", "xz", "html", "htm", "tiff", "svg", "bmp",
)

def test_shared_constant_exists(self):
assert hasattr(BasePlatformAdapter, "_MEDIA_DELIVERY_EXTS")
exts = BasePlatformAdapter._MEDIA_DELIVERY_EXTS
assert isinstance(exts, tuple)
assert len(exts) > 30 # not an empty / stub value

def test_md_extension_in_shared_constant(self):
# #34321 was opened against missing .md specifically.
assert ".md" in BasePlatformAdapter._MEDIA_DELIVERY_EXTS

@pytest.mark.parametrize("ext", _PREVIOUSLY_MISSING)
def test_previously_missing_extension_now_in_shared_constant(self, ext):
assert f".{ext}" in BasePlatformAdapter._MEDIA_DELIVERY_EXTS, (
f".{ext} should be in _MEDIA_DELIVERY_EXTS per #34321"
)

@pytest.mark.parametrize("ext", _PREVIOUSLY_MISSING)
def test_extract_media_accepts_previously_missing_extension(self, ext):
"""The actual bug fix: MEDIA:<path> with these extensions extracts."""
content = f"MEDIA:/tmp/test.{ext}"
media, cleaned = BasePlatformAdapter.extract_media(content)
assert len(media) == 1, (
f"extract_media should match .{ext} after #34321 fix "
f"(got {media!r}, cleaned={cleaned!r})"
)
assert media[0][0] == f"/tmp/test.{ext}"
# Tag was stripped from cleaned text.
assert "MEDIA:" not in cleaned

def test_extract_media_accepts_real_world_md_doc_path(self):
# Match the reproduction from #34321: a documents-cache path.
content = "Here you go:\nMEDIA:~/.hermes/cache/documents/report.md"
media, cleaned = BasePlatformAdapter.extract_media(content)
assert len(media) == 1
assert media[0][0].endswith("/report.md")
assert "Here you go:" in cleaned
assert "MEDIA:" not in cleaned

def test_extract_media_and_extract_local_files_extension_sets_match(self):
"""The whole point of the fix: both methods accept identical exts."""
import tempfile
import os

media_exts = set(BasePlatformAdapter._MEDIA_DELIVERY_EXTS)

# Probe extract_local_files indirectly: pick a few representative
# extensions and confirm a real file at /tmp gets picked up. We
# don't probe all 50+ exts because that's slow, but we cover the
# categories: doc/data/archive/web.
for ext in (".md", ".json", ".tar", ".html", ".tiff"):
assert ext in media_exts
with tempfile.NamedTemporaryFile(
suffix=ext, delete=False, dir="/tmp"
) as tf:
tf.write(b"x")
path = tf.name
try:
# extract_media should match a MEDIA:<this path> tag
media, _ = BasePlatformAdapter.extract_media(f"MEDIA:{path}")
assert len(media) == 1, f".{ext}: extract_media missed it"
# extract_local_files should also pick up the bare path
paths, _ = BasePlatformAdapter.extract_local_files(
f"Here is the file: {path}"
)
assert path in paths, f".{ext}: extract_local_files missed it"
finally:
try:
os.unlink(path)
except OSError:
pass

def test_extract_media_still_strips_voice_tag(self):
# Behavior preserved: [[audio_as_voice]] still applies to .md too.
content = "[[audio_as_voice]]\nMEDIA:/tmp/test.md"
media, cleaned = BasePlatformAdapter.extract_media(content)
assert len(media) == 1
assert media[0][1] is True
assert "[[audio_as_voice]]" not in cleaned

def test_extract_media_still_rejects_unknown_extension(self):
# Guardrail: not every extension is acceptable. .exe / .scr / random
# tokens should NOT match — the bug was missing extensions, not
# opening the gate to anything.
content = "MEDIA:/tmp/malware.exe"
media, _ = BasePlatformAdapter.extract_media(content)
assert media == []


class TestMediaDeliveryPathValidation:
def _patch_roots(self, monkeypatch, *roots):
monkeypatch.setattr(
Expand Down
Loading