feat(smolmachines): backend skeleton + Smolfile/gvproxy renderers (PRD 0023 chunk 1) #62

Merged
didericis merged 2 commits from prd-0023-chunk-1-skeleton into main 2026-05-27 03:18:48 -04:00
15 changed files with 1084 additions and 16 deletions
+2
View File
@@ -281,6 +281,7 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
# each backend module can pull BottleSpec / BottlePlan / BottleBackend
# via `from . import ...` without hitting a partially-initialized module.
from .docker import DockerBottleBackend # noqa: E402
from .smolmachines import SmolmachinesBottleBackend # noqa: E402
# The dict is heterogeneous: each value is a BottleBackend specialized
@@ -289,6 +290,7 @@ from .docker import DockerBottleBackend # noqa: E402
# unparameterized methods (prepare → plan → launch(plan), cleanup, etc.).
_BACKENDS: dict[str, BottleBackend[Any, Any]] = {
"docker": DockerBottleBackend(),
"smolmachines": SmolmachinesBottleBackend(),
}
+5 -16
View File
@@ -17,6 +17,7 @@ from ...log import info
from ...pipelock import PipelockProxyPlan
from ...supervise import SupervisePlan
from .. import BottlePlan
from ..print_util import print_multi
@dataclass(frozen=True)
@@ -70,22 +71,10 @@ class DockerBottlePlan(BottlePlan):
# from the agent to the proxy is needed.
env_names = sorted(set(bottle.env.keys()) | set(self.forwarded_env.keys()))
def _multi(label: str, values: list[str]) -> None:
"""Print a label with N continuation-indented values. Used
for env / skills / git-gate / egress where one item
per line keeps the summary scannable."""
if not values:
info(f"{label}: (none)")
return
info(f"{label}: {values[0]}")
indent = " " * (len(label) + 2)
for v_ in values[1:]:
info(f"{indent}{v_}")
print(file=sys.stderr)
info(f"agent : {spec.agent_name}")
_multi("env ", env_names)
_multi("skills ", list(agent.skills))
print_multi("env ", env_names)
print_multi("skills ", list(agent.skills))
info(f"bottle : {agent.bottle}")
git_lines = [
@@ -93,13 +82,13 @@ class DockerBottlePlan(BottlePlan):
for u in self.git_gate_plan.upstreams
]
if git_lines:
_multi(" git gate ", git_lines)
print_multi(" git gate ", git_lines)
if self.egress_plan.routes:
egress_lines = []
for r in self.egress_plan.routes:
auth = f" [auth:{r.auth_scheme}]" if r.auth_scheme else ""
egress_lines.append(f"{r.host}{auth}")
_multi(" egress ", egress_lines)
print_multi(" egress ", egress_lines)
print(file=sys.stderr)
+28
View File
@@ -0,0 +1,28 @@
"""Shared print helpers for BottlePlan.print implementations.
Lifts the multi-value label printer out of DockerBottlePlan so the
smolmachines backend (and any future backend) renders the same
two-column scannable preflight without duplicating the indent
math."""
from __future__ import annotations
from typing import Sequence
from ..log import info
def print_multi(label: str, values: Sequence[str]) -> None:
"""Print `label: <value>` with continuation lines indented to
align under the first value. Empty `values` renders `(none)`.
Used by every backend's `BottlePlan.print` for env / skills /
git / egress — one item per line keeps the preflight summary
scannable when an agent has many of any of these."""
if not values:
info(f"{label}: (none)")
return
info(f"{label}: {values[0]}")
indent = " " * (len(label) + 2)
for v in values[1:]:
info(f"{indent}{v}")
@@ -0,0 +1,15 @@
"""smolmachines bottle backend (PRD 0023).
Selectable via `CLAUDE_BOTTLE_BACKEND=smolmachines`. Runs each
bottle inside a per-agent microVM (libkrun / Hypervisor.framework
on macOS) with a userspace gvproxy gateway as the egress
primitive. The sidecar bundle (PRD 0024) runs as a host-side
docker container reached only through gvproxy's port-forward list.
Chunk 1 (this commit) ships the backend skeleton + Smolfile +
gvproxy renderers + preflight check. VM lifecycle, sidecar
bringup, and provisioning land in later chunks."""
from .backend import SmolmachinesBottleBackend # noqa: F401
__all__ = ["SmolmachinesBottleBackend"]
@@ -0,0 +1,76 @@
"""SmolmachinesBottleBackend — the smolmachines implementation of
BottleBackend (PRD 0023). Chunk 1 ships prepare-only; launch raises
NotImplementedError until chunk 2."""
from __future__ import annotations
from contextlib import contextmanager
from pathlib import Path
from typing import Generator
from .. import BottleBackend, BottleSpec
from . import prepare as _prepare
from .bottle import SmolmachinesBottle
from .bottle_cleanup_plan import SmolmachinesBottleCleanupPlan
from .bottle_plan import SmolmachinesBottlePlan
class SmolmachinesBottleBackend(
BottleBackend["SmolmachinesBottlePlan", "SmolmachinesBottleCleanupPlan"]
):
"""smolmachines backend. Selected by
`CLAUDE_BOTTLE_BACKEND=smolmachines`."""
name = "smolmachines"
def _resolve_plan(
self, spec: BottleSpec, *, stage_dir: Path
) -> SmolmachinesBottlePlan:
return _prepare.resolve_plan(spec, stage_dir=stage_dir)
@contextmanager
def launch(
self, plan: SmolmachinesBottlePlan
) -> Generator[SmolmachinesBottle, None, None]:
del plan
raise NotImplementedError(
"smolmachines launch is implemented in PRD 0023 chunk 2; "
"chunk 1 ships prepare-only (the Smolfile + gvproxy "
"config are written, but no VM is brought up)."
)
# The generator never gets here, but the type checker wants
# to see the yield:
yield # type: ignore[unreachable]
# The four `provision_*` methods land in chunk 4 alongside the
# `smolvm machine exec`-based copy-in flow. Stubs raise so any
# caller that reaches them before chunk 4 gets a clear pointer.
def provision_prompt(
self, plan: SmolmachinesBottlePlan, target: str
) -> str | None:
raise NotImplementedError("smolmachines provision_prompt → chunk 4")
def provision_skills(
self, plan: SmolmachinesBottlePlan, target: str
) -> None:
raise NotImplementedError("smolmachines provision_skills → chunk 4")
def provision_git(
self, plan: SmolmachinesBottlePlan, target: str
) -> None:
raise NotImplementedError("smolmachines provision_git → chunk 4")
def prepare_cleanup(self) -> SmolmachinesBottleCleanupPlan:
return SmolmachinesBottleCleanupPlan()
def cleanup(self, plan: SmolmachinesBottleCleanupPlan) -> None:
del plan
# Nothing to clean in chunk 1 — see SmolmachinesBottleCleanupPlan
# docstring.
def list_active(self) -> None:
from ...log import info
info(
"smolmachines list_active: not implemented (chunk 4 wires "
"it to `smolvm machine list`)"
)
@@ -0,0 +1,41 @@
"""SmolmachinesBottle — runtime handle stub (PRD 0023 chunk 1).
The chunk-1 backend doesn't launch VMs yet, so this class only
exists to make `SmolmachinesBottleBackend.launch` resolvable at
import time. Every method raises NotImplementedError; chunk 2
gives it real `smolvm machine exec` plumbing."""
from __future__ import annotations
from .. import Bottle, ExecResult
class SmolmachinesBottle(Bottle):
"""Stub. Real impl lands in chunk 2."""
def __init__(self, name: str) -> None:
self.name = name
def exec_claude(self, argv: list[str], *, tty: bool = True) -> int:
del argv, tty
raise NotImplementedError(
"smolmachines backend chunk 1 ships prepare-only; "
"exec_claude lands in chunk 2"
)
def exec(self, script: str) -> ExecResult:
del script
raise NotImplementedError(
"smolmachines backend chunk 1 ships prepare-only; "
"exec lands in chunk 2"
)
def cp_in(self, host_path: str, container_path: str) -> None:
del host_path, container_path
raise NotImplementedError(
"smolmachines backend chunk 1 ships prepare-only; "
"cp_in lands in chunk 2"
)
def close(self) -> None:
pass
@@ -0,0 +1,25 @@
"""SmolmachinesBottleCleanupPlan — concrete BottleCleanupPlan stub
(PRD 0023 chunk 1).
Chunk 1 always reports nothing-to-clean. Real enumeration —
orphaned smolvm machines, stranded gvproxy sockets, leftover
sidecar bundle containers — lands in chunk 4 alongside the
integration-test sweep that exercises teardown."""
from __future__ import annotations
from dataclasses import dataclass
from ...log import info
from .. import BottleCleanupPlan
@dataclass(frozen=True)
class SmolmachinesBottleCleanupPlan(BottleCleanupPlan):
def print(self) -> None:
info("smolmachines cleanup: nothing to remove (chunk 4 will "
"enumerate orphan machines + gvproxy sockets)")
@property
def empty(self) -> bool:
return True
@@ -0,0 +1,61 @@
"""SmolmachinesBottlePlan — concrete BottlePlan for the smolmachines
backend (PRD 0023).
Chunk 1 fields: slug, smolfile_path, gvproxy_config_path, gvproxy
subnet + socket, and the per-bottle port map. VM lifecycle fields
(machine name, OCI archive path, etc.) land in later chunks as the
launch flow grows."""
from __future__ import annotations
import sys
from dataclasses import dataclass
from pathlib import Path
from ...log import info
from .. import BottlePlan
from ..print_util import print_multi
@dataclass(frozen=True)
class SmolmachinesBottlePlan(BottlePlan):
"""Resolved fields the launch step needs to bring up the bottle.
Inherits `spec` and `stage_dir` from BottlePlan."""
slug: str
smolfile_path: Path
gvproxy_config_path: Path
gvproxy_socket: Path
gvproxy_subnet: str
gvproxy_gateway: str
# Daemon name → host-side loopback port the bundle binds.
# Always includes "pipelock"; "git-gate" and "supervise"
# conditional on the bottle's manifest.
host_port_map: dict[str, int]
def print(self, *, remote_control: bool) -> None:
"""Compact y/N preflight. Same shape as the Docker
backend's so operators see one format across backends."""
del remote_control # not surfaced in the compact summary
spec = self.spec
manifest = spec.manifest
agent = manifest.agents[spec.agent_name]
bottle = manifest.bottle_for(spec.agent_name)
env_names = sorted(bottle.env.keys())
upstreams = [
f"{g.Name}{g.Upstream}" for g in bottle.git
]
didericis marked this conversation as resolved Outdated
Outdated
Review

remove this

remove this
routes = [r.host for r in bottle.egress.routes]
print(file=sys.stderr)
info(f"agent : {spec.agent_name}")
print_multi("env ", env_names)
print_multi("skills ", list(agent.skills))
didericis marked this conversation as resolved Outdated
Outdated
Review

can you print this multiline like for docker? use a shared util method

can you print this multiline like for docker? use a shared util method
info(f"bottle : {agent.bottle}")
if upstreams:
print_multi(" git gate ", upstreams)
didericis marked this conversation as resolved Outdated
Outdated
Review

Actually use the shared util method for all of these

Actually use the shared util method for all of these
if routes:
didericis marked this conversation as resolved Outdated
Outdated
Review

Remove this

Remove this
print_multi(" egress ", routes)
didericis marked this conversation as resolved Outdated
Outdated
Review

Don't need this either

Don't need this either
print(file=sys.stderr)
@@ -0,0 +1,101 @@
"""gvproxy YAML config renderer (PRD 0023).
The gvproxy config defines:
- The per-bottle subnet + gateway IP — derived from the slug so
parallel bottles don't collide on 192.168.X.0/24.
- A DNS rule that resolves only `proxy.internal` to the gateway.
Every other hostname returns NXDOMAIN. This is the load-bearing
rule for PRD 0022's DNS-exfil attack: the guest can't `dig`
arbitrary names.
- `port_forwards` — one entry per sidecar daemon the bottle uses.
Only what's listed here is reachable from the guest. The
host-side sidecar bundle listens on the resolved `host_port`s;
gvproxy port-forwards `gateway_port` (what the guest dials) →
host `host_port`.
This is the file the PRD 0023 design names as the network
primitive. TSI is explicitly NOT used — see PRD 0023 "Why gvproxy,
not TSI".
The renderer is pure; disk writes happen in prepare.py."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any
@dataclass(frozen=True)
class PortForward:
"""One gvproxy port forward. The guest dials `gateway_port`; the
host receives the connection at `127.0.0.1:host_port`."""
gateway_port: int
host_port: int
def gvproxy_config_build(
*,
subnet: str,
gateway: str,
port_forwards: tuple[PortForward, ...],
) -> dict[str, Any]:
"""Build the gvproxy YAML config dict. The shape matches the
recipe in `agent-vm-isolation.md` § "Full Setup": top-level
`subnet`, `gateway`, `dns:` with the `proxy.internal` carve-out,
and `port_forwards`."""
return {
"subnet": subnet,
"gateway": gateway,
"dns": [
{
"zone": ".",
"records": [
{"name": "proxy.internal", "ip": gateway},
],
},
],
"port_forwards": [
{
"gateway_port": p.gateway_port,
"host": "127.0.0.1",
"host_port": p.host_port,
}
for p in port_forwards
],
}
def gvproxy_config_render(cfg: dict[str, Any]) -> str:
"""Render the gvproxy config as a small YAML subset (the same
shape `pipelock_render_yaml` uses). Stdlib has no YAML writer
and the config shape is narrow — strings + ints + lists of dicts
— so rendering by hand is straightforward and keeps the project
stdlib-only."""
lines: list[str] = []
lines.append(f'subnet: "{cfg["subnet"]}"')
lines.append(f'gateway: "{cfg["gateway"]}"')
lines.append("dns:")
for zone in cfg["dns"]:
lines.append(f' - zone: "{zone["zone"]}"')
lines.append(" records:")
for record in zone["records"]:
lines.append(f' - name: "{record["name"]}"')
lines.append(f' ip: "{record["ip"]}"')
if cfg["port_forwards"]:
lines.append("port_forwards:")
for pf in cfg["port_forwards"]:
lines.append(f' - gateway_port: {pf["gateway_port"]}')
lines.append(f' host: "{pf["host"]}"')
lines.append(f' host_port: {pf["host_port"]}')
else:
lines.append("port_forwards: []")
return "\n".join(lines) + "\n"
def gvproxy_config_write(cfg: dict[str, Any], path: Path) -> Path:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(gvproxy_config_render(cfg))
path.chmod(0o600)
return path
@@ -0,0 +1,166 @@
"""smolmachines `_resolve_plan` (PRD 0023 chunk 1).
Lays down the two config files the launch step will consume —
Smolfile (TOML) and gvproxy YAML — under the bottle's stage dir.
No VM bringup. The plan it returns is enough for the y/N
preflight to render."""
from __future__ import annotations
from datetime import datetime, timezone
from pathlib import Path
from ...backend import BottleSpec
from ...backend.docker.bottle_state import (
BottleMetadata,
bottle_identity,
write_metadata,
)
from ...backend.docker.sidecar_bundle import sidecar_bundle_container_name
from .bottle_plan import SmolmachinesBottlePlan
from .gvproxy_config import (
PortForward,
gvproxy_config_build,
gvproxy_config_write,
)
from .smolfile import (
GVPROXY_GIT_GATE_GATEWAY_PORT,
GVPROXY_PIPELOCK_GATEWAY_PORT,
GVPROXY_SUPERVISE_GATEWAY_PORT,
smolfile_build,
smolfile_write,
)
from .util import (
allocate_loopback_port,
smolmachines_gvproxy_subnet,
smolmachines_preflight,
)
def resolve_plan(
spec: BottleSpec, *, stage_dir: Path
) -> SmolmachinesBottlePlan:
"""Materialize the smolmachines plan. Three things land on disk
under stage_dir:
- `gvproxy.sock` path (created at launch time by gvproxy,
not by prepare — prepare only records where it'll go).
- `gvproxy.yaml` — subnet + DNS rule + port_forwards.
- `smolfile.toml` — guest command/env + virtio-net device
wired to the gvproxy unixgram socket.
The y/N preflight reads from the returned plan; chunk-2 launch
consumes the file paths it points at."""
smolmachines_preflight()
manifest = spec.manifest
agent = manifest.agents[spec.agent_name]
bottle = manifest.bottle_for(spec.agent_name)
slug = spec.identity or bottle_identity(spec.agent_name)
Review

why not just use the bottle identity?

why not just use the bottle identity?
# Record minimal metadata. `resume` (PRD 0016) reuses the slug
# via spec.identity, so the metadata file is the recoverability
# contract — keep it in lockstep with the docker backend's
# bottle_state schema.
write_metadata(BottleMetadata(
identity=slug,
agent_name=spec.agent_name,
cwd=spec.user_cwd if spec.copy_cwd else "",
copy_cwd=spec.copy_cwd,
started_at=datetime.now(timezone.utc).isoformat(),
# No compose project for smolmachines bottles. Empty string
# so the dashboard's compose-project-based discovery skips
# these entries cleanly until chunk 4 teaches it about
# `smolvm machine list`.
compose_project="",
))
# Per-bottle gvproxy subnet + gateway. Deterministic from slug;
# collisions surface at launch time (chunk 2).
subnet, gateway = smolmachines_gvproxy_subnet(slug)
# Allocate one host-side loopback port per active sidecar
# daemon. The bundle (PRD 0024) binds these; gvproxy
# port-forwards from a fixed gateway port -> host port.
host_port_map: dict[str, int] = {
"pipelock": allocate_loopback_port(),
}
port_forwards: list[PortForward] = [PortForward(
gateway_port=GVPROXY_PIPELOCK_GATEWAY_PORT,
host_port=host_port_map["pipelock"],
)]
if bottle.git:
host_port_map["git-gate"] = allocate_loopback_port()
port_forwards.append(PortForward(
gateway_port=GVPROXY_GIT_GATE_GATEWAY_PORT,
host_port=host_port_map["git-gate"],
))
if bottle.supervise:
host_port_map["supervise"] = allocate_loopback_port()
port_forwards.append(PortForward(
gateway_port=GVPROXY_SUPERVISE_GATEWAY_PORT,
host_port=host_port_map["supervise"],
))
# Render + write the two config files.
gvproxy_socket = stage_dir / "gvproxy.sock"
gvproxy_yaml = stage_dir / "gvproxy.yaml"
smolfile = stage_dir / "smolfile.toml"
gvproxy_config_write(
gvproxy_config_build(
subnet=subnet,
gateway=gateway,
port_forwards=tuple(port_forwards),
),
gvproxy_yaml,
)
# Build the guest env. proxy_url points at the gvproxy gateway;
# gvproxy resolves `proxy.internal` to the gateway IP via its
# DNS rule, then port-forwards to the host sidecar bundle.
guest_env: dict[str, str] = {
**bottle.env,
"HTTPS_PROXY": f"http://proxy.internal:{GVPROXY_PIPELOCK_GATEWAY_PORT}",
"HTTP_PROXY": f"http://proxy.internal:{GVPROXY_PIPELOCK_GATEWAY_PORT}",
"NO_PROXY": "localhost,127.0.0.1",
}
if bottle.git:
guest_env["GIT_GATE_URL"] = (
f"git://proxy.internal:{GVPROXY_GIT_GATE_GATEWAY_PORT}"
)
if bottle.supervise:
guest_env["MCP_SUPERVISE_URL"] = (
f"http://proxy.internal:{GVPROXY_SUPERVISE_GATEWAY_PORT}"
)
smolfile_write(
smolfile_build(
slug=slug,
gvproxy_socket=gvproxy_socket,
env=guest_env,
),
smolfile,
)
return SmolmachinesBottlePlan(
spec=spec,
stage_dir=stage_dir,
slug=slug,
smolfile_path=smolfile,
gvproxy_config_path=gvproxy_yaml,
gvproxy_socket=gvproxy_socket,
gvproxy_subnet=subnet,
gvproxy_gateway=gateway,
host_port_map=host_port_map,
)
# Used by future cleanup logic — the bundle container that runs on
# the host carries this name even when its host is a smolmachines
# microVM. Re-exported here so chunk 3's sidecar-bringup path has
# a single import target.
__all__ = [
"resolve_plan",
"sidecar_bundle_container_name",
]
@@ -0,0 +1,111 @@
"""Smolfile (TOML) renderer for the smolmachines backend (PRD 0023).
The Smolfile pins the per-bottle microVM's command + env + virtio-net
device. Three fields drive what we emit:
- `command` — the entrypoint claude-bottle runs inside the guest.
Chunk 1 ships a placeholder (`sleep infinity`); chunk 4 wires
the real `claude` entrypoint once provisioning is in place.
- `env` — the agent's HTTP_PROXY / NO_PROXY / CA paths, pointing
at `proxy.internal:<gateway-port>`. gvproxy resolves
`proxy.internal` to the gateway IP and port-forwards to the
host-side sidecar bundle.
- `[[net]]` — a virtio-net device backed by gvproxy's unixgram
socket via the VFKT handshake. This is the line that rejects
libkrun's TSI mode: TSI's CIDR allowlist permits the entire
127.0.0.0/8 of host loopback, which exposes every host-side
service; gvproxy's explicit port-forward list is the only thing
the guest can reach.
The renderer is a pure function. Disk writes happen in
`prepare.py` via `smolfile_write`."""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any, Mapping
# Default port assignments INSIDE the gvproxy network — what the
# guest dials. The agent's HTTPS_PROXY etc. resolve to
# `proxy.internal:<one of these>`. Host-side mapping is dynamic
# (chunk 3 allocates loopback ports per bottle).
GVPROXY_PIPELOCK_GATEWAY_PORT = 8888
GVPROXY_GIT_GATE_GATEWAY_PORT = 8889
GVPROXY_SUPERVISE_GATEWAY_PORT = 8890
def smolfile_build(
*,
slug: str,
gvproxy_socket: Path,
env: Mapping[str, str],
command: tuple[str, ...] = ("sleep", "infinity"),
) -> dict[str, Any]:
"""Build the Smolfile config dict.
`gvproxy_socket` is the unixgram socket gvproxy listens on; the
guest's virtio-net device handshakes (VFKT magic) with it on
start. `env` is `{NAME: VALUE}` for the guest's process env.
`command` is the entrypoint argv inside the guest (placeholder
until chunk 4 — see module docstring).
Returns a TOML-shaped dict; render with `smolfile_render`."""
return {
"name": f"claude-bottle-{slug}",
"command": list(command),
"env": [f"{k}={v}" for k, v in sorted(env.items())],
"net": [
{
"type": "virtio-net",
"attachment": "unixgram",
"socket": str(gvproxy_socket),
},
],
}
def smolfile_render(cfg: dict[str, Any]) -> str:
"""Render the Smolfile dict as TOML. Stdlib has `tomllib` for
reading TOML but no writer; the smolmachines schema we emit is
narrow enough (string scalars + string lists + one inline table
per net device) to render by hand. Avoids a `tomli_w` runtime
dep and keeps the project stdlib-only."""
lines: list[str] = []
lines.append(f'name = {_toml_str(cfg["name"])}')
lines.append(f'command = {_toml_array(cfg["command"])}')
lines.append(f'env = {_toml_array(cfg["env"])}')
lines.append("")
for net in cfg.get("net", ()):
lines.append("[[net]]")
for key, value in net.items():
lines.append(f'{key} = {_toml_str(value)}')
lines.append("")
return "\n".join(lines).rstrip("\n") + "\n"
def smolfile_write(cfg: dict[str, Any], path: Path) -> Path:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(smolfile_render(cfg))
path.chmod(0o600)
return path
def _toml_str(value: Any) -> str:
"""TOML basic string: double-quoted with backslash + double-quote
escapes. The smolmachines fields we emit (slugs, paths, env
pairs) are ASCII-safe; the escape table covers what's reachable."""
s = str(value)
s = s.replace("\\", "\\\\").replace('"', '\\"')
return f'"{s}"'
def _toml_array(values: list[Any]) -> str:
"""TOML inline array. Uses `_toml_str` so quoting is consistent;
the alternative would be `json.dumps(values)` which renders
identical text for ASCII-only lists, but going through the same
quoter is one less surprise on future inputs."""
return "[" + ", ".join(_toml_str(v) for v in values) + "]"
@@ -0,0 +1,84 @@
"""Slug / preflight helpers for the smolmachines backend (PRD 0023).
Backend-specific utilities the prepare/launch flow reaches for.
Kept in its own module so the renderers can be unit-tested without
importing the docker subprocess paths."""
from __future__ import annotations
import hashlib
import os
import shutil
import socket
from contextlib import closing
from ...log import die
def smolmachines_preflight() -> None:
"""Ensure the binaries the smolmachines backend shells out to are
on PATH. Called from the backend's `_resolve_plan` before any
file writes; gives the operator a clear pointer rather than a
cryptic FileNotFoundError later.
`smolvm` drives the libkrun microVM lifecycle. `gvproxy`
(gvisor-tap-vsock) is the userspace TCP/IP stack the guest's
virtio-net device hooks into."""
missing: list[tuple[str, str]] = []
if shutil.which("smolvm") is None:
missing.append((
"smolvm",
"brew install smolmachines-dev/smolmachines/smolmachines "
"# or build from https://smolmachines.com/",
))
if shutil.which("gvproxy") is None:
missing.append((
"gvproxy",
"go install "
"github.com/containers/gvisor-tap-vsock/cmd/gvproxy@latest "
"# requires Go + ensure $GOPATH/bin on PATH",
))
if not missing:
return
lines = [
f"CLAUDE_BOTTLE_BACKEND=smolmachines requires the following "
f"binar{'y' if len(missing) == 1 else 'ies'} on PATH:",
]
for name, install in missing:
lines.append(f" - {name} install: {install}")
die("\n".join(lines))
def smolmachines_gvproxy_subnet(slug: str) -> tuple[str, str]:
"""Derive a per-bottle subnet + gateway IP from the slug.
Returns `(subnet_cidr, gateway_ip)`. The third octet comes from
SHA-256 of the slug mod 254 (1..254, skipping 0 and 255 + the
127.x.x.x range) — collision-free for any reasonable concurrent-
bottle count and stable across `start` / `resume` of the same
bottle. Bottles that DO collide would fail at gvproxy bringup;
`launch.py` (chunk 2) detects the conflict and surfaces a clear
error rather than silently reusing another bottle's gateway."""
digest = hashlib.sha256(slug.encode("utf-8")).digest()
octet = (digest[0] % 254) + 1
# Skip docker-default 17 to dodge the usual second-bridge
# collision. Operators with docker's default bridge at
# 172.17.x.x — common Linux setup — would otherwise see a
# collision the moment they run a parallel docker bottle.
if octet == 17:
octet = 18
subnet = f"192.168.{octet}.0/24"
gateway = f"192.168.{octet}.1"
return subnet, gateway
def allocate_loopback_port() -> int:
"""Bind / release dance to grab a free TCP port on 127.0.0.1.
Race window: a parallel allocator could grab the same port
between our close and the bundle's bind. The window is small
enough that chunk 1 doesn't address it; chunk 3 will add a
retry loop if it shows up in practice."""
with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
s.bind(("127.0.0.1", 0))
return int(s.getsockname()[1])
+123
View File
@@ -0,0 +1,123 @@
"""Unit: Smolfile renderer for the smolmachines backend (PRD 0023).
Pure-function tests on `smolfile_build` + `smolfile_render`. The
schema we emit is narrow (name + command + env + one inline-table
per net device), so the tests exhaustively cover what lands on
disk."""
from __future__ import annotations
import unittest
from pathlib import Path
from claude_bottle.backend.smolmachines.smolfile import (
GVPROXY_PIPELOCK_GATEWAY_PORT,
smolfile_build,
smolfile_render,
)
class TestSmolfileBuild(unittest.TestCase):
def _build(self, **kwargs):
defaults = dict(
slug="demo-abc12",
gvproxy_socket=Path("/tmp/cb-stage/gvproxy.sock"),
env={"HTTPS_PROXY": "http://proxy.internal:8888"},
)
defaults.update(kwargs)
return smolfile_build(**defaults)
def test_name_uses_claude_bottle_prefix(self):
cfg = self._build(slug="myagent-xyz")
self.assertEqual("claude-bottle-myagent-xyz", cfg["name"])
def test_command_defaults_to_sleep_infinity(self):
# Chunk 1 placeholder; chunk 4 swaps in the real claude
# entrypoint.
cfg = self._build()
self.assertEqual(["sleep", "infinity"], cfg["command"])
def test_command_can_be_overridden(self):
cfg = self._build(command=("claude", "--no-banner"))
self.assertEqual(["claude", "--no-banner"], cfg["command"])
def test_env_renders_as_sorted_KEY_VALUE_list(self):
cfg = self._build(env={
"ZED": "one",
"ALPHA": "two",
"HTTPS_PROXY": "http://proxy.internal:8888",
})
# Sorted by key so renderer output is deterministic.
self.assertEqual(
["ALPHA=two", "HTTPS_PROXY=http://proxy.internal:8888", "ZED=one"],
cfg["env"],
)
def test_net_device_points_at_gvproxy_socket(self):
cfg = self._build(gvproxy_socket=Path("/state/foo/gv.sock"))
self.assertEqual(1, len(cfg["net"]))
net = cfg["net"][0]
self.assertEqual("virtio-net", net["type"])
self.assertEqual("unixgram", net["attachment"])
self.assertEqual("/state/foo/gv.sock", net["socket"])
def test_no_tsi_flags(self):
# PRD 0023: TSI is explicitly rejected. The Smolfile must
# never carry --allow-cidr / --allow-host /
# --outbound-localhost-only — gvproxy is the policy layer.
cfg = self._build()
rendered = smolfile_render(cfg)
self.assertNotIn("--allow-cidr", rendered)
self.assertNotIn("--allow-host", rendered)
self.assertNotIn("--outbound-localhost-only", rendered)
self.assertNotIn("tsi", rendered.lower())
class TestSmolfileRender(unittest.TestCase):
"""The rendered TOML must be parseable by stdlib `tomllib` and
contain the keys the smolmachines schema expects."""
def _render(self, **kwargs):
cfg = smolfile_build(
slug="demo-abc12",
gvproxy_socket=Path("/tmp/gvp.sock"),
env={"HTTPS_PROXY": "http://proxy.internal:8888"},
**kwargs,
)
return smolfile_render(cfg)
def test_round_trip_through_tomllib(self):
import tomllib # stdlib in 3.11+
rendered = self._render()
parsed = tomllib.loads(rendered)
self.assertEqual("claude-bottle-demo-abc12", parsed["name"])
self.assertEqual(["sleep", "infinity"], parsed["command"])
self.assertIn("HTTPS_PROXY=http://proxy.internal:8888", parsed["env"])
# net is an array of tables → list of dicts post-parse.
self.assertEqual(1, len(parsed["net"]))
self.assertEqual("/tmp/gvp.sock", parsed["net"][0]["socket"])
def test_special_chars_in_values_escape_correctly(self):
import tomllib
cfg = smolfile_build(
slug="demo",
gvproxy_socket=Path("/tmp/path with spaces/gv.sock"),
env={"WITH_QUOTES": 'has "double" quotes'},
)
rendered = smolfile_render(cfg)
parsed = tomllib.loads(rendered)
self.assertEqual(
"/tmp/path with spaces/gv.sock",
parsed["net"][0]["socket"],
)
# The env entry survives the quote escape.
self.assertIn('WITH_QUOTES=has "double" quotes', parsed["env"])
def test_constants_match_what_prepare_uses(self):
# Lock the gateway-port constant so the prepare side and the
# config-render side don't drift out of sync.
self.assertEqual(8888, GVPROXY_PIPELOCK_GATEWAY_PORT)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,117 @@
"""Unit: gvproxy YAML renderer for the smolmachines backend
(PRD 0023). The config shape comes from the recipe in
`docs/research/agent-vm-isolation.md` § "Full Setup". Tests pin
the load-bearing rules: only `proxy.internal` resolves; only
explicit port forwards are reachable."""
from __future__ import annotations
import unittest
from claude_bottle.backend.smolmachines.gvproxy_config import (
PortForward,
gvproxy_config_build,
gvproxy_config_render,
)
class TestGvproxyConfigBuild(unittest.TestCase):
def test_subnet_and_gateway_pass_through(self):
cfg = gvproxy_config_build(
subnet="192.168.50.0/24",
gateway="192.168.50.1",
port_forwards=(),
)
self.assertEqual("192.168.50.0/24", cfg["subnet"])
self.assertEqual("192.168.50.1", cfg["gateway"])
def test_dns_resolves_only_proxy_internal(self):
# Load-bearing for PRD 0022's DNS-exfil attack: anything
# other than `proxy.internal` MUST return NXDOMAIN.
cfg = gvproxy_config_build(
subnet="192.168.50.0/24",
gateway="192.168.50.1",
port_forwards=(),
)
self.assertEqual(1, len(cfg["dns"]))
zone = cfg["dns"][0]
self.assertEqual(".", zone["zone"])
self.assertEqual(
[{"name": "proxy.internal", "ip": "192.168.50.1"}],
zone["records"],
)
def test_port_forwards_render_one_per_entry(self):
cfg = gvproxy_config_build(
subnet="192.168.50.0/24",
gateway="192.168.50.1",
port_forwards=(
PortForward(gateway_port=8888, host_port=51001),
PortForward(gateway_port=8889, host_port=51002),
PortForward(gateway_port=8890, host_port=51003),
),
)
self.assertEqual(3, len(cfg["port_forwards"]))
# All forwards land on host loopback.
for pf in cfg["port_forwards"]:
self.assertEqual("127.0.0.1", pf["host"])
def test_no_port_forwards_renders_empty_list(self):
# A bottle that somehow had no forwards (none in practice
# since pipelock is always allocated) must not silently
# default to permissive — explicit empty list keeps the
# guest with literally no outbound destinations.
cfg = gvproxy_config_build(
subnet="192.168.50.0/24",
gateway="192.168.50.1",
port_forwards=(),
)
self.assertEqual([], cfg["port_forwards"])
class TestGvproxyConfigRender(unittest.TestCase):
def _render(self, **kwargs):
defaults = dict(
subnet="192.168.50.0/24",
gateway="192.168.50.1",
port_forwards=(PortForward(gateway_port=8888, host_port=51001),),
)
defaults.update(kwargs)
return gvproxy_config_render(gvproxy_config_build(**defaults))
def test_subnet_and_gateway_quoted_strings(self):
text = self._render()
self.assertIn('subnet: "192.168.50.0/24"', text)
self.assertIn('gateway: "192.168.50.1"', text)
def test_dns_records_emit_in_yaml_list_form(self):
text = self._render()
self.assertIn('dns:', text)
self.assertIn('- zone: "."', text)
self.assertIn('- name: "proxy.internal"', text)
self.assertIn('ip: "192.168.50.1"', text)
def test_port_forwards_emit_inline_ints(self):
text = self._render(port_forwards=(
PortForward(gateway_port=8888, host_port=51001),
))
self.assertIn('- gateway_port: 8888', text)
self.assertIn('host_port: 51001', text)
self.assertIn('host: "127.0.0.1"', text)
def test_empty_port_forwards_uses_empty_list_syntax(self):
text = self._render(port_forwards=())
self.assertIn("port_forwards: []", text)
def test_no_tsi_or_allowlist_leak(self):
# gvproxy's job is the explicit port-forward allowlist. No
# mention of TSI primitives that the smolmachines research
# note recommended and PRD 0023 explicitly rejected.
text = self._render()
for forbidden in ("allow-cidr", "allow-host", "TSI", "tsi"):
self.assertNotIn(forbidden, text,
f"gvproxy config leaked {forbidden!r}")
if __name__ == "__main__":
unittest.main()
+129
View File
@@ -0,0 +1,129 @@
"""Unit: smolmachines backend util helpers (PRD 0023 chunk 1)."""
from __future__ import annotations
import socket
import unittest
from unittest.mock import patch
from claude_bottle.backend.smolmachines.util import (
allocate_loopback_port,
smolmachines_gvproxy_subnet,
smolmachines_preflight,
)
class TestGvproxySubnet(unittest.TestCase):
def test_returns_192_168_X_format(self):
subnet, gateway = smolmachines_gvproxy_subnet("demo-abc12")
self.assertTrue(subnet.startswith("192.168."))
self.assertTrue(subnet.endswith(".0/24"))
self.assertTrue(gateway.startswith("192.168."))
self.assertTrue(gateway.endswith(".1"))
# The subnet and gateway share the same third octet.
sub_octet = subnet.split(".")[2]
gw_octet = gateway.split(".")[2]
self.assertEqual(sub_octet, gw_octet)
def test_stable_for_same_slug(self):
# Recoverability: `resume` reuses the slug + expects the
# same subnet so a re-attach doesn't try to grab a fresh
# network range from gvproxy.
a = smolmachines_gvproxy_subnet("demo-abc12")
b = smolmachines_gvproxy_subnet("demo-abc12")
self.assertEqual(a, b)
def test_different_slugs_likely_differ(self):
# Not a guarantee (it's hash-mod-254 so collisions exist),
# but two arbitrary slugs shouldn't share a subnet in the
# typical case.
seen = {
smolmachines_gvproxy_subnet(s)
for s in ("a", "b", "c", "d", "e", "alpha", "beta", "gamma")
}
self.assertGreater(len(seen), 1)
def test_never_collides_with_docker_default_bridge(self):
# docker's default bridge sits at 172.17.x.x but operators
# commonly also see 192.168.17.x from VPN clients on macOS.
# The util explicitly skips octet 17 → 18 so the smolmachines
# subnet doesn't collide with that historical pain point.
for slug in (f"slug-{i}" for i in range(500)):
subnet, gateway = smolmachines_gvproxy_subnet(slug)
self.assertNotEqual("192.168.17.0/24", subnet,
f"slug {slug!r} landed on the skipped octet")
class TestAllocateLoopbackPort(unittest.TestCase):
def test_returns_in_ephemeral_range(self):
port = allocate_loopback_port()
# Linux ephemeral starts at 32768; macOS at 49152. Either
# way it's >1024, which is what matters.
self.assertGreater(port, 1024)
self.assertLess(port, 65536)
def test_port_is_free_at_return(self):
# The dance is bind-with-port-0 + getsockname + close. By
# the time we return, the kernel has the port back in the
# free pool. We confirm by binding it ourselves immediately
# (we'll race with anyone else who races for it; the
# race-window caveat lives in the docstring).
port = allocate_loopback_port()
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
s.bind(("127.0.0.1", port))
finally:
s.close()
def test_multiple_calls_return_distinct_ports(self):
# The kernel rotates ephemeral ports; consecutive calls
# almost certainly land different ports.
ports = {allocate_loopback_port() for _ in range(8)}
self.assertGreater(len(ports), 1)
class TestPreflight(unittest.TestCase):
def test_both_binaries_present_returns_none(self):
with patch(
"claude_bottle.backend.smolmachines.util.shutil.which",
side_effect=lambda name: f"/usr/local/bin/{name}",
):
self.assertIsNone(smolmachines_preflight())
def test_missing_smolvm_dies_with_pointer(self):
with patch(
"claude_bottle.backend.smolmachines.util.shutil.which",
side_effect=lambda name: None if name == "smolvm" else f"/x/{name}",
):
with self.assertRaises(SystemExit) as cm:
smolmachines_preflight()
self.assertNotEqual(0, cm.exception.code)
def test_missing_gvproxy_dies_with_pointer(self):
with patch(
"claude_bottle.backend.smolmachines.util.shutil.which",
side_effect=lambda name: None if name == "gvproxy" else f"/x/{name}",
):
with self.assertRaises(SystemExit):
smolmachines_preflight()
def test_missing_both_lists_both_in_message(self):
# When both are gone, the message names both binaries and
# gives both install pointers — operator shouldn't have to
# re-run to discover the second missing dep.
import io, sys
with patch(
"claude_bottle.backend.smolmachines.util.shutil.which",
return_value=None,
):
captured = io.StringIO()
with patch.object(sys, "stderr", captured):
with self.assertRaises(SystemExit):
smolmachines_preflight()
msg = captured.getvalue()
self.assertIn("smolvm", msg)
self.assertIn("gvproxy", msg)
if __name__ == "__main__":
unittest.main()