didericis/bot-bottle
1
0
Fork 0
Code Issues 7 Pull Requests 15 Actions Packages Projects Releases Wiki Activity

feat(smolmachines): rewrite Smolfile to smolvm 0.8.0 schema + drop gvproxy (PRD 0023 chunk 2a) #64

Merged
didericis merged 1 commits from prd-0023-chunk-2a-smolfile-rewrite into main 2026-05-27 04:08:33 -04:00
Conversation 0 Commits 1 Files Changed 8
+203 -613
8 changed files with 203 additions and 613 deletions
Download Patch File Download Diff File Expand all files Collapse all files
claude_bottle/backend/smolmachines/bottle_plan.py
+10 -12
View File
@@ -1,10 +1,10 @@
"""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."""
Chunk 1 + 2a fields: slug, smolfile_path, bundle docker subnet /
gateway / pinned IP. VM lifecycle + provisioning fields (machine
name, `.smolmachine` artifact path, etc.) land in later chunks
as the launch flow grows."""
from __future__ import annotations
@@ -25,14 +25,12 @@ class SmolmachinesBottlePlan(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]
# Per-bottle docker subnet for the sidecar bundle container.
# The bundle runs at `bundle_ip` (always `.2`); the gateway is
# at `.1`. smolvm's TSI allowlist is set to `bundle_ip/32`.
bundle_subnet: str
bundle_gateway: str
bundle_ip: str
def print(self, *, remote_control: bool) -> None:
"""Compact y/N preflight. Same shape as the Docker
claude_bottle/backend/smolmachines/gvproxy_config.py
-101
View File
@@ -1,101 +0,0 @@
"""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
claude_bottle/backend/smolmachines/prepare.py
+37 -112
View File
@@ -1,9 +1,8 @@
"""smolmachines `_resolve_plan` (PRD 0023 chunk 1).
"""smolmachines `_resolve_plan` (PRD 0023 chunk 2a).
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."""
Resolves the per-bottle docker subnet + bundle IP and writes the
Smolfile to the stage dir. No VM bringup. The plan it returns is
enough for the y/N preflight to render."""
from __future__ import annotations
@@ -16,151 +15,77 @@ from ...backend.docker.bottle_state import (
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,
)
from .smolfile import smolfile_build, smolfile_write
from .util import smolmachines_bundle_subnet, smolmachines_preflight
# Gateway ports the bundle exposes inside its container — pipelock
# HTTPS proxy, git-gate's git-daemon, supervise's MCP. The agent
# inside the smolvm guest dials these on the bundle's pinned IP.
_BUNDLE_PIPELOCK_PORT = 8888
_BUNDLE_GIT_GATE_PORT = 9418
_BUNDLE_SUPERVISE_PORT = 9100
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."""
"""Materialize the smolmachines plan. The Smolfile lands at
`<stage>/smolfile.toml`; the bundle's docker subnet + pinned
IP are derived from the slug and carried on the plan for
launch to consume."""
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)
# 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.
# Record minimal metadata so `cli.py resume` can recover the
# slug. Same schema as the docker backend.
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`.
# No compose project for smolmachines bottles; chunk 4
# will give dashboard discovery a backend-specific path.
compose_project="",
))
# Per-bottle gvproxy subnet + gateway. Deterministic from slug;
# collisions surface at launch time (chunk 2).
subnet, gateway = smolmachines_gvproxy_subnet(slug)
subnet, gateway, bundle_ip = smolmachines_bundle_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.
# Agent's env. IP literals; no DNS resolution inside the guest
# (TSI allowlist contains only `<bundle_ip>/32` — no resolver).
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}",
"HTTPS_PROXY": f"http://{bundle_ip}:{_BUNDLE_PIPELOCK_PORT}",
"HTTP_PROXY": f"http://{bundle_ip}:{_BUNDLE_PIPELOCK_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}"
f"git://{bundle_ip}:{_BUNDLE_GIT_GATE_PORT}"
)
if bottle.supervise:
guest_env["MCP_SUPERVISE_URL"] = (
f"http://proxy.internal:{GVPROXY_SUPERVISE_GATEWAY_PORT}"
f"http://{bundle_ip}:{_BUNDLE_SUPERVISE_PORT}"
)
smolfile_path = stage_dir / "smolfile.toml"
smolfile_write(
smolfile_build(
slug=slug,
gvproxy_socket=gvproxy_socket,
env=guest_env,
),
smolfile,
smolfile_build(env=guest_env, bundle_ip=bundle_ip),
smolfile_path,
)
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,
smolfile_path=smolfile_path,
bundle_subnet=subnet,
bundle_gateway=gateway,
bundle_ip=bundle_ip,
)
# 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",
]
claude_bottle/backend/smolmachines/smolfile.py
+37 -67
View File
@@ -1,90 +1,65 @@
"""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:
The Smolfile shape comes from `smolvm machine create --smolfile`
in smolvm 0.8.0. The renderer emits only the runtime overrides
that vary per bottle:
- `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 = ["K=V", ...]` — agent's HTTPS_PROXY / NO_PROXY /
NODE_EXTRA_CA_CERTS, all using IP literals pointing at the
per-bottle sidecar bundle's pinned docker IP.
- `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.
- `[network] allow_cidrs = ["<bundle-ip>/32"]` — TSI's single-IP
allowlist. With this and no other `allow_*` or
`outbound-localhost-only`, the agent can dial exactly one IP:
the bundle. Host loopback, LAN, and the public internet
directly are all refused at the VMM layer.
- `[[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.
What the renderer does NOT emit:
The renderer is a pure function. Disk writes happen in
`prepare.py` via `smolfile_write`."""
- `image` / `entrypoint` / `cmd` — those come from the
`.smolmachine` artifact (produced by `smolvm pack create
--image claude-bottle:latest`) and don't vary across bottles
of the same agent image.
- `cpus` / `memory` — left at smolvm defaults until the
operator surfaces a need to override per bottle (the manifest
has no such field today).
Pure function; disk writes happen 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"),
bundle_ip: str,
) -> 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`."""
"""Build the Smolfile config dict. `env` is `{NAME: VALUE}` for
the guest's process env (IP-literal HTTPS_PROXY etc.).
`bundle_ip` is the pinned docker IP of the per-bottle sidecar
bundle; it lands in `[network] allow_cidrs` as a /32."""
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),
},
],
"network": {
"allow_cidrs": [f"{bundle_ip}/32"],
},
}
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."""
"""Render the Smolfile dict as TOML. Schema is narrow (string
list + one table with a string list) so we render by hand and
stay 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"
lines.append("[network]")
lines.append(f'allow_cidrs = {_toml_array(cfg["network"]["allow_cidrs"])}')
return "\n".join(lines) + "\n"
def smolfile_write(cfg: dict[str, Any], path: Path) -> Path:
@@ -96,16 +71,11 @@ def smolfile_write(cfg: dict[str, Any], path: Path) -> 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."""
escapes."""
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) + "]"
claude_bottle/backend/smolmachines/util.py
+28 -64
View File
@@ -1,84 +1,48 @@
"""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."""
"""Slug / preflight / subnet helpers for the smolmachines backend
(PRD 0023). 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:
"""Ensure `smolvm` is on PATH before the launch flow runs.
Called from `_resolve_plan`; gives the operator a clear
install pointer rather than a cryptic FileNotFoundError
later. `gvproxy` is no longer required — see the PRD's design
pivot section."""
if shutil.which("smolvm") is not None:
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))
die(
"CLAUDE_BOTTLE_BACKEND=smolmachines requires `smolvm` on "
"PATH. Install with: "
"curl -sSL https://smolmachines.com/install.sh | sh"
)
def smolmachines_gvproxy_subnet(slug: str) -> tuple[str, str]:
"""Derive a per-bottle subnet + gateway IP from the slug.
def smolmachines_bundle_subnet(slug: str) -> tuple[str, str, str]:
"""Derive a per-bottle docker subnet + gateway IP + bundle 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."""
Returns `(subnet_cidr, gateway_ip, bundle_ip)`. The third
octet comes from SHA-256 of the slug mod 254 (skipping 17 to
avoid the docker-default bridge), so parallel bottles get
distinct /24s and `resume` reuses the same /24. The bundle
container always lands at `.2`; gateway is `.1`; the smolvm
Smolfile's `allow_cidrs` is `<bundle_ip>/32`."""
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.
# Skip the docker-default bridge to dodge the most common
# collision (operators with `docker0` at 172.17.x.x or a
# 192.168.17.x VPN client).
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])
bundle_ip = f"192.168.{octet}.2"
return subnet, gateway, bundle_ip
tests/unit/test_smolfile.py
+58 -69
View File
@@ -1,17 +1,14 @@
"""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."""
schema we emit is narrow (env list + `[network] allow_cidrs`), 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,
)
@@ -20,104 +17,96 @@ from claude_bottle.backend.smolmachines.smolfile import (
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"},
env={"HTTPS_PROXY": "http://192.168.50.2:8888"},
bundle_ip="192.168.50.2",
)
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):
# Sorted by key so renderer output is deterministic.
cfg = self._build(env={
"ZED": "one",
"ALPHA": "two",
"HTTPS_PROXY": "http://proxy.internal:8888",
"HTTPS_PROXY": "http://192.168.50.2:8888",
})
# Sorted by key so renderer output is deterministic.
self.assertEqual(
["ALPHA=two", "HTTPS_PROXY=http://proxy.internal:8888", "ZED=one"],
[
"ALPHA=two",
"HTTPS_PROXY=http://192.168.50.2: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_allow_cidrs_is_single_slash_32(self):
# TSI's single-IP allowlist. Anything else would
# re-introduce the loopback / LAN reachability the PRD
# design carefully avoids.
cfg = self._build(bundle_ip="10.20.30.40")
self.assertEqual(
{"allow_cidrs": ["10.20.30.40/32"]},
cfg["network"],
)
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.
def test_no_image_or_command_emitted(self):
# The chunk-1 renderer (under the abandoned gvproxy design)
# emitted `name = ...` + `[[net]] attachment="unixgram"`.
# The new renderer carries only the per-bottle overrides;
# image / entrypoint / cmd come from the .smolmachine
# artifact, not the Smolfile.
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())
self.assertNotIn("image", cfg)
self.assertNotIn("entrypoint", cfg)
self.assertNotIn("cmd", cfg)
self.assertNotIn("command", cfg)
self.assertNotIn("name", cfg)
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,
defaults = dict(
env={"HTTPS_PROXY": "http://192.168.50.2:8888"},
bundle_ip="192.168.50.2",
)
return smolfile_render(cfg)
defaults.update(kwargs)
return smolfile_render(smolfile_build(**defaults))
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"])
self.assertIn(
"HTTPS_PROXY=http://192.168.50.2:8888",
parsed["env"],
)
self.assertEqual(
["192.168.50.2/32"],
parsed["network"]["allow_cidrs"],
)
def test_special_chars_in_values_escape_correctly(self):
def test_no_tsi_outbound_localhost_only(self):
# Whole point of the design pivot: never emit
# `--outbound-localhost-only` or similar that would
# re-open host loopback.
text = self._render()
self.assertNotIn("outbound_localhost_only", text)
self.assertNotIn("outbound-localhost-only", text)
# And no gvproxy / virtio-net carve-out leaked from the
# abandoned first draft.
self.assertNotIn("unixgram", text)
self.assertNotIn("gvproxy", text.lower())
def test_special_chars_in_env_value_escape(self):
import tomllib
cfg = smolfile_build(
slug="demo",
gvproxy_socket=Path("/tmp/path with spaces/gv.sock"),
env={"WITH_QUOTES": 'has "double" quotes'},
bundle_ip="10.0.0.1",
)
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()
tests/unit/test_smolmachines_gvproxy_config.py
-117
View File
@@ -1,117 +0,0 @@
"""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()
tests/unit/test_smolmachines_util.py
+33 -71
View File
@@ -1,117 +1,79 @@
"""Unit: smolmachines backend util helpers (PRD 0023 chunk 1)."""
"""Unit: smolmachines backend util helpers (PRD 0023)."""
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_bundle_subnet,
smolmachines_preflight,
)
class TestGvproxySubnet(unittest.TestCase):
def test_returns_192_168_X_format(self):
subnet, gateway = smolmachines_gvproxy_subnet("demo-abc12")
class TestBundleSubnet(unittest.TestCase):
def test_returns_subnet_gateway_and_bundle_ip(self):
subnet, gateway, bundle_ip = smolmachines_bundle_subnet("demo-abc12")
self.assertTrue(subnet.startswith("192.168."))
self.assertTrue(subnet.endswith(".0/24"))
self.assertTrue(gateway.startswith("192.168."))
# Gateway at .1, bundle at .2 — fixed convention.
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)
self.assertTrue(bundle_ip.endswith(".2"))
# All three share the same third octet.
third = subnet.split(".")[2]
self.assertEqual(third, gateway.split(".")[2])
self.assertEqual(third, bundle_ip.split(".")[2])
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")
# Recoverability: `cli.py resume` reuses the slug and
# expects to find the same per-bottle subnet (a fresh
# docker bridge would mean a different IP, and smolvm's
# allow_cidrs would no longer match).
a = smolmachines_bundle_subnet("demo-abc12")
b = smolmachines_bundle_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),
# Not a guarantee — it's hash-mod-254, collisions exist —
# but two arbitrary slugs shouldn't share a subnet in the
# typical case.
seen = {
smolmachines_gvproxy_subnet(s)
smolmachines_bundle_subnet(s)[0]
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.
def test_skips_docker_default_octet(self):
# docker's default bridge sits at 172.17.x.x; operators
# often also see 192.168.17.x from VPN clients on macOS.
# The util 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)
subnet, _, _ = smolmachines_bundle_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):
def test_smolvm_present_returns_none(self):
with patch(
"claude_bottle.backend.smolmachines.util.shutil.which",
side_effect=lambda name: f"/usr/local/bin/{name}",
return_value="/usr/local/bin/smolvm",
):
self.assertIsNone(smolmachines_preflight())
def test_missing_smolvm_dies_with_pointer(self):
def test_missing_smolvm_dies(self):
with patch(
"claude_bottle.backend.smolmachines.util.shutil.which",
side_effect=lambda name: None if name == "smolvm" else f"/x/{name}",
return_value=None,
):
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
def test_install_pointer_in_error(self):
import io
import sys
with patch(
"claude_bottle.backend.smolmachines.util.shutil.which",
return_value=None,
@@ -122,7 +84,7 @@ class TestPreflight(unittest.TestCase):
smolmachines_preflight()
msg = captured.getvalue()
self.assertIn("smolvm", msg)
self.assertIn("gvproxy", msg)
self.assertIn("smolmachines.com/install.sh", msg)
if __name__ == "__main__":