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
73 changes: 70 additions & 3 deletions gateway/platforms/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -888,6 +888,9 @@ def cache_video_from_bytes(data: bytes, ext: str = ".mp4") -> str:
def _media_delivery_allowed_roots() -> List[Path]:
"""Return roots from which model-emitted local media may be delivered."""
roots = [Path(root) for root in MEDIA_DELIVERY_SAFE_ROOTS]
docker_workspace = _docker_workspace_host_root()
if docker_workspace is not None:
roots.append(docker_workspace)
extra_roots = os.environ.get(MEDIA_DELIVERY_ALLOW_DIRS_ENV, "")
for chunk in extra_roots.split(os.pathsep):
for raw_root in chunk.split(","):
Expand Down Expand Up @@ -969,6 +972,64 @@ def _path_is_within(path: Path, root: Path) -> bool:
return False


def _docker_workspace_container_path(candidate: Path) -> bool:
"""Return True for absolute container paths rooted at ``/workspace``."""
if not candidate.is_absolute():
return False
try:
candidate.relative_to("/workspace")
return True
except ValueError:
return False


def _docker_workspace_host_root() -> Optional[Path]:
"""Return the host path backing Docker's default persistent ``/workspace``."""
if os.getenv("TERMINAL_ENV", "").strip().lower() != "docker":
return None
if os.getenv("TERMINAL_CONTAINER_PERSISTENT", "true").strip().lower() not in {"1", "true", "yes", "on"}:
return None
if os.getenv("TERMINAL_DOCKER_MOUNT_CWD_TO_WORKSPACE", "false").strip().lower() in {"1", "true", "yes", "on"}:
return None

raw_volumes = os.getenv("TERMINAL_DOCKER_VOLUMES", "").strip()
if raw_volumes:
try:
import json as _json
parsed = _json.loads(raw_volumes)
except Exception:
parsed = []
if isinstance(parsed, list):
for spec in parsed:
if isinstance(spec, str) and ":/workspace" in spec:
return None

try:
from tools.environments.base import get_sandbox_dir

root = (get_sandbox_dir() / "docker" / "default" / "workspace").resolve(strict=True)
except (ImportError, OSError, RuntimeError, ValueError):
return None
return root if root.is_dir() else None


def _translate_docker_workspace_media_path(candidate: Path) -> Optional[Path]:
"""Translate ``/workspace/...`` from a Docker container to its host path."""
if not _docker_workspace_container_path(candidate):
return None
host_workspace = _docker_workspace_host_root()
if host_workspace is None:
return None
try:
relative = candidate.relative_to("/workspace")
translated = (host_workspace / relative).resolve(strict=True)
except (OSError, RuntimeError, ValueError):
return None
if translated != host_workspace and not _path_is_within(translated, host_workspace):
return None
return translated


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

Expand All @@ -991,10 +1052,16 @@ def validate_media_delivery_path(path: str) -> Optional[str]:
if not expanded.is_absolute():
return None

try:
resolved = expanded.resolve(strict=True)
except (OSError, RuntimeError, ValueError):
translated = _translate_docker_workspace_media_path(expanded)
if translated is not None:
resolved = translated
elif _docker_workspace_container_path(expanded) and os.getenv("TERMINAL_ENV", "").strip().lower() == "docker":
return None
else:
try:
resolved = expanded.resolve(strict=True)
except (OSError, RuntimeError, ValueError):
return None

if not resolved.is_file():
return None
Expand Down
28 changes: 27 additions & 1 deletion plugins/image_gen/openai/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,13 @@
2. ``image_gen.openai.model`` in ``config.yaml``
3. ``image_gen.model`` in ``config.yaml`` (when it's one of our tier IDs)
4. :data:`DEFAULT_MODEL` — ``gpt-image-2-medium``

Endpoint override:

- ``image_gen.openai.base_url`` in ``config.yaml`` is passed explicitly to
``openai.OpenAI(base_url=...)`` when set, so image generation can target a
provider-specific OpenAI-compatible endpoint without relying on the global
``OPENAI_BASE_URL`` environment variable.
"""

from __future__ import annotations
Expand Down Expand Up @@ -117,6 +124,21 @@ def _resolve_model() -> Tuple[str, Dict[str, Any]]:
return DEFAULT_MODEL, _MODELS[DEFAULT_MODEL]


def _resolve_base_url() -> Optional[str]:
"""Return a normalized explicit base URL for the OpenAI image client."""
cfg = _load_openai_config()
openai_cfg = cfg.get("openai") if isinstance(cfg.get("openai"), dict) else {}
if not isinstance(openai_cfg, dict):
return None

value = openai_cfg.get("base_url")
if not isinstance(value, str):
return None

cleaned = value.strip().rstrip("/")
return cleaned or None


# ---------------------------------------------------------------------------
# Provider
# ---------------------------------------------------------------------------
Expand Down Expand Up @@ -224,7 +246,11 @@ def generate(
}

try:
client = openai.OpenAI()
base_url = _resolve_base_url()
if base_url is not None:
client = openai.OpenAI(base_url=base_url)
else:
client = openai.OpenAI()
response = client.images.generate(**payload)
except Exception as exc:
logger.debug("OpenAI image generation failed", exc_info=True)
Expand Down
94 changes: 94 additions & 0 deletions tests/gateway/test_platform_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@

import pytest

from hermes_constants import reset_hermes_home_override, set_hermes_home_override
from gateway.platforms.base import (
BasePlatformAdapter,
GATEWAY_SECRET_CAPTURE_UNSUPPORTED_MESSAGE,
Expand Down Expand Up @@ -373,6 +374,10 @@ def _patch_roots(self, monkeypatch, *roots):
# specifically cover recency trust re-enable it themselves.
monkeypatch.setenv("HERMES_MEDIA_TRUST_RECENT_FILES", "0")

def _set_hermes_home(self, monkeypatch, hermes_home):
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
return set_hermes_home_override(hermes_home)

def test_allows_existing_file_inside_safe_root(self, tmp_path, monkeypatch):
root = tmp_path / "media-cache"
media_file = root / "voice.ogg"
Expand Down Expand Up @@ -535,6 +540,95 @@ def test_filter_keeps_recently_produced_files(self, tmp_path, monkeypatch):
out = BasePlatformAdapter.filter_local_delivery_paths([str(fresh)])
assert out == [str(fresh.resolve())]

def test_docker_workspace_media_path_translates_to_host_workspace(self, tmp_path, monkeypatch):
hermes_home = tmp_path / "hermes"
workspace = hermes_home / "sandboxes" / "docker" / "default" / "workspace"
image = workspace / "foo.png"
image.parent.mkdir(parents=True)
image.write_bytes(b"fake image")
self._patch_roots(monkeypatch)
monkeypatch.setenv("TERMINAL_ENV", "docker")
monkeypatch.setenv("TERMINAL_CONTAINER_PERSISTENT", "1")
token = self._set_hermes_home(monkeypatch, hermes_home)

try:
assert BasePlatformAdapter.validate_media_delivery_path("/workspace/foo.png") == str(image.resolve())
finally:
reset_hermes_home_override(token)

def test_docker_root_media_path_is_not_translated_or_allowed(self, tmp_path, monkeypatch):
hermes_home = tmp_path / "hermes"
docker_home = hermes_home / "sandboxes" / "docker" / "default" / "home"
secret = docker_home / "foo.png"
secret.parent.mkdir(parents=True)
secret.write_bytes(b"do not send")
self._patch_roots(monkeypatch)
monkeypatch.setenv("TERMINAL_ENV", "docker")
monkeypatch.setenv("TERMINAL_CONTAINER_PERSISTENT", "1")
token = self._set_hermes_home(monkeypatch, hermes_home)

try:
assert BasePlatformAdapter.validate_media_delivery_path("/root/foo.png") is None
finally:
reset_hermes_home_override(token)

def test_docker_workspace_media_path_cannot_escape_workspace_via_dotdot(self, tmp_path, monkeypatch):
hermes_home = tmp_path / "hermes"
workspace = hermes_home / "sandboxes" / "docker" / "default" / "workspace"
docker_home = hermes_home / "sandboxes" / "docker" / "default" / "home"
secret = docker_home / "secret.png"
workspace.mkdir(parents=True)
secret.parent.mkdir(parents=True)
secret.write_bytes(b"secret")
self._patch_roots(monkeypatch)
monkeypatch.setenv("TERMINAL_ENV", "docker")
monkeypatch.setenv("TERMINAL_CONTAINER_PERSISTENT", "1")
token = self._set_hermes_home(monkeypatch, hermes_home)

try:
assert BasePlatformAdapter.validate_media_delivery_path("/workspace/../home/secret.png") is None
finally:
reset_hermes_home_override(token)

def test_docker_workspace_media_path_cannot_escape_workspace_via_symlink(self, tmp_path, monkeypatch):
hermes_home = tmp_path / "hermes"
workspace = hermes_home / "sandboxes" / "docker" / "default" / "workspace"
docker_home = hermes_home / "sandboxes" / "docker" / "default" / "home"
secret = docker_home / "secret.png"
workspace.mkdir(parents=True)
secret.parent.mkdir(parents=True)
secret.write_bytes(b"secret")
link = workspace / "escape"
try:
link.symlink_to(docker_home, target_is_directory=True)
except OSError:
pytest.skip("symlink creation is unavailable")
self._patch_roots(monkeypatch)
monkeypatch.setenv("TERMINAL_ENV", "docker")
monkeypatch.setenv("TERMINAL_CONTAINER_PERSISTENT", "1")
token = self._set_hermes_home(monkeypatch, hermes_home)

try:
assert BasePlatformAdapter.validate_media_delivery_path("/workspace/escape/secret.png") is None
finally:
reset_hermes_home_override(token)

def test_workspace_media_path_not_translated_for_non_docker_backend(self, tmp_path, monkeypatch):
hermes_home = tmp_path / "hermes"
workspace = hermes_home / "sandboxes" / "docker" / "default" / "workspace"
image = workspace / "foo.png"
image.parent.mkdir(parents=True)
image.write_bytes(b"fake image")
self._patch_roots(monkeypatch)
monkeypatch.setenv("TERMINAL_ENV", "local")
monkeypatch.setenv("TERMINAL_CONTAINER_PERSISTENT", "1")
token = self._set_hermes_home(monkeypatch, hermes_home)

try:
assert BasePlatformAdapter.validate_media_delivery_path("/workspace/foo.png") is None
finally:
reset_hermes_home_override(token)


# ---------------------------------------------------------------------------
# should_send_media_as_audio
Expand Down
37 changes: 37 additions & 0 deletions tests/plugins/image_gen/test_openai_provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -164,6 +164,43 @@ def test_b64_saves_to_cache(self, provider, tmp_path):
# gpt-image-2 rejects response_format — we must NOT send it.
assert "response_format" not in call_kwargs

def test_config_base_url_is_passed_to_openai_client(self, provider, tmp_path):
import yaml

(tmp_path / "config.yaml").write_text(
yaml.safe_dump({"image_gen": {"openai": {"base_url": "https://images.example.test/v1/"}}})
)

fake_client = MagicMock()
fake_client.images.generate.return_value = _fake_response(b64=_b64_png())
fake_openai = MagicMock()
fake_openai.OpenAI.return_value = fake_client

with patch.dict("sys.modules", {"openai": fake_openai}):
result = provider.generate("a cat")

assert result["success"] is True
assert fake_openai.OpenAI.call_args.kwargs["base_url"] == "https://images.example.test/v1"

def test_config_base_url_overrides_openai_base_url_env(self, provider, monkeypatch, tmp_path):
import yaml

monkeypatch.setenv("OPENAI_BASE_URL", "https://env.example.test/v1")
(tmp_path / "config.yaml").write_text(
yaml.safe_dump({"image_gen": {"openai": {"base_url": "https://config.example.test/v1/"}}})
)

fake_client = MagicMock()
fake_client.images.generate.return_value = _fake_response(b64=_b64_png())
fake_openai = MagicMock()
fake_openai.OpenAI.return_value = fake_client

with patch.dict("sys.modules", {"openai": fake_openai}):
result = provider.generate("a cat")

assert result["success"] is True
assert fake_openai.OpenAI.call_args.kwargs["base_url"] == "https://config.example.test/v1"

@pytest.mark.parametrize("tier,expected_quality", [
("gpt-image-2-low", "low"),
("gpt-image-2-medium", "medium"),
Expand Down