diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index eb393a7a611..c5ed7c7baba 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -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`[^`\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]: """ diff --git a/gateway/stream_consumer.py b/gateway/stream_consumer.py index 63853b24645..9c6d1280875 100644 --- a/gateway/stream_consumer.py +++ b/gateway/stream_consumer.py @@ -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,