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
62 changes: 62 additions & 0 deletions gateway/platforms/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -1024,6 +1024,57 @@ def _path_is_within(path: Path, root: Path) -> bool:
return False


def _translate_docker_container_path(candidate: str) -> Optional[str]:
"""Translate a Docker container path to its host-side equivalent.

When ``terminal.backend`` is ``docker``, the agent emits paths that
exist inside the container (e.g. ``/output/report.pdf``). The gateway
runs on the host and cannot resolve those paths directly. This helper
maps the container prefix to the host prefix using the longest-prefix
match against ``terminal.docker_volumes`` entries (``host:container``
format).

Returns the translated host path string, or ``None`` if translation
is not applicable (backend is not docker, no matching volume, etc.).
"""
try:
from hermes_cli.config import load_config as _load_config
cfg = _load_config()
terminal_cfg = cfg.get("terminal", {})
if not isinstance(terminal_cfg, dict):
return None
if terminal_cfg.get("backend") != "docker":
return None
volumes = terminal_cfg.get("docker_volumes", [])
if not isinstance(volumes, list):
return None
except Exception:
return None

# Longest-prefix match on container mount points.
best_match_len = 0
best_host_prefix = None
for vol in volumes:
if not isinstance(vol, str) or ":" not in vol:
continue
host_part, _, container_part = vol.rpartition(":")

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This misparses supported host:container[:options] specs: /host/data:/data:ro produces container_part == "ro", so paths below /data never translate. Please parse the optional mode separately and add a :ro regression test.

if not container_part or not host_part:
continue
container_prefix = container_part.rstrip("/")
if not container_prefix:
continue
if candidate == container_prefix or candidate.startswith(container_prefix + "/"):
if len(container_prefix) > best_match_len:
best_match_len = len(container_prefix)
best_host_prefix = host_part.rstrip("/")

if best_host_prefix is None:
return None

suffix = candidate[best_match_len:]
return best_host_prefix + suffix

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The suffix is unnormalized, so /output/../../home/user/file under /host/export:/output becomes /host/export/../../home/user/file and can resolve outside the configured bind mount. Resolve the host root and candidate and reject results not contained in that root before returning them.



def validate_media_delivery_path(path: str) -> Optional[str]:
"""Return a safe absolute file path for native media delivery, else None.

Expand Down Expand Up @@ -1062,6 +1113,17 @@ def validate_media_delivery_path(path: str) -> Optional[str]:
if not expanded.is_absolute():
return None

# When running inside a Docker sandbox, the agent emits container-local
# paths (e.g. /output/report.pdf) that don't exist on the host.
# Translate them to the host-side path using docker_volumes config
# before resolving.
translated = _translate_docker_container_path(str(expanded))
if translated is not None:
try:
expanded = Path(translated)
except (OSError, RuntimeError, ValueError):
return None

try:
resolved = expanded.resolve(strict=True)
except (OSError, RuntimeError, ValueError):
Expand Down
146 changes: 146 additions & 0 deletions tests/gateway/test_platform_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -1485,3 +1485,149 @@ def test_canonical_cache_roots_present(self):
assert any(r.endswith("cache/documents") for r in roots)
# Legacy layout still present.
assert any(r.endswith("image_cache") for r in roots)


class TestDockerContainerPathTranslation:
"""Tests for _translate_docker_container_path and docker-aware media delivery."""

def _make_config(self, backend="docker", volumes=None):
"""Return a mock load_config result."""
return {
"terminal": {
"backend": backend,
"docker_volumes": volumes or [],
}
}

def test_translates_output_path_to_host(self, monkeypatch):
"""Container /output/file.txt -> host ~/cache/documents/file.txt."""
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: self._make_config(volumes=["/home/user/.hermes/cache/documents:/output"]),
)
from gateway.platforms.base import _translate_docker_container_path
result = _translate_docker_container_path("/output/report.pdf")
assert result == "/home/user/.hermes/cache/documents/report.pdf"

def test_translates_nested_container_path(self, monkeypatch):
"""Container /output/subdir/file.txt -> host path with subdir."""
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: self._make_config(volumes=["/home/user/.hermes/cache/documents:/output"]),
)
from gateway.platforms.base import _translate_docker_container_path
result = _translate_docker_container_path("/output/subdir/data.csv")
assert result == "/home/user/.hermes/cache/documents/subdir/data.csv"

def test_longest_prefix_match(self, monkeypatch):
"""More specific volume mount wins over shorter one."""
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: self._make_config(volumes=[
"/host/a:/output",
"/host/b:/output/special",
]),
)
from gateway.platforms.base import _translate_docker_container_path
# /output/special/file.txt should match /host/b:/output/special (longer prefix)
result = _translate_docker_container_path("/output/special/file.txt")
assert result == "/host/b/file.txt"
# /output/other.txt should match /host/a:/output (shorter prefix)
result2 = _translate_docker_container_path("/output/other.txt")
assert result2 == "/host/a/other.txt"

def test_returns_none_for_non_docker_backend(self, monkeypatch):
"""No translation when backend is not docker."""
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: self._make_config(backend="local", volumes=["/host:/output"]),
)
from gateway.platforms.base import _translate_docker_container_path
result = _translate_docker_container_path("/output/report.pdf")
assert result is None

def test_returns_none_for_no_matching_volume(self, monkeypatch):
"""No translation when path doesn't match any volume mount."""
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: self._make_config(volumes=["/host/data:/workspace"]),
)
from gateway.platforms.base import _translate_docker_container_path
result = _translate_docker_container_path("/output/report.pdf")
assert result is None

def test_returns_none_for_empty_volumes(self, monkeypatch):
"""No translation when docker_volumes is empty."""
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: self._make_config(volumes=[]),
)
from gateway.platforms.base import _translate_docker_container_path
result = _translate_docker_container_path("/output/report.pdf")
assert result is None

def test_returns_none_on_config_error(self, monkeypatch):
"""No translation when config loading fails."""
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: (_ for _ in ()).throw(RuntimeError("config error")),
)
from gateway.platforms.base import _translate_docker_container_path
result = _translate_docker_container_path("/output/report.pdf")
assert result is None

def test_exact_mount_point_match(self, monkeypatch):
"""Container path exactly at mount point translates correctly."""
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: self._make_config(volumes=["/host/data:/data"]),
)
from gateway.platforms.base import _translate_docker_container_path
result = _translate_docker_container_path("/data")
assert result == "/host/data"

def test_validate_with_docker_translation(self, tmp_path, monkeypatch):
"""Integration: validate_media_delivery_path translates container paths."""
host_dir = tmp_path / "cache" / "documents"
host_dir.mkdir(parents=True)
host_file = host_dir / "report.pdf"
host_file.write_bytes(b"%PDF-1.4")

monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: self._make_config(volumes=[f"{host_dir}:/output"]),
)
# Non-strict mode so any existing file passes
monkeypatch.setenv("HERMES_MEDIA_DELIVERY_STRICT", "0")

from gateway.platforms.base import validate_media_delivery_path
result = validate_media_delivery_path("/output/report.pdf")
assert result == str(host_file.resolve())

def test_validate_rejects_container_path_without_docker_backend(self, tmp_path, monkeypatch):
"""Non-docker backend: container path is not translated, so resolve fails."""
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: self._make_config(backend="local", volumes=[f"{tmp_path}:/output"]),
)
monkeypatch.setenv("HERMES_MEDIA_DELIVERY_STRICT", "0")

from gateway.platforms.base import validate_media_delivery_path
result = validate_media_delivery_path("/output/report.pdf")
# /output/report.pdf doesn't exist on the host, so resolve fails
assert result is None

def test_validate_docker_denylist_still_applies(self, tmp_path, monkeypatch):
"""Docker translation doesn't bypass the system-path denylist."""
# Map container /hostroot to host / so /hostroot/etc/passwd -> /etc/passwd.
# /etc is on _MEDIA_DELIVERY_DENIED_PREFIXES, so it should be blocked
# even after docker path translation.
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: self._make_config(volumes=["/:/hostroot"]),
)
monkeypatch.setenv("HERMES_MEDIA_DELIVERY_STRICT", "0")

from gateway.platforms.base import validate_media_delivery_path
result = validate_media_delivery_path("/hostroot/etc/passwd")
assert result is None
Loading