diff --git a/claude_bottle/backend/docker/backend.py b/claude_bottle/backend/docker/backend.py index 6d2f5c4..0a34288 100644 --- a/claude_bottle/backend/docker/backend.py +++ b/claude_bottle/backend/docker/backend.py @@ -39,6 +39,7 @@ class DockerBottleBackend(BottleBackend): (default).""" name = "docker" + _proxy: pipelock.PipelockProxy = pipelock.PipelockProxy() def prepare(self, spec: BottleSpec, *, stage_dir: Path) -> DockerBottlePlan: """Resolve names, validate, write scratch files. No Docker @@ -128,49 +129,10 @@ class DockerBottleBackend(BottleBackend): ) def prepare_proxy(self, spec: BottleSpec, yaml_path: Path) -> None: - """Write the pipelock proxy's yaml config (mode 600) to - `yaml_path` for the sidecar to consume when it boots in - `launch`. Carries the effective allowlist (bottle.egress.allowlist - UNION claude-bottle defaults UNION ssh hostnames), a fixed - listen port, strict mode + forward_proxy + DLP defaults + - scan_env. Deliberately contains no env values, no secrets, no - per-agent customization beyond the hostname list.""" + """Delegate to PipelockProxy to write the sidecar's yaml + config. Stage-only: no Docker resources created yet.""" bottle_name = spec.manifest.agents[spec.agent_name].bottle - allowlist = pipelock.pipelock_effective_allowlist(spec.manifest, bottle_name) - trusted = pipelock.pipelock_bottle_ssh_trusted_domains(spec.manifest, bottle_name) - ip_cidrs = pipelock.pipelock_bottle_ssh_ip_cidrs(spec.manifest, bottle_name) - - lines: list[str] = [] - lines.append("version: 1") - lines.append("mode: strict") - lines.append("enforce: true") - lines.append("") - lines.append("# Hostnames the agent is allowed to reach. Effective list is") - lines.append("# claude-bottle defaults UNION bottle.egress.allowlist (sorted, deduped).") - lines.append("api_allowlist:") - for h in allowlist: - lines.append(f' - "{h}"') - lines.append("") - lines.append("forward_proxy:") - lines.append(" enabled: true") - lines.append("") - if trusted: - lines.append("trusted_domains:") - for td in trusted: - lines.append(f' - "{td}"') - lines.append("") - if ip_cidrs: - lines.append("ssrf:") - lines.append(" ip_allowlist:") - for cidr in ip_cidrs: - lines.append(f' - "{cidr}"') - lines.append("") - lines.append("dlp:") - lines.append(" include_defaults: true") - lines.append(" scan_env: true") - - yaml_path.write_text("\n".join(lines) + "\n") - yaml_path.chmod(0o600) + self._proxy.prepare(spec.manifest, bottle_name, yaml_path) @contextmanager def launch(self, plan: BottlePlan) -> Iterator[DockerBottle]: diff --git a/claude_bottle/pipelock.py b/claude_bottle/pipelock.py index ba94416..10afcce 100644 --- a/claude_bottle/pipelock.py +++ b/claude_bottle/pipelock.py @@ -120,6 +120,60 @@ def pipelock_allowlist_summary(manifest: Manifest, bottle_name: str) -> str: +# --- Proxy class ----------------------------------------------------------- + + +class PipelockProxy: + """The pipelock egress proxy. Encapsulates the YAML-config + generation (and is the natural home for any future proxy-level + state). Backends that use pipelock hold a PipelockProxy instance + and delegate the prepare step to it.""" + + def prepare(self, manifest: Manifest, bottle_name: str, yaml_path: Path) -> None: + """Write the pipelock yaml config (mode 600) to `yaml_path` + for the sidecar to consume when it boots. Carries the + effective allowlist (bottle.egress.allowlist UNION + claude-bottle defaults UNION ssh hostnames), a fixed listen + port, strict mode + forward_proxy + DLP defaults + scan_env. + Deliberately contains no env values, no secrets, no per-agent + customization beyond the hostname list.""" + allowlist = pipelock_effective_allowlist(manifest, bottle_name) + trusted = pipelock_bottle_ssh_trusted_domains(manifest, bottle_name) + ip_cidrs = pipelock_bottle_ssh_ip_cidrs(manifest, bottle_name) + + lines: list[str] = [] + lines.append("version: 1") + lines.append("mode: strict") + lines.append("enforce: true") + lines.append("") + lines.append("# Hostnames the agent is allowed to reach. Effective list is") + lines.append("# claude-bottle defaults UNION bottle.egress.allowlist (sorted, deduped).") + lines.append("api_allowlist:") + for h in allowlist: + lines.append(f' - "{h}"') + lines.append("") + lines.append("forward_proxy:") + lines.append(" enabled: true") + lines.append("") + if trusted: + lines.append("trusted_domains:") + for td in trusted: + lines.append(f' - "{td}"') + lines.append("") + if ip_cidrs: + lines.append("ssrf:") + lines.append(" ip_allowlist:") + for cidr in ip_cidrs: + lines.append(f' - "{cidr}"') + lines.append("") + lines.append("dlp:") + lines.append(" include_defaults: true") + lines.append(" scan_env: true") + + yaml_path.write_text("\n".join(lines) + "\n") + yaml_path.chmod(0o600) + + # --- Sidecar lifecycle ----------------------------------------------------- diff --git a/tests/test_pipelock_sidecar_smoke.py b/tests/test_pipelock_sidecar_smoke.py index 873e1c2..1e5bfea 100644 --- a/tests/test_pipelock_sidecar_smoke.py +++ b/tests/test_pipelock_sidecar_smoke.py @@ -12,9 +12,7 @@ import unittest import urllib.request from pathlib import Path -from claude_bottle.backend import BottleSpec -from claude_bottle.backend.docker import DockerBottleBackend -from claude_bottle.pipelock import PIPELOCK_IMAGE +from claude_bottle.pipelock import PIPELOCK_IMAGE, PipelockProxy from tests._docker import skip_unless_docker from tests.fixtures import fixture_minimal @@ -40,14 +38,7 @@ class TestPipelockSidecarSmoke(unittest.TestCase): ) def test_smoke(self): yaml_path = self.work_dir / "pipelock.yaml" - spec = BottleSpec( - manifest=fixture_minimal(), - agent_name="demo", - copy_cwd=False, - user_cwd="/tmp", - forward_oauth_token=False, - ) - DockerBottleBackend().prepare_proxy(spec, yaml_path) + PipelockProxy().prepare(fixture_minimal(), "dev", yaml_path) create = subprocess.run( [ diff --git a/tests/test_pipelock_yaml.py b/tests/test_pipelock_yaml.py index 3647921..9514110 100644 --- a/tests/test_pipelock_yaml.py +++ b/tests/test_pipelock_yaml.py @@ -1,34 +1,21 @@ -"""Unit: DockerBottleBackend.prepare_proxy — produces a pipelock YAML -config containing the expected top-level keys and per-bottle entries. -We don't fully parse YAML; we grep for content shape.""" +"""Unit: PipelockProxy.prepare — produces a pipelock YAML config +containing the expected top-level keys and per-bottle entries. We +don't fully parse YAML; we grep for content shape.""" import os import tempfile import unittest from pathlib import Path -from claude_bottle.backend import BottleSpec -from claude_bottle.backend.docker import DockerBottleBackend from claude_bottle.manifest import Manifest +from claude_bottle.pipelock import PipelockProxy from tests.fixtures import fixture_minimal, fixture_with_ssh -def _spec(manifest: Manifest) -> BottleSpec: - """Construct a minimal BottleSpec around a fixture manifest. The - fixtures all define an agent named 'demo' on a bottle named 'dev'.""" - return BottleSpec( - manifest=manifest, - agent_name="demo", - copy_cwd=False, - user_cwd="/tmp", - forward_oauth_token=False, - ) - - -class TestPrepareProxyYaml(unittest.TestCase): +class TestPipelockProxyPrepare(unittest.TestCase): def setUp(self): self.out_dir = Path(tempfile.mkdtemp()) - self.backend = DockerBottleBackend() + self.proxy = PipelockProxy() def tearDown(self): import shutil @@ -36,7 +23,7 @@ class TestPrepareProxyYaml(unittest.TestCase): def test_minimal(self): yaml_path = self.out_dir / "min.yaml" - self.backend.prepare_proxy(_spec(fixture_minimal()), yaml_path) + self.proxy.prepare(fixture_minimal(), "dev", yaml_path) content = yaml_path.read_text() self.assertIn("mode: strict", content) self.assertIn("enforce: true", content) @@ -54,7 +41,7 @@ class TestPrepareProxyYaml(unittest.TestCase): def test_ssh_blocks(self): yaml_path = self.out_dir / "ssh.yaml" - self.backend.prepare_proxy(_spec(fixture_with_ssh()), yaml_path) + self.proxy.prepare(fixture_with_ssh(), "dev", yaml_path) content = yaml_path.read_text() self.assertIn("trusted_domains:", content) self.assertIn("github.com", content) @@ -78,7 +65,7 @@ class TestPrepareProxyYaml(unittest.TestCase): "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}}, }) yaml_path = self.out_dir / "secret.yaml" - self.backend.prepare_proxy(_spec(manifest), yaml_path) + self.proxy.prepare(manifest, "dev", yaml_path) content = yaml_path.read_text() self.assertNotIn("literal-value-should-not-appear", content) self.assertNotIn("MY_SECRET", content) @@ -86,7 +73,7 @@ class TestPrepareProxyYaml(unittest.TestCase): def test_file_mode_is_600(self): yaml_path = self.out_dir / "min.yaml" - self.backend.prepare_proxy(_spec(fixture_minimal()), yaml_path) + self.proxy.prepare(fixture_minimal(), "dev", yaml_path) mode = os.stat(yaml_path).st_mode & 0o777 self.assertEqual(0o600, mode)