feat(smolmachines): provision_ca + provision_git + provision_supervise (PRD 0023 chunk 4d) #72

Merged
didericis merged 1 commits from prd-0023-chunk-4d-provision-ca-git-supervise into main 2026-05-27 14:27:30 -04:00
10 changed files with 661 additions and 77 deletions
+2 -23
View File
@@ -19,9 +19,8 @@ import os
import subprocess
from pathlib import Path
from ....git_gate import GIT_GATE_HOSTNAME
from ....git_gate import GIT_GATE_HOSTNAME, git_gate_render_gitconfig
from ....log import info
from ....manifest import GitEntry
from .. import util as docker_mod
from ..bottle_plan import DockerBottlePlan
@@ -56,26 +55,6 @@ def _provision_cwd_git(plan: DockerBottlePlan, target: str) -> None:
)
def render_git_gate_gitconfig(entries: tuple[GitEntry, ...]) -> str:
"""Render the ~/.gitconfig content for git-gate `insteadOf`
rewrites. Pure host-side, no docker; exposed for tests.
Empty `entries` returns an empty string so callers can no-op
cleanly without conditional formatting at the call site."""
if not entries:
return ""
out = [
"# claude-bottle git-gate (PRD 0008): every git operation against\n",
"# a declared upstream routes through the gate, which mirrors\n",
"# the upstream bidirectionally (gitleaks-scanned push;\n",
"# fetch-from-upstream-before-every-upload-pack via access-hook).\n",
]
for entry in entries:
out.append(f'[url "git://{GIT_GATE_HOSTNAME}/{entry.Name}.git"]\n')
out.append(f"\tinsteadOf = {entry.Upstream}\n")
return "".join(out)
def _provision_git_gate_config(plan: DockerBottlePlan, target: str) -> None:
"""Write ~/.gitconfig in the bottle with the git-gate
insteadOf rules. No-op when the bottle has no `git` entries."""
@@ -86,7 +65,7 @@ def _provision_git_gate_config(plan: DockerBottlePlan, target: str) -> None:
container_home = os.environ.get("CLAUDE_BOTTLE_CONTAINER_HOME", "/home/node")
container_gitconfig = f"{container_home}/.gitconfig"
content = render_git_gate_gitconfig(bottle.git)
content = git_gate_render_gitconfig(bottle.git, GIT_GATE_HOSTNAME)
config_file = plan.stage_dir / "agent_gitconfig"
config_file.write_text(content)
config_file.chmod(0o600)
+14 -4
View File
@@ -13,8 +13,11 @@ from . import prepare as _prepare
from .bottle import SmolmachinesBottle
from .bottle_cleanup_plan import SmolmachinesBottleCleanupPlan
from .bottle_plan import SmolmachinesBottlePlan
from .provision import ca as _ca
from .provision import git as _git
from .provision import prompt as _prompt
from .provision import skills as _skills
from .provision import supervise as _supervise
class SmolmachinesBottleBackend(
@@ -37,6 +40,11 @@ class SmolmachinesBottleBackend(
with _launch.launch(plan, provision=self.provision) as bottle:
yield bottle
def provision_ca(
self, plan: SmolmachinesBottlePlan, target: str
) -> None:
_ca.provision_ca(plan, target)
def provision_prompt(
self, plan: SmolmachinesBottlePlan, target: str
) -> str | None:
@@ -50,10 +58,12 @@ class SmolmachinesBottleBackend(
def provision_git(
self, plan: SmolmachinesBottlePlan, target: str
) -> None:
# Chunk 4 follow-on: needs the git-gate inner Plan (so the
# gitconfig insteadOf URL points at the gate's host) and
# the agent image must contain `git`. Stub for chunk 4a.
pass
_git.provision_git(plan, target)
def provision_supervise(
self, plan: SmolmachinesBottlePlan, target: str
) -> None:
_supervise.provision_supervise(plan, target)
def prepare_cleanup(self) -> SmolmachinesBottleCleanupPlan:
return SmolmachinesBottleCleanupPlan()
@@ -0,0 +1,83 @@
"""Install the per-bottle MITM CA into the smolmachines guest's
trust store (PRD 0023 chunk 4d).
Mirrors `backend.docker.provision.ca`: select the right CA (egress
when the bottle has routes, else pipelock), `smolvm machine cp` it
to Debian's `/usr/local/share/ca-certificates/` path,
`update-ca-certificates` to rebuild the trust bundle, and log the
fingerprint once. The selected cert depends on the agent's
HTTP_PROXY target — same logic as the docker backend, since the
agent dials the same daemons through the same bundle.
`smolvm machine exec` runs commands as root in the VM (no `-u`
flag exists; the VM init is root), so we don't need the explicit
`-u 0` the docker backend uses on its `docker exec` calls."""
from __future__ import annotations
import hashlib
import ssl
from pathlib import Path
from ....log import die, info
from ...docker.provision.ca import AGENT_CA_BUNDLE, AGENT_CA_PATH
from .. import smolvm as _smolvm
from ..bottle_plan import SmolmachinesBottlePlan
def _select_ca_cert(plan: SmolmachinesBottlePlan) -> tuple[Path, str]:
"""Pick the CA cert (and a short label for the log line) that
matches the proxy the agent's HTTP_PROXY points at. Egress-proxy
wins when the bottle declares any routes; else pipelock.
The launch step minted both CAs (pipelock always; egress when
routes are declared) and stored their host paths back into the
inner Plans via `dataclasses.replace`. If those paths are empty
here something has gone wrong in launch's bringup."""
if plan.egress_plan.routes:
cert = plan.egress_plan.mitmproxy_ca_cert_only_host_path
if cert == Path() or not cert.is_file():
die(
f"egress CA cert missing at {cert or '(empty)'}; "
f"launch must have called egress_tls_init and "
f"re-bound the plan before provision"
)
return cert, "egress"
cert = plan.proxy_plan.ca_cert_host_path
if not cert or not cert.is_file():
die(
f"pipelock CA cert missing at {cert or '(empty)'}; "
f"launch must have called pipelock_tls_init and re-bound "
f"the plan before provision"
)
return cert, "pipelock"
def provision_ca(plan: SmolmachinesBottlePlan, target: str) -> None:
"""Copy the agent-facing CA cert into the guest, rebuild the
trust bundle, emit a one-line fingerprint log. Called from
`BottleBackend.provision` after the smolvm guest is up."""
cert_host_path, label = _select_ca_cert(plan)
_smolvm.machine_cp(str(cert_host_path), f"{target}:{AGENT_CA_PATH}")
# Mode 0644 — readable to non-root tools in the guest.
# update-ca-certificates rebuilds the bundle at AGENT_CA_BUNDLE,
# which is what curl / Python ssl / OpenSSL-based tools read by
# default. The env trio (NODE_EXTRA_CA_CERTS / SSL_CERT_FILE /
# REQUESTS_CA_BUNDLE) on the guest_env covers Node + Python
# `requests` / libraries that don't load the system bundle.
_smolvm.machine_exec(target, ["chmod", "644", AGENT_CA_PATH])
_smolvm.machine_exec(target, ["update-ca-certificates"])
# Stdlib SHA-256 of the cert's DER bytes — the standard
# fingerprint form. Never the private key.
der = ssl.PEM_cert_to_DER_cert(cert_host_path.read_text())
fingerprint = hashlib.sha256(der).hexdigest()
info(f"{label} ca fingerprint: sha256:{fingerprint[:32]}...")
# Re-exported for the launch/provision_ca caller + tests. The path
# constants come from the docker module because they're tied to
# Debian's `update-ca-certificates` layout — same in both backends
# since both guest images are Debian-family.
__all__ = ["AGENT_CA_BUNDLE", "AGENT_CA_PATH", "provision_ca"]
@@ -0,0 +1,102 @@
"""Git provisioning inside a running smolmachines bottle
(PRD 0023 chunk 4d).
Two concerns, both about git in the agent:
1. If --cwd was passed AND the host cwd has a .git, copy that
.git into /home/node/workspace/.git so the agent operates on
the user's repo.
2. If the bottle declares `git` entries (PRD 0008), write a
~/.gitconfig with insteadOf rules so every git operation
against a declared upstream transparently hits the per-bottle
git-gate. The gate mirrors the upstream in both directions,
so URL rewriting is symmetric.
Differs from `backend.docker.provision.git` in one address detail:
the TSI-allowlisted guest can only reach the bundle's pinned IP
(no DNS resolver in the /32 allowlist), so the insteadOf URLs
are `git://<bundle_ip>:<port>/<name>.git` rather than the
docker backend's `git://git-gate/<name>.git`. The render itself
is the shared `git_gate_render_gitconfig` on the platform-neutral
git_gate module."""
from __future__ import annotations
import os
import tempfile
from pathlib import Path
from ....git_gate import git_gate_render_gitconfig
from ....log import info
from ...docker.git_gate import GIT_GATE_PORT
from .. import smolvm as _smolvm
from ..bottle_plan import SmolmachinesBottlePlan
# `node` is the agent user from the repo Dockerfile. Override via
# CLAUDE_BOTTLE_GUEST_HOME mirrors the docker backend's
# CLAUDE_BOTTLE_CONTAINER_HOME knob — same purpose, different
# transport.
_DEFAULT_GUEST_HOME = "/home/node"
def _guest_home() -> str:
return os.environ.get("CLAUDE_BOTTLE_GUEST_HOME", _DEFAULT_GUEST_HOME)
def provision_git(plan: SmolmachinesBottlePlan, target: str) -> None:
"""Set up git inside the guest. Runs both subcases; each
no-ops when its condition isn't met."""
_provision_cwd_git(plan, target)
_provision_git_gate_config(plan, target)
def _provision_cwd_git(plan: SmolmachinesBottlePlan, target: str) -> None:
"""If --cwd was set and the host cwd has a .git directory, copy
it into <guest_home>/workspace/.git and fix ownership. No-op
otherwise."""
if not (plan.spec.copy_cwd and Path(plan.spec.user_cwd, ".git").is_dir()):
return
guest_workspace_git = f"{_guest_home()}/workspace/.git"
info(f"copying {plan.spec.user_cwd}/.git -> {target}:{guest_workspace_git}")
# mkdir -p the workspace dir so `machine cp` lands the .git
# directly there even on first-time bottles.
_smolvm.machine_exec(target, ["mkdir", "-p", f"{_guest_home()}/workspace"])
_smolvm.machine_cp(
f"{plan.spec.user_cwd}/.git", f"{target}:{guest_workspace_git}",
)
# `machine cp` lands files as root; the agent runs as node so
# the workspace tree must be chowned over.
_smolvm.machine_exec(
target, ["chown", "-R", "node:node", guest_workspace_git],
)
def _provision_git_gate_config(plan: SmolmachinesBottlePlan, target: str) -> None:
"""Write ~/.gitconfig in the guest with the git-gate insteadOf
rules. No-op when the bottle has no `git` entries."""
bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
if not bottle.git:
return
# IP-literal form: the TSI allowlist passes <bundle_ip>/32 and
# nothing else, so the agent has to dial the gate by IP+port.
gate_host = f"{plan.bundle_ip}:{GIT_GATE_PORT}"
content = git_gate_render_gitconfig(bottle.git, gate_host)
guest_gitconfig = f"{_guest_home()}/.gitconfig"
# Stage the file under the plan's stage_dir so `machine cp`
# has a stable host path. The plan's stage_dir is cleaned up
# by start.py's session-end teardown.
with tempfile.NamedTemporaryFile(
"w", dir=str(plan.stage_dir), prefix="gitconfig.",
delete=False,
) as f:
f.write(content)
config_file = Path(f.name)
os.chmod(config_file, 0o600)
info(f"writing {guest_gitconfig} with {len(bottle.git)} insteadOf rule(s)")
_smolvm.machine_cp(str(config_file), f"{target}:{guest_gitconfig}")
_smolvm.machine_exec(target, ["chown", "node:node", guest_gitconfig])
_smolvm.machine_exec(target, ["chmod", "644", guest_gitconfig])
@@ -3,30 +3,40 @@
The prompt file is always copied (so the in-guest path always
exists) but `--append-system-prompt-file` only fires when the
agent actually has a prompt — the return value signals which
case, mirroring the docker backend's contract."""
case, mirroring the docker backend's contract.
`smolvm machine cp` lands files as root inside the VM; the claude
process runs as `node`, so we chown + chmod the prompt after the
copy. Same flow as the docker backend's provision_prompt."""
from __future__ import annotations
import os
from .. import smolvm as _smolvm
from ..bottle_plan import SmolmachinesBottlePlan
# In-guest path for the prompt. Smolvm's default agent image
# (alpine for now; the real claude-bottle image later) runs as
# root with $HOME=/root. The path is also surfaced as the return
# value so the caller can pass it via --append-system-prompt-file.
_IN_GUEST_PROMPT_PATH = "/root/.claude-bottle-prompt.txt"
# `node` is the agent user from the repo Dockerfile.
# CLAUDE_BOTTLE_GUEST_HOME mirrors the docker backend's
# CLAUDE_BOTTLE_CONTAINER_HOME knob.
_DEFAULT_GUEST_HOME = "/home/node"
def provision_prompt(plan: SmolmachinesBottlePlan, target: str) -> str | None:
"""Copy the prompt file into the running smolvm guest. Returns
the in-guest path if the agent has a non-empty prompt (drives
--append-system-prompt-file), else None. The file is copied
either way so the path always exists — mirrors the docker
backend's behavior."""
_smolvm.machine_cp(
str(plan.prompt_file),
f"{target}:{_IN_GUEST_PROMPT_PATH}",
)
"""Copy the prompt file into the running smolvm guest, fix
ownership/mode. Returns the in-guest path if the agent has a
non-empty prompt (drives --append-system-prompt-file), else
None. The file is copied either way so the path always
exists — mirrors the docker backend's behavior."""
guest_home = os.environ.get("CLAUDE_BOTTLE_GUEST_HOME", _DEFAULT_GUEST_HOME)
in_guest_prompt_path = f"{guest_home}/.claude-bottle-prompt.txt"
_smolvm.machine_cp(str(plan.prompt_file), f"{target}:{in_guest_prompt_path}")
# machine cp lands as root, source's 0o600 mode is preserved —
# node can't read its own prompt without these two.
_smolvm.machine_exec(target, ["chown", "node:node", in_guest_prompt_path])
_smolvm.machine_exec(target, ["chmod", "600", in_guest_prompt_path])
agent = plan.spec.manifest.agents[plan.spec.agent_name]
return _IN_GUEST_PROMPT_PATH if agent.prompt else None
return in_guest_prompt_path if agent.prompt else None
@@ -18,23 +18,26 @@ from ..bottle_plan import SmolmachinesBottlePlan
# In-guest path mirrors the docker backend's claude-skills
# convention (~/.claude/skills/<name>/). For smolmachines the
# agent is root by default; chunk 5+ may swap to a node user
# in the real claude-bottle image, at which point this path
# follows /home/node/ — the env knob below provides the override.
_DEFAULT_SKILLS_DIR = "/root/.claude/skills"
# convention (~/.claude/skills/<name>/) under the node user's
# home — same path as the real claude-bottle image's
# /home/node/.claude/skills (pre-created in the Dockerfile).
_DEFAULT_SKILLS_DIR = "/home/node/.claude/skills"
def provision_skills(plan: SmolmachinesBottlePlan, target: str) -> None:
"""Copy each of the agent's named skills from the host's
~/.claude/skills/<name>/ into the guest's equivalent path.
For each skill: `mkdir -p` the destination, then
`smolvm machine cp` the host source dir over. No-op when the
agent has no skills.
For each skill: `mkdir -p` the destination, `smolvm machine cp`
the host source dir over, then chown the result to node:node so
the agent can read it. No-op when the agent has no skills.
smolvm machine cp on a directory copies recursively (same
semantics as `cp -r`); unlike docker cp's trailing-slash
convention, smolvm doesn't need the `/.` suffix dance."""
convention, smolvm doesn't need the `/.` suffix dance.
machine cp lands files as root inside the VM, so we chown each
skill tree over to node:node after the copy — same pattern as
the docker backend's provision_prompt."""
agent = plan.spec.manifest.agents[plan.spec.agent_name]
if not agent.skills:
return
@@ -57,3 +60,4 @@ def provision_skills(plan: SmolmachinesBottlePlan, target: str) -> None:
# Wipe any prior copy so re-runs don't accumulate.
_smolvm.machine_exec(target, ["rm", "-rf", dst])
_smolvm.machine_cp(src, f"{target}:{dst}")
_smolvm.machine_exec(target, ["chown", "-R", "node:node", dst])
@@ -0,0 +1,60 @@
"""Supervise sidecar provisioning inside a running smolmachines
bottle (PRD 0023 chunk 4d; PRD 0013 supervise plane).
Registers the per-bottle supervise sidecar as an HTTP MCP server
in the agent's claude-code config so the agent discovers the
stuck-recovery MCP tools (pipelock-block, capability-block) at
startup.
Mirrors `backend.docker.provision.supervise` same `claude mcp
add` call, just dispatched via `smolvm machine exec` instead of
`docker exec`, and against `<bundle_ip>:<port>` instead of the
short `supervise` alias (no DNS in the TSI-allowlisted guest)."""
from __future__ import annotations
from ....log import info, warn
from ....supervise import SUPERVISE_PORT
from .. import smolvm as _smolvm
from ..bottle_plan import SmolmachinesBottlePlan
_SUPERVISE_MCP_NAME = "supervise"
def supervise_mcp_url(bundle_ip: str) -> str:
return f"http://{bundle_ip}:{SUPERVISE_PORT}/"
def provision_supervise(plan: SmolmachinesBottlePlan, target: str) -> None:
"""Run `claude mcp add` inside the guest to register the
supervise sidecar in claude-code's user config. No-op when
bottle.supervise is False.
Failure is logged but not fatal: the bottle still works (you
just can't call supervise tools from the agent until the entry
is added manually). The operator sees the warning at launch."""
if plan.supervise_plan is None:
return
url = supervise_mcp_url(plan.bundle_ip)
info(f"registering supervise MCP server in agent claude config → {url}")
r = _smolvm.machine_exec(
target,
[
"claude", "mcp", "add",
"--scope", "user",
"--transport", "http",
_SUPERVISE_MCP_NAME,
url,
],
)
if r.returncode != 0:
warn(
f"`claude mcp add supervise` failed (exit {r.returncode}): "
f"{(r.stderr or r.stdout or '').strip()}. Inside the bottle, "
f"register manually with: "
f"claude mcp add --scope user --transport http supervise {url}"
)
__all__ = ["provision_supervise", "supervise_mcp_url"]
+30 -1
View File
@@ -35,7 +35,7 @@ from pathlib import Path
from typing import Mapping
from .log import die
from .manifest import Bottle
from .manifest import Bottle, GitEntry
# Short network alias for git-gate inside the sidecar bundle. The
@@ -140,6 +140,35 @@ def git_gate_aggregate_extra_hosts(
return merged
def git_gate_render_gitconfig(
entries: tuple[GitEntry, ...], gate_host: str
) -> str:
"""Render the agent's ~/.gitconfig content for git-gate
`insteadOf` rewrites. Pure host-side, no docker / smolvm;
exposed for tests + reuse across backends.
`gate_host` is the part of the URL between `git://` and the
repo path backends differ here:
- docker: `git-gate` (the short network alias)
- smolmachines: `<bundle_ip>:<port>` (no DNS in the
TSI-allowlisted guest)
Empty `entries` returns an empty string so callers can no-op
cleanly without conditional formatting at the call site."""
if not entries:
return ""
out = [
"# claude-bottle git-gate (PRD 0008): every git operation against\n",
"# a declared upstream routes through the gate, which mirrors\n",
"# the upstream bidirectionally (gitleaks-scanned push;\n",
"# fetch-from-upstream-before-every-upload-pack via access-hook).\n",
]
for entry in entries:
out.append(f'[url "git://{gate_host}/{entry.Name}.git"]\n')
out.append(f"\tinsteadOf = {entry.Upstream}\n")
return "".join(out)
def git_gate_known_hosts_line(host: str, port: str, key: str) -> str:
"""Format `host[:port] key` for OpenSSH's known_hosts. Non-default
ports use the bracketed `[host]:port` form (the form OpenSSH writes
+23 -5
View File
@@ -1,19 +1,28 @@
"""Unit: render of ~/.gitconfig pushInsteadOf rules (PRD 0008)."""
"""Unit: render of ~/.gitconfig insteadOf rules (PRD 0008).
The render moved to `claude_bottle.git_gate` so both backends
share it; tests live here because docker's provision_git is the
original consumer."""
import unittest
from claude_bottle.backend.docker.provision.git import render_git_gate_gitconfig
from claude_bottle.git_gate import (
GIT_GATE_HOSTNAME,
git_gate_render_gitconfig,
)
from tests.fixtures import fixture_minimal, fixture_with_git
class TestGitGateGitconfigRender(unittest.TestCase):
def test_empty_entries_renders_nothing(self):
bottle = fixture_minimal().bottles["dev"]
self.assertEqual("", render_git_gate_gitconfig(bottle.git))
self.assertEqual(
"", git_gate_render_gitconfig(bottle.git, GIT_GATE_HOSTNAME),
)
def test_one_block_per_entry(self):
bottle = fixture_with_git().bottles["dev"]
out = render_git_gate_gitconfig(bottle.git)
out = git_gate_render_gitconfig(bottle.git, GIT_GATE_HOSTNAME)
# Both entries map to a [url ...] block keyed on the gate's
# short network alias (`git-gate`) inside the sidecar bundle.
self.assertIn(
@@ -37,10 +46,19 @@ class TestGitGateGitconfigRender(unittest.TestCase):
# gate push and leave fetch on the original URL — exactly the
# v1 design we've moved past.
bottle = fixture_with_git().bottles["dev"]
out = render_git_gate_gitconfig(bottle.git)
out = git_gate_render_gitconfig(bottle.git, GIT_GATE_HOSTNAME)
self.assertIn("\tinsteadOf", out)
self.assertNotIn("pushInsteadOf", out)
def test_gate_host_can_be_ip_port_form(self):
# The smolmachines backend's TSI-allowlisted guest has no
# DNS, so it dials git-gate via `<bundle_ip>:<port>`.
bottle = fixture_with_git().bottles["dev"]
out = git_gate_render_gitconfig(bottle.git, "192.168.20.2:9418")
self.assertIn(
'[url "git://192.168.20.2:9418/claude-bottle.git"]', out,
)
if __name__ == "__main__":
unittest.main()
+308 -19
View File
@@ -1,4 +1,4 @@
"""Unit: smolmachines provisioning helpers (PRD 0023 chunk 4a).
"""Unit: smolmachines provisioning helpers (PRD 0023 chunks 4a + 4d).
Tests mock `smolvm.machine_cp` / `smolvm.machine_exec` and assert
on the dispatched call shape. The real round-trip lives in the
@@ -6,6 +6,8 @@ chunk-4 integration smoke."""
from __future__ import annotations
import subprocess
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
@@ -15,22 +17,48 @@ from claude_bottle.backend.smolmachines.bottle_plan import (
SmolmachinesBottlePlan,
)
from claude_bottle.backend.smolmachines.provision import (
ca as _ca,
git as _git,
prompt as _prompt,
skills as _skills,
supervise as _supervise,
)
from claude_bottle.egress import EgressPlan
from claude_bottle.backend.smolmachines.smolvm import SmolvmRunResult
from claude_bottle.egress import EgressPlan, EgressRoute
from claude_bottle.git_gate import GitGatePlan
from claude_bottle.manifest import Manifest
from claude_bottle.manifest import GitEntry, Manifest
from claude_bottle.pipelock import PipelockProxyPlan
from claude_bottle.supervise import SupervisePlan
def _plan(
*,
agent_prompt: str = "",
skills: list[str] | None = None,
git: list[GitEntry] = (),
copy_cwd: bool = False,
user_cwd: str = "/tmp/x",
stage_dir: Path | None = None,
egress_routes: tuple[EgressRoute, ...] = (),
egress_ca_path: Path = Path(),
pipelock_ca_path: Path = Path(),
supervise: bool = False,
bundle_ip: str = "192.168.50.2",
) -> SmolmachinesBottlePlan:
bottle_json: dict = {}
if git:
bottle_json["git"] = [
{
"Name": g.Name,
"Upstream": g.Upstream,
"IdentityFile": g.IdentityFile,
}
for g in git
]
if supervise:
bottle_json["supervise"] = True
manifest = Manifest.from_json_obj({
"bottles": {"dev": {}},
"bottles": {"dev": bottle_json},
"agents": {
"demo": {
"skills": list(skills or []),
@@ -42,16 +70,23 @@ def _plan(
spec = BottleSpec(
manifest=manifest,
agent_name="demo",
copy_cwd=False,
user_cwd="/tmp/x",
copy_cwd=copy_cwd,
user_cwd=user_cwd,
)
supervise_plan = None
if supervise:
supervise_plan = SupervisePlan(
slug="demo-abc12",
queue_dir=Path("/tmp/queue"),
current_config_dir=Path("/tmp/current-config"),
)
return SmolmachinesBottlePlan(
spec=spec,
stage_dir=Path("/tmp/stage"),
stage_dir=stage_dir or Path("/tmp/stage"),
slug="demo-abc12",
bundle_subnet="192.168.50.0/24",
bundle_gateway="192.168.50.1",
bundle_ip="192.168.50.2",
bundle_ip=bundle_ip,
machine_name="claude-bottle-demo-abc12",
agent_from_path=Path("/tmp/agent.smolmachine"),
guest_env={},
@@ -59,6 +94,7 @@ def _plan(
proxy_plan=PipelockProxyPlan(
yaml_path=Path("/tmp/pipelock.yaml"),
slug="demo-abc12",
ca_cert_host_path=pipelock_ca_path,
),
git_gate_plan=GitGatePlan(
slug="demo-abc12",
@@ -70,10 +106,11 @@ def _plan(
egress_plan=EgressPlan(
slug="demo-abc12",
routes_path=Path("/tmp/routes.yaml"),
routes=(),
routes=egress_routes,
token_env_map={},
mitmproxy_ca_cert_only_host_path=egress_ca_path,
),
supervise_plan=None,
supervise_plan=supervise_plan,
)
@@ -81,33 +118,58 @@ class TestProvisionPrompt(unittest.TestCase):
def test_cp_uses_smolvm_machine_cp_with_machine_path_syntax(self):
with patch(
"claude_bottle.backend.smolmachines.provision.prompt._smolvm.machine_cp"
) as cp:
) as cp, patch(
"claude_bottle.backend.smolmachines.provision.prompt._smolvm.machine_exec"
):
_prompt.provision_prompt(_plan(), "claude-bottle-demo-abc12")
cp.assert_called_once_with(
"/tmp/state/demo-abc12/agent/prompt.txt",
"claude-bottle-demo-abc12:/root/.claude-bottle-prompt.txt",
"claude-bottle-demo-abc12:/home/node/.claude-bottle-prompt.txt",
)
def test_returns_path_when_agent_has_prompt(self):
with patch(
"claude_bottle.backend.smolmachines.provision.prompt._smolvm.machine_cp"
), patch(
"claude_bottle.backend.smolmachines.provision.prompt._smolvm.machine_exec"
):
r = _prompt.provision_prompt(
_plan(agent_prompt="You are a helpful assistant."),
"claude-bottle-demo-abc12",
)
self.assertEqual("/root/.claude-bottle-prompt.txt", r)
self.assertEqual("/home/node/.claude-bottle-prompt.txt", r)
def test_returns_none_when_agent_has_no_prompt(self):
# The file is still copied (path-must-exist contract);
# only the return value differs.
with patch(
"claude_bottle.backend.smolmachines.provision.prompt._smolvm.machine_cp"
) as cp:
) as cp, patch(
"claude_bottle.backend.smolmachines.provision.prompt._smolvm.machine_exec"
):
r = _prompt.provision_prompt(_plan(agent_prompt=""), "claude-bottle-demo-abc12")
self.assertIsNone(r)
cp.assert_called_once()
def test_chowns_to_node_after_copy(self):
# machine cp lands as root; without the chown, the node user
# can't read its own mode-600 prompt.
with patch(
"claude_bottle.backend.smolmachines.provision.prompt._smolvm.machine_cp"
), patch(
"claude_bottle.backend.smolmachines.provision.prompt._smolvm.machine_exec"
) as ex:
_prompt.provision_prompt(_plan(), "claude-bottle-demo-abc12")
argv_seen = [call.args[1] for call in ex.call_args_list]
self.assertIn(
["chown", "node:node", "/home/node/.claude-bottle-prompt.txt"],
argv_seen,
)
self.assertIn(
["chmod", "600", "/home/node/.claude-bottle-prompt.txt"],
argv_seen,
)
class TestProvisionSkills(unittest.TestCase):
def _patch_host_skill_dir(self, returns: dict[str, str]):
@@ -143,11 +205,11 @@ class TestProvisionSkills(unittest.TestCase):
"claude-bottle-demo-abc12",
)
# mkdir -p the skills dir once + rm -rf per skill = 3 exec calls.
self.assertEqual(3, ex.call_count)
# mkdir -p once + (rm -rf + chown) per skill = 5 exec calls.
self.assertEqual(5, ex.call_count)
mkdir_call = ex.call_args_list[0]
self.assertEqual(
("claude-bottle-demo-abc12", ["mkdir", "-p", "/root/.claude/skills"]),
("claude-bottle-demo-abc12", ["mkdir", "-p", "/home/node/.claude/skills"]),
mkdir_call.args,
)
# Two cp calls, one per skill, into the per-skill subdir.
@@ -155,11 +217,25 @@ class TestProvisionSkills(unittest.TestCase):
cp_targets = {call.args[1] for call in cp.call_args_list}
self.assertEqual(
{
"claude-bottle-demo-abc12:/root/.claude/skills/init-prd",
"claude-bottle-demo-abc12:/root/.claude/skills/verify",
"claude-bottle-demo-abc12:/home/node/.claude/skills/init-prd",
"claude-bottle-demo-abc12:/home/node/.claude/skills/verify",
},
cp_targets,
)
# Each skill gets a chown -R node:node so claude can read it.
chown_argvs = [
call.args[1] for call in ex.call_args_list
if call.args[1][:1] == ["chown"]
]
self.assertEqual(2, len(chown_argvs))
chown_targets = {argv[-1] for argv in chown_argvs}
self.assertEqual(
{
"/home/node/.claude/skills/init-prd",
"/home/node/.claude/skills/verify",
},
chown_targets,
)
def test_skills_dir_overridable_via_env(self):
import os
@@ -197,5 +273,218 @@ class TestProvisionSkills(unittest.TestCase):
_skills.provision_skills(_plan(skills=["init-prd"]), "claude-bottle-demo-abc12")
def _write_self_signed_cert(path: Path) -> None:
"""Drop a real self-signed PEM at `path` so provision_ca's
fingerprint computation (PEM_cert_to_DER_cert + sha256) has
actual bytes to chew on. Generated once per test via openssl."""
subprocess.run(
["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes",
"-keyout", "/dev/null",
"-out", str(path),
"-days", "1",
"-subj", "/CN=test"],
check=True, capture_output=True,
)
class TestProvisionCA(unittest.TestCase):
"""provision_ca selects the right CA cert (egress when the
bottle has routes, else pipelock) and dispatches
machine_cp + machine_exec in the right order."""
def setUp(self):
self._tmp = tempfile.TemporaryDirectory(prefix="cb-prov-ca.")
self.tmp = Path(self._tmp.name)
self.pipelock_ca = self.tmp / "pipelock-ca.pem"
self.egress_ca = self.tmp / "egress-ca.pem"
_write_self_signed_cert(self.pipelock_ca)
_write_self_signed_cert(self.egress_ca)
def tearDown(self):
self._tmp.cleanup()
def test_pipelock_path_when_no_routes(self):
plan = _plan(pipelock_ca_path=self.pipelock_ca)
with patch(
"claude_bottle.backend.smolmachines.provision.ca._smolvm.machine_cp"
) as cp, patch(
"claude_bottle.backend.smolmachines.provision.ca._smolvm.machine_exec"
) as ex:
_ca.provision_ca(plan, "claude-bottle-demo-abc12")
cp.assert_called_once_with(
str(self.pipelock_ca),
"claude-bottle-demo-abc12:" + _ca.AGENT_CA_PATH,
)
argvs = [c.args[1] for c in ex.call_args_list]
self.assertIn(["chmod", "644", _ca.AGENT_CA_PATH], argvs)
self.assertIn(["update-ca-certificates"], argvs)
def test_egress_path_when_routes_declared(self):
plan = _plan(
egress_routes=(EgressRoute(host="api.anthropic.com"),),
egress_ca_path=self.egress_ca,
pipelock_ca_path=self.pipelock_ca,
)
with patch(
"claude_bottle.backend.smolmachines.provision.ca._smolvm.machine_cp"
) as cp, patch(
"claude_bottle.backend.smolmachines.provision.ca._smolvm.machine_exec"
):
_ca.provision_ca(plan, "claude-bottle-demo-abc12")
# When routes are declared, egress is the agent's first hop,
# so egress's CA is the one that gets installed.
cp.assert_called_once_with(
str(self.egress_ca),
"claude-bottle-demo-abc12:" + _ca.AGENT_CA_PATH,
)
def test_dies_when_selected_cert_missing(self):
# Plan claims a pipelock cert at a path that doesn't exist —
# something went wrong in launch's pipelock_tls_init.
plan = _plan(pipelock_ca_path=self.tmp / "does-not-exist.pem")
with patch(
"claude_bottle.backend.smolmachines.provision.ca._smolvm.machine_cp"
), patch(
"claude_bottle.backend.smolmachines.provision.ca._smolvm.machine_exec"
):
with self.assertRaises(SystemExit):
_ca.provision_ca(plan, "claude-bottle-demo-abc12")
class TestProvisionGit(unittest.TestCase):
"""provision_git dispatches two independent passes (cwd .git
copy + gitconfig insteadOf write); each no-ops on its own
when its condition doesn't hold."""
def setUp(self):
self._tmp = tempfile.TemporaryDirectory(prefix="cb-prov-git.")
self.stage = Path(self._tmp.name)
def tearDown(self):
self._tmp.cleanup()
def test_noop_when_no_cwd_and_no_git_entries(self):
with patch(
"claude_bottle.backend.smolmachines.provision.git._smolvm.machine_cp"
) as cp, patch(
"claude_bottle.backend.smolmachines.provision.git._smolvm.machine_exec"
) as ex:
_git.provision_git(
_plan(stage_dir=self.stage), "claude-bottle-demo-abc12",
)
cp.assert_not_called()
ex.assert_not_called()
def test_copies_cwd_git_when_copy_cwd_and_git_present(self):
# Stage a fake host .git dir under user_cwd so the path-
# check in _provision_cwd_git fires.
cwd = self.stage / "cwd"
(cwd / ".git").mkdir(parents=True)
plan = _plan(
copy_cwd=True, user_cwd=str(cwd), stage_dir=self.stage,
)
with patch(
"claude_bottle.backend.smolmachines.provision.git._smolvm.machine_cp"
) as cp, patch(
"claude_bottle.backend.smolmachines.provision.git._smolvm.machine_exec"
) as ex:
_git.provision_git(plan, "claude-bottle-demo-abc12")
cp.assert_called_once_with(
f"{cwd}/.git",
"claude-bottle-demo-abc12:/home/node/workspace/.git",
)
argvs = [c.args[1] for c in ex.call_args_list]
self.assertIn(["mkdir", "-p", "/home/node/workspace"], argvs)
# chown the workspace tree so the agent (node) owns it.
self.assertIn(
["chown", "-R", "node:node", "/home/node/workspace/.git"],
argvs,
)
def test_skips_cwd_when_copy_cwd_false(self):
plan = _plan(copy_cwd=False, stage_dir=self.stage)
with patch(
"claude_bottle.backend.smolmachines.provision.git._smolvm.machine_cp"
) as cp, patch(
"claude_bottle.backend.smolmachines.provision.git._smolvm.machine_exec"
):
_git.provision_git(plan, "claude-bottle-demo-abc12")
cp.assert_not_called()
def test_writes_gitconfig_with_ip_port_form_for_smolmachines(self):
# Smolmachines's TSI-allowlisted guest has no DNS resolver,
# so the insteadOf URL has to be IP+port rather than the
# docker backend's `git-gate` short alias.
plan = _plan(
git=[GitEntry(
Name="claude-bottle",
Upstream="ssh://git@host/repo.git",
IdentityFile="~/.ssh/id_ed25519",
)],
stage_dir=self.stage,
bundle_ip="192.168.99.2",
)
with patch(
"claude_bottle.backend.smolmachines.provision.git._smolvm.machine_cp"
) as cp, patch(
"claude_bottle.backend.smolmachines.provision.git._smolvm.machine_exec"
):
_git.provision_git(plan, "claude-bottle-demo-abc12")
# The staged gitconfig path is whatever NamedTemporaryFile
# picked; we read its contents.
cp_call = cp.call_args
staged_path = Path(cp_call.args[0])
self.assertEqual(self.stage, staged_path.parent)
content = staged_path.read_text()
self.assertIn(
'[url "git://192.168.99.2:9418/claude-bottle.git"]', content,
)
self.assertIn(
"\tinsteadOf = ssh://git@host/repo.git", content,
)
class TestProvisionSupervise(unittest.TestCase):
def test_noop_when_supervise_not_enabled(self):
with patch(
"claude_bottle.backend.smolmachines.provision.supervise._smolvm.machine_exec"
) as ex:
_supervise.provision_supervise(_plan(), "claude-bottle-demo-abc12")
ex.assert_not_called()
def test_calls_claude_mcp_add_when_supervise_enabled(self):
plan = _plan(supervise=True, bundle_ip="192.168.50.2")
with patch(
"claude_bottle.backend.smolmachines.provision.supervise._smolvm.machine_exec",
return_value=SmolvmRunResult(returncode=0, stdout="", stderr=""),
) as ex:
_supervise.provision_supervise(plan, "claude-bottle-demo-abc12")
ex.assert_called_once()
argv = ex.call_args.args[1]
# claude mcp add --scope user --transport http supervise <url>
self.assertEqual(
[
"claude", "mcp", "add",
"--scope", "user",
"--transport", "http",
"supervise",
"http://192.168.50.2:9100/",
],
argv,
)
def test_non_zero_exit_logs_warning_but_does_not_raise(self):
plan = _plan(supervise=True)
with patch(
"claude_bottle.backend.smolmachines.provision.supervise._smolvm.machine_exec",
return_value=SmolvmRunResult(
returncode=1, stdout="", stderr="boom",
),
):
# No raise — the bottle still works without the MCP
# entry, so we log and move on.
_supervise.provision_supervise(plan, "claude-bottle-demo-abc12")
if __name__ == "__main__":
unittest.main()