feat(runtime): add Podman as a first-class runtime backend

Selectable via STRIX_RUNTIME_BACKEND=podman, with auto-detection of Podman sockets across Linux (rootless/rootful) and macOS (applehv/libkrun podman machine), correct host-gateway hostname (host.containers.internal), and multi-layer socket fallthrough: STRIX_RUNTIME_SOCKET -> DOCKER_HOST -> per-backend auto-detection -> docker.from_env() default.

Adds backend-aware CLI/connection checks (check_docker_installed, check_docker_connection) and 28 unit tests covering backend registry, socket candidate generation, podman machine inspect JSON parsing, and socket fallthrough. Adds make test target wired into make check-all.

Closes #106
This commit is contained in:
B661LP 2026-09-09 13:55:38 +02:00
parent 52b1923347
commit fb5fcb17ad
13 changed files with 890 additions and 43 deletions

View file

@ -1,4 +1,4 @@
.PHONY: help install dev-install format lint type-check security check-all clean pre-commit setup-dev dev viewer wheel tui-build tui-test tui-lint
.PHONY: help install dev-install format lint type-check security test check-all clean pre-commit setup-dev dev viewer wheel tui-build tui-test tui-lint
TUI_BINARY := build/sidecar/strix-tui$(if $(filter Windows_NT,$(OS)),.exe)
@ -13,6 +13,7 @@ help:
@echo " lint - Lint code with ruff"
@echo " type-check - Run type checking with mypy and pyright"
@echo " security - Run security checks with bandit"
@echo " test - Run unit tests with pytest"
@echo " check-all - Run all code quality checks"
@echo ""
@echo "Development:"
@ -57,7 +58,12 @@ security:
uv run bandit -r strix/ -c pyproject.toml
@echo "✅ Security checks complete!"
check-all: format lint type-check security
test:
@echo "🧪 Running unit tests with pytest..."
uv run pytest tests/test_podman.py -v
@echo "✅ Tests complete!"
check-all: format lint type-check security test
@echo "✅ All code quality checks passed!"
pre-commit:

View file

@ -124,20 +124,57 @@ strix_runs/<run_name>/events.jsonl
When remote vars are set, Strix dual-writes telemetry to both local JSONL and the remote OTEL endpoint.
## Docker Configuration
## Container Runtime Configuration
<ParamField path="STRIX_IMAGE" default="ghcr.io/usestrix/strix-sandbox:1.3.0" type="string">
Docker image to use for the sandbox container.
Strix supports both **Docker** and **Podman** as sandbox runtime backends.
<ParamField path="STRIX_RUNTIME_BACKEND" default="docker" type="string">
Runtime backend for the sandbox environment (`docker` or `podman`).
</ParamField>
<ParamField path="STRIX_RUNTIME_SOCKET" type="string">
Direct socket path or URL for the container runtime (e.g. `unix:///run/user/1000/podman/podman.sock`). Takes precedence over `DOCKER_HOST` and auto-detection.
</ParamField>
<ParamField path="DOCKER_HOST" type="string">
Docker daemon socket path. Use for remote Docker hosts or custom configurations.
Container daemon socket path. Used as fallback when `STRIX_RUNTIME_SOCKET` is unset.
</ParamField>
<ParamField path="STRIX_RUNTIME_BACKEND" default="docker" type="string">
Runtime backend for the sandbox environment.
<ParamField path="STRIX_IMAGE" default="ghcr.io/usestrix/strix-sandbox:1.3.0" type="string">
Container image to use for the sandbox.
</ParamField>
### Using Podman
To use Podman instead of Docker:
```bash
export STRIX_RUNTIME_BACKEND=podman
strix --target https://example.com
```
Or point Strix at a specific Podman socket:
```bash
export STRIX_RUNTIME_SOCKET=unix:///run/user/1000/podman/podman.sock
strix --target https://example.com
```
#### How Socket Detection Works
When `STRIX_RUNTIME_BACKEND=podman` is set, Strix uses multi-layer socket fallthrough:
1. `STRIX_RUNTIME_SOCKET` (environment variable or config file)
2. `DOCKER_HOST` (environment variable)
3. **Automatic socket detection**:
- **Linux rootless**: `$XDG_RUNTIME_DIR/podman/podman.sock`, `/run/user/<uid>/podman/podman.sock`, and `/tmp/podman-run-<uid>/podman/podman.sock`
- **Linux rootful**: `/run/podman/podman.sock` and `/var/run/podman/podman.sock`
- **macOS**: Queries active Podman machines via `podman machine inspect` (applehv, libkrun, qemu) and common machine socket paths under `~/.local/share/containers/podman/machine/`
4. Falls back to standard Docker environment default (`docker.from_env()`) if no specific socket is reachable
#### Host Gateway & Networking
With Podman, Strix automatically uses `host.containers.internal` as the container-to-host gateway (mapping host-bound endpoints cleanly into Podman's built-in DNS), while retaining `host.docker.internal` compatibility.
## Sandbox Configuration
<ParamField path="STRIX_SANDBOX_EXECUTION_TIMEOUT" default="120" type="integer">

View file

@ -9,7 +9,7 @@ description: "Contribute to Strix development"
- Python 3.12+
- Latest Go 1.24.x patch (only for Bubble Tea TUI development and release artifacts)
- Docker (running)
- Docker or Podman (running)
- [uv](https://docs.astral.sh/uv/)
- Git

View file

@ -5,7 +5,7 @@ description: "Install Strix and run your first security scan"
## Prerequisites
- Docker (running)
- Docker or Podman (running)
- An LLM API key from any [supported provider](/llm-providers/overview) (OpenAI, Anthropic, Google, etc.)
## Installation

View file

@ -110,6 +110,7 @@ class RuntimeSettings(BaseSettings):
alias="STRIX_IMAGE",
)
backend: str = Field(default="docker", alias="STRIX_RUNTIME_BACKEND")
socket: str | None = Field(default=None, alias="STRIX_RUNTIME_SOCKET")
# Max screenshot/image tool outputs kept live per agent context (0 = none).
max_context_images: int = Field(default=3, ge=0, alias="STRIX_MAX_CONTEXT_IMAGES")

View file

@ -1,6 +1,7 @@
"""Startup environment validation and Docker image management."""
import logging
import os
import shutil
import sys
@ -164,15 +165,31 @@ def validate_environment() -> None:
def check_docker_installed() -> None:
if shutil.which("docker") is None:
logger.debug("Docker CLI not found in PATH")
backend = os.environ.get("STRIX_RUNTIME_BACKEND", "").strip()
if not backend:
try:
backend = getattr(load_settings().runtime, "backend", "docker")
except Exception:
backend = "docker"
backend = (backend or "docker").lower()
cli_name = "podman" if backend == "podman" else "docker"
display_name = "Podman" if backend == "podman" else "Docker"
if backend == "podman":
installed = shutil.which("podman") is not None or shutil.which("docker") is not None
else:
installed = shutil.which("docker") is not None
if not installed:
logger.debug("%s CLI not found in PATH", display_name)
console = Console()
error_text = Text()
error_text.append("DOCKER NOT INSTALLED", style="bold red")
error_text.append(f"{display_name.upper()} NOT INSTALLED", style="bold red")
error_text.append("\n\n", style="white")
error_text.append("The 'docker' CLI was not found in your PATH.\n", style="white")
error_text.append(f"The '{cli_name}' CLI was not found in your PATH.\n", style="white")
error_text.append(
"Please install Docker and ensure the 'docker' command is available.\n\n", style="white"
f"Please install {display_name} and ensure the '{cli_name}' command is available.\n\n",
style="white",
)
panel = Panel(
@ -183,24 +200,32 @@ def check_docker_installed() -> None:
padding=(1, 2),
)
console.print("\n", panel, "\n")
report_error("docker_not_installed")
report_error(f"{backend}_not_installed")
sys.exit(1)
logger.debug("Docker CLI present")
logger.debug("%s CLI present", display_name)
def pull_docker_image() -> None:
from docker.errors import DockerException
console = Console()
client = check_docker_connection()
backend = os.environ.get("STRIX_RUNTIME_BACKEND", "").strip()
if not backend:
try:
backend = getattr(load_settings().runtime, "backend", "docker")
except Exception:
backend = "docker"
backend = (backend or "docker").lower()
display_name = "Podman" if backend == "podman" else "Docker"
client = check_docker_connection(backend)
image = load_settings().runtime.image
if image_exists(client, image):
logger.debug("Docker image already present locally: %s", image)
logger.debug("%s image already present locally: %s", display_name, image)
return
logger.info("Pulling docker image: %s", image)
logger.info("Pulling %s image: %s", display_name.lower(), image)
console.print()
console.print(f"[dim]Pulling image[/] {image}")
console.print("[dim yellow]This only happens on first run and may take a few minutes...[/]")
@ -215,7 +240,7 @@ def pull_docker_image() -> None:
last_update = process_pull_line(line, layers_info, status, last_update)
except DockerException as e:
logger.debug("Failed to pull docker image %s", image, exc_info=True)
logger.debug("Failed to pull %s image %s", display_name.lower(), image, exc_info=True)
console.print()
error_text = Text()
error_text.append("FAILED TO PULL IMAGE", style="bold red")
@ -234,8 +259,8 @@ def pull_docker_image() -> None:
report_error("image_pull_failed", e)
sys.exit(1)
logger.info("Docker image %s ready", image)
logger.info("%s image %s ready", display_name, image)
success_text = Text()
success_text.append("Docker image ready", style="#22c55e")
success_text.append(f"{display_name} image ready", style="#22c55e")
console.print(success_text)
console.print()

View file

@ -31,6 +31,7 @@ from strix.interface.utils import (
stage_api_specs,
write_fetched_collection,
)
from strix.runtime.backends import get_host_gateway
from strix.telemetry import posthog, scarf
from strix.utils.api_spec import (
SpecParseError,
@ -130,7 +131,10 @@ def build_targets_info(args: argparse.Namespace) -> None:
args.targets_info = dedupe_local_targets(args.targets_info)
assign_workspace_subdirs(args.targets_info)
rewrite_localhost_targets(args.targets_info, HOST_GATEWAY_HOSTNAME)
rewrite_localhost_targets(
args.targets_info,
get_host_gateway(load_settings().runtime.backend),
)
def _resolve_api_spec(target: str, details: dict[str, Any]) -> None:

View file

@ -1597,23 +1597,44 @@ def clone_repository(repo_url: str, run_name: str, dest_name: str | None = None)
) from e
def check_docker_connection() -> Any:
import docker
from docker.errors import DockerException
def check_docker_connection(backend: str | None = None) -> Any:
import os
from strix.config import load_settings
from strix.runtime.backends import get_docker_client
if backend is None:
backend = os.environ.get("STRIX_RUNTIME_BACKEND", "").strip()
if not backend:
try:
backend = getattr(load_settings().runtime, "backend", "docker")
except Exception:
backend = "docker"
resolved_backend = (backend or "docker").lower()
display_name = "Podman" if resolved_backend == "podman" else "Docker"
try:
return docker.from_env()
except DockerException as exc:
report_error("docker_unavailable", exc)
client = get_docker_client(resolved_backend)
client.ping()
except Exception as exc:
report_error(f"{resolved_backend}_unavailable", exc)
console = Console()
error_text = Text()
error_text.append("DOCKER NOT AVAILABLE", style="bold red")
error_text.append(f"{display_name.upper()} NOT AVAILABLE", style="bold red")
error_text.append("\n\n", style="white")
error_text.append("Cannot connect to Docker daemon.\n", style="white")
error_text.append(
"Please ensure Docker Desktop is installed and running, and try running strix again.\n",
style="white",
)
error_text.append(f"Cannot connect to {display_name} daemon.\n", style="white")
if resolved_backend == "podman":
error_text.append(
"Please ensure Podman is running (e.g. 'podman machine start' or system service),\n"
"and try running strix again.\n",
style="white",
)
else:
error_text.append(
"Please ensure Docker Desktop is installed and running, "
"and try running strix again.\n",
style="white",
)
panel = Panel(
error_text,
@ -1623,7 +1644,9 @@ def check_docker_connection() -> Any:
padding=(1, 2),
)
console.print("\n", panel, "\n")
raise RuntimeError("Docker not available") from None
raise RuntimeError(f"{display_name} not available") from None
else:
return client
def image_exists(client: Any, image_name: str) -> bool:

View file

@ -1 +1,26 @@
"""Pluggable sandbox lifecycle on top of the Agents SDK."""
from strix.runtime.backends import (
backend_supports_bind_mounts,
get_backend,
get_docker_client,
get_host_gateway,
get_podman_socket_candidates,
parse_podman_machine_inspect,
register_backend,
resolve_runtime_socket,
supported_backends,
)
__all__ = [
"backend_supports_bind_mounts",
"get_backend",
"get_docker_client",
"get_host_gateway",
"get_podman_socket_candidates",
"parse_podman_machine_inspect",
"register_backend",
"resolve_runtime_socket",
"supported_backends",
]

View file

@ -2,8 +2,15 @@
from __future__ import annotations
import contextlib
import json
import logging
import os
import shutil
import subprocess # nosec B404
import sys
from collections.abc import Awaitable, Callable
from pathlib import Path
from typing import TYPE_CHECKING, Any
@ -17,6 +24,308 @@ logger = logging.getLogger(__name__)
SandboxBackend = Callable[..., Awaitable[tuple[Any, Any]]]
def get_host_gateway(backend: str | None = None) -> str:
"""Return the container-to-host gateway hostname for ``backend``.
For docker, returns ``"host.docker.internal"``.
For podman, returns ``"host.containers.internal"`` so container-to-host
networking works out of the box with Podman's built-in DNS.
"""
if backend is None:
try:
from strix.config import load_settings
backend = load_settings().runtime.backend
except Exception: # noqa: BLE001
backend = "docker"
if (backend or "").lower() == "podman":
return "host.containers.internal"
return "host.docker.internal"
def _extract_machine_socket(m: dict[str, Any]) -> str | None:
"""Extract socket path from a machine inspect dictionary."""
conn_info: object = m.get("ConnectionInfo")
if isinstance(conn_info, dict):
podman_sock: object = conn_info.get("PodmanSocket") # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]
if isinstance(podman_sock, dict):
p: object = podman_sock.get("Path") # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]
if isinstance(p, str) and p.strip():
return p.strip()
elif isinstance(podman_sock, str) and podman_sock.strip():
return podman_sock.strip()
elif isinstance(conn_info, str) and conn_info.strip():
return conn_info.strip()
direct_sock: object = m.get("PodmanSocket")
if isinstance(direct_sock, dict):
direct_path: object = direct_sock.get("Path") # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]
if isinstance(direct_path, str) and direct_path.strip():
return direct_path.strip()
elif isinstance(direct_sock, str) and direct_sock.strip():
return direct_sock.strip()
return None
def _normalize_inspect_payload(
output: str | bytes | list[Any] | dict[str, Any],
) -> list[dict[str, Any]]:
"""Convert raw input into a list of machine inspection dictionaries."""
if isinstance(output, bytes | bytearray):
output = output.decode("utf-8", errors="replace")
data: object
if isinstance(output, str):
text = output.strip()
if not text:
return []
try:
data = json.loads(text)
except (json.JSONDecodeError, ValueError):
return []
else:
data = output
if isinstance(data, dict):
return [data] # pyright: ignore[reportUnknownVariableType]
if isinstance(data, list):
return [m for m in data if isinstance(m, dict)] # pyright: ignore[reportUnknownVariableType]
return []
def parse_podman_machine_inspect(output: str | bytes | list[Any] | dict[str, Any]) -> list[str]:
"""Parse JSON output from ``podman machine inspect``.
Returns a list of discovered socket paths. Handles errors, missing keys,
single-machine dicts, and multi-machine arrays (prioritizing running machines).
"""
machines = _normalize_inspect_payload(output)
if not machines:
return []
running_sockets: list[str] = []
other_sockets: list[str] = []
for m in machines:
is_running = bool(m.get("Running")) or str(m.get("State") or "").lower() == "running"
socket_path = _extract_machine_socket(m)
if socket_path:
if is_running:
running_sockets.append(socket_path)
else:
other_sockets.append(socket_path)
seen: set[str] = set()
result: list[str] = []
for s in running_sockets + other_sockets:
if s not in seen:
seen.add(s)
result.append(s)
return result
def _run_podman_machine_inspect() -> list[str]:
"""Execute ``podman machine inspect`` and return discovered socket paths."""
podman_bin = shutil.which("podman")
if not podman_bin:
return []
try:
proc = subprocess.run( # nosec B603 # noqa: S603
[podman_bin, "machine", "inspect"],
capture_output=True,
text=True,
timeout=3,
check=False,
)
if proc.returncode != 0:
return []
return parse_podman_machine_inspect(proc.stdout)
except Exception: # noqa: BLE001
logger.debug("Failed to run podman machine inspect", exc_info=True)
return []
def get_podman_socket_candidates(
platform: str | None = None,
*,
uid: int | None = None,
xdg_runtime_dir: str | None = None,
home: Path | str | None = None,
) -> list[Path]:
"""Generate candidate filesystem paths for the Podman socket.
Covers Linux (rootless and rootful) and macOS (applehv, libkrun, podman machine).
"""
plat = (platform or sys.platform).lower()
candidates: list[Path] = []
if plat == "linux":
# 1. Linux rootless via XDG_RUNTIME_DIR
xdg_dir = (
xdg_runtime_dir if xdg_runtime_dir is not None else os.environ.get("XDG_RUNTIME_DIR")
)
if xdg_dir:
candidates.append(Path(xdg_dir) / "podman" / "podman.sock")
# 2. Linux rootless via UID
effective_uid = uid
if effective_uid is None and hasattr(os, "getuid"):
try:
effective_uid = os.getuid()
except (AttributeError, OSError):
effective_uid = None
if effective_uid is not None:
candidates.append(Path(f"/run/user/{effective_uid}/podman/podman.sock"))
candidates.append(
Path(f"/tmp/podman-run-{effective_uid}/podman/podman.sock") # nosec B108 # noqa: S108
)
# 3. Linux rootful
candidates.append(Path("/run/podman/podman.sock"))
candidates.append(Path("/var/run/podman/podman.sock"))
elif plat == "darwin":
user_home = Path(home) if home is not None else Path.home()
# Dynamic machine inspect discovery (applehv / libkrun / qemu VMs)
candidates.extend(Path(sock) for sock in _run_podman_machine_inspect())
# Standard macOS VM socket locations (applehv, libkrun, qemu, default)
machine_dir = user_home / ".local" / "share" / "containers" / "podman" / "machine"
candidates.append(machine_dir / "applehv" / "podman.sock")
candidates.append(machine_dir / "libkrun" / "podman.sock")
candidates.append(machine_dir / "qemu" / "podman.sock")
candidates.append(machine_dir / "podman-machine-default" / "podman.sock")
candidates.append(machine_dir / "podman.sock")
else:
user_home = Path(home) if home is not None else Path.home()
machine_dir = user_home / ".local" / "share" / "containers" / "podman" / "machine"
candidates.append(machine_dir / "podman-machine-default" / "podman.sock")
candidates.append(machine_dir / "podman.sock")
seen: set[Path] = set()
result: list[Path] = []
for c in candidates:
if c not in seen:
seen.add(c)
result.append(c)
return result
def auto_detect_podman_socket() -> str | None:
"""Look for an existing Podman socket on the host."""
try:
candidates = get_podman_socket_candidates()
for candidate in candidates:
try:
if candidate.exists() or candidate.is_socket():
return f"unix://{candidate.resolve()}"
except OSError:
continue
except Exception: # noqa: BLE001
logger.debug("Podman socket auto-detection failed", exc_info=True)
return None
def auto_detect_docker_socket() -> str | None:
"""Look for an existing Docker daemon socket on the host."""
candidates = [
Path("/var/run/docker.sock"),
Path("/run/docker.sock"),
Path.home() / ".docker" / "run" / "docker.sock",
Path.home() / ".docker" / "desktop" / "docker.sock",
]
for c in candidates:
try:
if c.exists() or c.is_socket():
return f"unix://{c.resolve()}"
except OSError:
continue
return None
def normalize_socket_url(socket_path_or_url: str) -> str:
"""Normalize a socket path or URL into a docker-compatible base_url."""
s = socket_path_or_url.strip()
if not s:
return ""
if "://" in s:
return s
if s.startswith((r"\\.\pipe", "//./pipe")):
return f"npipe://{s}"
return f"unix://{s}"
def resolve_runtime_socket(backend: str = "docker") -> str | None:
"""Resolve the runtime socket URL using multi-layer fallthrough:
1. STRIX_RUNTIME_SOCKET (env var or settings)
2. DOCKER_HOST (env var)
3. Per-backend auto-detection (for podman or docker)
4. None (falls back to docker.from_env() default)
"""
# Layer 1: STRIX_RUNTIME_SOCKET
runtime_socket = os.environ.get("STRIX_RUNTIME_SOCKET", "").strip()
if not runtime_socket:
with contextlib.suppress(Exception):
from strix.config import load_settings
cfg_socket = getattr(load_settings().runtime, "socket", None)
if cfg_socket:
runtime_socket = str(cfg_socket).strip()
if runtime_socket:
return normalize_socket_url(runtime_socket)
# Layer 2: DOCKER_HOST
docker_host = os.environ.get("DOCKER_HOST", "").strip()
if docker_host:
return normalize_socket_url(docker_host)
# Layer 3: Per-backend auto-detection
b = backend.lower()
if b == "podman":
detected = auto_detect_podman_socket()
if detected:
return detected
elif b == "docker":
detected = auto_detect_docker_socket()
if detected:
return detected
# Layer 4: Fall back to default
return None
def get_docker_client(backend: str = "docker") -> Any:
"""Create a Docker client for ``backend`` using multi-layer socket fallthrough:
STRIX_RUNTIME_SOCKET → DOCKER_HOST → per-backend auto-detection → docker.from_env() default.
Gracefully falls through to docker.from_env() on connection/ping failure.
"""
import docker
socket_url = resolve_runtime_socket(backend)
if socket_url:
try:
client: Any = docker.DockerClient(base_url=socket_url)
client.ping()
except Exception: # noqa: BLE001
logger.warning(
"Failed to connect to %s via socket %s; falling through to default",
backend,
socket_url,
exc_info=True,
)
else:
logger.info("Connected to %s runtime via socket: %s", backend, socket_url)
return client
logger.debug("Using docker.from_env() default for backend %s", backend)
return docker.from_env()
async def _docker_backend(
*,
image: str,
@ -37,12 +346,39 @@ async def _docker_backend(
Strix manages session lifetime explicitly via ``client.delete()`` so we
trigger ``start()`` ourselves.
"""
import docker
from agents.sandbox.sandboxes.docker import DockerSandboxClientOptions
from strix.runtime.docker_client import StrixDockerSandboxClient
client = StrixDockerSandboxClient(docker.from_env())
raw_client = get_docker_client("docker")
client = StrixDockerSandboxClient(raw_client)
client.host_gateway = get_host_gateway("docker")
client.strix_bind_mounts = bind_mounts or []
options = DockerSandboxClientOptions(image=image, exposed_ports=exposed_ports)
session = await client.create(options=options, manifest=manifest)
await session.start()
return client, session
async def _podman_backend(
*,
image: str,
manifest: Manifest,
exposed_ports: tuple[int, ...],
bind_mounts: list[dict[str, Any]] | None = None,
) -> tuple[Any, Any]:
"""Bring up a session backed by the local Podman engine.
Uses :class:`StrixDockerSandboxClient` with Podman socket detection
and host-gateway hostname (``host.containers.internal``).
"""
from agents.sandbox.sandboxes.docker import DockerSandboxClientOptions
from strix.runtime.docker_client import StrixDockerSandboxClient
raw_client = get_docker_client("podman")
client = StrixDockerSandboxClient(raw_client)
client.host_gateway = get_host_gateway("podman")
client.strix_bind_mounts = bind_mounts or []
options = DockerSandboxClientOptions(image=image, exposed_ports=exposed_ports)
session = await client.create(options=options, manifest=manifest)
@ -52,9 +388,10 @@ async def _docker_backend(
_BACKENDS: dict[str, SandboxBackend] = {
"docker": _docker_backend,
"podman": _podman_backend,
}
_BIND_MOUNT_BACKENDS: set[str] = {"docker"}
_BIND_MOUNT_BACKENDS: set[str] = {"docker", "podman"}
def get_backend(name: str) -> SandboxBackend:

View file

@ -165,6 +165,7 @@ class StrixDockerSandboxClient(DockerSandboxClient):
# Host directories to bind-mount into the container, set by the docker
# backend before ``create()``. Each item is ``{source, target, read_only}``.
strix_bind_mounts: list[dict[str, Any]] | None = None
host_gateway: str = "host.docker.internal"
async def _create_container(
self,
@ -230,6 +231,8 @@ class StrixDockerSandboxClient(DockerSandboxClient):
cap_add.append(cap)
extra_hosts = create_kwargs.setdefault("extra_hosts", {})
host_gw = getattr(self, "host_gateway", "host.docker.internal")
extra_hosts[host_gw] = "host-gateway"
extra_hosts["host.docker.internal"] = "host-gateway"
_apply_sandbox_network(create_kwargs)

View file

@ -15,7 +15,11 @@ from agents.sandbox.entries import BaseEntry, File, LocalDir
from agents.sandbox.manifest import Environment, Manifest
from strix.config import load_settings
from strix.runtime.backends import backend_supports_bind_mounts, get_backend
from strix.runtime.backends import (
backend_supports_bind_mounts,
get_backend,
get_host_gateway,
)
from strix.runtime.caido_bootstrap import bootstrap_caido
from strix.runtime.caido_handle import CaidoBootstrapHandle
@ -48,7 +52,10 @@ def _host_identity_env() -> dict[str, str]:
return {}
# Bind-mount ownership only needs mapping on Linux, where the container uid
# must match the host's.
return {"STRIX_HOST_UID": str(os.getuid()), "STRIX_HOST_GID": str(os.getgid())}
return {
"STRIX_HOST_UID": str(os.getuid()), # type: ignore[attr-defined]
"STRIX_HOST_GID": str(os.getgid()), # type: ignore[attr-defined]
}
def build_bind_mounts(local_sources: list[dict[str, Any]]) -> list[dict[str, Any]]:
@ -318,7 +325,7 @@ async def create_or_reuse(
environment=Environment(
value={
"PYTHONUNBUFFERED": "1",
"HOST_GATEWAY": "host.docker.internal",
"HOST_GATEWAY": get_host_gateway(backend_name),
**_host_identity_env(),
"http_proxy": container_caido_url,
"https_proxy": container_caido_url,

379
tests/test_podman.py Normal file
View file

@ -0,0 +1,379 @@
"""Unit tests for Podman runtime backend, socket candidate generation,
host-gateway resolution, and multi-layer socket fallthrough.
"""
from __future__ import annotations
import json
from pathlib import Path
from types import SimpleNamespace
from typing import Any
from unittest.mock import MagicMock, patch
import pytest
from strix.interface.environment import check_docker_installed
from strix.interface.utils import check_docker_connection
from strix.runtime.backends import (
_BACKENDS,
_BIND_MOUNT_BACKENDS,
auto_detect_podman_socket,
backend_supports_bind_mounts,
get_backend,
get_docker_client,
get_host_gateway,
get_podman_socket_candidates,
normalize_socket_url,
parse_podman_machine_inspect,
register_backend,
resolve_runtime_socket,
supported_backends,
)
# ============================================================================
# 1. Host Gateway Resolution (3 tests)
# ============================================================================
def test_get_host_gateway_docker() -> None:
"""Docker backend maps to host.docker.internal."""
assert get_host_gateway("docker") == "host.docker.internal"
def test_get_host_gateway_podman() -> None:
"""Podman backend maps to host.containers.internal (case-insensitive)."""
assert get_host_gateway("podman") == "host.containers.internal"
assert get_host_gateway("Podman") == "host.containers.internal"
assert get_host_gateway("PODMAN") == "host.containers.internal"
def test_get_host_gateway_default_and_fallback(monkeypatch: pytest.MonkeyPatch) -> None:
"""Default or unknown backend falls back to host.docker.internal."""
monkeypatch.delenv("STRIX_RUNTIME_BACKEND", raising=False)
monkeypatch.setattr(
"strix.config.load_settings",
lambda: SimpleNamespace(runtime=SimpleNamespace(backend="docker")),
)
assert get_host_gateway(None) == "host.docker.internal"
assert get_host_gateway("unknown_backend") == "host.docker.internal"
# ============================================================================
# 2. Backend Registry (4 tests)
# ============================================================================
def test_backend_registry_get_docker() -> None:
"""Docker backend is registered and callable."""
backend = get_backend("docker")
assert callable(backend)
assert "docker" in supported_backends()
assert backend_supports_bind_mounts("docker") is True
def test_backend_registry_get_podman() -> None:
"""Podman backend is registered, callable, and supports bind mounts."""
backend = get_backend("podman")
assert callable(backend)
assert "podman" in supported_backends()
assert backend_supports_bind_mounts("podman") is True
def test_backend_registry_unknown_raises() -> None:
"""Querying an unknown backend raises ValueError listing supported options."""
with pytest.raises(ValueError, match="Unknown STRIX_RUNTIME_BACKEND: 'unknown_rt'"):
get_backend("unknown_rt")
def test_backend_registry_custom_registration() -> None:
"""Custom backend can be registered and looked up with bind-mount support."""
async def _dummy_backend(**_kwargs: Any) -> tuple[Any, Any]:
return MagicMock(), MagicMock()
try:
register_backend("test_custom", _dummy_backend, supports_bind_mounts=True)
assert get_backend("test_custom") is _dummy_backend
assert backend_supports_bind_mounts("test_custom") is True
assert "test_custom" in supported_backends()
finally:
_BACKENDS.pop("test_custom", None)
_BIND_MOUNT_BACKENDS.discard("test_custom")
# ============================================================================
# 3. podman machine inspect JSON Parsing (6 tests)
# ============================================================================
def test_parse_machine_inspect_single_machine() -> None:
"""Extract socket path from a single machine inspect dictionary or list."""
sock_path = "/Users/user/.local/share/containers/podman/machine/applehv/podman.sock"
payload = json.dumps(
[
{
"Name": "podman-machine-default",
"Running": True,
"ConnectionInfo": {
"PodmanSocket": {
"Path": sock_path,
}
},
}
]
)
result = parse_podman_machine_inspect(payload)
assert result == [sock_path]
def test_parse_machine_inspect_multi_machine_running_priority() -> None:
"""When multiple machines exist, running machines take priority."""
payload = json.dumps(
[
{
"Name": "machine-stopped",
"Running": False,
"State": "stopped",
"ConnectionInfo": {"PodmanSocket": {"Path": "/path/to/stopped.sock"}},
},
{
"Name": "machine-active",
"Running": True,
"State": "running",
"ConnectionInfo": {"PodmanSocket": {"Path": "/path/to/active.sock"}},
},
]
)
result = parse_podman_machine_inspect(payload)
assert result == ["/path/to/active.sock", "/path/to/stopped.sock"]
def test_parse_machine_inspect_string_socket_path() -> None:
"""Handle ConnectionInfo where PodmanSocket is directly a string path."""
payload = json.dumps(
[
{
"Name": "default",
"Running": True,
"ConnectionInfo": {"PodmanSocket": "/var/run/podman-direct.sock"},
}
]
)
result = parse_podman_machine_inspect(payload)
assert result == ["/var/run/podman-direct.sock"]
def test_parse_machine_inspect_invalid_json() -> None:
"""Invalid JSON returns empty list without raising."""
assert parse_podman_machine_inspect("Error: machine not found") == []
assert parse_podman_machine_inspect("{not-valid-json") == []
def test_parse_machine_inspect_empty_or_missing_keys() -> None:
"""Empty string, empty array, or machines without socket path return empty list."""
assert parse_podman_machine_inspect("") == []
assert parse_podman_machine_inspect(" ") == []
assert parse_podman_machine_inspect("[]") == []
assert parse_podman_machine_inspect(json.dumps([{"Name": "empty"}])) == []
assert (
parse_podman_machine_inspect(
json.dumps([{"Name": "null_path", "ConnectionInfo": {"PodmanSocket": None}}])
)
== []
)
def test_parse_machine_inspect_non_dict_elements() -> None:
"""Lists with non-dict elements are handled gracefully."""
payload = json.dumps([None, 123, "random-string", False])
assert parse_podman_machine_inspect(payload) == []
# ============================================================================
# 4. Podman Socket Candidates Across Platform Variants (6 tests)
# ============================================================================
def test_socket_candidates_linux_rootless_xdg() -> None:
"""Linux rootless candidate incorporates XDG_RUNTIME_DIR."""
candidates = get_podman_socket_candidates(
platform="linux",
xdg_runtime_dir="/run/user/1000",
)
expected = Path("/run/user/1000/podman/podman.sock")
assert expected in candidates
def test_socket_candidates_linux_rootless_uid() -> None:
"""Linux rootless candidate incorporates UID and /tmp fallback."""
candidates = get_podman_socket_candidates(platform="linux", uid=1001)
assert Path("/run/user/1001/podman/podman.sock") in candidates
assert Path("/tmp/podman-run-1001/podman/podman.sock") in candidates # noqa: S108
def test_socket_candidates_linux_rootful() -> None:
"""Linux rootful candidates include system socket paths."""
candidates = get_podman_socket_candidates(platform="linux")
assert Path("/run/podman/podman.sock") in candidates
assert Path("/var/run/podman/podman.sock") in candidates
def test_socket_candidates_darwin_applehv() -> None:
"""macOS candidate includes applehv VM socket path."""
home = Path("/Users/developer")
candidates = get_podman_socket_candidates(platform="darwin", home=home)
expected = home / ".local/share/containers/podman/machine/applehv/podman.sock"
assert expected in candidates
def test_socket_candidates_darwin_libkrun() -> None:
"""macOS candidate includes libkrun VM socket path."""
home = Path("/Users/developer")
candidates = get_podman_socket_candidates(platform="darwin", home=home)
expected = home / ".local/share/containers/podman/machine/libkrun/podman.sock"
assert expected in candidates
def test_socket_candidates_darwin_machine_inspect_integration() -> None:
"""macOS candidates prioritize sockets discovered via machine inspect."""
home = Path("/Users/developer")
with patch(
"strix.runtime.backends._run_podman_machine_inspect",
return_value=["/custom/machine/inspect.sock"],
):
candidates = get_podman_socket_candidates(platform="darwin", home=home)
assert candidates[0] == Path("/custom/machine/inspect.sock")
# ============================================================================
# 5. Socket Detection and Multi-Layer Fallthrough (5 tests)
# ============================================================================
def test_socket_fallthrough_strix_runtime_socket(monkeypatch: pytest.MonkeyPatch) -> None:
"""STRIX_RUNTIME_SOCKET takes highest precedence in socket resolution."""
monkeypatch.setenv("STRIX_RUNTIME_SOCKET", "unix:///custom/strix.sock")
monkeypatch.setenv("DOCKER_HOST", "unix:///custom/docker_host.sock")
resolved = resolve_runtime_socket("podman")
assert resolved == "unix:///custom/strix.sock"
def test_socket_fallthrough_docker_host(monkeypatch: pytest.MonkeyPatch) -> None:
"""DOCKER_HOST is used when STRIX_RUNTIME_SOCKET is unset."""
monkeypatch.delenv("STRIX_RUNTIME_SOCKET", raising=False)
monkeypatch.setenv("DOCKER_HOST", "unix:///custom/docker_host.sock")
resolved = resolve_runtime_socket("podman")
assert resolved == "unix:///custom/docker_host.sock"
def test_socket_fallthrough_autodetect_podman(monkeypatch: pytest.MonkeyPatch) -> None:
"""Auto-detects first existing socket candidate when env vars are unset."""
monkeypatch.delenv("STRIX_RUNTIME_SOCKET", raising=False)
monkeypatch.delenv("DOCKER_HOST", raising=False)
fake_sock = Path("/run/user/1000/podman/podman.sock")
with (
patch(
"strix.runtime.backends.get_podman_socket_candidates",
return_value=[fake_sock],
),
patch.object(Path, "exists", return_value=True),
patch.object(Path, "resolve", return_value=fake_sock),
):
detected = auto_detect_podman_socket()
assert detected == f"unix://{fake_sock}"
def test_socket_fallthrough_graceful_on_missing_or_failed_socket(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Connection failure on detected socket gracefully falls through to docker.from_env()."""
monkeypatch.delenv("STRIX_RUNTIME_SOCKET", raising=False)
monkeypatch.delenv("DOCKER_HOST", raising=False)
mock_docker = MagicMock()
mock_bad_client = MagicMock()
mock_bad_client.ping.side_effect = ConnectionRefusedError("Daemon unreachable")
mock_docker.DockerClient.return_value = mock_bad_client
mock_default_client = MagicMock()
mock_docker.from_env.return_value = mock_default_client
with (
patch.dict("sys.modules", {"docker": mock_docker}),
patch(
"strix.runtime.backends.resolve_runtime_socket",
return_value="unix:///unreachable.sock",
),
):
client = get_docker_client("podman")
assert client is mock_default_client
mock_docker.from_env.assert_called_once()
def test_socket_fallthrough_strix_runtime_socket_raw_path_normalization() -> None:
"""Raw socket path is normalized to unix:// URI scheme."""
assert normalize_socket_url("/run/podman/podman.sock") == "unix:///run/podman/podman.sock"
assert normalize_socket_url("unix:///var/run/podman.sock") == "unix:///var/run/podman.sock"
assert normalize_socket_url("tcp://127.0.0.1:2375") == "tcp://127.0.0.1:2375"
# ============================================================================
# 7. CLI Check and Connection for Podman Runtime (4 tests)
# ============================================================================
def test_check_docker_installed_podman_success(monkeypatch: pytest.MonkeyPatch) -> None:
"""When STRIX_RUNTIME_BACKEND=podman, succeeds if podman is in PATH even if docker is not."""
monkeypatch.setenv("STRIX_RUNTIME_BACKEND", "podman")
def fake_which(cmd: str) -> str | None:
if cmd == "podman":
return "/usr/bin/podman"
return None
monkeypatch.setattr("shutil.which", fake_which)
# Should not raise or sys.exit
check_docker_installed()
def test_check_docker_installed_podman_missing(monkeypatch: pytest.MonkeyPatch) -> None:
"""When STRIX_RUNTIME_BACKEND=podman and neither podman nor docker in PATH, exits with error."""
monkeypatch.setenv("STRIX_RUNTIME_BACKEND", "podman")
monkeypatch.setattr("shutil.which", lambda _cmd: None)
with pytest.raises(SystemExit) as exc_info:
check_docker_installed()
assert exc_info.value.code == 1
def test_check_docker_installed_docker_missing(monkeypatch: pytest.MonkeyPatch) -> None:
"""When backend is docker and docker is missing from PATH, exits with error."""
monkeypatch.setenv("STRIX_RUNTIME_BACKEND", "docker")
def fake_which(cmd: str) -> str | None:
if cmd == "podman":
return "/usr/bin/podman"
return None
monkeypatch.setattr("shutil.which", fake_which)
with pytest.raises(SystemExit) as exc_info:
check_docker_installed()
assert exc_info.value.code == 1
def test_check_docker_connection_podman_uses_podman_backend(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""check_docker_connection connects and pings backend client."""
monkeypatch.setenv("STRIX_RUNTIME_BACKEND", "podman")
mock_client = MagicMock()
with patch("strix.runtime.backends.get_docker_client", return_value=mock_client) as mock_get:
client = check_docker_connection()
mock_get.assert_called_once_with("podman")
mock_client.ping.assert_called_once()
assert client is mock_client