fix(egress): remove implicit provider routes
test / unit (pull_request) Successful in 33s
test / integration (pull_request) Successful in 58s

This commit is contained in:
2026-05-28 19:04:49 -04:00
parent 9399626ba6
commit c31845a5b8
5 changed files with 35 additions and 90 deletions
+8 -39
View File
@@ -127,23 +127,6 @@ class EgressPlan:
pipelock_proxy_url: str = ""
# Hosts the agent needs by default for claude-code itself. Folded
# into every bottle's egress routes table as bare-pass entries
# (no auth, no path filter) so the agent reaches them without each
# bottle having to opt in. Pipelock used to own this list; PRD 0017
# moves it to egress because egress is the primary gate
# now and pipelock's allowlist is mirrored from egress.
DEFAULT_ALLOWLIST: tuple[str, ...] = (
"api.anthropic.com",
"statsig.anthropic.com",
"sentry.io",
"claude.ai",
"platform.claude.com",
"downloads.claude.ai",
"raw.githubusercontent.com",
)
def egress_manifest_routes(
bottle: Bottle,
) -> tuple[EgressRoute, ...]:
@@ -157,10 +140,9 @@ def egress_manifest_routes(
shares slot 0. Unauthenticated routes (`auth` omitted) contribute
no slot.
Does NOT include the folded-in DEFAULT_ALLOWLIST /
bottle.egress.allowlist bare-pass entries — see
`egress_routes_for_bottle` for the effective set the
addon enforces."""
This is the effective set the addon enforces. Provider runtime
routes are intentionally not injected implicitly; every allowed
host must come from the home-owned bottle manifest."""
out: list[EgressRoute] = []
slot_for_token: dict[str, str] = {}
for r in bottle.egress.routes:
@@ -189,26 +171,14 @@ def egress_manifest_routes(
def egress_routes_for_bottle(
bottle: Bottle,
) -> tuple[EgressRoute, ...]:
"""Effective egress routes: manifest routes followed by
bare-pass entries for DEFAULT_ALLOWLIST hosts. This is what
gets rendered into routes.yaml + what the addon enforces.
"""Effective egress routes. 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.
Operators that want to allow an arbitrary host that isn't in
DEFAULT_ALLOWLIST declare it directly in
`bottle.egress.routes` as a bare-pass entry
Operators that want to allow a host declare it directly in
`bottle.egress.routes` as an authenticated route or bare-pass entry
(`- host: <name>`). The legacy `bottle.egress.allowlist`
folding is gone — egress is the single allowlist surface."""
out: list[EgressRoute] = list(egress_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(EgressRoute(host=host))
claimed.add(host.lower())
return tuple(out)
return egress_manifest_routes(bottle)
def egress_token_env_map(
@@ -327,7 +297,6 @@ class Egress(ABC):
)
__all__ = [
"DEFAULT_ALLOWLIST",
"EGRESS_HOSTNAME",
"EGRESS_ROUTES_IN_CONTAINER",
"Egress",
+6 -12
View File
@@ -21,11 +21,7 @@ from dataclasses import dataclass
from pathlib import Path
from typing import cast
from .egress import (
DEFAULT_ALLOWLIST,
EGRESS_HOSTNAME,
egress_routes_for_bottle,
)
from .egress import EGRESS_HOSTNAME, egress_routes_for_bottle
from .supervise import SUPERVISE_HOSTNAME
from .manifest import Bottle
@@ -67,12 +63,11 @@ PIPELOCK_HOSTNAME = "pipelock"
def pipelock_effective_allowlist(bottle: Bottle) -> list[str]:
"""Hostnames pipelock allows. Sorted for stability.
Always mirrors `egress_routes_for_bottle(bottle)` — the
egress is the single allowlist surface; pipelock's
allowlist is the downstream copy for defense-in-depth + DLP
body scanning. For bottles without any `egress.routes[]`
declared, this is just the baked DEFAULT_ALLOWLIST that
egress_routes_for_bottle always folds in.
Always mirrors `egress_routes_for_bottle(bottle)` — egress is the
single allowlist surface, and pipelock's allowlist is the downstream
copy for defense-in-depth + DLP body scanning. For bottles without
any `egress.routes[]` declared, this is empty except for supervise
sidecar traffic when `supervise: true`.
The supervise sidecar's hostname is auto-added when supervise
is enabled (sibling-sidecar traffic that flows through pipelock
@@ -354,4 +349,3 @@ class PipelockProxy:
yaml_path.write_text(pipelock_render_yaml(cfg))
yaml_path.chmod(0o600)
return PipelockProxyPlan(yaml_path=yaml_path, slug=slug)
+12 -22
View File
@@ -4,7 +4,6 @@ resolution (PRD 0017)."""
import unittest
from bot_bottle.egress import (
DEFAULT_ALLOWLIST,
egress_manifest_routes,
egress_render_routes,
egress_resolve_token_values,
@@ -85,37 +84,28 @@ class TestRoutesForBottle(unittest.TestCase):
self.assertEqual("", routes[1].token_env)
class TestRoutesForBottleFoldsDefaults(unittest.TestCase):
"""The effective route table includes DEFAULT_ALLOWLIST +
bottle.egress.allowlist as bare-pass entries — pipelock's
allowlist is a mirror of this set."""
class TestRoutesForBottleUsesManifestOnly(unittest.TestCase):
"""The effective route table is exactly the manifest-declared
routes. Provider defaults are not injected implicitly."""
def test_defaults_present_when_no_manifest_routes(self):
def test_no_manifest_routes_means_no_effective_routes(self):
b = _bottle([])
hosts = [r.host for r in egress_routes_for_bottle(b)]
for default in DEFAULT_ALLOWLIST:
self.assertIn(default, hosts)
self.assertEqual((), egress_routes_for_bottle(b))
def test_manifest_route_wins_over_default(self):
# api.anthropic.com is in DEFAULT_ALLOWLIST. A manifest
# route for the same host takes precedence — we want the
# auth config to apply, not a duplicate bare-pass entry.
def test_manifest_route_preserved_with_auth(self):
b = _bottle([{
"host": "api.anthropic.com",
"auth": {"scheme": "Bearer", "token_ref": "T"},
}])
routes = egress_routes_for_bottle(b)
anthropic = [r for r in routes if r.host == "api.anthropic.com"]
self.assertEqual(1, len(anthropic))
self.assertEqual("Bearer", anthropic[0].auth_scheme)
self.assertEqual(1, len(routes))
self.assertEqual("api.anthropic.com", routes[0].host)
self.assertEqual("Bearer", routes[0].auth_scheme)
def test_manifest_only_when_no_defaults_or_allowlist(self):
# Sanity: egress_manifest_routes returns just the
# manifest entries — defaults are added by the
# _routes_for_bottle wrapper.
def test_manifest_only(self):
b = _bottle([{"host": "x.example"}])
manifest = [r.host for r in egress_manifest_routes(b)]
self.assertEqual(["x.example"], manifest)
effective = [r.host for r in egress_routes_for_bottle(b)]
self.assertEqual(["x.example"], effective)
class TestTokenEnvMap(unittest.TestCase):
+7 -14
View File
@@ -1,7 +1,7 @@
"""Unit: pipelock_effective_allowlist — pipelock's allowlist
mirrors `egress_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)."""
mirrors manifest-declared egress routes. Git upstreams declared in
`bottle.git` don't contribute; they flow through the per-agent
git-gate (PRD 0008)."""
import unittest
@@ -24,16 +24,11 @@ def _routes(routes):
class TestEffectiveAllowlist(unittest.TestCase):
def test_default_allowlist_present_without_any_manifest_routes(self):
# No egress routes declared → pipelock allowlist is
# just the baked DEFAULT_ALLOWLIST (folded in by
# egress_routes_for_bottle).
def test_empty_without_any_manifest_routes(self):
eff = pipelock_effective_allowlist(_bottle({}))
self.assertIn("api.anthropic.com", eff)
self.assertIn("sentry.io", eff)
self.assertEqual([], eff)
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"}},
@@ -53,14 +48,12 @@ 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):
def test_no_baked_defaults_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)
self.assertEqual(["x.example"], eff)
def test_egress_hostname_NOT_in_pipelock_allowlist(self):
# The agent never dials egress via the proxy mechanism
+2 -3
View File
@@ -42,9 +42,8 @@ class TestBuildConfig(unittest.TestCase):
},
cfg["request_body_scanning"],
)
# Baked defaults always present.
self.assertIn("api.anthropic.com", cast(list[str], cfg["api_allowlist"]))
self.assertIn("raw.githubusercontent.com", cast(list[str], cfg["api_allowlist"]))
# No provider defaults are injected implicitly.
self.assertEqual([], cast(list[str], cfg["api_allowlist"]))
# pipelock has no SSH carve-outs at all — neither
# trusted_domains nor ssrf are emitted from bottle data.
self.assertNotIn("trusted_domains", cfg)