feat: add Docker network support for sandbox connections

This commit is contained in:
TheArtificialQ 2026-07-01 10:35:39 +02:00
parent f342808d2b
commit 3e4372d6f5
10 changed files with 296 additions and 3 deletions

View file

@ -177,6 +177,9 @@ strix --target api.your-app.com --instruction "Focus on business logic flaws and
# Provide detailed instructions through file (e.g., rules of engagement, scope, exclusions)
strix --target api.your-app.com --instruction-file ./instruction.md
# Connect the sandbox to a Docker network
strix --target https://app.internal --docker-network my-network
# Force PR diff-scope against a specific base branch
strix -n --target ./ --scan-mode quick --scope-mode diff --diff-base origin/main
```

View file

@ -29,6 +29,13 @@ strix --target <target> [options]
</Note>
</ParamField>
<ParamField path="--docker-network" type="string">
Docker network to connect the container to. Accepts a network name such as `my-network`
or a built-in mode such as `host` or `bridge`.
Use this when the target is only reachable through a specific Docker network.
</ParamField>
<ParamField path="--instruction" type="string">
Custom instructions for the scan. Use for credentials, focus areas, or specific testing approaches.
</ParamField>
@ -98,6 +105,9 @@ strix -t https://github.com/org/app -t https://staging.example.com
# Large local repository — bind-mount instead of copying it in
strix --mount ./huge-monorepo
# Connect the sandbox to a Docker network
strix --target https://app.internal --docker-network my-network
```
## Exit Codes

View file

@ -144,6 +144,7 @@ async def run_strix_scan(
scan_id,
image=image,
local_sources=local_sources or [],
docker_network=scan_config.get("docker_network"),
)
logger.info("Sandbox ready for scan %s", scan_id)

View file

@ -94,6 +94,7 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
"scope_mode": getattr(args, "scope_mode", "auto"),
"diff_base": getattr(args, "diff_base", None),
"resume_instruction": getattr(args, "user_explicit_instruction", None) or "",
"docker_network": getattr(args, "docker_network", None),
}
report_state = ReportState(args.run_name)

View file

@ -315,6 +315,26 @@ def _positive_budget(value: str) -> float:
return budget
def _normalize_docker_network(
value: str | None,
parser: argparse.ArgumentParser,
) -> str | None:
if value is None:
return None
docker_network = value.strip()
if not docker_network:
parser.error("--docker-network requires a non-empty Docker network name or mode")
if docker_network.lower() == "none":
parser.error("--docker-network none is not supported because Strix needs networking")
if docker_network.lower() in {"bridge", "host"}:
return docker_network.lower()
return docker_network
def parse_arguments() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Strix Multi-Agent Cybersecurity Penetration Testing Tool",
@ -334,6 +354,9 @@ Examples:
# Large local repository (bind-mounted read-only instead of copied)
strix --mount ./huge-monorepo
# Connect the sandbox to a Docker network
strix --target https://app.internal --docker-network my-network
# Domain penetration test
strix --target example.com
@ -378,6 +401,17 @@ Examples:
"copying it file-by-file. Use this for large repositories that are too big to "
"stream into the container. Can be specified multiple times.",
)
parser.add_argument(
"--docker-network",
type=str,
dest="docker_network",
metavar="NETWORK",
help=(
"Docker network to connect the container to. Accepts a network name "
"(e.g., 'my-network') or a built-in mode such as 'host' or 'bridge'. "
"Useful when the target is only reachable through a specific Docker network."
),
)
parser.add_argument(
"--instruction",
type=str,
@ -548,6 +582,8 @@ Examples:
"--mount <path> to bind-mount the directory instead of copying it."
)
args.docker_network = _normalize_docker_network(args.docker_network, parser)
return args
@ -568,6 +604,7 @@ def _persist_run_record(args: argparse.Namespace) -> None:
"diff_scope": getattr(args, "diff_scope", {"active": False}),
"scope_mode": args.scope_mode,
"diff_base": args.diff_base,
"docker_network": args.docker_network,
}
write_run_record(run_dir, run_record)
@ -612,6 +649,8 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser
args.local_sources = state.get("local_sources")
if state.get("diff_scope"):
args.diff_scope = state.get("diff_scope")
if args.docker_network is None and state.get("docker_network"):
args.docker_network = state.get("docker_network")
persisted_scan_mode = state.get("scan_mode")
if persisted_scan_mode and args.scan_mode == "deep":
args.scan_mode = persisted_scan_mode

View file

@ -745,6 +745,7 @@ class StrixTUIApp(App): # type: ignore[misc]
"scope_mode": getattr(args, "scope_mode", "auto"),
"diff_base": getattr(args, "diff_base", None),
"resume_instruction": getattr(args, "user_explicit_instruction", None) or "",
"docker_network": getattr(args, "docker_network", None),
}
def _setup_cleanup_handlers(self) -> None:

View file

@ -23,6 +23,7 @@ async def _docker_backend(
manifest: Manifest,
exposed_ports: tuple[int, ...],
bind_mounts: list[dict[str, Any]] | None = None,
docker_network: str | None = None,
) -> tuple[Any, Any]:
"""Bring up a session backed by the local Docker daemon.
@ -41,6 +42,10 @@ async def _docker_backend(
``bind_mounts`` are host directories (e.g. large repos passed via
``--mount``) bind-mounted read-only; unlike manifest entries they are
applied by Docker at container-create time, not by ``start()``.
``docker_network`` is passed through to Docker. ``host`` changes the
container's network mode; custom networks are attached while preserving the
default bridge network used for Strix's control port.
"""
import docker
from agents.sandbox.sandboxes.docker import DockerSandboxClientOptions
@ -49,6 +54,7 @@ async def _docker_backend(
client = StrixDockerSandboxClient(docker.from_env())
client.strix_bind_mounts = bind_mounts or []
client.strix_docker_network = docker_network
options = DockerSandboxClientOptions(image=image, exposed_ports=exposed_ports)
session = await client.create(options=options, manifest=manifest)
await session.start()

View file

@ -3,8 +3,8 @@ NET_ADMIN/NET_RAW capabilities + host-gateway.
The SDK's ``DockerSandboxClient._create_container`` does not expose a hook for
extending ``create_kwargs`` before ``containers.create`` is called. We subclass
and reimplement the method body verbatim from the SDK source, with three
deltas:
and reimplement the method body verbatim from the SDK source, with
Strix-specific deltas:
1. Drop the SDK's ``entrypoint=["tail"]`` override; supply ``["tail", "-f",
"/dev/null"]`` as ``command`` instead. This lets our image's
@ -15,6 +15,7 @@ deltas:
other raw-socket tools).
3. Add ``host.docker.internal`` → host-gateway to ``extra_hosts`` so the
agent can reach host-served apps.
4. Attach the container to a configured Docker network, if requested.
Pinned to ``openai-agents==0.14.6``. Bumping the SDK requires
re-merging the parent body. Track upstream for an injection hook.
@ -49,6 +50,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]] = [] # overridden per-instance in backends.py
strix_docker_network: str | None = None # overridden per-instance in backends.py
async def _create_container(
self,
@ -104,6 +106,16 @@ class StrixDockerSandboxClient(DockerSandboxClient):
}
# ----- END VERBATIM COPY -----
docker_network = getattr(self, "strix_docker_network", None)
post_create_network: str | None = None
if docker_network:
docker_network_mode = docker_network.strip().lower()
if docker_network_mode == "host":
create_kwargs["network_mode"] = "host"
create_kwargs.pop("ports", None)
elif docker_network_mode != "bridge":
post_create_network = docker_network
# Strix injections — append, don't overwrite, so FUSE/SYS_ADMIN survives.
cap_add = create_kwargs.setdefault("cap_add", [])
if not isinstance(cap_add, list):
@ -132,12 +144,20 @@ class StrixDockerSandboxClient(DockerSandboxClient):
)
logger.debug(
"Creating sandbox container: image=%s caps=%s exposed_ports=%s",
"Creating sandbox container: image=%s caps=%s exposed_ports=%s docker_network=%s",
image,
cap_add,
list(exposed_ports),
docker_network,
)
container = self.docker_client.containers.create(**create_kwargs)
if post_create_network:
try:
self.docker_client.networks.get(post_create_network).connect(container)
except Exception:
with contextlib.suppress(Exception):
container.remove(force=True)
raise
logger.info(
"Sandbox container created: id=%s image=%s",
container.short_id if hasattr(container, "short_id") else "?",

View file

@ -63,6 +63,7 @@ async def create_or_reuse(
*,
image: str,
local_sources: list[dict[str, Any]],
docker_network: str | None = None,
) -> dict[str, Any]:
"""Return the existing session bundle for ``scan_id`` or create a new one.
@ -111,6 +112,7 @@ async def create_or_reuse(
manifest=manifest,
exposed_ports=(_CONTAINER_CAIDO_PORT,),
bind_mounts=bind_mounts,
docker_network=docker_network,
)
caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)

View file

@ -0,0 +1,210 @@
"""Tests for Docker network selection."""
from __future__ import annotations
from types import SimpleNamespace
from typing import Any
import pytest
from strix.runtime import session_manager
from strix.runtime.docker_client import StrixDockerSandboxClient
class FakeSession:
def __init__(self) -> None:
self.resolve_calls: list[int] = []
async def resolve_exposed_port(self, port: int) -> SimpleNamespace:
self.resolve_calls.append(port)
return SimpleNamespace(host="127.0.0.1", port=12345)
class FakeContainers:
def __init__(self) -> None:
self.create_kwargs: dict[str, Any] | None = None
def create(self, **kwargs: Any) -> SimpleNamespace:
self.create_kwargs = kwargs
return SimpleNamespace(short_id="abc123", removed=False, remove=lambda **_kwargs: None)
class FakeNetwork:
def __init__(self, name: str) -> None:
self.name = name
self.connected: list[Any] = []
def connect(self, container: Any) -> None:
self.connected.append(container)
class FakeNetworks:
def __init__(self) -> None:
self.requested: list[str] = []
self.networks: dict[str, FakeNetwork] = {}
def get(self, name: str) -> FakeNetwork:
self.requested.append(name)
network = self.networks.get(name)
if network is None:
network = FakeNetwork(name)
self.networks[name] = network
return network
class FakeDockerClient:
def __init__(self) -> None:
self.containers = FakeContainers()
self.networks = FakeNetworks()
self.images = SimpleNamespace(pull=lambda *_args, **_kwargs: None)
def _docker_sandbox_client(
docker_network: str | None,
) -> tuple[StrixDockerSandboxClient, FakeDockerClient]:
docker_client = FakeDockerClient()
sandbox_client = object.__new__(StrixDockerSandboxClient)
sandbox_client.docker_client = docker_client
sandbox_client.strix_bind_mounts = []
sandbox_client.strix_docker_network = docker_network
sandbox_client.image_exists = lambda _image: True # type: ignore[method-assign]
return sandbox_client, docker_client
@pytest.mark.asyncio
async def test_create_or_reuse_passes_docker_network_to_backend(
monkeypatch: pytest.MonkeyPatch,
) -> None:
captured: dict[str, Any] = {}
session = FakeSession()
async def fake_backend(
*,
image: str,
manifest: Any,
exposed_ports: tuple[int, ...],
bind_mounts: list[dict[str, Any]] | None = None,
docker_network: str | None = None,
) -> tuple[object, FakeSession]:
del image, manifest, exposed_ports, bind_mounts
captured["docker_network"] = docker_network
return object(), session
async def fake_bootstrap_caido(
sandbox_session: FakeSession,
*,
host_url: str,
container_url: str,
) -> object:
del sandbox_session, host_url, container_url
return object()
monkeypatch.setattr(session_manager, "_SESSION_CACHE", {})
monkeypatch.setattr(
session_manager,
"load_settings",
lambda: SimpleNamespace(runtime=SimpleNamespace(backend="docker")),
)
monkeypatch.setattr(session_manager, "get_backend", lambda _name: fake_backend)
monkeypatch.setattr(session_manager, "bootstrap_caido", fake_bootstrap_caido)
await session_manager.create_or_reuse(
"scan-with-docker-network",
image="strix-test:latest",
local_sources=[],
docker_network="my-network",
)
assert captured["docker_network"] == "my-network"
@pytest.mark.asyncio
async def test_create_or_reuse_uses_local_caido_url_for_host_network(
monkeypatch: pytest.MonkeyPatch,
) -> None:
captured: dict[str, Any] = {}
session = FakeSession()
async def fake_backend(
*,
image: str,
manifest: Any,
exposed_ports: tuple[int, ...],
bind_mounts: list[dict[str, Any]] | None = None,
docker_network: str | None = None,
) -> tuple[object, FakeSession]:
del image, manifest, exposed_ports, bind_mounts, docker_network
return object(), session
async def fake_bootstrap_caido(
sandbox_session: FakeSession,
*,
host_url: str,
container_url: str,
) -> object:
del sandbox_session, container_url
captured["host_url"] = host_url
return object()
monkeypatch.setattr(session_manager, "_SESSION_CACHE", {})
monkeypatch.setattr(
session_manager,
"load_settings",
lambda: SimpleNamespace(runtime=SimpleNamespace(backend="docker")),
)
monkeypatch.setattr(session_manager, "get_backend", lambda _name: fake_backend)
monkeypatch.setattr(session_manager, "bootstrap_caido", fake_bootstrap_caido)
await session_manager.create_or_reuse(
"scan-with-host-network",
image="strix-test:latest",
local_sources=[],
docker_network="host",
)
assert session.resolve_calls == []
assert captured["host_url"] == "http://127.0.0.1:48080"
@pytest.mark.asyncio
async def test_docker_client_passes_docker_network_to_container_create() -> None:
sandbox_client, docker_client = _docker_sandbox_client("my-network")
await sandbox_client._create_container("strix-test:latest", exposed_ports=(48080,))
assert docker_client.containers.create_kwargs is not None
assert "network" not in docker_client.containers.create_kwargs
assert "network_mode" not in docker_client.containers.create_kwargs
assert docker_client.containers.create_kwargs["ports"] == {
"48080/tcp": ("127.0.0.1", None),
}
assert docker_client.networks.requested == ["my-network"]
assert len(docker_client.networks.networks["my-network"].connected) == 1
@pytest.mark.asyncio
async def test_docker_client_drops_port_bindings_for_host_network() -> None:
sandbox_client, docker_client = _docker_sandbox_client("host")
await sandbox_client._create_container("strix-test:latest", exposed_ports=(48080,))
assert docker_client.containers.create_kwargs is not None
assert docker_client.containers.create_kwargs["network_mode"] == "host"
assert "network" not in docker_client.containers.create_kwargs
assert "ports" not in docker_client.containers.create_kwargs
assert docker_client.networks.requested == []
@pytest.mark.asyncio
async def test_docker_client_keeps_default_bridge_port_bindings() -> None:
sandbox_client, docker_client = _docker_sandbox_client("bridge")
await sandbox_client._create_container("strix-test:latest", exposed_ports=(48080,))
assert docker_client.containers.create_kwargs is not None
assert "network" not in docker_client.containers.create_kwargs
assert "network_mode" not in docker_client.containers.create_kwargs
assert docker_client.containers.create_kwargs["ports"] == {
"48080/tcp": ("127.0.0.1", None),
}
assert docker_client.networks.requested == []