This commit is contained in:
sam kerr 2026-09-25 14:00:40 +00:00 • committed by GitHub
commit 62f0520024
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 1052 additions and 57 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 test-podman 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 the unit test suite with pytest"
@echo " check-all - Run all code quality checks"
@echo ""
@echo "Development:"
@ -57,7 +58,17 @@ security:
uv run bandit -r strix/ -c pyproject.toml
@echo "✅ Security checks complete!"
check-all: format lint type-check security
test-podman:
@echo "🧪 Running Podman backend unit tests with pytest..."
uv run pytest tests/test_podman.py -v
@echo "✅ Tests complete!"
test:
@echo "🧪 Running unit tests with pytest..."
uv run pytest tests/ -v
@echo "✅ Tests complete!"
check-all: format lint type-check security test
@echo "✅ All code quality checks passed!"
pre-commit:

View file

@ -130,20 +130,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

@ -4,7 +4,7 @@ from __future__ import annotations
from typing import Literal
from pydantic import AliasChoices, Field
from pydantic import AliasChoices, Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
@ -116,9 +116,17 @@ 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")
@field_validator("backend", mode="after")
@classmethod
def _normalize_backend(cls, value: str) -> str:
# Normalize once here so every consumer (registry lookup, install
# checks, socket detection) agrees on the same casing.
return value.strip().lower() or "docker"
class TelemetrySettings(BaseSettings):
model_config = _BASE_CONFIG

View file

@ -1,6 +1,7 @@
"""Startup environment validation and Docker image management."""
"""Startup environment validation and sandbox image management."""
import logging
import os
import shutil
import sys
@ -10,7 +11,7 @@ from rich.text import Text
from strix.config import IntegrationSettings, codex, load_settings
from strix.interface.utils import (
check_docker_connection,
check_runtime_connection,
image_exists,
process_pull_line,
)
@ -163,16 +164,32 @@ def validate_environment() -> None:
)
def check_docker_installed() -> None:
if shutil.which("docker") is None:
logger.debug("Docker CLI not found in PATH")
def check_runtime_installed() -> None:
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:
def pull_runtime_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_runtime_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

@ -17,8 +17,8 @@ from strix.config import codex, load_settings, persist_current
from strix.core.paths import run_dir_for
from strix.interface.cli_args import parse_arguments
from strix.interface.environment import (
check_docker_installed,
pull_docker_image,
check_runtime_installed,
pull_runtime_image,
validate_environment,
)
from strix.interface.interactive import (
@ -469,8 +469,8 @@ def main() -> None:
restart_after_update()
sys.exit(0)
check_docker_installed()
pull_docker_image()
check_runtime_installed()
pull_runtime_image()
validate_environment()
# Everything below imports the scan engine; do not race the warm-up thread.

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,
@ -47,8 +48,6 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
HOST_GATEWAY_HOSTNAME = "host.docker.internal"
class ModelConnectionError(RuntimeError):
"""An ordinary model preflight failure, annotated with its model route."""
@ -130,7 +129,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

@ -1598,23 +1598,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_runtime_connection(backend: str | None = None) -> Any:
import os
from strix.config import load_settings
from strix.runtime.backends import get_runtime_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_runtime_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,
@ -1624,7 +1645,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_host_gateway,
get_podman_socket_candidates,
get_runtime_client,
parse_podman_machine_inspect,
register_backend,
resolve_runtime_socket,
supported_backends,
)
__all__ = [
"backend_supports_bind_mounts",
"get_backend",
"get_host_gateway",
"get_podman_socket_candidates",
"get_runtime_client",
"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,349 @@ 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 _socket_is_live(socket_url: str) -> bool:
"""Return True if a docker-compatible client can connect to and ping ``socket_url``."""
import docker
try:
client: Any = docker.DockerClient(base_url=socket_url)
try:
client.ping()
finally:
with contextlib.suppress(Exception):
client.close()
except Exception: # noqa: BLE001
return False
return True
def auto_detect_podman_socket() -> str | None:
"""Look for a live Podman socket on the host.
Candidates are tried in order; a candidate that exists but does not
respond to a ping (stale socket file, wrong machine, etc.) is skipped
in favor of the next one instead of being returned as-is.
"""
try:
candidates = get_podman_socket_candidates()
for candidate in candidates:
try:
if not (candidate.exists() or candidate.is_socket()):
continue
except OSError:
continue
url = f"unix://{candidate.resolve()}"
if _socket_is_live(url):
return url
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_runtime_client(backend: str = "docker") -> Any:
"""Create a container runtime client for ``backend`` using multi-layer socket fallthrough:
STRIX_RUNTIME_SOCKET → DOCKER_HOST → per-backend auto-detection → docker.from_env() default.
The ``docker.from_env()`` default is only used for the ``docker`` backend itself: falling
back to it for a non-docker backend (e.g. ``podman``) would silently run the sandbox on the
wrong runtime, so that case raises instead.
"""
import docker
normalized_backend = (backend or "docker").strip().lower()
socket_url = resolve_runtime_socket(normalized_backend)
if socket_url:
try:
client: Any = docker.DockerClient(base_url=socket_url)
client.ping()
except Exception as exc:
if normalized_backend != "docker":
raise RuntimeError(
f"Could not connect to the {normalized_backend} runtime via socket "
f"{socket_url}. Set STRIX_RUNTIME_SOCKET to a reachable {normalized_backend} "
"socket, or check that the daemon is running."
) from exc
logger.warning(
"Failed to connect to %s via socket %s; falling through to default",
normalized_backend,
socket_url,
exc_info=True,
)
else:
logger.info("Connected to %s runtime via socket: %s", normalized_backend, socket_url)
return client
if normalized_backend != "docker":
raise RuntimeError(
f"No reachable {normalized_backend} socket found (checked STRIX_RUNTIME_SOCKET, "
"DOCKER_HOST, and auto-detection). Set STRIX_RUNTIME_SOCKET to the "
f"{normalized_backend} socket path, or set STRIX_RUNTIME_BACKEND=docker to use Docker."
)
logger.debug("Using docker.from_env() default for backend %s", normalized_backend)
return docker.from_env()
async def _docker_backend(
*,
image: str,
@ -37,12 +387,41 @@ 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_runtime_client("docker")
client = StrixDockerSandboxClient(raw_client)
client.host_gateway = get_host_gateway("docker")
client.backend_name = "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_runtime_client("podman")
client = StrixDockerSandboxClient(raw_client)
client.host_gateway = get_host_gateway("podman")
client.backend_name = "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 +431,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

@ -125,6 +125,7 @@ def _apply_run_labels(create_kwargs: dict[str, Any]) -> None:
class StrixDockerSandboxSession(DockerSandboxSession):
sandbox_network: str = ""
backend_name: str = "docker"
async def _resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
try:
@ -135,7 +136,7 @@ class StrixDockerSandboxSession(DockerSandboxSession):
exposed_ports=self.state.exposed_ports,
reason="backend_unavailable",
context={
"backend": "docker",
"backend": self.backend_name,
"detail": "container_reload_failed",
"network": self.sandbox_network,
},
@ -152,7 +153,7 @@ class StrixDockerSandboxSession(DockerSandboxSession):
exposed_ports=self.state.exposed_ports,
reason="backend_unavailable",
context={
"backend": "docker",
"backend": self.backend_name,
"detail": "container_not_on_network",
"network": self.sandbox_network,
},
@ -165,6 +166,8 @@ 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"
backend_name: str = "docker"
async def _create_container(
self,
@ -230,6 +233,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)
@ -272,7 +277,9 @@ class StrixDockerSandboxClient(DockerSandboxClient):
inner = session._inner
if network and isinstance(inner, DockerSandboxSession):
inner.__class__ = StrixDockerSandboxSession
cast("StrixDockerSandboxSession", inner).sandbox_network = network
strix_inner = cast("StrixDockerSandboxSession", inner)
strix_inner.sandbox_network = network
strix_inner.backend_name = self.backend_name
return session
async def delete(self, session: SandboxSession) -> SandboxSession:

View file

@ -16,7 +16,11 @@ from agents.sandbox.entries import BaseEntry, 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
@ -58,7 +62,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]]:
@ -307,7 +314,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,

470
tests/test_podman.py Normal file
View file

@ -0,0 +1,470 @@
"""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.config.settings import RuntimeSettings
from strix.interface.environment import check_runtime_installed
from strix.interface.utils import check_runtime_connection
from strix.runtime.backends import (
_BACKENDS,
_BIND_MOUNT_BACKENDS,
auto_detect_podman_socket,
backend_supports_bind_mounts,
get_backend,
get_host_gateway,
get_podman_socket_candidates,
get_runtime_client,
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, live 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),
patch("strix.runtime.backends._socket_is_live", return_value=True),
):
detected = auto_detect_podman_socket()
assert detected == f"unix://{fake_sock}"
def test_socket_fallthrough_autodetect_podman_skips_stale_candidate(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A stale/unreachable candidate is skipped in favor of the next live one."""
monkeypatch.delenv("STRIX_RUNTIME_SOCKET", raising=False)
monkeypatch.delenv("DOCKER_HOST", raising=False)
stale_sock = Path("/run/user/1000/podman/podman.sock")
live_sock = Path("/run/podman/podman.sock")
def fake_is_live(url: str) -> bool:
return url == f"unix://{live_sock}"
with (
patch(
"strix.runtime.backends.get_podman_socket_candidates",
return_value=[stale_sock, live_sock],
),
patch.object(Path, "exists", return_value=True),
patch.object(Path, "resolve", side_effect=[stale_sock, live_sock]),
patch("strix.runtime.backends._socket_is_live", side_effect=fake_is_live),
):
detected = auto_detect_podman_socket()
assert detected == f"unix://{live_sock}"
def test_socket_fallthrough_graceful_on_missing_or_failed_socket(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Docker backend connection failure 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_runtime_client("docker")
assert client is mock_default_client
mock_docker.from_env.assert_called_once()
def test_get_runtime_client_podman_raises_instead_of_falling_back_to_docker(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A failed Podman socket connection raises rather than silently using Docker."""
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
with (
patch.dict("sys.modules", {"docker": mock_docker}),
patch(
"strix.runtime.backends.resolve_runtime_socket",
return_value="unix:///unreachable.sock",
),
pytest.raises(RuntimeError, match="podman"),
):
get_runtime_client("podman")
mock_docker.from_env.assert_not_called()
def test_get_runtime_client_podman_raises_when_no_socket_resolved(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""No resolvable Podman socket at all raises rather than defaulting to Docker."""
monkeypatch.delenv("STRIX_RUNTIME_SOCKET", raising=False)
monkeypatch.delenv("DOCKER_HOST", raising=False)
mock_docker = MagicMock()
with (
patch.dict("sys.modules", {"docker": mock_docker}),
patch("strix.runtime.backends.resolve_runtime_socket", return_value=None),
pytest.raises(RuntimeError, match="podman"),
):
get_runtime_client("podman")
mock_docker.from_env.assert_not_called()
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_runtime_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_runtime_installed()
def test_check_runtime_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_runtime_installed()
assert exc_info.value.code == 1
def test_check_runtime_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_runtime_installed()
assert exc_info.value.code == 1
def test_check_runtime_connection_podman_uses_podman_backend(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""check_runtime_connection connects and pings backend client."""
monkeypatch.setenv("STRIX_RUNTIME_BACKEND", "podman")
mock_client = MagicMock()
with patch("strix.runtime.backends.get_runtime_client", return_value=mock_client) as mock_get:
client = check_runtime_connection()
mock_get.assert_called_once_with("podman")
mock_client.ping.assert_called_once()
assert client is mock_client
# ============================================================================
# 8. Runtime Backend Setting Normalization (2 tests)
# ============================================================================
def test_runtime_settings_backend_is_lowercased(monkeypatch: pytest.MonkeyPatch) -> None:
"""A mixed-case STRIX_RUNTIME_BACKEND is normalized so get_backend() finds it."""
monkeypatch.setenv("STRIX_RUNTIME_BACKEND", "Podman")
settings = RuntimeSettings()
assert settings.backend == "podman"
assert get_backend(settings.backend) is not None
def test_runtime_settings_backend_defaults_when_blank(monkeypatch: pytest.MonkeyPatch) -> None:
"""A blank/whitespace-only STRIX_RUNTIME_BACKEND still normalizes to the docker default."""
monkeypatch.setenv("STRIX_RUNTIME_BACKEND", " ")
settings = RuntimeSettings()
assert settings.backend == "docker"