fix(gateway): deliver MEDIA tags for extension-less files when path validates

Files like Caddyfile or Makefile have no extension, so MEDIA_TAG_CLEANUP_RE
never matched them and Telegram showed the raw MEDIA: line as text. Extract
and strip validated extension-less tags via a second pass.
This commit is contained in:
HexLab98 2026-06-30 21:38:13 +07:00 committed by Teknium
parent dc3d435f9d
commit 6da181062b
2 changed files with 79 additions and 16 deletions

View file

@ -1442,6 +1442,48 @@ 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.
MEDIA_EXTENSIONLESS_TAG_RE = re.compile(
r'''[`"']?MEDIA:\s*'''
r'''(?P<path>`[^`\n]+`|"[^"\n]+"|'[^'\n]+'|'''
r'''(?:~/|/|[A-Za-z]:[/\\])[^\s\n`"']+)'''
r'''[`"']?\s*''',
re.IGNORECASE,
)
def _normalize_media_tag_path(raw: str) -> str:
path = str(raw or "").strip()
if len(path) >= 2 and path[0] == path[-1] and path[0] in "`\"'":
path = path[1:-1].strip()
return path.lstrip("`\"'").rstrip("`\"',.;:)}]")
def _path_lacks_deliverable_extension(path: str) -> bool:
"""True only when the basename has no extension (Caddyfile, Makefile, …)."""
return not Path(path).suffix
def _strip_media_tag_directives(text: str) -> str:
"""Remove MEDIA: tags and [[audio_as_voice]] / [[as_document]] markers."""
if "MEDIA:" not in text and "[[audio_as_voice]]" not in text:
return text
cleaned = text.replace("[[audio_as_voice]]", "").replace("[[as_document]]", "")
def _strip_extensionless(match: re.Match) -> str:
path = _normalize_media_tag_path(match.group("path"))
if not path or not _path_lacks_deliverable_extension(path):
return match.group(0)
return "" if validate_media_delivery_path(path) else match.group(0)
cleaned = MEDIA_TAG_CLEANUP_RE.sub("", cleaned)
return MEDIA_EXTENSIONLESS_TAG_RE.sub(_strip_extensionless, cleaned)
def get_document_cache_dir() -> Path:
"""Return the document cache directory, creating it if it doesn't exist."""
@ -2118,13 +2160,13 @@ def _strip_media_directives(text: str) -> str:
Backstop only: run ``extract_media`` first. MEDIA cleanup uses the shared
``MEDIA_TAG_CLEANUP_RE`` (only tags whose path has a known deliverable
extension are removed; an unknown-extension tag is intentionally left so the
bare-path detector downstream can still pick it up, per #34517). [[...]] is
exact.
bare-path detector downstream can still pick it up, per #34517). Validated
extension-less tags (e.g. ``MEDIA:/output/Caddyfile``) are also removed.
[[...]] is exact.
"""
if not text:
return text
text = text.replace("[[audio_as_voice]]", "").replace("[[as_document]]", "")
return MEDIA_TAG_CLEANUP_RE.sub("", text)
return _strip_media_tag_directives(text)
class BasePlatformAdapter(ABC):
@ -3447,10 +3489,7 @@ class BasePlatformAdapter(ABC):
scan_content = BasePlatformAdapter._mask_protected_spans(content)
scan_content = BasePlatformAdapter._mask_json_string_media(scan_content)
for match in media_pattern.finditer(scan_content):
path = match.group("path").strip()
if len(path) >= 2 and path[0] == path[-1] and path[0] in "`\"'":
path = path[1:-1].strip()
path = path.lstrip("`\"'").rstrip("`\"',.;:)}]")
path = _normalize_media_tag_path(match.group("path"))
if path:
try:
media.append((os.path.expanduser(path), has_voice_tag))
@ -3459,6 +3498,16 @@ class BasePlatformAdapter(ABC):
# and dropping every other attachment in the response.
continue
seen_paths = {p for p, _ in media}
for match in MEDIA_EXTENSIONLESS_TAG_RE.finditer(scan_content):
path = _normalize_media_tag_path(match.group("path"))
if not path or not _path_lacks_deliverable_extension(path):
continue
safe = validate_media_delivery_path(path)
if safe and safe not in seen_paths:
media.append((safe, has_voice_tag))
seen_paths.add(safe)
# Remove the delivered MEDIA tags from the user-visible text. Mask a
# length-equal copy of ``cleaned`` (same union of protected regions) to
# *locate* the real tag spans, then delete exactly those spans from the
@ -3471,6 +3520,12 @@ class BasePlatformAdapter(ABC):
masked_cleaned = BasePlatformAdapter._mask_protected_spans(cleaned)
masked_cleaned = BasePlatformAdapter._mask_json_string_media(masked_cleaned)
spans = [m.span() for m in media_pattern.finditer(masked_cleaned)]
for match in MEDIA_EXTENSIONLESS_TAG_RE.finditer(masked_cleaned):
path = _normalize_media_tag_path(match.group("path"))
if not path or not _path_lacks_deliverable_extension(path):
continue
if validate_media_delivery_path(path):
spans.append(match.span())
if spans:
chars = list(cleaned)
for start, end in sorted(spans, reverse=True):
@ -3480,6 +3535,21 @@ class BasePlatformAdapter(ABC):
return media, cleaned
@staticmethod
def strip_media_directives_for_display(text: str) -> str:
"""Strip MEDIA: directives from streamed/display text.
Known-extension tags are removed unconditionally (same as
``MEDIA_TAG_CLEANUP_RE``). Extension-less tags are removed only when
``validate_media_delivery_path`` accepts the path so undeliverable
paths stay visible for debugging.
"""
if "MEDIA:" not in text and "[[audio_as_voice]]" not in text:
return text
cleaned = _strip_media_tag_directives(text)
cleaned = re.sub(r'\n{3,}', '\n\n', cleaned)
return cleaned.rstrip()
@staticmethod
def extract_local_files(content: str) -> Tuple[List[str], str]:
"""

View file

@ -813,14 +813,7 @@ class GatewayStreamConsumer:
stream finishes we just need to hide the raw directives from the
user.
"""
if "MEDIA:" not in text and "[[audio_as_voice]]" not in text:
return text
cleaned = text.replace("[[audio_as_voice]]", "")
cleaned = GatewayStreamConsumer._MEDIA_RE.sub("", cleaned)
# Collapse excessive blank lines left behind by removed tags
cleaned = re.sub(r'\n{3,}', '\n\n', cleaned)
# Strip trailing whitespace/newlines but preserve leading content
return cleaned.rstrip()
return _BasePlatformAdapter.strip_media_directives_for_display(text)
async def _send_new_chunk(
self,