From 4fada1651bf43d35fbcc2208f7fba7c62e1e2f87 Mon Sep 17 00:00:00 2001 From: didericis Date: Mon, 25 May 2026 05:07:26 -0400 Subject: [PATCH] test(pipelock): integration test for apply_allowlist_change (PRD 0015) Phase 4 of PRD 0015. End-to-end test against real Docker: - Brings up a real pipelock sidecar via the production DockerPipelockProxy bring-up + pipelock_tls_init. - Calls apply_allowlist_change to add a new host. - Polls the live /etc/pipelock.yaml until the new host shows up (bridging the docker-restart window). - Verifies api_allowlist contains both old + new hosts and tls_interception block is preserved. - Smaller cases: invalid hostname raises, missing sidecar raises, fetch_current_allowlist returns one-per-line format. Skipped under GITEA_ACTIONS because pipelock_tls_init bind-mounts a host path that doesn't share fs in the runner, matching the existing pipelock smoke test's skip pattern. Drive-by fix: fetch_current_yaml now uses `docker cp` (daemon-API tarball copy) instead of `docker exec cat` because the pipelock image is distroless and has no shell utilities. Co-Authored-By: Claude Opus 4.7 --- .../backend/docker/pipelock_apply.py | 39 +++-- tests/integration/test_pipelock_apply.py | 152 ++++++++++++++++++ 2 files changed, 180 insertions(+), 11 deletions(-) create mode 100644 tests/integration/test_pipelock_apply.py diff --git a/claude_bottle/backend/docker/pipelock_apply.py b/claude_bottle/backend/docker/pipelock_apply.py index 641c5ca..9a27cd2 100644 --- a/claude_bottle/backend/docker/pipelock_apply.py +++ b/claude_bottle/backend/docker/pipelock_apply.py @@ -65,19 +65,36 @@ def render_allowlist_content(hosts: list[str]) -> str: def fetch_current_yaml(slug: str) -> str: - """Read the live /etc/pipelock.yaml from the running pipelock - sidecar. Raises PipelockApplyError if the read fails.""" + """Read the live /etc/pipelock.yaml from the pipelock sidecar. + + Uses `docker cp` (not `docker exec cat`) because the pipelock + image is distroless and has no shell utilities. `docker cp` is a + daemon-API tarball copy — works on stopped containers too, and + doesn't need anything in the container's PATH. + + Raises PipelockApplyError if the read fails.""" container = pipelock_container_name(slug) - r = subprocess.run( - ["docker", "exec", container, "cat", PIPELOCK_YAML_IN_CONTAINER], - capture_output=True, text=True, check=False, - ) - if r.returncode != 0: - raise PipelockApplyError( - f"could not read pipelock.yaml from {container}: " - f"{(r.stderr or '').strip() or 'container not running?'}" + fd, tmp_path = tempfile.mkstemp(prefix="cb-pipelock-fetch.", suffix=".yaml") + os.close(fd) + try: + r = subprocess.run( + [ + "docker", "cp", + f"{container}:{PIPELOCK_YAML_IN_CONTAINER}", tmp_path, + ], + capture_output=True, text=True, check=False, ) - return r.stdout + if r.returncode != 0: + raise PipelockApplyError( + f"could not fetch pipelock.yaml from {container}: " + f"{(r.stderr or '').strip() or 'container not running?'}" + ) + return Path(tmp_path).read_text() + finally: + try: + Path(tmp_path).unlink() + except OSError: + pass def fetch_current_allowlist(slug: str) -> str: diff --git a/tests/integration/test_pipelock_apply.py b/tests/integration/test_pipelock_apply.py new file mode 100644 index 0000000..3f0181b --- /dev/null +++ b/tests/integration/test_pipelock_apply.py @@ -0,0 +1,152 @@ +"""Integration: drive `apply_allowlist_change` against a real +pipelock sidecar (PRD 0015). + +Brings up a real pipelock sidecar (via the production DockerPipelockProxy +bring-up), calls apply_allowlist_change to swap the api_allowlist, +restarts pipelock, and verifies the running container now serves the +new yaml. + +Setup uses pipelock_tls_init which bind-mounts a host path into a +one-shot pipelock container — that doesn't work in DinD, so the test +skips under GITEA_ACTIONS the same way the existing pipelock smoke +test does. +""" + +from __future__ import annotations + +import dataclasses +import os +import shutil +import subprocess +import tempfile +import time +import unittest +from pathlib import Path + +from claude_bottle.backend.docker.network import ( + network_create_egress, + network_create_internal, + network_remove, +) +from claude_bottle.backend.docker.pipelock import ( + DockerPipelockProxy, + pipelock_container_name, + pipelock_tls_init, +) +from claude_bottle.backend.docker.pipelock_apply import ( + PipelockApplyError, + apply_allowlist_change, + fetch_current_allowlist, + fetch_current_yaml, +) +from claude_bottle.yaml_subset import parse_yaml_subset +from tests._docker import skip_unless_docker +from tests.fixtures import fixture_minimal + + +@skip_unless_docker() +@unittest.skipIf( + os.environ.get("GITEA_ACTIONS") == "true", + "skipped under act_runner: pipelock_tls_init uses a host bind mount " + "that doesn't share fs with the runner container", +) +class TestPipelockApply(unittest.TestCase): + def setUp(self): + self.slug = f"cb-test-pla-{os.getpid()}-{int(time.time())}" + self.sidecar_name = "" + self.internal_net = "" + self.egress_net = "" + self.work_dir = Path(tempfile.mkdtemp(prefix="pipelock-apply.")) + + def tearDown(self): + if self.sidecar_name: + DockerPipelockProxy().stop(self.sidecar_name) + for n in (self.internal_net, self.egress_net): + if n: + network_remove(n) + shutil.rmtree(self.work_dir, ignore_errors=True) + + def _bring_up(self) -> None: + proxy = DockerPipelockProxy() + prep = proxy.prepare(fixture_minimal().bottles["dev"], self.slug, self.work_dir) + self.internal_net = network_create_internal(self.slug) + self.egress_net = network_create_egress(self.slug) + ca_cert_host, ca_key_host = pipelock_tls_init(self.work_dir) + plan = dataclasses.replace( + prep, + internal_network=self.internal_net, + egress_network=self.egress_net, + ca_cert_host_path=ca_cert_host, + ca_key_host_path=ca_key_host, + ) + self.sidecar_name = proxy.start(plan) + self.assertEqual(pipelock_container_name(self.slug), self.sidecar_name) + # Wait until docker exec succeeds — the container is up but + # pipelock may still be initializing. fetch_current_yaml is + # itself a docker exec, so retrying it doubles as a readiness + # probe. + deadline = time.monotonic() + 15.0 + while time.monotonic() < deadline: + try: + fetch_current_yaml(self.slug) + return + except PipelockApplyError: + pass + time.sleep(0.25) + raise AssertionError("pipelock sidecar never became reachable") + + def _wait_for_yaml(self, contains: str, *, deadline_s: float = 15.0) -> str: + """Poll docker exec until /etc/pipelock.yaml contains `contains`, + returning the yaml. Used to bridge the docker-restart window.""" + deadline = time.monotonic() + deadline_s + while time.monotonic() < deadline: + try: + yaml = fetch_current_yaml(self.slug) + if contains in yaml: + return yaml + except PipelockApplyError: + pass + time.sleep(0.25) + self.fail(f"never saw {contains!r} in /etc/pipelock.yaml") + + def test_apply_swaps_api_allowlist(self): + self._bring_up() + + initial_yaml = fetch_current_yaml(self.slug) + # fixture_minimal yields the baked-in DEFAULT_ALLOWLIST in + # pipelock.py; api.anthropic.com is in there. + self.assertIn("api.anthropic.com", initial_yaml) + + new_content = "api.anthropic.com\nnew-host.example\n" + before, after = apply_allowlist_change(self.slug, new_content) + self.assertIn("api.anthropic.com", before) + self.assertNotIn("new-host.example", before) + self.assertIn("new-host.example", after) + + updated = self._wait_for_yaml("new-host.example") + cfg = parse_yaml_subset(updated) + self.assertIn("new-host.example", cfg["api_allowlist"]) # type: ignore[operator] + self.assertIn("api.anthropic.com", cfg["api_allowlist"]) # type: ignore[operator] + # tls_interception block (set up by the production prepare + # via pipelock_build_config) is preserved across the swap. + self.assertIn("tls_interception", cfg) + + def test_apply_with_invalid_host_raises(self): + self._bring_up() + with self.assertRaises(PipelockApplyError): + apply_allowlist_change(self.slug, "host with space.example\n") + + def test_fetch_current_allowlist_renders_one_per_line(self): + self._bring_up() + listing = fetch_current_allowlist(self.slug) + self.assertTrue(listing.endswith("\n")) + self.assertIn("api.anthropic.com\n", listing) + + def test_apply_against_missing_sidecar_raises(self): + # Don't bring up — the slug points at nothing. + with self.assertRaises(PipelockApplyError): + apply_allowlist_change(self.slug, "x.example\n") + + +if __name__ == "__main__": + unittest.main()