refactor(manifest): drop bottle.egress field, egress_proxy is the only allowlist
test / unit (pull_request) Successful in 17s
test / integration (pull_request) Successful in 1m4s

Goal: one allowlist surface (egress_proxy.routes), no second
free-form `egress:` knob. Anything that used to live there now
goes in `egress_proxy.routes` as a bare-pass entry
(`- host: <name>`).

Removed:
  - `BottleEgress` dataclass + DLP_ACTIONS constant + bottle.egress
    field on `Bottle`.
  - `pipelock_bottle_allowlist` helper.
  - `pipelock_allowlist_summary` helper (the compact preflight
    summary stopped using it after PR #31).
  - `allowlist_summary` field on `DockerBottlePlan`.
  - `bottle.egress.allowlist` folding in
    `egress_proxy_routes_for_bottle` — only DEFAULT_ALLOWLIST
    auto-folds now.
  - The two-branch logic in `pipelock_effective_allowlist`
    (egress-proxy-present vs not) — pipelock now just mirrors
    `egress_proxy_routes_for_bottle` unconditionally.

Hard-coded:
  - `request_body_scanning.action = "block"` in
    `pipelock_build_config` (was driven by
    `bottle.egress.dlp_action`). The previous default was already
    "block" — the knob to switch to "warn" was a foot-gun in a
    sandboxed agent context, so it's gone.

Tests:
  - `test_pipelock_allowlist.py` rewritten to assert the
    mirrored-from-egress-proxy semantics directly.
  - `test_manifest_md_load.py`, `test_pipelock_yaml.py`,
    `test_egress_proxy.py` fixtures migrated to put hosts in
    `egress_proxy.routes` instead of `egress.allowlist`.

Local bottle migrated too: `~/.claude-bottle/bottles/dev.md`
loses the `egress: { allowlist: [example.com] }` block, picks up
a bare-pass `- host: example.com` route.

409 unit + integration tests pass.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-25 21:12:56 -04:00
parent d79a976999
commit 6456904763
9 changed files with 56 additions and 209 deletions
@@ -49,7 +49,6 @@ class DockerBottlePlan(BottlePlan):
# None when bottle.supervise is False. PRD 0013 supervise sidecar
# is opt-in via the manifest's bottle.supervise field.
supervise_plan: SupervisePlan | None
allowlist_summary: str
use_runsc: bool
def print(self, *, remote_control: bool) -> None:
-3
View File
@@ -14,7 +14,6 @@ import os
from datetime import datetime, timezone
from pathlib import Path
from ... import pipelock
from ...env import ResolvedEnv, resolve_env
from ...log import die
from .. import BottleSpec
@@ -192,7 +191,6 @@ def resolve_plan(
_write_env_file(resolved, env_file)
prompt_file.write_text(agent.prompt)
allowlist_summary = pipelock.pipelock_allowlist_summary(bottle)
use_runsc = docker_mod.runsc_available()
return DockerBottlePlan(
@@ -212,7 +210,6 @@ def resolve_plan(
git_gate_plan=git_gate_plan,
egress_proxy_plan=egress_proxy_plan,
supervise_plan=supervise_plan,
allowlist_summary=allowlist_summary,
use_runsc=use_runsc,
)
+9 -8
View File
@@ -190,23 +190,24 @@ def egress_proxy_routes_for_bottle(
bottle: Bottle,
) -> tuple[EgressProxyRoute, ...]:
"""Effective egress-proxy routes: manifest routes followed by
bare-pass entries for DEFAULT_ALLOWLIST hosts and
`bottle.egress.allowlist` hosts. This is what gets rendered into
routes.yaml + what the addon enforces.
bare-pass entries for DEFAULT_ALLOWLIST hosts. This is what
gets rendered into routes.yaml + what the addon enforces.
Manifest routes win over defaults on host collision (manifest
routes carry more specific config — auth, path filter, role
markers). Hostname comparison is case-insensitive."""
markers). Hostname comparison is case-insensitive.
Operators that want to allow an arbitrary host that isn't in
DEFAULT_ALLOWLIST declare it directly in
`bottle.egress_proxy.routes` as a bare-pass entry
(`- host: <name>`). The legacy `bottle.egress.allowlist`
folding is gone — egress_proxy is the single allowlist surface."""
out: list[EgressProxyRoute] = list(egress_proxy_manifest_routes(bottle))
claimed: set[str] = {r.host.lower() for r in out}
for host in DEFAULT_ALLOWLIST:
if host.lower() not in claimed:
out.append(EgressProxyRoute(host=host))
claimed.add(host.lower())
for host in bottle.egress.allowlist:
if host and host.lower() not in claimed:
out.append(EgressProxyRoute(host=host))
claimed.add(host.lower())
return tuple(out)
+2 -62
View File
@@ -15,7 +15,6 @@ Bottle schema (frontmatter):
env: { <NAME>: <env-entry>, ... }
git: [ <git-entry>, ... ]
egress_proxy: { routes: [ <egress-route>, ... ] }
egress: { allowlist: [ <hostname>, ... ] }
Agent schema (frontmatter):
bottle: <bottle-name> # required
@@ -341,63 +340,11 @@ class EgressProxyConfig:
return cls(routes=routes)
DLP_ACTIONS = ("block", "warn")
@dataclass(frozen=True)
class BottleEgress:
allowlist: tuple[str, ...] = ()
# Action pipelock takes when its DLP layer matches a credential
# pattern in a request body. "block" → 403 from the proxy, the
# request never leaves the egress network. "warn" → forward the
# request and emit a log line. Default is "block": detect-only
# would let real secrets escape under the agent's compromised
# tooling, which is the threat model claude-bottle was built for.
dlp_action: str = "block"
@classmethod
def from_dict(cls, bottle_name: str, raw: object) -> "BottleEgress":
d = _as_json_object(raw, f"bottle '{bottle_name}' egress")
allow = d.get("allowlist")
items: list[str] = []
if allow is not None:
if not isinstance(allow, list):
die(
f"bottle '{bottle_name}' egress.allowlist must be an array "
f"(was {type(allow).__name__})"
)
allow_list = cast(list[object], allow)
for i, host in enumerate(allow_list):
if not isinstance(host, str):
die(
f"bottle '{bottle_name}' egress.allowlist[{i}] must be a string "
f"(was {type(host).__name__})"
)
items.append(host)
dlp_action_raw = d.get("dlp_action")
if dlp_action_raw is None:
dlp_action = "block"
elif isinstance(dlp_action_raw, str):
if dlp_action_raw not in DLP_ACTIONS:
die(
f"bottle '{bottle_name}' egress.dlp_action must be one of "
f"{', '.join(DLP_ACTIONS)} (was {dlp_action_raw!r})"
)
dlp_action = dlp_action_raw
else:
die(
f"bottle '{bottle_name}' egress.dlp_action must be a string "
f"(was {type(dlp_action_raw).__name__})"
)
return cls(allowlist=tuple(items), dlp_action=dlp_action)
@dataclass(frozen=True)
class Bottle:
env: Mapping[str, str] = field(default_factory=_empty_str_dict)
git: tuple[GitEntry, ...] = ()
egress_proxy: EgressProxyConfig = field(default_factory=EgressProxyConfig)
egress: BottleEgress = field(default_factory=BottleEgress)
# Opt-in per-bottle stuck-recovery sidecar (PRD 0013). When true,
# the launch step brings up a supervise sidecar that exposes three
# MCP tools to the agent (cred-proxy-block, pipelock-block,
@@ -481,13 +428,6 @@ class Bottle:
else EgressProxyConfig()
)
egress_raw = d.get("egress")
egress = (
BottleEgress.from_dict(name, egress_raw)
if egress_raw is not None
else BottleEgress()
)
supervise_raw = d.get("supervise", False)
if not isinstance(supervise_raw, bool):
die(
@@ -496,7 +436,7 @@ class Bottle:
)
return cls(
env=env, git=git, egress_proxy=egress_proxy, egress=egress,
env=env, git=git, egress_proxy=egress_proxy,
supervise=supervise_raw,
)
@@ -832,7 +772,7 @@ _FILENAME_RX = re.compile(r"^[a-z][a-z0-9-]*$")
# sets dies with a "did you mean" pointer — typos shouldn't silently
# ghost into an empty config.
_BOTTLE_KEYS = frozenset(
{"env", "git", "egress_proxy", "egress", "supervise"}
{"env", "git", "egress_proxy", "supervise"}
)
_AGENT_KEYS_REQUIRED = frozenset({"bottle"})
_AGENT_KEYS_OPTIONAL = frozenset({"skills"})
+11 -60
View File
@@ -50,38 +50,15 @@ DEFAULT_TLS_PASSTHROUGH: tuple[str, ...] = (
# --- Allowlist resolution --------------------------------------------------
def pipelock_bottle_allowlist(bottle: Bottle) -> list[str]:
"""Hostnames in bottle.egress.allowlist."""
return list(bottle.egress.allowlist)
def pipelock_route_hosts(bottle: Bottle) -> list[str]:
"""Hostnames declared in `bottle.egress_proxy.routes`. Returned
sorted + deduped. Used by the no-egress-proxy fallback path
below; bottles that DO use egress-proxy include the same hosts
via `egress_proxy_routes_for_bottle`."""
hosts = {r.Host for r in bottle.egress_proxy.routes if r.Host}
return sorted(hosts)
def pipelock_effective_allowlist(bottle: Bottle) -> list[str]:
"""Hostnames pipelock allows. Sorted for stability.
Two paths, depending on whether the bottle uses egress-proxy:
- Bottle declares `egress_proxy.routes[]` agent's HTTPS_PROXY
points at egress-proxy. Egress-proxy is the bottle's primary
allowlist gate (DEFAULT_ALLOWLIST + bottle.egress.allowlist +
manifest routes all live there as bare-pass or full routes,
folded in by `egress_proxy_routes_for_bottle`). Pipelock's
allowlist is then a MIRROR of egress-proxy's hosts — same
set, just serving as the defense-in-depth hostname gate +
DLP scanner on the upstream leg.
- Bottle has no `egress_proxy.routes[]` agent talks straight
to pipelock. Pipelock keeps its previous behavior: bake in
DEFAULT_ALLOWLIST + bottle.egress.allowlist for claude-code
defaults.
Always mirrors `egress_proxy_routes_for_bottle(bottle)` the
egress-proxy is the single allowlist surface; pipelock's
allowlist is the downstream copy for defense-in-depth + DLP
body scanning. For bottles without any `egress_proxy.routes[]`
declared, this is just the baked DEFAULT_ALLOWLIST that
egress_proxy_routes_for_bottle always folds in.
The supervise sidecar's hostname is auto-added when supervise
is enabled (sibling-sidecar traffic that flows through pipelock
@@ -89,19 +66,9 @@ def pipelock_effective_allowlist(bottle: Bottle) -> list[str]:
`bottle.git` do NOT contribute here git traffic flows
through git-gate (PRD 0008), not pipelock."""
seen: dict[str, None] = {}
if bottle.egress_proxy.routes:
# Mirror egress-proxy's effective host set — same defaults
# and bottle.egress.allowlist entries are already folded in
# at the egress-proxy layer; we don't add them twice.
for r in egress_proxy_routes_for_bottle(bottle):
if r.host:
seen.setdefault(r.host, None)
else:
for h in DEFAULT_ALLOWLIST:
seen.setdefault(h, None)
for h in pipelock_bottle_allowlist(bottle):
if h:
seen.setdefault(h, None)
for r in egress_proxy_routes_for_bottle(bottle):
if r.host:
seen.setdefault(r.host, None)
if bottle.supervise:
seen.setdefault(SUPERVISE_HOSTNAME, None)
return sorted(seen.keys())
@@ -160,22 +127,6 @@ def pipelock_effective_tls_passthrough(bottle: Bottle) -> list[str]:
return sorted(DEFAULT_TLS_PASSTHROUGH)
def pipelock_allowlist_summary(bottle: Bottle) -> str:
"""One-line summary for the y/N preflight display:
"<N> hosts allowed (host1, host2, host3, +M more)"."""
hosts = pipelock_effective_allowlist(bottle)
count = len(hosts)
if count == 0:
return "0 hosts allowed (none)"
show = count
more = 0
if count > 5:
show = 3
more = count - show
joined = ", ".join(hosts[:show])
if more > 0:
return f"{count} hosts allowed ({joined}, +{more} more)"
return f"{count} hosts allowed ({joined})"
@@ -226,9 +177,9 @@ def pipelock_build_config(
# Body-scan enforcement is a separate pipelock section (each DLP
# "surface" — body, MCP, response — has its own action). Pipelock's
# built-in default for request_body_scanning is "warn" (forward
# with a log line); claude-bottle's default is "block" so a hit
# with a log line); claude-bottle hard-codes "block" so a hit
# actually stops the request from leaving the egress network.
cfg["request_body_scanning"] = {"action": bottle.egress.dlp_action}
cfg["request_body_scanning"] = {"action": "block"}
if ca_cert_path or ca_key_path:
if not (ca_cert_path and ca_key_path):
raise ValueError(
-11
View File
@@ -109,17 +109,6 @@ class TestRoutesForBottleFoldsDefaults(unittest.TestCase):
self.assertEqual(1, len(anthropic))
self.assertEqual("Bearer", anthropic[0].auth_scheme)
def test_bottle_egress_allowlist_folded_in(self):
m = Manifest.from_json_obj({
"bottles": {"dev": {
"egress_proxy": {"routes": []},
"egress": {"allowlist": ["example.com"]},
}},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
})
hosts = [r.host for r in egress_proxy_routes_for_bottle(m.bottles["dev"])]
self.assertIn("example.com", hosts)
def test_manifest_only_when_no_defaults_or_allowlist(self):
# Sanity: egress_proxy_manifest_routes returns just the
# manifest entries — defaults are added by the
+4 -6
View File
@@ -28,9 +28,7 @@ _BOTTLE_DEV = """
auth:
scheme: Bearer
token_ref: CLAUDE_CODE_OAUTH_TOKEN
egress:
allowlist:
- example.com
- host: example.com
---
The dev bottle. Anthropic OAuth via egress-proxy.
@@ -88,11 +86,11 @@ class TestBottleFileParses(_ResolveCase):
m = self.resolve()
self.assertIn("dev", m.bottles)
routes = m.bottles["dev"].egress_proxy.routes
self.assertEqual(1, len(routes))
self.assertEqual(2, len(routes))
self.assertEqual("api.anthropic.com", routes[0].Host)
self.assertEqual("Bearer", routes[0].AuthScheme)
self.assertEqual("CLAUDE_CODE_OAUTH_TOKEN", routes[0].TokenRef)
self.assertEqual(["example.com"], list(m.bottles["dev"].egress.allowlist))
self.assertEqual("example.com", routes[1].Host)
class TestAgentFileParses(_ResolveCase):
@@ -134,7 +132,7 @@ class TestCwdAgentOverridesHome(_ResolveCase):
m = self.resolve()
self.assertIn("CWD-OVERRIDE-PROMPT", m.agents["implementer"].prompt)
# Home bottle still present
self.assertEqual(1, len(m.bottles["dev"].egress_proxy.routes))
self.assertEqual(2, len(m.bottles["dev"].egress_proxy.routes))
class TestCwdBottlesIgnored(_ResolveCase):
+28 -56
View File
@@ -1,8 +1,7 @@
"""Unit: pipelock_effective_allowlist — the union of baked-in defaults,
bottle.egress.allowlist, and egress-proxy route hosts derived from
bottle.egress_proxy.routes (PRD 0017). Git upstreams declared in
bottle.git do not contribute here; they flow through the per-agent
git-gate (PRD 0008)."""
"""Unit: pipelock_effective_allowlist — pipelock's allowlist
mirrors `egress_proxy_routes_for_bottle` (which folds in
DEFAULT_ALLOWLIST). Git upstreams declared in `bottle.git` don't
contribute; they flow through the per-agent git-gate (PRD 0008)."""
import unittest
@@ -10,7 +9,6 @@ from claude_bottle.manifest import Manifest
from claude_bottle.pipelock import (
pipelock_effective_allowlist,
pipelock_effective_tls_passthrough,
pipelock_route_hosts,
)
@@ -26,38 +24,26 @@ def _routes(routes):
class TestEffectiveAllowlist(unittest.TestCase):
def test_union_and_dedup(self):
eff = pipelock_effective_allowlist(_bottle({
"egress": {
"allowlist": [
"registry.npmjs.org",
# Duplicate of a baked default; the union must dedupe.
"api.anthropic.com",
],
},
}))
self.assertIn("api.anthropic.com", eff, "baked default present")
self.assertIn("registry.npmjs.org", eff, "egress.allowlist present")
self.assertEqual(len(eff), len(set(eff)), "deduplicated")
self.assertEqual(eff, sorted(eff), "sorted")
def test_default_allowlist_present_without_any_manifest_routes(self):
# No egress_proxy routes declared → pipelock allowlist is
# just the baked DEFAULT_ALLOWLIST (folded in by
# egress_proxy_routes_for_bottle).
eff = pipelock_effective_allowlist(_bottle({}))
self.assertIn("api.anthropic.com", eff)
self.assertIn("sentry.io", eff)
class TestRouteHosts(unittest.TestCase):
def test_each_route_contributes_its_host(self):
hosts = pipelock_route_hosts(_bottle(_routes([
{"host": "api.github.com",
"auth": {"scheme": "Bearer", "token_ref": "GH"}},
{"host": "github.com",
"auth": {"scheme": "Bearer", "token_ref": "GH"}},
def test_sorted_and_deduped(self):
# Manifest route for a default host collapses to one entry.
eff = pipelock_effective_allowlist(_bottle(_routes([
{"host": "api.anthropic.com",
"auth": {"scheme": "Bearer", "token_ref": "T"}},
])))
self.assertEqual(["api.github.com", "github.com"], hosts)
def test_no_routes_empty(self):
self.assertEqual([], pipelock_route_hosts(_bottle({})))
self.assertEqual(len(eff), len(set(eff)))
self.assertEqual(eff, sorted(eff))
class TestAllowlistWithRoutes(unittest.TestCase):
def test_route_hosts_added_to_allowlist(self):
def test_manifest_route_hosts_present(self):
eff = pipelock_effective_allowlist(_bottle(_routes([
{"host": "registry.npmjs.org",
"auth": {"scheme": "Bearer", "token_ref": "N"}},
@@ -67,6 +53,15 @@ class TestAllowlistWithRoutes(unittest.TestCase):
self.assertIn("registry.npmjs.org", eff)
self.assertIn("api.github.com", eff)
def test_baked_defaults_still_present_alongside_manifest_routes(self):
eff = pipelock_effective_allowlist(_bottle(_routes([
{"host": "x.example",
"auth": {"scheme": "Bearer", "token_ref": "T"}},
])))
for default in ("api.anthropic.com", "sentry.io"):
self.assertIn(default, eff)
self.assertIn("x.example", eff)
def test_egress_proxy_hostname_NOT_in_pipelock_allowlist(self):
# The agent never dials egress-proxy via the proxy mechanism
# — it IS the proxy. Pipelock receives upstream hostnames
@@ -78,25 +73,7 @@ class TestAllowlistWithRoutes(unittest.TestCase):
])))
self.assertNotIn("egress-proxy", eff)
def test_pipelock_mirrors_egress_proxy_defaults_when_routes_present(self):
# When egress_proxy is in use, pipelock's allowlist mirrors
# the egress-proxy effective routes — which fold in
# DEFAULT_ALLOWLIST + bottle.egress.allowlist.
eff = pipelock_effective_allowlist(_bottle(_routes([
{"host": "x.example",
"auth": {"scheme": "Bearer", "token_ref": "T"}},
])))
for default in ("api.anthropic.com", "sentry.io"):
self.assertIn(default, eff)
self.assertIn("x.example", eff)
def test_supervise_hostname_auto_added_when_supervise_enabled(self):
# The agent's MCP client opens long-polled requests to
# http://supervise:9100/. They bypass the agent's HTTP_PROXY
# (via NO_PROXY=supervise) and shouldn't traverse pipelock;
# but for the launch path where supervise traffic does flow
# through pipelock (egress-proxy → ... → supervise edge
# cases), the hostname needs to be on the allowlist anyway.
eff = pipelock_effective_allowlist(_bottle({"supervise": True}))
self.assertIn("supervise", eff)
@@ -114,7 +91,6 @@ class TestAllowlistWithRoutes(unittest.TestCase):
{"host": "github.com", "path_allowlist": ["/x/", "/y/"]},
])))
self.assertIn("github.com", eff)
# The path strings don't leak into the allowlist.
for entry in eff:
self.assertFalse(entry.startswith("/"))
@@ -125,10 +101,6 @@ class TestTlsPassthrough(unittest.TestCase):
self.assertEqual(["api.anthropic.com"], passthrough)
def test_route_hosts_NOT_added_to_passthrough(self):
# egress-proxy trusts pipelock's per-bottle CA, so pipelock
# MITMs and body-scans the egress-proxy → upstream leg the
# same way it scanned direct agent traffic before. Auto-adding
# route hosts to passthrough would silently disable that.
passthrough = pipelock_effective_tls_passthrough(_bottle(_routes([
{"host": "api.github.com",
"auth": {"scheme": "Bearer", "token_ref": "G"}},
+2 -2
View File
@@ -31,7 +31,7 @@ class TestBuildConfig(unittest.TestCase):
self.assertEqual(
{"include_defaults": True, "scan_env": True}, cfg["dlp"]
)
# Default body-scan action is "block" — see BottleEgress.dlp_action.
# Body-scan action is hard-coded "block" in pipelock_build_config.
self.assertEqual(
{"action": "block"}, cfg["request_body_scanning"]
)
@@ -158,7 +158,7 @@ class TestRenderAndWrite(unittest.TestCase):
"MY_SECRET": "literal-value-should-not-appear",
"ANOTHER": "?prompt-message",
},
"egress": {"allowlist": ["github.com"]},
"egress_proxy": {"routes": [{"host": "github.com"}]},
}
},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},