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

chore: remove all pipelock references from tests, docs, and non-pipelock source
lint / lint (push) Failing after 1m26s
Details
test / unit (pull_request) Failing after 35s
Details
test / integration (pull_request) Successful in 44s
Details

Browse Source 
- Strip pipelock from all unit and integration test fixtures:
  proxy_plan fields removed from DockerBottlePlan/SmolmachinesBottlePlan
  constructors; pipelock-specific test classes deleted or renamed
- Update test_sidecar_init: remove test_pipelock_loses_egress_tokens,
  rename "pipelock" daemon fixtures to "git-gate" throughout
- Remove test_pipelock_binary_present_and_versioned from integration test
- Remove test_pipelock_answers_on_bundle_ip from smolmachines launch test
- Update _SANDBOX_BLOCK_MARKERS: remove "pipelock" marker (egress blocks)
- Dockerfile.sidecars: remove pipelock build stage and COPY; update layout
  comments and port table
- egress_entrypoint.sh: update comments now that egress is sole proxy
- Clean up pipelock references in comments/docstrings across backend,
  network, manifest, supervise, git_gate, yaml_subset, agent_provider,
  sidecar_bundle, sidecar_init, egress_addon_core modules

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
didericis (claude)
2026-06-04 21:54:06 +00:00
parent bbd6ec85ac
commit a59da9921e
53 changed files with 266 additions and 945 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/integration/test_sandbox_escape.py
+20 -24
View File
@@ -53,7 +53,7 @@ _FAKE_SECRETS = {
@skip_unless_docker()
@unittest.skipIf(
os.environ.get("GITEA_ACTIONS") == "true",
"skipped under act_runner: pipelock_tls_init uses a host bind mount "
"skipped under act_runner: egress_tls_init uses a host bind mount "
"the runner container can't see, and the network topology hides "
"sibling-sidecar visibility — same constraint as the other "
"bottle-bringup integration tests",
@@ -256,14 +256,11 @@ class TestSandboxEscape(unittest.TestCase):
# ---- attack 3: HTTP exfil shapes ---------------------------------
# Sandbox-block signature: pipelock / egress return HTTP 403 on
# policy reject; the response body carries a recognizable
# marker. Egress's reject message starts `"egress: host '...'
# is not in the bottle's egress.routes allowlist"`; pipelock's
# DLP rejects start `"blocked: "` (e.g.
# `"blocked: DLP match: Anthropic API Key (critical)"`,
# `"blocked: request body contains secret"`).
_SANDBOX_BLOCK_MARKERS = ("egress:", "pipelock", "blocked:")
# Sandbox-block signature: egress returns HTTP 403 on policy
# reject; the response body carries a recognizable marker.
# Egress's reject message starts with `"egress: host '...'
# is not in the bottle's egress.routes allowlist"`.
_SANDBOX_BLOCK_MARKERS = ("egress:", "blocked:")
def _assert_sandbox_block(self, label: str, r: object) -> None: # type: ignore
"""A real sandbox block produces an HTTP 403 with a
@@ -286,7 +283,7 @@ class TestSandboxEscape(unittest.TestCase):
self.assertTrue(
has_marker and http_code == "403",
f"{label}: expected a sandbox-level block (HTTP 403 with "
f"egress/pipelock marker), got HTTP {http_code!r}. "
f"egress marker), got HTTP {http_code!r}. "
f"If the response came from the actual upstream, the "
f"secret REACHED the network — that's the leak this "
f"test exists to catch. body={body!r} "
@@ -297,9 +294,8 @@ class TestSandboxEscape(unittest.TestCase):
"""The agent tries to leak `$TEST_SECRET_ANTHROPIC` to an
ALLOWLISTED destination (api.anthropic.com) via four
different request shapes. Each must be blocked by a
sandbox layer (pipelock DLP / egress filter / etc.) —
producing an HTTP 403 with an `egress:` or `pipelock`
marker in the response body.
sandbox layer (egress allowlist filter) — producing an
HTTP 403 with an `egress:` marker in the response body.
Anything else means the request reached upstream and the
secret leaked, even if the upstream responded with its
@@ -310,10 +306,10 @@ class TestSandboxEscape(unittest.TestCase):
remediation lands as its own PRD before this test merges.
DON'T mark expectedFailure to silence it.
Destination note: we use `raw.githubusercontent.com`, one
of the DEFAULT_ALLOWLIST hosts. It is not route-configured
for pipelock TLS passthrough, so pipelock MITMs it and the
DLP scan applies, which is what this attack exercises."""
Destination note: we use `raw.githubusercontent.com`, which
is not in the bottle's egress.routes allowlist, so egress
blocks every request to it regardless of the secret's
location in the request."""
# Capture HTTP code via curl's -w; don't use --fail so
# we get the response body even on 4xx.
url_base = "https://raw.githubusercontent.com"
@@ -352,13 +348,13 @@ class TestSandboxEscape(unittest.TestCase):
def test_4_dns_exfil_blocked(self) -> None:
"""Two sub-attacks against DNS:
4a — crafted subdomain that pipelock would resolve. The
hostname `<SECRET>.api.anthropic.com` looks "under"
the allowlisted apex but pipelock's allowlist is
exact-match — it should reject the host BEFORE
issuing the DNS query, so the secret never reaches
an external resolver.
4b — direct DNS query bypassing pipelock entirely. The
4a — crafted subdomain attack. The hostname
`<SECRET>.api.anthropic.com` looks "under" the
allowlisted apex but egress's allowlist is
exact-match — it rejects the host before issuing
a DNS query, so the secret never reaches an
external resolver.
4b — direct DNS query bypassing egress entirely. The
agent's internal network has no default gateway;
even an explicit resolver like 8.8.8.8 should be
unreachable. Confirms the network isolation is
tests/integration/test_sidecar_bundle_compose.py
+12 -20
View File
@@ -2,8 +2,8 @@
Verifies that flipping `BOT_BOTTLE_SIDECAR_BUNDLE=1` produces a
working bottle: `docker compose up` brings the agent + bundle pair
online, the four daemons inside the bundle bind their ports, and
the agent can reach pipelock + supervise via the bundle's network
online, the daemons inside the bundle bind their ports, and the
agent can reach egress + supervise via the bundle's network
aliases (no agent-side config changes between flag positions).
Skipped under GITEA_ACTIONS — the bundle image is a multi-stage
@@ -27,11 +27,9 @@ from tests._docker import skip_unless_docker
def _manifest() -> Manifest:
"""Bottle with supervise on so the bundle exercises three of
the four daemons (pipelock, egress, supervise). Git is off
because a meaningful git-gate test needs a real upstream and
SSH keys — out of scope for a bundle smoke. Egress is
implicitly on as pipelock's upstream regardless of routes."""
"""Bottle with supervise on so the bundle exercises egress +
supervise. Git is off because a meaningful git-gate test needs
a real upstream and SSH keys — out of scope for a bundle smoke."""
return Manifest.from_json_obj({
"bottles": {
"dev": {
@@ -68,21 +66,16 @@ class TestSidecarBundleCompose(unittest.TestCase):
plan = backend.prepare(spec, stage_dir=stage_dir)
with backend.launch(plan) as bottle:
# The agent's HTTPS_PROXY URL (resolved at
# renderer-time, unchanged from the legacy
# shape) should reach pipelock inside the
# bundle. We probe by asking for the proxy's
# listening port from inside the agent.
# renderer-time) should reach egress inside
# the bundle. A bare CONNECT with no upstream
# URL gets rejected with 400 or 405 but proves
# the listener is alive at the alias.
probe = bottle.exec(
"set -eu\n"
"echo HTTPS_PROXY=$HTTPS_PROXY\n"
"PORT=$(echo \"$HTTPS_PROXY\" | sed -E 's|.*:([0-9]+).*|\\1|')\n"
"HOST=$(echo \"$HTTPS_PROXY\" | sed -E 's|http://([^:]+):.*|\\1|')\n"
"echo HOST=$HOST PORT=$PORT\n"
# nc is not in the agent image but curl is —
# a CONNECT with no upstream URL will get
# rejected by pipelock with 400 or 405 but
# confirms the listener is alive at the
# alias.
"curl -sS --max-time 5 -o /dev/null -w 'http=%{http_code}\\n' "
" \"http://$HOST:$PORT/\" || true\n"
)
@@ -98,11 +91,10 @@ class TestSidecarBundleCompose(unittest.TestCase):
shutil.rmtree(stage_dir, ignore_errors=True)
self.assertEqual(0, probe.returncode, msg=probe.stderr)
# pipelock answered SOMETHING — any 4xx is fine, just proves
# the bundle's pipelock daemon is listening at the
# `pipelock` alias on port 8888 (or whatever the env says).
# egress answered SOMETHING — any 4xx is fine, just proves
# the egress daemon is listening at the proxy address.
self.assertIn("http=", probe.stdout,
f"no HTTP response from pipelock: {probe.stdout!r}")
f"no HTTP response from egress: {probe.stdout!r}")
# supervise's /health endpoint exists (PRD 0013); it should
# answer 200 or similar — anything non-empty proves the
# third daemon's alias resolves to the same bundle.
tests/integration/test_sidecar_bundle_image.py
+6 -11
View File
@@ -1,15 +1,15 @@
"""Integration: PRD 0024 chunk 1 — the sidecar bundle image builds
and the four daemon binaries are present + executable inside it.
and the daemon binaries are present + executable inside it.
This test does NOT exercise the daemons running against real
config (pipelock.yaml, routes.yaml, etc) — that lands in chunk 2
when the renderer wires the bundle into compose. What we verify
here is the chunk-1 contract:
config (routes.yaml, etc) — that lands in chunk 2 when the
renderer wires the bundle into compose. What we verify here is
the chunk-1 contract:
- Dockerfile.sidecars builds (multi-stage works, base layers
pull, COPYs resolve).
- pipelock, gitleaks, mitmdump are at the documented paths and
answer `--version`.
- gitleaks, mitmdump are at the documented paths and answer
`--version`.
- The Python init at /app/sidecar_init.py runs and prints the
expected "no daemons selected" line when the supervisor is
pointed at an empty daemon set.
@@ -74,11 +74,6 @@ class TestSidecarBundleImage(unittest.TestCase):
)
return proc.returncode, proc.stdout.decode("utf-8", errors="replace")
def test_pipelock_binary_present_and_versioned(self):
rc, out = self._run_in_image("/usr/local/bin/pipelock", "version")
self.assertEqual(0, rc, msg=out)
self.assertIn("pipelock version", out)
def test_gitleaks_binary_present_and_versioned(self):
rc, out = self._run_in_image("/usr/bin/gitleaks", "version")
self.assertEqual(0, rc, msg=out)
tests/integration/test_smolmachines_bundle_bringup.py
+3 -7
View File
@@ -81,13 +81,9 @@ class TestBundleBringup(unittest.TestCase):
subnet=subnet,
gateway=gateway,
bundle_ip=bundle_ip,
# Only run the pipelock daemon for this smoke — it's
# the lightest of the four and doesn't need bind
# mounts beyond what we'd skip without
# BOT_BOTTLE_SIDECAR_DAEMONS. (The init
# supervisor will exit if pipelock fails to find its
# yaml — that's expected here; we just need the
# container to land on the network at the right IP.)
# Empty daemons_csv → init exits "no daemons selected"
# immediately. We just need the container to land on
# the network at the right IP before it exits.
daemons_csv="", # empty → init exits "no daemons selected"
)
start_bundle(spec)
tests/integration/test_smolmachines_launch.py
-26
View File
@@ -124,32 +124,6 @@ class TestSmolmachinesLaunch(unittest.TestCase):
f"expected a connect-refusal message; got: {r.stdout!r}",
)
def test_pipelock_answers_on_bundle_ip(self):
# Chunk 4b: the bundle's pipelock daemon is now actually
# running (was daemons_csv="" in chunks 2d/3). From inside
# the guest, a TCP connect to <bundle-ip>:8888 must succeed
# — distinct from the egress-port-bypass probe below where
# the connect must FAIL.
#
# We don't try to speak proxy protocol here — pipelock will
# 4xx a bare GET — we just verify the socket answers.
r = self.bottle.exec(
f"wget -T 5 -t 1 -O - http://{self.plan.bundle_ip}:8888/ "
"2>&1 || true"
)
# Any HTTP response (even a 4xx) proves pipelock is up.
# "connection refused" / "unable to connect" / "timed out"
# would mean it isn't.
msg = r.stdout.lower()
self.assertNotIn(
"connection refused", msg,
f"pipelock connect refused — daemon not listening? {r.stdout!r}",
)
self.assertNotIn(
"timed out", msg,
f"pipelock connect timed out: {r.stdout!r}",
)
def test_prompt_file_lands_in_guest(self):
# provision_prompt copies the host-side prompt.txt into the
# guest at /root/.bot-bottle-prompt.txt. The content
tests/unit/test_backend_selection.py
+1 -1
View File
@@ -57,7 +57,7 @@ class TestEnumerateActiveAgents(unittest.TestCase):
def test_concatenates_per_backend(self):
a = ActiveAgent(
backend_name="docker", slug="a-1", agent_name="impl",
started_at="", services=("pipelock",),
started_at="", services=("egress",),
)
b = ActiveAgent(
backend_name="smolmachines", slug="b-2", agent_name="research",
tests/unit/test_compose.py
+22 -55
View File
@@ -32,7 +32,6 @@ from bot_bottle.egress import (
)
from bot_bottle.git_gate import GitGatePlan, GitGateUpstream
from bot_bottle.manifest import Manifest
from bot_bottle.pipelock import PipelockProxyPlan
from bot_bottle.supervise import SupervisePlan
from bot_bottle.workspace import workspace_plan
@@ -80,18 +79,6 @@ def _spec(*, supervise: bool, with_git: bool, with_egress: bool) -> BottleSpec:
)
def _proxy_plan() -> PipelockProxyPlan:
return PipelockProxyPlan(
yaml_path=STATE / "pipelock.yaml",
slug=SLUG,
internal_network=f"bot-bottle-net-{SLUG}",
internal_network_cidr="10.1.2.0/24",
egress_network=f"bot-bottle-egress-{SLUG}",
ca_cert_host_path=STATE / "pipelock-ca" / "ca.pem",
ca_key_host_path=STATE / "pipelock-ca" / "ca-key.pem",
)
def _git_gate_plan(upstreams: tuple[GitGateUpstream, ...] = ()) -> GitGatePlan:
return GitGatePlan(
slug=SLUG,
@@ -119,8 +106,6 @@ def _egress_plan(routes: tuple[EgressRoute, ...] = ()) -> EgressPlan:
egress_network=f"bot-bottle-egress-{SLUG}",
mitmproxy_ca_host_path=STATE / "egress-ca" / "mitmproxy-ca.pem",
mitmproxy_ca_cert_only_host_path=STATE / "egress-ca" / "ca.pem",
pipelock_ca_host_path=STATE / "pipelock-ca" / "ca.pem",
pipelock_proxy_url="http://127.0.0.1:8888",
)
@@ -178,7 +163,6 @@ def _plan(
env_file=Path("/dev/null"), # exists, size 0 → renderer skips env_file
forwarded_env={"CLAUDE_CODE_OAUTH_TOKEN": "x"},
prompt_file=STAGE / "prompt",
proxy_plan=_proxy_plan(),
git_gate_plan=_git_gate_plan(upstreams),
egress_plan=_egress_plan(routes),
supervise_plan=_supervise_plan() if supervise else None,
@@ -233,16 +217,15 @@ class TestAgentAlwaysPresent(unittest.TestCase):
s = bottle_plan_to_compose(_plan())["services"]["agent"]
self.assertEqual({"internal"}, set(s["networks"].keys()))
def test_agent_proxy_via_pipelock_when_no_egress(self):
s = bottle_plan_to_compose(_plan(with_egress=False))["services"]["agent"]
env = s["environment"]
# Looking for HTTPS_PROXY pointing at pipelock's container name.
proxy_lines = [e for e in env if e.startswith("HTTPS_PROXY=")]
self.assertEqual(1, len(proxy_lines))
self.assertEqual(
"HTTPS_PROXY=http://pipelock:8888",
proxy_lines[0],
)
def test_agent_proxy_always_via_egress(self):
for with_egress in (False, True):
with self.subTest(with_egress=with_egress):
s = bottle_plan_to_compose(
_plan(with_egress=with_egress)
)["services"]["agent"]
proxy_lines = [e for e in s["environment"] if e.startswith("HTTPS_PROXY=")]
self.assertEqual(1, len(proxy_lines))
self.assertEqual("HTTPS_PROXY=http://egress:9099", proxy_lines[0])
def test_agent_proxy_via_egress_when_egress_present(self):
s = bottle_plan_to_compose(_plan(with_egress=True))["services"]["agent"]
@@ -306,9 +289,9 @@ class TestAgentAlwaysPresent(unittest.TestCase):
class TestSidecarBundleShape(unittest.TestCase):
"""The compose renderer emits exactly one `sidecars` service in
place of the four daemons it owns (pipelock + egress + git-gate
+ supervise). PRD 0024 chunk 5 dropped the legacy four-sidecar
shape entirely, so the bundle is the only thing exercised here."""
place of the daemons it owns (egress + git-gate + supervise).
PRD 0024 chunk 5 dropped the legacy four-sidecar shape entirely,
so the bundle is the only thing exercised here."""
def _render(self, **plan_kwargs: object) -> Any: # type: ignore
return bottle_plan_to_compose(_plan(**plan_kwargs)) # type: ignore
@@ -335,13 +318,10 @@ class TestSidecarBundleShape(unittest.TestCase):
sc = self._render()["services"]["sidecars"]
self.assertEqual({"internal", "egress"}, set(sc["networks"].keys()))
def test_internal_aliases_cover_pipelock_and_egress_shortnames(self):
# The agent's HTTPS_PROXY url references either `egress` or
# `pipelock`. Both must resolve to the bundle.
def test_internal_aliases_include_egress_shortname(self):
sc = self._render()["services"]["sidecars"]
aliases = set(sc["networks"]["internal"]["aliases"])
self.assertIn("egress", aliases)
self.assertIn("pipelock", aliases)
def test_internal_aliases_omit_inactive_sidecars(self):
# With no git-gate / supervise, those names are NOT aliased
@@ -359,16 +339,13 @@ class TestSidecarBundleShape(unittest.TestCase):
self.assertIn("supervise", aliases)
def test_daemons_csv_lists_only_active(self):
# Egress + pipelock are always in the daemon set even when
# the bottle has no routes (egress falls back to regular@9099
# and is just unused; cheaper than special-casing).
sc = self._render()["services"]["sidecars"]
daemons = {
line.split("=", 1)[1]
for line in sc["environment"]
if line.startswith("BOT_BOTTLE_SIDECAR_DAEMONS=")
}
self.assertEqual({"egress,pipelock"}, daemons)
self.assertEqual({"egress"}, daemons)
def test_daemons_csv_expands_with_optional_sidecars(self):
sc = self._render(with_git=True, supervise=True)["services"]["sidecars"]
@@ -379,13 +356,13 @@ class TestSidecarBundleShape(unittest.TestCase):
else:
self.fail("BOT_BOTTLE_SIDECAR_DAEMONS not in env")
self.assertEqual(
["egress", "pipelock", "git-gate", "supervise"],
["egress", "git-gate", "supervise"],
csv.split(","),
)
def test_bundle_env_does_not_set_https_proxy(self):
# HTTPS_PROXY at the container level would route git-gate's
# git fetches through pipelock. Scoping it to mitmdump is
# git fetches through the proxy. Scoping it to mitmdump is
# the job of egress_entrypoint.sh; the bundle env must not
# leak it.
sc = self._render(with_egress=True)["services"]["sidecars"]
@@ -397,22 +374,15 @@ class TestSidecarBundleShape(unittest.TestCase):
f"bundle env must not set {line!r}",
)
def test_egress_env_present_when_routes_declared(self):
def test_egress_token_env_present_when_routes_declared(self):
sc = self._render(with_egress=True)["services"]["sidecars"]
env_strings = sc["environment"]
self.assertTrue(any(
e.startswith("EGRESS_UPSTREAM_PROXY=") for e in env_strings))
self.assertTrue(any(
e.startswith("EGRESS_UPSTREAM_CA=") for e in env_strings))
# Token env name is forwarded as a bare entry.
self.assertIn("EGRESS_TOKEN_0", env_strings)
def test_egress_env_omitted_when_no_routes(self):
def test_egress_token_env_omitted_when_no_routes(self):
sc = self._render()["services"]["sidecars"]
env_strings = sc["environment"]
for e in env_strings:
self.assertFalse(e.startswith("EGRESS_UPSTREAM_PROXY="))
self.assertFalse(e.startswith("EGRESS_UPSTREAM_CA="))
self.assertNotIn("EGRESS_TOKEN_0", env_strings)
def test_supervise_env_present_when_active(self):
sc = self._render(supervise=True)["services"]["sidecars"]
@@ -421,22 +391,19 @@ class TestSidecarBundleShape(unittest.TestCase):
self.assertTrue(any(e.startswith("SUPERVISE_QUEUE_DIR=") for e in env_strings))
self.assertTrue(any(e.startswith("SUPERVISE_PORT=") for e in env_strings))
def test_volumes_union_minimal_includes_pipelock(self):
def test_volumes_always_includes_egress_ca(self):
sc = self._render()["services"]["sidecars"]
targets = {v["target"] for v in sc["volumes"]}
self.assertIn("/etc/pipelock.yaml", targets)
self.assertIn("/home/mitmproxy/.mitmproxy/mitmproxy-ca.pem", targets)
def test_volumes_union_full_matrix(self):
sc = self._render(with_git=True, with_egress=True, supervise=True)[
"services"]["sidecars"]
targets = {v["target"] for v in sc["volumes"]}
# Pipelock + egress + git-gate + supervise paths all
# present.
self.assertIn("/etc/pipelock.yaml", targets)
self.assertIn("/home/mitmproxy/.mitmproxy/mitmproxy-ca.pem", targets)
self.assertIn("/etc/egress/routes.yaml", targets)
self.assertIn("/git-gate-entrypoint.sh", targets)
self.assertIn("/git-gate/creds/upstream-known_hosts", targets)
# supervise queue dir target = QUEUE_DIR_IN_CONTAINER
self.assertTrue(any("supervise/queue" in t or t.startswith("/run/supervise")
for t in targets))
tests/unit/test_contrib_claude_provider.py
-4
View File
@@ -23,7 +23,6 @@ from bot_bottle.contrib.claude.agent_provider import ClaudeAgentProvider
from bot_bottle.egress import EgressPlan
from bot_bottle.git_gate import GitGatePlan
from bot_bottle.manifest import Manifest
from bot_bottle.pipelock import PipelockProxyPlan
from bot_bottle.supervise import SupervisePlan
from bot_bottle.workspace import workspace_plan
@@ -90,9 +89,6 @@ def _plan(
env_file=Path("/tmp/agent.env"),
forwarded_env={},
prompt_file=Path("/tmp/state/demo-abc12/agent/prompt.txt"),
proxy_plan=PipelockProxyPlan(
yaml_path=Path("/tmp/pipelock.yaml"), slug="demo-abc12",
),
git_gate_plan=GitGatePlan(
slug="demo-abc12",
entrypoint_script=Path("/tmp/git-gate-entrypoint.sh"),
tests/unit/test_contrib_codex_provider.py
-4
View File
@@ -24,7 +24,6 @@ from bot_bottle.contrib.codex.agent_provider import CodexAgentProvider
from bot_bottle.egress import EgressPlan
from bot_bottle.git_gate import GitGatePlan
from bot_bottle.manifest import Manifest
from bot_bottle.pipelock import PipelockProxyPlan
from bot_bottle.supervise import SupervisePlan
from bot_bottle.workspace import workspace_plan
@@ -91,9 +90,6 @@ def _plan(
env_file=Path("/tmp/agent.env"),
forwarded_env={},
prompt_file=Path("/tmp/state/demo-abc12/agent/prompt.txt"),
proxy_plan=PipelockProxyPlan(
yaml_path=Path("/tmp/pipelock.yaml"), slug="demo-abc12",
),
git_gate_plan=GitGatePlan(
slug="demo-abc12",
entrypoint_script=Path("/tmp/git-gate-entrypoint.sh"),
tests/unit/test_docker_enumerate_active.py
+8 -8
View File
@@ -40,22 +40,22 @@ class TestParseServicesByProject(unittest.TestCase):
def test_multiple_services_per_project(self):
out = _enumerate._parse_services_by_project(
"bot-bottle-dev-abc\tegress\n"
"bot-bottle-dev-abc\tpipelock\n"
"bot-bottle-dev-abc\tgit-gate\n"
"bot-bottle-dev-abc\tsupervise\n"
)
self.assertEqual(
{"bot-bottle-dev-abc": {"egress", "pipelock", "supervise"}},
{"bot-bottle-dev-abc": {"egress", "git-gate", "supervise"}},
out,
)
def test_multiple_projects(self):
out = _enumerate._parse_services_by_project(
"proj-a\tegress\n"
"proj-b\tpipelock\n"
"proj-b\tgit-gate\n"
"proj-a\tsupervise\n"
)
self.assertEqual(
{"proj-a": {"egress", "supervise"}, "proj-b": {"pipelock"}},
{"proj-a": {"egress", "supervise"}, "proj-b": {"git-gate"}},
out,
)
@@ -117,7 +117,7 @@ class TestEnumerateActive(_FakeHomeMixin, unittest.TestCase):
))
self._stub(
["dev-abc"],
{"bot-bottle-dev-abc": {"pipelock", "egress", "supervise"}},
{"bot-bottle-dev-abc": {"egress", "git-gate", "supervise"}},
)
active = _enumerate.enumerate_active()
self.assertEqual(1, len(active))
@@ -126,17 +126,17 @@ class TestEnumerateActive(_FakeHomeMixin, unittest.TestCase):
self.assertEqual("dev-abc", a.slug)
self.assertEqual("implementer", a.agent_name)
self.assertEqual("2026-05-26T03:00:00+00:00", a.started_at)
self.assertEqual(("egress", "pipelock", "supervise"), a.services)
self.assertEqual(("egress", "git-gate", "supervise"), a.services)
def test_missing_metadata_renders_question_mark(self):
# State dir doesn't exist for this slug — agent_name falls
# back to "?" rather than dropping the row.
self._stub(["mystery-zzz"], {"bot-bottle-mystery-zzz": {"pipelock"}})
self._stub(["mystery-zzz"], {"bot-bottle-mystery-zzz": {"egress"}})
active = _enumerate.enumerate_active()
self.assertEqual(1, len(active))
self.assertEqual("?", active[0].agent_name)
self.assertEqual("", active[0].started_at)
self.assertEqual(("pipelock",), active[0].services)
self.assertEqual(("egress",), active[0].services)
def test_no_services_for_project_yields_empty_tuple(self):
# Race window between `compose up` returning and the actual
tests/unit/test_docker_launch_teardown.py
-9
View File
@@ -22,7 +22,6 @@ from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
from bot_bottle.egress import EgressPlan
from bot_bottle.git_gate import GitGatePlan
from bot_bottle.manifest import Manifest
from bot_bottle.pipelock import PipelockProxyPlan
from bot_bottle.workspace import workspace_plan
@@ -80,10 +79,6 @@ def _plan(tmp: str) -> DockerBottlePlan:
env_file=stage / "env",
forwarded_env={},
prompt_file=stage / "prompt.txt",
proxy_plan=PipelockProxyPlan(
yaml_path=stage / "pipelock.yaml",
slug="test-teardown-00001",
),
use_runsc=False,
)
@@ -101,10 +96,6 @@ class TestTeardownWarning(unittest.TestCase):
buf = io.StringIO()
with mock.patch.object(launch_mod.docker_mod, "build_image"), \
mock.patch.object(
launch_mod, "pipelock_tls_init",
return_value=(Path("/ca.crt"), Path("/ca.key")),
), \
mock.patch.object(
launch_mod, "egress_tls_init",
return_value=(Path("/egress_ca"), Path("/egress_cert")),
tests/unit/test_docker_provision_git_user.py
-4
View File
@@ -20,7 +20,6 @@ from bot_bottle.backend.docker.provision import git as _git
from bot_bottle.egress import EgressPlan
from bot_bottle.git_gate import GitGatePlan
from bot_bottle.manifest import Manifest
from bot_bottle.pipelock import PipelockProxyPlan
from bot_bottle.workspace import workspace_plan
@@ -53,9 +52,6 @@ def _plan(*, git_user: dict | None = None, # type: ignore
env_file=Path("/tmp/agent.env"),
forwarded_env={},
prompt_file=Path("/tmp/prompt.txt"),
proxy_plan=PipelockProxyPlan(
yaml_path=Path("/tmp/pipelock.yaml"), slug="demo-abc12",
),
git_gate_plan=GitGatePlan(
slug="demo-abc12",
entrypoint_script=Path("/tmp/git-gate-entrypoint.sh"),
tests/unit/test_egress.py
-13
View File
@@ -133,19 +133,6 @@ class TestRoutesForBottleManifestOnly(unittest.TestCase):
effective = [r.host for r in egress_routes_for_bottle(b)]
self.assertEqual(["x.example"], effective)
def test_tls_passthrough_lifted_from_manifest(self):
b = _bottle([{
"host": "api.openai.com",
"auth": {"scheme": "Bearer", "token_ref": "T"},
"pipelock": {"tls_passthrough": True},
}])
routes = egress_routes_for_bottle(b)
self.assertTrue(routes[0].tls_passthrough)
def test_tls_passthrough_false_by_default(self):
b = _bottle([{"host": "api.github.com"}])
routes = egress_routes_for_bottle(b)
self.assertFalse(routes[0].tls_passthrough)
class TestProviderRouteMerge(unittest.TestCase):
tests/unit/test_egress_addon_core.py
+3 -5
View File
@@ -180,7 +180,7 @@ class TestMatchRoute(unittest.TestCase):
def test_wildcard_hosts_not_supported(self):
# `*.example.com` is treated as a literal host string by
# the exact-only matcher. Removed from the design after
# the apex/RFC-6125/pipelock-mirror edge cases stacked up.
# the apex/RFC-6125 edge cases stacked up.
routes = (Route(host="*.example.com"),)
self.assertIsNone(match_route(routes, "foo.example.com"))
self.assertIsNone(match_route(routes, "example.com"))
@@ -191,10 +191,8 @@ class TestMatchRoute(unittest.TestCase):
class TestDecide(unittest.TestCase):
def test_no_matching_route_blocks(self):
# Defense-in-depth: egress gates the bottle's allowlist
# too, not just pipelock. Any host the operator didn't declare
# in egress.routes is 403'd at egress before it
# ever reaches pipelock.
# Egress gates the bottle's allowlist. Any host the operator
# didn't declare in egress.routes is 403'd at egress.
d = decide((), "elsewhere.example", "/anything", {})
self.assertEqual("block", d.action)
self.assertIn("allowlist", d.reason)
tests/unit/test_egress_apply.py
-75
View File
@@ -6,9 +6,7 @@ import unittest
from bot_bottle.backend.docker.egress_apply import (
EgressApplyError,
_hosts_in_routes,
_merge_single_route,
_pipelock_safe_hosts,
validate_routes_content,
)
from bot_bottle.yaml_subset import parse_yaml_subset
@@ -66,44 +64,6 @@ class TestValidateRoutesContent(unittest.TestCase):
)
class TestHostsInRoutes(unittest.TestCase):
def test_extracts_each_unique_host(self):
hosts = _hosts_in_routes(
'routes:\n'
' - host: "api.github.com"\n'
' - host: "github.com"\n'
' - host: "api.anthropic.com"\n'
)
# Sorted+deduped.
self.assertEqual(
["api.anthropic.com", "api.github.com", "github.com"],
hosts,
)
def test_dedupes_same_host(self):
hosts = _hosts_in_routes(
'routes:\n'
' - host: "x.example"\n'
' path_allowlist:\n'
' - "/a/"\n'
' - host: "x.example"\n'
' path_allowlist:\n'
' - "/b/"\n'
)
self.assertEqual(["x.example"], hosts)
def test_empty_routes_returns_empty(self):
self.assertEqual([], _hosts_in_routes(_ROUTES_EMPTY))
def test_invalid_routes_raises(self):
# The mirror helper relies on parsing succeeding; bad input
# should error before pipelock is touched.
with self.assertRaises(EgressApplyError):
_hosts_in_routes(
'routes:\n - path_allowlist:\n - "/no-host/"\n'
)
class TestMergeSingleRoute(unittest.TestCase):
BASE = _ROUTES_ONE
@@ -214,40 +174,5 @@ class TestMergeSingleRoute(unittest.TestCase):
_merge_single_route("routes:\n\tbad", {"host": "x.example"})
class TestPipelockSafeHosts(unittest.TestCase):
def test_passes_normal_hostnames_through(self):
self.assertEqual(
["api.github.com", "registry.npmjs.org"],
_pipelock_safe_hosts(["api.github.com", "registry.npmjs.org"]),
)
def test_drops_wildcards(self):
# Wildcard host matching was removed from egress too,
# so a `*.foo.com` route is dead weight anyway; we drop it
# entirely from the pipelock mirror so the apply doesn't
# fail parse.
self.assertEqual(
["api.github.com"],
_pipelock_safe_hosts(["*.example.com", "api.github.com"]),
)
def test_drops_bare_wildcard(self):
self.assertEqual([], _pipelock_safe_hosts(["*"]))
def test_drops_ipv6_literals(self):
self.assertEqual(
["api.example.com"],
_pipelock_safe_hosts(["[::1]", "api.example.com"]),
)
def test_preserves_order(self):
self.assertEqual(
["a.example", "b.example", "c.example"],
_pipelock_safe_hosts([
"a.example", "*.junk", "b.example", "weird host", "c.example",
]),
)
if __name__ == "__main__":
unittest.main()
tests/unit/test_manifest_egress.py
+3 -50
View File
@@ -219,57 +219,10 @@ class TestRole(unittest.TestCase):
_bottle([{"host": "x.example", "role": ["x", 42]}])
class TestPipelockPolicy(unittest.TestCase):
def test_tls_passthrough_route_policy(self):
b = _bottle([{
"host": "api.openai.com",
"pipelock": {"tls_passthrough": True},
}])
self.assertTrue(b.egress.routes[0].Pipelock.TlsPassthrough)
def test_ssrf_ip_allowlist_route_policy(self):
b = _bottle([{
"host": "gitea.dideric.is",
"pipelock": {"ssrf_ip_allowlist": ["100.78.141.42/32"]},
}])
self.assertEqual(
("100.78.141.42/32",),
b.egress.routes[0].Pipelock.SsrfIpAllowlist,
)
def test_tls_passthrough_defaults_false(self):
b = _bottle([{"host": "api.openai.com"}])
self.assertFalse(b.egress.routes[0].Pipelock.TlsPassthrough)
self.assertEqual((), b.egress.routes[0].Pipelock.SsrfIpAllowlist)
def test_pipelock_policy_must_be_object(self):
class TestPipelockKeyRejected(unittest.TestCase):
def test_pipelock_key_rejected_as_unknown(self):
with self.assertRaises(ManifestError):
_bottle([{"host": "x.example", "pipelock": True}])
def test_tls_passthrough_must_be_bool(self):
with self.assertRaises(ManifestError):
_bottle([{
"host": "x.example",
"pipelock": {"tls_passthrough": "yes"},
}])
def test_ssrf_ip_allowlist_must_be_array(self):
with self.assertRaises(ManifestError):
_bottle([{
"host": "x.example",
"pipelock": {"ssrf_ip_allowlist": "100.78.141.42/32"},
}])
def test_ssrf_ip_allowlist_items_must_be_cidr_or_ip(self):
with self.assertRaises(ManifestError):
_bottle([{
"host": "x.example",
"pipelock": {"ssrf_ip_allowlist": ["not-an-ip"]},
}])
def test_unknown_pipelock_key_rejected(self):
with self.assertRaises(ManifestError):
_bottle([{"host": "x.example", "pipelock": {"wat": True}}])
_bottle([{"host": "x.example", "pipelock": {"tls_passthrough": True}}])
class TestRouteValidation(unittest.TestCase):
tests/unit/test_plan_print_parity.py
-10
View File
@@ -20,7 +20,6 @@ from bot_bottle.backend.smolmachines.bottle_plan import SmolmachinesBottlePlan
from bot_bottle.egress import EgressPlan, EgressRoute
from bot_bottle.git_gate import GitGatePlan, GitGateUpstream
from bot_bottle.manifest import Manifest
from bot_bottle.pipelock import PipelockProxyPlan
from bot_bottle.workspace import workspace_plan
@@ -93,13 +92,6 @@ def _agent_provision() -> AgentProvisionPlan:
)
def _proxy_plan(tmp: str) -> PipelockProxyPlan:
return PipelockProxyPlan(
yaml_path=Path(tmp) / "pipelock.yaml",
slug="test-00001",
)
def _docker_plan(spec: BottleSpec, tmp: str) -> DockerBottlePlan:
stage = Path(tmp)
return DockerBottlePlan(
@@ -121,7 +113,6 @@ def _docker_plan(spec: BottleSpec, tmp: str) -> DockerBottlePlan:
env_file=stage / "env",
forwarded_env={},
prompt_file=stage / "prompt.txt",
proxy_plan=_proxy_plan(tmp),
use_runsc=False,
)
@@ -145,7 +136,6 @@ def _smolmachines_plan(spec: BottleSpec, tmp: str) -> SmolmachinesBottlePlan:
agent_image_ref="bot-bottle-claude:latest",
guest_env={"HTTPS_PROXY": "http://127.0.0.1:9999"},
prompt_file=stage / "prompt.txt",
proxy_plan=_proxy_plan(tmp),
)
tests/unit/test_sidecar_init.py
+40 -52
View File
@@ -29,11 +29,8 @@ from bot_bottle.sidecar_init import (
class TestEnvForDaemon(unittest.TestCase):
"""Scope egress-only credential env vars to the egress daemon.
Regression for issue #84: pipelock's `scan_env: true` matched
`EGRESS_TOKEN_*` against egress's just-injected Authorization
header and 403-blocked the legitimate request. The agent
never has access to these slots, so stripping them from
non-egress daemons loses no DLP coverage."""
The agent never has access to EGRESS_TOKEN_* slots, so stripping
them from non-egress daemons loses no DLP coverage."""
_BASE = {
"PATH": "/usr/bin",
@@ -47,26 +44,20 @@ class TestEnvForDaemon(unittest.TestCase):
env = _env_for_daemon("egress", self._BASE)
self.assertEqual(self._BASE, env)
def test_pipelock_loses_egress_tokens(self):
env = _env_for_daemon("pipelock", self._BASE)
self.assertNotIn("EGRESS_TOKEN_0", env)
self.assertNotIn("EGRESS_TOKEN_1", env)
# Non-token bundle env stays — supervise / git-gate / git-http / the
# upstream proxy URL are all load-bearing for other
# daemons.
self.assertEqual("/usr/bin", env["PATH"])
self.assertEqual("http://127.0.0.1:8888", env["EGRESS_UPSTREAM_PROXY"])
self.assertEqual("9100", env["SUPERVISE_PORT"])
def test_git_daemons_and_supervise_also_lose_egress_tokens(self):
def test_git_daemons_and_supervise_lose_egress_tokens(self):
for name in ("git-gate", "git-http", "supervise"):
env = _env_for_daemon(name, self._BASE)
self.assertNotIn("EGRESS_TOKEN_0", env)
self.assertNotIn("EGRESS_TOKEN_1", env)
# Non-token bundle env stays — supervise / git-gate / git-http are
# all load-bearing for other daemons.
self.assertEqual("/usr/bin", env["PATH"])
self.assertEqual("http://127.0.0.1:8888", env["EGRESS_UPSTREAM_PROXY"])
self.assertEqual("9100", env["SUPERVISE_PORT"])
def test_returns_independent_dict(self):
# Caller mutation mustn't affect the original.
env = _env_for_daemon("pipelock", self._BASE)
env = _env_for_daemon("git-gate", self._BASE)
env["X"] = "y"
self.assertNotIn("X", self._BASE)
@@ -78,7 +69,6 @@ class TestSelectedDaemons(unittest.TestCase):
_DAEMONS = (
_DaemonSpec("egress", ("/bin/sh", "-c", ":")),
_DaemonSpec("pipelock", ("/bin/sh", "-c", ":")),
_DaemonSpec("git-gate", ("/bin/sh", "-c", ":")),
_DaemonSpec("supervise", ("/bin/sh", "-c", ":")),
)
@@ -86,35 +76,34 @@ class TestSelectedDaemons(unittest.TestCase):
def test_unset_returns_all(self):
got = _selected_daemons({}, all_daemons=self._DAEMONS)
self.assertEqual([d.name for d in got],
["egress", "pipelock", "git-gate", "supervise"])
["egress", "git-gate", "supervise"])
def test_empty_returns_all(self):
got = _selected_daemons({"BOT_BOTTLE_SIDECAR_DAEMONS": ""},
all_daemons=self._DAEMONS)
self.assertEqual(4, len(got))
self.assertEqual(3, len(got))
def test_whitespace_only_returns_all(self):
got = _selected_daemons({"BOT_BOTTLE_SIDECAR_DAEMONS": " "},
all_daemons=self._DAEMONS)
self.assertEqual(4, len(got))
self.assertEqual(3, len(got))
def test_explicit_subset(self):
got = _selected_daemons(
{"BOT_BOTTLE_SIDECAR_DAEMONS": "egress,pipelock"},
{"BOT_BOTTLE_SIDECAR_DAEMONS": "egress,git-gate"},
all_daemons=self._DAEMONS,
)
self.assertEqual([d.name for d in got], ["egress", "pipelock"])
self.assertEqual([d.name for d in got], ["egress", "git-gate"])
def test_preserves_canonical_order(self):
# Order in the env var doesn't matter; the result follows
# the canonical _DAEMONS order so egress starts before
# pipelock (race-window reason).
# the canonical _DAEMONS order so egress starts first.
got = _selected_daemons(
{"BOT_BOTTLE_SIDECAR_DAEMONS": "supervise,pipelock,egress"},
{"BOT_BOTTLE_SIDECAR_DAEMONS": "supervise,git-gate,egress"},
all_daemons=self._DAEMONS,
)
self.assertEqual([d.name for d in got],
["egress", "pipelock", "supervise"])
["egress", "git-gate", "supervise"])
def test_unknown_names_ignored(self):
got = _selected_daemons(
@@ -125,10 +114,10 @@ class TestSelectedDaemons(unittest.TestCase):
def test_whitespace_in_names_stripped(self):
got = _selected_daemons(
{"BOT_BOTTLE_SIDECAR_DAEMONS": " egress , pipelock "},
{"BOT_BOTTLE_SIDECAR_DAEMONS": " egress , git-gate "},
all_daemons=self._DAEMONS,
)
self.assertEqual([d.name for d in got], ["egress", "pipelock"])
self.assertEqual([d.name for d in got], ["egress", "git-gate"])
class TestSupervisor(unittest.TestCase):
@@ -279,25 +268,24 @@ class TestSupervisor(unittest.TestCase):
self._drive(sup)
def test_restart_daemon_replaces_in_place(self):
# pipelock_apply.py sends SIGUSR1 to the bundle, supervisor
# restarts the pipelock daemon, supervise (the other
# daemon's MCP server in production) stays up.
# Restart one daemon; the other (supervise, the MCP server
# in production) must remain untouched.
specs = [
_DaemonSpec("pipelock", ("/bin/sleep", "30")),
_DaemonSpec("git-gate", ("/bin/sleep", "30")),
_DaemonSpec("supervise", ("/bin/sleep", "30")),
]
sup = _Supervisor(specs)
sup.start_all()
time.sleep(0.1)
old_pipelock_pid = sup.procs[0][1].pid
old_git_gate_pid = sup.procs[0][1].pid
supervise_pid = sup.procs[1][1].pid
ok = sup.restart_daemon("pipelock", grace=2.0)
ok = sup.restart_daemon("git-gate", grace=2.0)
self.assertTrue(ok)
# Pipelock got a fresh PID — different process.
new_pipelock_pid = sup.procs[0][1].pid
self.assertNotEqual(old_pipelock_pid, new_pipelock_pid)
# git-gate got a fresh PID — different process.
new_git_gate_pid = sup.procs[0][1].pid
self.assertNotEqual(old_git_gate_pid, new_git_gate_pid)
# Supervise's PID is unchanged — it was NOT restarted.
self.assertEqual(supervise_pid, sup.procs[1][1].pid)
self.assertIsNone(sup.procs[1][1].poll(),
@@ -308,38 +296,38 @@ class TestSupervisor(unittest.TestCase):
def test_request_restart_is_drained_by_tick(self):
specs = [
_DaemonSpec("pipelock", ("/bin/sleep", "30")),
_DaemonSpec("git-gate", ("/bin/sleep", "30")),
_DaemonSpec("supervise", ("/bin/sleep", "30")),
]
sup = _Supervisor(specs)
sup.start_all()
time.sleep(0.1)
old_pipelock_pid = sup.procs[0][1].pid
old_git_gate_pid = sup.procs[0][1].pid
supervise_pid = sup.procs[1][1].pid
ok = sup.request_restart("pipelock")
ok = sup.request_restart("git-gate")
self.assertTrue(ok)
# The non-blocking request path only records intent.
self.assertEqual(old_pipelock_pid, sup.procs[0][1].pid)
self.assertEqual(old_git_gate_pid, sup.procs[0][1].pid)
done = sup.tick()
self.assertFalse(done)
self.assertNotEqual(old_pipelock_pid, sup.procs[0][1].pid)
self.assertNotEqual(old_git_gate_pid, sup.procs[0][1].pid)
self.assertEqual(supervise_pid, sup.procs[1][1].pid)
sup.request_shutdown(reason="cleanup")
self._drive(sup)
def test_repeated_restart_requests_coalesce(self):
specs = [_DaemonSpec("pipelock", ("/bin/sleep", "30"))]
specs = [_DaemonSpec("git-gate", ("/bin/sleep", "30"))]
sup = _Supervisor(specs)
sup.start_all()
time.sleep(0.1)
self.assertTrue(sup.request_restart("pipelock"))
self.assertTrue(sup.request_restart("pipelock"))
self.assertEqual({"pipelock"}, sup._restart_requested)
self.assertTrue(sup.request_restart("git-gate"))
self.assertTrue(sup.request_restart("git-gate"))
self.assertEqual({"git-gate"}, sup._restart_requested)
old_pid = sup.procs[0][1].pid
sup.tick()
@@ -374,23 +362,23 @@ class TestSupervisor(unittest.TestCase):
self._drive(sup)
def test_restart_during_shutdown_is_no_op(self):
specs = [_DaemonSpec("pipelock", ("/bin/sleep", "30"))]
specs = [_DaemonSpec("git-gate", ("/bin/sleep", "30"))]
sup = _Supervisor(specs)
sup.start_all()
sup.request_shutdown(reason="test")
ok = sup.restart_daemon("pipelock")
ok = sup.restart_daemon("git-gate")
self.assertFalse(ok,
"must not respawn a daemon during teardown")
self._drive(sup)
def test_pending_restart_dropped_during_shutdown(self):
specs = [_DaemonSpec("pipelock", ("/bin/sleep", "30"))]
specs = [_DaemonSpec("git-gate", ("/bin/sleep", "30"))]
sup = _Supervisor(specs)
sup.start_all()
time.sleep(0.1)
old_pid = sup.procs[0][1].pid
self.assertTrue(sup.request_restart("pipelock"))
self.assertTrue(sup.request_restart("git-gate"))
sup.request_shutdown(reason="test")
self.assertEqual(set(), sup._restart_requested)
self._drive(sup)
tests/unit/test_smolmachines_prepare.py
-2
View File
@@ -56,7 +56,6 @@ class TestSmolmachinesResolveEnv(unittest.TestCase):
patch("bot_bottle.backend.smolmachines.prepare.smolmachines_bundle_subnet",
return_value=("10.99.0.0/24", "10.99.0.1", "10.99.0.2")),
patch("bot_bottle.backend.smolmachines.prepare.GitGate") as mock_gg,
patch("bot_bottle.backend.smolmachines.prepare.PipelockProxy") as mock_pl,
patch("bot_bottle.backend.smolmachines.prepare.Egress") as mock_eg,
patch("bot_bottle.backend.smolmachines.prepare.Supervise"),
patch(
@@ -65,7 +64,6 @@ class TestSmolmachinesResolveEnv(unittest.TestCase):
patch("bot_bottle.backend.smolmachines.prepare.runtime_for"),
):
mock_gg.return_value.prepare.return_value = MagicMock()
mock_pl.return_value.prepare.return_value = MagicMock()
mock_eg.return_value.prepare.return_value = MagicMock()
def _make_provision(**kwargs): # type: ignore
return AgentProvisionPlan(
tests/unit/test_smolmachines_provision.py
+8 -38
View File
@@ -32,7 +32,6 @@ from bot_bottle.backend.smolmachines.launch import _bundle_launch_spec
from bot_bottle.egress import EgressPlan, EgressRoute
from bot_bottle.git_gate import GitGatePlan, GitGateUpstream
from bot_bottle.manifest import GitEntry, Manifest
from bot_bottle.pipelock import PipelockProxyPlan
from bot_bottle.supervise import SupervisePlan
from bot_bottle.workspace import workspace_plan
@@ -71,7 +70,6 @@ def _plan(
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",
agent_git_gate_host: str = "127.0.0.1:55555",
@@ -131,11 +129,6 @@ def _plan(
agent_image_ref="bot-bottle-claude:latest",
guest_env=dict(guest_env or {}),
prompt_file=Path("/tmp/state/demo-abc12/agent/prompt.txt"),
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",
entrypoint_script=Path("/tmp/git-gate-entrypoint.sh"),
@@ -235,16 +228,13 @@ def _write_self_signed_cert(path: Path) -> None:
class TestProvisionCA(unittest.TestCase):
"""provision_ca selects the right CA cert (egress when the
bottle has routes, else pipelock) and dispatches
"""provision_ca always uses the egress MITM CA and dispatches
cp_in + 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):
@@ -259,40 +249,22 @@ class TestProvisionCA(unittest.TestCase):
stderr="",
)
def test_pipelock_path_when_no_routes(self):
plan = _plan(pipelock_ca_path=self.pipelock_ca)
def test_egress_ca_always_installed(self):
plan = _plan(egress_ca_path=self.egress_ca)
bottle = _make_bottle(exec_result=self._UPDATE_OK)
_ca.provision_ca(plan, bottle)
bottle.cp_in.assert_called_once_with(
str(self.pipelock_ca),
str(self.egress_ca),
_ca.AGENT_CA_PATH,
)
# chmod + chown + update-ca-certificates are folded into
# one exec invocation; look at the single exec's script
# rather than expecting separate calls.
bottle.exec.assert_called_once()
script = bottle.exec.call_args.args[0]
self.assertIn("chmod 644", script)
self.assertIn("update-ca-certificates", script)
self.assertEqual("root", bottle.exec.call_args.kwargs.get("user"))
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,
)
bottle = _make_bottle(exec_result=self._UPDATE_OK)
_ca.provision_ca(plan, bottle)
# When routes are declared, egress is the agent's first hop,
# so egress's CA is the one that gets installed.
bottle.cp_in.assert_called_once_with(
str(self.egress_ca),
_ca.AGENT_CA_PATH,
)
def test_retries_smolvm_sigkill_during_update_ca(self):
plan = _plan(pipelock_ca_path=self.pipelock_ca)
plan = _plan(egress_ca_path=self.egress_ca)
killed = ExecResult(
returncode=137,
stdout="Updating certificates in /etc/ssl/certs...\n",
@@ -308,10 +280,8 @@ class TestProvisionCA(unittest.TestCase):
self.assertEqual(2, bottle.exec.call_count)
sleep.assert_called_once_with(1.0)
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")
def test_dies_when_egress_cert_missing(self):
plan = _plan(egress_ca_path=self.tmp / "does-not-exist.pem")
bottle = _make_bottle()
with self.assertRaises(SystemExit):
_ca.provision_ca(plan, bottle)
@@ -414,7 +384,7 @@ class TestBundleLaunchSpec(unittest.TestCase):
spec = _bundle_launch_spec(plan, "net", "127.0.0.16")
self.assertEqual(
"egress,pipelock,git-gate,git-http",
"egress,git-gate,git-http",
spec.daemons_csv,
)
self.assertIn(9420, spec.ports_to_publish)
tests/unit/test_smolmachines_sidecar_bundle.py
+7 -6
View File
@@ -134,34 +134,35 @@ class TestStartBundle(unittest.TestCase):
def test_daemons_env_passed_in(self):
with self._patch_run() as m:
start_bundle(_spec(daemons_csv="egress,pipelock,supervise"))
start_bundle(_spec(daemons_csv="egress,supervise"))
argv = m.call_args.args[0]
self.assertIn("-e", argv)
self.assertIn(
"BOT_BOTTLE_SIDECAR_DAEMONS=egress,pipelock,supervise",
"BOT_BOTTLE_SIDECAR_DAEMONS=egress,supervise",
argv,
)
def test_environment_entries_pass_through(self):
with self._patch_run() as m:
start_bundle(_spec(environment=(
"EGRESS_UPSTREAM_PROXY=http://...",
"SUPERVISE_BOTTLE_SLUG=demo-abc12",
"EGRESS_TOKEN_0", # bare-name → host env inherit
)))
argv = m.call_args.args[0]
self.assertIn("EGRESS_UPSTREAM_PROXY=http://...", argv)
self.assertIn("SUPERVISE_BOTTLE_SLUG=demo-abc12", argv)
self.assertIn("EGRESS_TOKEN_0", argv)
def test_volumes_render_with_ro_flag(self):
with self._patch_run() as m:
start_bundle(_spec(volumes=(
("/host/pipelock.yaml", "/etc/pipelock.yaml", True),
("/host/egress-ca.pem", "/home/mitmproxy/.mitmproxy/mitmproxy-ca.pem", True),
("/host/queue", "/run/supervise/queue", False),
)))
argv = m.call_args.args[0]
self.assertIn("/host/pipelock.yaml:/etc/pipelock.yaml:ro", argv)
self.assertIn(
"/host/egress-ca.pem:/home/mitmproxy/.mitmproxy/mitmproxy-ca.pem:ro",
argv,
)
self.assertIn("/host/queue:/run/supervise/queue", argv)
def test_failure_dies(self):
tests/unit/test_supervise.py
+4 -4
View File
@@ -247,13 +247,13 @@ class TestAuditLog(unittest.TestCase):
write_audit_entry(AuditEntry(
timestamp=f"2026-05-25T12:00:0{i}+00:00",
bottle_slug="dev",
component="pipelock",
component="egress",
operator_action=STATUS_APPROVED,
operator_notes=f"n{i}",
justification="",
diff="",
))
path = audit_log_path("pipelock", "dev")
path = audit_log_path("egress", "dev")
with path.open() as f:
lines = [line for line in f if line.strip()]
self.assertEqual(3, len(lines))
@@ -273,7 +273,7 @@ class TestAuditLog(unittest.TestCase):
write_audit_entry(AuditEntry(
timestamp="t",
bottle_slug="dev",
component="pipelock",
component="egress",
operator_action=STATUS_APPROVED,
operator_notes="",
justification="",
@@ -289,7 +289,7 @@ class TestAuditLog(unittest.TestCase):
diff="",
))
self.assertEqual(1, len(read_audit_entries("cred-proxy", "dev")))
self.assertEqual(1, len(read_audit_entries("pipelock", "dev")))
self.assertEqual(1, len(read_audit_entries("egress", "dev")))
self.assertEqual(1, len(read_audit_entries("cred-proxy", "other")))
def test_read_audit_entries_missing_log_returns_empty(self):
tests/unit/test_supervise_cli.py
-109
View File
@@ -18,7 +18,6 @@ from pathlib import Path
from bot_bottle import supervise
from bot_bottle.backend.docker.capability_apply import CapabilityApplyError
from bot_bottle.backend.docker.egress_apply import EgressApplyError
from bot_bottle.backend.docker.pipelock_apply import PipelockApplyError
from bot_bottle.cli import supervise as supervise_cli
from bot_bottle.supervise import (
Proposal,
@@ -27,7 +26,6 @@ from bot_bottle.supervise import (
STATUS_REJECTED,
TOOL_CAPABILITY_BLOCK,
TOOL_EGRESS_BLOCK,
TOOL_PIPELOCK_BLOCK,
read_audit_entries,
read_response,
sha256_hex,
@@ -38,13 +36,8 @@ FIXED = datetime(2026, 5, 25, 12, 0, 0, tzinfo=timezone.utc)
def _proposal(slug: str = "dev", tool: str = TOOL_EGRESS_BLOCK) -> Proposal:
# Per-tool payload shape: cred-proxy gets routes.yaml, pipelock
# gets a failed URL (PR #25 follow-up), capability gets a
# Dockerfile-ish blob. Match the production dispatch in
# PROPOSED_FILE_FIELD.
payloads = {
TOOL_EGRESS_BLOCK: '{"routes": []}\n',
TOOL_PIPELOCK_BLOCK: "https://example.com/path",
TOOL_CAPABILITY_BLOCK: "FROM python:3.13\n",
}
payload = payloads.get(tool, "")
@@ -128,26 +121,18 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
def setUp(self):
self._setup_fake_home()
self._original_add_route = supervise_cli.add_route
self._original_apply_allowlist = supervise_cli.apply_allowlist_change
self._original_fetch_allowlist = supervise_cli.fetch_current_allowlist
self._original_apply_capability = supervise_cli.apply_capability_change
# Default stubs: succeed with deterministic before/after so the
# audit log shows a non-empty diff.
supervise_cli.add_route = lambda slug, content: ( # type: ignore
'{"routes": []}\n', '{"routes": [{"host": "x"}]}\n',
)
supervise_cli.apply_allowlist_change = lambda slug, content: ( # type: ignore
"old.example\n", content,
)
supervise_cli.fetch_current_allowlist = lambda slug: "old.example\n" # type: ignore
supervise_cli.apply_capability_change = lambda slug, content: ( # type: ignore
"FROM old\n", content,
)
def tearDown(self):
supervise_cli.add_route = self._original_add_route
supervise_cli.apply_allowlist_change = self._original_apply_allowlist
supervise_cli.fetch_current_allowlist = self._original_fetch_allowlist
supervise_cli.apply_capability_change = self._original_apply_capability
self._teardown_fake_home()
@@ -192,15 +177,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
qp = self._enqueue(tool=TOOL_CAPABILITY_BLOCK)
supervise_cli.approve(qp)
# No audit log for capability-block (per PRD 0013 / 0016).
# cred-proxy and pipelock logs both empty.
self.assertEqual([], read_audit_entries("egress", "dev"))
self.assertEqual([], read_audit_entries("pipelock", "dev"))
def test_pipelock_audit_distinct_from_egress(self):
qp = self._enqueue(tool=TOOL_PIPELOCK_BLOCK)
supervise_cli.approve(qp)
self.assertEqual(1, len(read_audit_entries("pipelock", "dev")))
self.assertEqual(0, len(read_audit_entries("egress", "dev")))
class TestEgressApplyWiring(_FakeHomeMixin, unittest.TestCase):
@@ -299,91 +276,6 @@ class TestEgressApplyWiring(_FakeHomeMixin, unittest.TestCase):
self.assertEqual("", entries[0].diff)
class TestPipelockApplyWiring(_FakeHomeMixin, unittest.TestCase):
"""PRD 0015 Phase 2 + PR #25 follow-up: approve() on a
pipelock-block proposal carries the failed URL; the supervise TUI
extracts the host, merges it into the running allowlist, and
calls apply_allowlist_change with the merged content."""
def setUp(self):
self._setup_fake_home()
self._original_apply = supervise_cli.apply_allowlist_change
self._original_fetch = supervise_cli.fetch_current_allowlist
def tearDown(self):
supervise_cli.apply_allowlist_change = self._original_apply
supervise_cli.fetch_current_allowlist = self._original_fetch
self._teardown_fake_home()
def _enqueue_pipelock(self, failed_url: str = "https://api.github.com/repos/foo/bar"):
p = Proposal.new(
bottle_slug="dev", tool=TOOL_PIPELOCK_BLOCK,
proposed_file=failed_url,
justification="need to read PR metadata",
current_file_hash=sha256_hex(failed_url),
now=FIXED,
)
qdir = supervise.queue_dir_for_slug("dev")
qdir.mkdir(parents=True, exist_ok=True)
supervise.write_proposal(qdir, p)
return supervise_cli.QueuedProposal(proposal=p, queue_dir=qdir)
def test_url_host_merged_into_current_allowlist(self):
supervise_cli.fetch_current_allowlist = lambda slug: "existing.example\n" # type: ignore
applied = []
supervise_cli.apply_allowlist_change = lambda slug, content: ( # type: ignore
applied.append((slug, content))
or ("existing.example\n", content)
)
qp = self._enqueue_pipelock("https://api.github.com/repos/foo/bar")
supervise_cli.approve(qp)
# apply_allowlist_change was called with the merged content:
# existing host + the URL's host (no path, since pipelock is
# hostname-only).
self.assertEqual(1, len(applied))
slug, content = applied[0]
self.assertEqual("dev", slug)
self.assertIn("existing.example", content)
self.assertIn("api.github.com", content)
self.assertNotIn("/repos/foo/bar", content) # path stripped
def test_host_already_in_allowlist_is_idempotent(self):
supervise_cli.fetch_current_allowlist = lambda slug: "api.github.com\n" # type: ignore
applied = []
supervise_cli.apply_allowlist_change = lambda slug, content: ( # type: ignore
applied.append(content)
or ("api.github.com\n", content)
)
qp = self._enqueue_pipelock("https://api.github.com/some/path")
supervise_cli.approve(qp)
# Still applied, but the content is unchanged from current —
# before/after diff is empty.
self.assertEqual(1, len(applied))
self.assertEqual("api.github.com\n", applied[0])
def test_apply_failure_blocks_response_and_audit(self):
supervise_cli.fetch_current_allowlist = lambda slug: "existing.example\n" # type: ignore
supervise_cli.apply_allowlist_change = lambda slug, content: (_ for _ in ()).throw( # type: ignore
PipelockApplyError("docker exec failed")
)
qp = self._enqueue_pipelock()
with self.assertRaises(PipelockApplyError):
supervise_cli.approve(qp)
self.assertEqual(
[qp.proposal.id],
[p.id for p in supervise.list_pending_proposals(qp.queue_dir)],
)
self.assertEqual([], read_audit_entries("pipelock", "dev"))
def test_url_without_host_raises(self):
supervise_cli.fetch_current_allowlist = lambda slug: "" # type: ignore
# supervise_server's validator would catch this; if a broken
# URL ever makes it through, the supervise TUI surfaces it too.
qp = self._enqueue_pipelock("https:///nohost")
with self.assertRaises(PipelockApplyError):
supervise_cli.approve(qp)
class TestCapabilityApplyWiring(_FakeHomeMixin, unittest.TestCase):
"""PRD 0016 Phase 3: approve() on a capability-block proposal
calls apply_capability_change, archives the proposal afterward
@@ -439,7 +331,6 @@ class TestCapabilityApplyWiring(_FakeHomeMixin, unittest.TestCase):
# capability-block has no audit log per PRD 0013 — its record
# lives in the per-bottle Dockerfile + transcript state.
self.assertEqual([], read_audit_entries("egress", "dev"))
self.assertEqual([], read_audit_entries("pipelock", "dev"))
def test_proposal_archived_after_apply(self):
supervise_cli.apply_capability_change = lambda slug, content: ("FROM old\n", content) # type: ignore
tests/unit/test_supervise_cli_detail_lines.py
-99
View File
@@ -1,99 +0,0 @@
"""Unit: supervise's detail-view line builder.
_detail_lines returns (text, attr) tuples. Most are plain; for
pipelock-block proposals it appends a "→ would allow host: <host>"
line tagged with the green attr so the operator sees at a glance
which hostname will land in pipelock's allowlist on approval."""
import unittest
from bot_bottle.cli import supervise as supervise_cli
from bot_bottle.supervise import (
Proposal,
TOOL_CAPABILITY_BLOCK,
TOOL_EGRESS_BLOCK,
TOOL_PIPELOCK_BLOCK,
sha256_hex,
)
def _qp(tool: str, payload: str) -> supervise_cli.QueuedProposal:
from datetime import datetime, timezone
from pathlib import Path
p = Proposal.new(
bottle_slug="dev",
tool=tool,
proposed_file=payload,
justification="needs",
current_file_hash=sha256_hex(payload),
now=datetime(2026, 5, 25, 12, 0, 0, tzinfo=timezone.utc),
)
return supervise_cli.QueuedProposal(proposal=p, queue_dir=Path("/tmp/q"))
class TestPipelockHostHighlight(unittest.TestCase):
GREEN = 0xDEADBEEF # arbitrary sentinel; _detail_lines passes through
def test_appends_green_host_line_for_pipelock_block(self):
lines = supervise_cli._detail_lines(
_qp(TOOL_PIPELOCK_BLOCK, "https://api.github.com/repos/foo/bar"),
green_attr=self.GREEN,
)
# The host appears as its own green-tagged line — literal
# text of what gets appended to pipelock's allowlist on
# approve.
green_lines = [text for text, attr in lines if attr == self.GREEN]
self.assertEqual(["api.github.com"], green_lines)
def test_no_green_lines_for_egress_block(self):
lines = supervise_cli._detail_lines(
_qp(TOOL_EGRESS_BLOCK, '{"routes": []}'),
green_attr=self.GREEN,
)
self.assertEqual([], [t for t, a in lines if a == self.GREEN])
def test_no_green_lines_for_capability_block(self):
lines = supervise_cli._detail_lines(
_qp(TOOL_CAPABILITY_BLOCK, "FROM python:3.13\n"),
green_attr=self.GREEN,
)
self.assertEqual([], [t for t, a in lines if a == self.GREEN])
def test_skips_host_line_when_url_unparseable(self):
# Shouldn't happen in production — supervise_server validates
# the URL before queuing — but if a malformed payload ever
# reaches the supervise TUI, don't render a misleading host line.
lines = supervise_cli._detail_lines(
_qp(TOOL_PIPELOCK_BLOCK, "garbage-not-a-url"),
green_attr=self.GREEN,
)
self.assertEqual([], [t for t, a in lines if a == self.GREEN])
def test_no_green_attr_passed_still_renders_host(self):
# Even without color support (green_attr=0), the host line
# is still present — it just won't be coloured.
lines = supervise_cli._detail_lines(
_qp(TOOL_PIPELOCK_BLOCK, "https://api.github.com/x"),
green_attr=0,
)
# Last non-empty line should be the host.
non_empty = [t for t, _ in lines if t]
self.assertEqual("api.github.com", non_empty[-1])
class TestFailedUrlHost(unittest.TestCase):
def test_extracts_hostname(self):
self.assertEqual(
"api.github.com",
supervise_cli._failed_url_host("https://api.github.com/repos/foo"),
)
def test_returns_empty_for_unparseable(self):
self.assertEqual("", supervise_cli._failed_url_host("not a url"))
def test_returns_empty_for_url_without_host(self):
self.assertEqual("", supervise_cli._failed_url_host("https:///nohost"))
if __name__ == "__main__":
unittest.main()
tests/unit/test_supervise_server.py
-22
View File
@@ -47,28 +47,6 @@ from bot_bottle.supervise_server import (
class TestValidation(unittest.TestCase):
def test_pipelock_block_accepts_https_url(self):
validate_proposed_file(
_sv.TOOL_PIPELOCK_BLOCK,
"https://api.github.com/repos/foo/bar",
)
def test_pipelock_block_accepts_http_url(self):
validate_proposed_file(
_sv.TOOL_PIPELOCK_BLOCK,
"http://internal.example/path/to/thing",
)
def test_pipelock_block_rejects_missing_scheme(self):
with self.assertRaises(_RpcError) as cm:
validate_proposed_file(_sv.TOOL_PIPELOCK_BLOCK, "api.github.com/foo")
self.assertIn("http://", str(cm.exception.message))
def test_pipelock_block_rejects_missing_host(self):
with self.assertRaises(_RpcError) as cm:
validate_proposed_file(_sv.TOOL_PIPELOCK_BLOCK, "https:///just-a-path")
self.assertIn("hostname", str(cm.exception.message))
def test_capability_block_accepts_anything_nonempty(self):
validate_proposed_file(
_sv.TOOL_CAPABILITY_BLOCK,