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
11 changes: 11 additions & 0 deletions gateway/platforms/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -830,16 +830,27 @@ def cache_video_from_bytes(data: bytes, ext: str = ".mp4") -> str:
MEDIA_DELIVERY_TRUST_RECENT_ENV = "HERMES_MEDIA_TRUST_RECENT_FILES"
MEDIA_DELIVERY_TRUST_RECENT_SECONDS_ENV = "HERMES_MEDIA_TRUST_RECENT_SECONDS"
MEDIA_DELIVERY_SAFE_ROOTS = (
# get_hermes_dir() entries — resolve to legacy when it exists, canonical otherwise.
IMAGE_CACHE_DIR,
AUDIO_CACHE_DIR,
VIDEO_CACHE_DIR,
DOCUMENT_CACHE_DIR,
SCREENSHOT_CACHE_DIR,
# Explicit legacy paths (e.g. image_cache/, audio_cache/).
_HERMES_HOME / "image_cache",
_HERMES_HOME / "audio_cache",
_HERMES_HOME / "video_cache",
_HERMES_HOME / "document_cache",
_HERMES_HOME / "browser_screenshots",
# Explicit canonical paths (e.g. cache/images/, cache/audio/).
# Required because image_gen_provider writes to cache/images directly, and
# get_hermes_dir() resolves to the legacy path when it already exists on disk,
# leaving the canonical path uncovered. See issue #31733.
_HERMES_HOME / "cache" / "images",
_HERMES_HOME / "cache" / "audio",
_HERMES_HOME / "cache" / "videos",
_HERMES_HOME / "cache" / "documents",
_HERMES_HOME / "cache" / "screenshots",
)

# Default recency window for trusting freshly-produced files (seconds).
Expand Down
119 changes: 119 additions & 0 deletions tests/gateway/test_platform_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import os
import time
from pathlib import Path
from unittest.mock import patch

import pytest
Expand Down Expand Up @@ -536,6 +537,124 @@ def test_filter_keeps_recently_produced_files(self, tmp_path, monkeypatch):
assert out == [str(fresh.resolve())]


# ---------------------------------------------------------------------------
# Regression: canonical cache paths in MEDIA_DELIVERY_SAFE_ROOTS
# ---------------------------------------------------------------------------
# Issue #31733 — image_gen_provider writes to cache/images/ but the gateway's
# MEDIA_DELIVERY_SAFE_ROOTS only covered the legacy image_cache/ path when
# get_hermes_dir() resolved to it. The fix adds explicit canonical entries.
#
# Unlike the tests above, these do NOT monkeypatch MEDIA_DELIVERY_SAFE_ROOTS.
# They verify the actual default production list covers canonical cache paths
# even when legacy directories also exist on disk.
# ---------------------------------------------------------------------------


class TestCanonicalCachePathRegression:
"""Verify cache/<subdir> paths are accepted alongside legacy directories.

These tests exercise the DEFAULT ``MEDIA_DELIVERY_SAFE_ROOTS`` — no
monkeypatching of the tuple. Recency trust is disabled so every test
exercises the strict allowlist path.
"""

# Canonical subpath → legacy name pairs. Must match MEDIA_DELIVERY_SAFE_ROOTS.
_CACHE_PAIRS = (
("cache/images", "image_cache", "test.png", b"\x89PNG\r\n\x1a\n"),
("cache/audio", "audio_cache", "test.ogg", b"OggS"),
("cache/videos", "video_cache", "test.mp4", b"\x00\x00\x00\x18ftyp"),
("cache/documents", "document_cache", "test.pdf", b"%PDF-1.4"),
("cache/screenshots", "browser_screenshots", "test.png", b"\x89PNG\r\n\x1a\n"),
)

@pytest.fixture(autouse=True)
def _disable_recency_trust(self, monkeypatch):
monkeypatch.setenv("HERMES_MEDIA_TRUST_RECENT_FILES", "0")

def test_default_safe_roots_include_all_canonical_paths(self):
"""Structural: every canonical cache/<subdir> must appear in the default
``MEDIA_DELIVERY_SAFE_ROOTS`` tuple."""
from gateway.platforms.base import (
MEDIA_DELIVERY_SAFE_ROOTS,
_HERMES_HOME,
)

roots_set = {Path(r) for r in MEDIA_DELIVERY_SAFE_ROOTS}
for canonical_subpath, _legacy, _fname, _data in self._CACHE_PAIRS:
expected = _HERMES_HOME / canonical_subpath
assert expected in roots_set, (
f"Canonical path {expected} missing from MEDIA_DELIVERY_SAFE_ROOTS. "
f"Current roots: {sorted(str(r) for r in roots_set)}"
)

@pytest.mark.parametrize(
"canonical_subpath,legacy_name,fname,data",
_CACHE_PAIRS,
ids=[p[0] for p in _CACHE_PAIRS],
)
def test_canonical_cache_file_accepted_with_legacy_present(
self, tmp_path, monkeypatch, canonical_subpath, legacy_name, fname, data,
):
"""End-to-end: a file under cache/<subdir> is accepted even when the
legacy directory also exists on disk.

This reproduces the exact scenario from #31733: ``image_gen_provider``
writes to ``cache/images/`` but ``get_hermes_dir("cache/images",
"image_cache")`` resolves to the legacy ``image_cache/`` when it
exists. Without the explicit canonical entries, the file would be
rejected.
"""
from gateway.platforms.base import (
MEDIA_DELIVERY_SAFE_ROOTS,
_HERMES_HOME,
)

# Build the real canonical and legacy paths under the actual HERMES_HOME.
canonical_dir = _HERMES_HOME / canonical_subpath
legacy_dir = _HERMES_HOME / legacy_name

# Ensure both directories exist (simulating a migrated install).
canonical_dir.mkdir(parents=True, exist_ok=True)
legacy_dir.mkdir(parents=True, exist_ok=True)

# Write a test file to the canonical path.
test_file = canonical_dir / f"_pr_regression_{fname}"
try:
test_file.write_bytes(data)

# The canonical path must be in the default safe roots.
canonical_root = _HERMES_HOME / canonical_subpath
assert canonical_root in MEDIA_DELIVERY_SAFE_ROOTS, (
f"{canonical_root} not in MEDIA_DELIVERY_SAFE_ROOTS"
)

# End-to-end validation must accept the file.
result = BasePlatformAdapter.validate_media_delivery_path(str(test_file))
assert result == str(test_file.resolve()), (
f"validate_media_delivery_path rejected {test_file} "
f"(canonical_subpath={canonical_subpath}, legacy={legacy_name})"
)
finally:
# Clean up test artefacts (leave dirs — they may pre-exist).
test_file.unlink(missing_ok=True)

def test_legacy_cache_file_still_accepted(self):
"""Backward-compat: files under the legacy directory must still work."""
from gateway.platforms.base import _HERMES_HOME

legacy_dir = _HERMES_HOME / "image_cache"
legacy_dir.mkdir(parents=True, exist_ok=True)

test_file = legacy_dir / "_pr_regression_legacy.png"
try:
test_file.write_bytes(b"\x89PNG\r\n\x1a\n")

result = BasePlatformAdapter.validate_media_delivery_path(str(test_file))
assert result == str(test_file.resolve())
finally:
test_file.unlink(missing_ok=True)


# ---------------------------------------------------------------------------
# should_send_media_as_audio
# ---------------------------------------------------------------------------
Expand Down