refactor: rename egress-proxy → egress everywhere
test / unit (pull_request) Successful in 17s
test / integration (pull_request) Successful in 1m10s

The manifest key is `egress:` now; finish the rename so the rest of
the codebase matches. Files (Dockerfile.egress, claude_bottle/egress.py
etc.), classes (Egress, EgressConfig, EgressRoute, EgressPlan,
DockerEgress), constants (EGRESS_HOSTNAME, EGRESS_ROUTES, ...),
container name prefix (claude-bottle-egress-*), docker network alias
(egress), the introspection host (_egress.local), the MCP tool IDs
(egress-block, list-egress-routes), and the preflight label all drop
the `-proxy` suffix.
This commit is contained in:
2026-05-25 21:59:47 -04:00
parent 14c8a51c16
commit 1e5b0dcfca
30 changed files with 583 additions and 583 deletions
+9 -9
View File
@@ -196,29 +196,29 @@ class TestSuperviseSidecar(unittest.TestCase):
names = {t["name"] for t in result["result"]["tools"]}
self.assertEqual(
{
_sv.TOOL_EGRESS_PROXY_BLOCK,
_sv.TOOL_EGRESS_BLOCK,
_sv.TOOL_PIPELOCK_BLOCK,
_sv.TOOL_CAPABILITY_BLOCK,
_sv.TOOL_LIST_EGRESS_PROXY_ROUTES,
_sv.TOOL_LIST_EGRESS_ROUTES,
},
names,
)
def test_tools_call_round_trips_through_queue(self):
"""End-to-end: agent in the bottle calls egress-proxy-block;
"""End-to-end: agent in the bottle calls egress-block;
the call blocks on the queue; the host approves via the
dashboard helpers; the agent receives the approval.
This test focuses on the supervise sidecar's queue + response
plumbing, not the egress-proxy apply path itself. The apply
plumbing, not the egress apply path itself. The apply
function is stubbed so we don't need to bring up a real
egress-proxy sidecar (its docker lifecycle has its own
egress sidecar (its docker lifecycle has its own
integration coverage)."""
self._require_bind_mount_sharing()
self._bring_up_sidecar()
# Stub the apply step. The dashboard's approve() calls
# add_route to docker-exec into the egress-proxy sidecar;
# add_route to docker-exec into the egress sidecar;
# this test isn't exercising the real sidecar, so patch it
# to a no-op that returns plausible before/after strings
# the audit-log writer can render.
@@ -234,7 +234,7 @@ class TestSuperviseSidecar(unittest.TestCase):
captured["response"] = self._curl_jsonrpc({
"jsonrpc": "2.0", "id": 7, "method": "tools/call",
"params": {
"name": _sv.TOOL_EGRESS_PROXY_BLOCK,
"name": _sv.TOOL_EGRESS_BLOCK,
"arguments": {
"host": "api.example.com",
"justification": "integration test",
@@ -260,12 +260,12 @@ class TestSuperviseSidecar(unittest.TestCase):
self.assertIsNotNone(qp, "proposal never appeared in queue")
assert qp is not None # type-narrowing
self.assertEqual(
_sv.TOOL_EGRESS_PROXY_BLOCK, qp.proposal.tool,
_sv.TOOL_EGRESS_BLOCK, qp.proposal.tool,
)
self.assertEqual("integration test", qp.proposal.justification)
# Approve via the dashboard helper. The apply step (now
# stubbed) would docker-exec into the egress-proxy sidecar
# stubbed) would docker-exec into the egress sidecar
# and SIGHUP it. The supervise sidecar sees the response
# file and returns to the curl caller.
dashboard.approve(qp, notes="lgtm from integration test")
+35 -35
View File
@@ -17,7 +17,7 @@ from pathlib import Path
from claude_bottle import supervise
from claude_bottle.backend.docker.capability_apply import CapabilityApplyError
from claude_bottle.backend.docker.egress_proxy_apply import EgressProxyApplyError
from claude_bottle.backend.docker.egress_apply import EgressApplyError
from claude_bottle.backend.docker.pipelock_apply import PipelockApplyError
from claude_bottle.cli import dashboard
from claude_bottle.supervise import (
@@ -26,7 +26,7 @@ from claude_bottle.supervise import (
STATUS_MODIFIED,
STATUS_REJECTED,
TOOL_CAPABILITY_BLOCK,
TOOL_EGRESS_PROXY_BLOCK,
TOOL_EGRESS_BLOCK,
TOOL_PIPELOCK_BLOCK,
read_audit_entries,
read_response,
@@ -37,13 +37,13 @@ from claude_bottle.supervise import (
FIXED = datetime(2026, 5, 25, 12, 0, 0, tzinfo=timezone.utc)
def _proposal(slug: str = "dev", tool: str = TOOL_EGRESS_PROXY_BLOCK) -> Proposal:
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_PROXY_BLOCK: '{"routes": []}\n',
TOOL_EGRESS_BLOCK: '{"routes": []}\n',
TOOL_PIPELOCK_BLOCK: "https://example.com/path",
TOOL_CAPABILITY_BLOCK: "FROM python:3.13\n",
}
@@ -95,13 +95,13 @@ class TestDiscoverPending(_FakeHomeMixin, unittest.TestCase):
def test_sorted_by_arrival_across_bottles(self):
early = Proposal.new(
bottle_slug="api", tool=TOOL_EGRESS_PROXY_BLOCK,
bottle_slug="api", tool=TOOL_EGRESS_BLOCK,
proposed_file="{}", justification="early",
current_file_hash="h",
now=datetime(2026, 5, 25, 10, 0, 0, tzinfo=timezone.utc),
)
late = Proposal.new(
bottle_slug="dev", tool=TOOL_EGRESS_PROXY_BLOCK,
bottle_slug="dev", tool=TOOL_EGRESS_BLOCK,
proposed_file="{}", justification="late",
current_file_hash="h",
now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc),
@@ -151,7 +151,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
dashboard.apply_capability_change = self._original_apply_capability
self._teardown_fake_home()
def _enqueue(self, tool: str = TOOL_EGRESS_PROXY_BLOCK):
def _enqueue(self, tool: str = TOOL_EGRESS_BLOCK):
p = _proposal(tool=tool)
qdir = supervise.queue_dir_for_slug("dev")
qdir.mkdir(parents=True, exist_ok=True)
@@ -164,7 +164,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
resp = read_response(qp.queue_dir, qp.proposal.id)
self.assertEqual(STATUS_APPROVED, resp.status)
self.assertIsNone(resp.final_file)
entries = read_audit_entries("egress-proxy", "dev")
entries = read_audit_entries("egress", "dev")
self.assertEqual(1, len(entries))
self.assertEqual("approved", entries[0].operator_action)
@@ -175,7 +175,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
self.assertEqual(STATUS_MODIFIED, resp.status)
self.assertEqual('{"routes": [{"path": "/x/"}]}\n', resp.final_file)
self.assertEqual("tweaked", resp.notes)
entries = read_audit_entries("egress-proxy", "dev")
entries = read_audit_entries("egress", "dev")
self.assertEqual("modified", entries[0].operator_action)
def test_reject_writes_rejection(self):
@@ -184,7 +184,7 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
resp = read_response(qp.queue_dir, qp.proposal.id)
self.assertEqual(STATUS_REJECTED, resp.status)
self.assertEqual("nope", resp.notes)
entries = read_audit_entries("egress-proxy", "dev")
entries = read_audit_entries("egress", "dev")
self.assertEqual("rejected", entries[0].operator_action)
self.assertEqual("nope", entries[0].operator_notes)
@@ -193,18 +193,18 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
dashboard.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-proxy", "dev"))
self.assertEqual([], read_audit_entries("egress", "dev"))
self.assertEqual([], read_audit_entries("pipelock", "dev"))
def test_pipelock_audit_distinct_from_egress_proxy(self):
def test_pipelock_audit_distinct_from_egress(self):
qp = self._enqueue(tool=TOOL_PIPELOCK_BLOCK)
dashboard.approve(qp)
self.assertEqual(1, len(read_audit_entries("pipelock", "dev")))
self.assertEqual(0, len(read_audit_entries("egress-proxy", "dev")))
self.assertEqual(0, len(read_audit_entries("egress", "dev")))
class TestEgressProxyApplyWiring(_FakeHomeMixin, unittest.TestCase):
"""PRD 0017 chunk 3: approve() on an egress-proxy-block proposal
class TestEgressApplyWiring(_FakeHomeMixin, unittest.TestCase):
"""PRD 0017 chunk 3: approve() on an egress-block proposal
must call add_route (single-route merge) with the right args
and surface its failures."""
@@ -216,9 +216,9 @@ class TestEgressProxyApplyWiring(_FakeHomeMixin, unittest.TestCase):
dashboard.add_route = self._original_add_route
self._teardown_fake_home()
def _enqueue_egress_proxy(self, proposed: str = '{"host": "x.example"}\n'):
def _enqueue_egress(self, proposed: str = '{"host": "x.example"}\n'):
p = Proposal.new(
bottle_slug="dev", tool=TOOL_EGRESS_PROXY_BLOCK,
bottle_slug="dev", tool=TOOL_EGRESS_BLOCK,
proposed_file=proposed,
justification="need a route",
current_file_hash=sha256_hex(proposed),
@@ -229,12 +229,12 @@ class TestEgressProxyApplyWiring(_FakeHomeMixin, unittest.TestCase):
supervise.write_proposal(qdir, p)
return dashboard.QueuedProposal(proposal=p, queue_dir=qdir)
def test_egress_proxy_block_calls_add_route_with_proposed_json(self):
def test_egress_block_calls_add_route_with_proposed_json(self):
calls = []
dashboard.add_route = lambda slug, content: (
calls.append((slug, content)) or ("before", "after")
)
qp = self._enqueue_egress_proxy(
qp = self._enqueue_egress(
proposed='{"host": "new.example", "path_allowlist": ["/x/"]}\n'
)
dashboard.approve(qp)
@@ -253,7 +253,7 @@ class TestEgressProxyApplyWiring(_FakeHomeMixin, unittest.TestCase):
dashboard.add_route = lambda slug, content: (
calls.append(content) or ("before", "after")
)
qp = self._enqueue_egress_proxy()
qp = self._enqueue_egress()
dashboard.approve(
qp,
final_file='{"host": "edited.example"}\n',
@@ -263,10 +263,10 @@ class TestEgressProxyApplyWiring(_FakeHomeMixin, unittest.TestCase):
def test_apply_failure_blocks_response_and_audit(self):
dashboard.add_route = lambda slug, content: (_ for _ in ()).throw(
EgressProxyApplyError("docker exec failed")
EgressApplyError("docker exec failed")
)
qp = self._enqueue_egress_proxy()
with self.assertRaises(EgressProxyApplyError):
qp = self._enqueue_egress()
with self.assertRaises(EgressApplyError):
dashboard.approve(qp)
# No response file (proposal stays pending).
self.assertEqual(
@@ -274,16 +274,16 @@ class TestEgressProxyApplyWiring(_FakeHomeMixin, unittest.TestCase):
[p.id for p in supervise.list_pending_proposals(qp.queue_dir)],
)
# No audit entry.
self.assertEqual([], read_audit_entries("egress-proxy", "dev"))
self.assertEqual([], read_audit_entries("egress", "dev"))
def test_real_diff_lands_in_audit(self):
dashboard.add_route = lambda slug, content: (
'{"routes": []}\n', # before
'{"routes": [{"host": "new.example"}]}\n', # after
)
qp = self._enqueue_egress_proxy(proposed='{"host": "new.example"}\n')
qp = self._enqueue_egress(proposed='{"host": "new.example"}\n')
dashboard.approve(qp)
entries = read_audit_entries("egress-proxy", "dev")
entries = read_audit_entries("egress", "dev")
self.assertEqual(1, len(entries))
self.assertIn('+{"routes": [{"host": "new.example"}]}', entries[0].diff)
self.assertIn('-{"routes": []}', entries[0].diff)
@@ -293,13 +293,13 @@ class TestEgressProxyApplyWiring(_FakeHomeMixin, unittest.TestCase):
dashboard.apply_routes_change = lambda slug, content: (
called.append(True) or ("", content)
)
qp = self._enqueue_egress_proxy()
qp = self._enqueue_egress()
dashboard.reject(qp, reason="no thanks")
self.assertEqual([], called)
# Reject still writes a response + audit entry with empty diff.
resp = read_response(qp.queue_dir, qp.proposal.id)
self.assertEqual(STATUS_REJECTED, resp.status)
entries = read_audit_entries("egress-proxy", "dev")
entries = read_audit_entries("egress", "dev")
self.assertEqual(1, len(entries))
self.assertEqual("", entries[0].diff)
@@ -443,7 +443,7 @@ class TestCapabilityApplyWiring(_FakeHomeMixin, unittest.TestCase):
dashboard.approve(qp)
# 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-proxy", "dev"))
self.assertEqual([], read_audit_entries("egress", "dev"))
self.assertEqual([], read_audit_entries("pipelock", "dev"))
def test_proposal_archived_after_apply(self):
@@ -475,7 +475,7 @@ class TestOperatorEditRoutes(_FakeHomeMixin, unittest.TestCase):
'{"routes": []}\n', content,
)
dashboard.operator_edit_routes("dev", '{"routes": [{"path": "/x/"}]}\n')
entries = read_audit_entries("egress-proxy", "dev")
entries = read_audit_entries("egress", "dev")
self.assertEqual(1, len(entries))
self.assertEqual(supervise.ACTION_OPERATOR_EDIT, entries[0].operator_action)
self.assertEqual("", entries[0].justification)
@@ -483,14 +483,14 @@ class TestOperatorEditRoutes(_FakeHomeMixin, unittest.TestCase):
def test_failure_does_not_write_audit(self):
dashboard.apply_routes_change = lambda slug, content: (_ for _ in ()).throw(
EgressProxyApplyError("nope")
EgressApplyError("nope")
)
with self.assertRaises(EgressProxyApplyError):
with self.assertRaises(EgressApplyError):
dashboard.operator_edit_routes("dev", '{"routes": []}\n')
self.assertEqual([], read_audit_entries("egress-proxy", "dev"))
self.assertEqual([], read_audit_entries("egress", "dev"))
class TestDiscoverEgressProxySlugs(unittest.TestCase):
class TestDiscoverEgressSlugs(unittest.TestCase):
"""Slug-extraction parsing — exercises only the parsing path; the
docker ps invocation itself is environment-dependent (and tested
implicitly by the integration test)."""
@@ -502,7 +502,7 @@ class TestDiscoverEgressProxySlugs(unittest.TestCase):
original = os.environ.get("PATH", "")
os.environ["PATH"] = "/nonexistent-no-docker-here"
try:
self.assertEqual([], dashboard.discover_egress_proxy_slugs())
self.assertEqual([], dashboard.discover_egress_slugs())
self.assertEqual([], dashboard.discover_pipelock_slugs())
finally:
os.environ["PATH"] = original
+3 -3
View File
@@ -12,7 +12,7 @@ from claude_bottle.cli import dashboard
from claude_bottle.supervise import (
Proposal,
TOOL_CAPABILITY_BLOCK,
TOOL_EGRESS_PROXY_BLOCK,
TOOL_EGRESS_BLOCK,
TOOL_PIPELOCK_BLOCK,
sha256_hex,
)
@@ -46,9 +46,9 @@ class TestPipelockHostHighlight(unittest.TestCase):
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_proxy_block(self):
def test_no_green_lines_for_egress_block(self):
lines = dashboard._detail_lines(
_qp(TOOL_EGRESS_PROXY_BLOCK, '{"routes": []}'),
_qp(TOOL_EGRESS_BLOCK, '{"routes": []}'),
green_attr=self.GREEN,
)
self.assertEqual([], [t for t, a in lines if a == self.GREEN])
@@ -1,16 +1,16 @@
"""Unit: EgressProxy route lift + routes.yaml render + token
"""Unit: Egress route lift + routes.yaml render + token
resolution (PRD 0017)."""
import json
import unittest
from claude_bottle.egress_proxy import (
from claude_bottle.egress import (
DEFAULT_ALLOWLIST,
egress_proxy_manifest_routes,
egress_proxy_render_routes,
egress_proxy_resolve_token_values,
egress_proxy_routes_for_bottle,
egress_proxy_token_env_map,
egress_manifest_routes,
egress_render_routes,
egress_resolve_token_values,
egress_routes_for_bottle,
egress_token_env_map,
)
from claude_bottle.log import Die
from claude_bottle.manifest import Manifest
@@ -29,18 +29,18 @@ class TestRoutesForBottle(unittest.TestCase):
"host": "api.github.com",
"auth": {"scheme": "Bearer", "token_ref": "GH_PAT"},
}])
routes = egress_proxy_manifest_routes(b)
routes = egress_manifest_routes(b)
self.assertEqual(1, len(routes))
r = routes[0]
self.assertEqual("api.github.com", r.host)
self.assertEqual("Bearer", r.auth_scheme)
self.assertEqual("EGRESS_PROXY_TOKEN_0", r.token_env)
self.assertEqual("EGRESS_TOKEN_0", r.token_env)
self.assertEqual("GH_PAT", r.token_ref)
self.assertEqual((), r.path_allowlist)
def test_unauthenticated_route_has_empty_auth_fields(self):
b = _bottle([{"host": "github.com", "path_allowlist": ["/x/"]}])
routes = egress_proxy_manifest_routes(b)
routes = egress_manifest_routes(b)
r = routes[0]
self.assertEqual("", r.auth_scheme)
self.assertEqual("", r.token_env)
@@ -54,9 +54,9 @@ class TestRoutesForBottle(unittest.TestCase):
{"host": "github.com",
"auth": {"scheme": "Bearer", "token_ref": "GH_PAT"}},
])
routes = egress_proxy_manifest_routes(b)
routes = egress_manifest_routes(b)
slots = {r.token_env for r in routes}
self.assertEqual({"EGRESS_PROXY_TOKEN_0"}, slots)
self.assertEqual({"EGRESS_TOKEN_0"}, slots)
def test_distinct_token_refs_get_distinct_slots(self):
b = _bottle([
@@ -65,9 +65,9 @@ class TestRoutesForBottle(unittest.TestCase):
{"host": "b.example",
"auth": {"scheme": "Bearer", "token_ref": "T2"}},
])
routes = egress_proxy_manifest_routes(b)
routes = egress_manifest_routes(b)
slots = [r.token_env for r in routes]
self.assertEqual(["EGRESS_PROXY_TOKEN_0", "EGRESS_PROXY_TOKEN_1"], slots)
self.assertEqual(["EGRESS_TOKEN_0", "EGRESS_TOKEN_1"], slots)
def test_unauthenticated_routes_dont_consume_slots(self):
# A bare-pass route between two authenticated routes mustn't
@@ -79,9 +79,9 @@ class TestRoutesForBottle(unittest.TestCase):
{"host": "b.example",
"auth": {"scheme": "Bearer", "token_ref": "T2"}},
])
routes = egress_proxy_manifest_routes(b)
routes = egress_manifest_routes(b)
authed = [r.token_env for r in routes if r.token_env]
self.assertEqual(["EGRESS_PROXY_TOKEN_0", "EGRESS_PROXY_TOKEN_1"], authed)
self.assertEqual(["EGRESS_TOKEN_0", "EGRESS_TOKEN_1"], authed)
self.assertEqual("", routes[1].token_env)
@@ -92,7 +92,7 @@ class TestRoutesForBottleFoldsDefaults(unittest.TestCase):
def test_defaults_present_when_no_manifest_routes(self):
b = _bottle([])
hosts = [r.host for r in egress_proxy_routes_for_bottle(b)]
hosts = [r.host for r in egress_routes_for_bottle(b)]
for default in DEFAULT_ALLOWLIST:
self.assertIn(default, hosts)
@@ -104,17 +104,17 @@ class TestRoutesForBottleFoldsDefaults(unittest.TestCase):
"host": "api.anthropic.com",
"auth": {"scheme": "Bearer", "token_ref": "T"},
}])
routes = egress_proxy_routes_for_bottle(b)
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)
def test_manifest_only_when_no_defaults_or_allowlist(self):
# Sanity: egress_proxy_manifest_routes returns just the
# Sanity: egress_manifest_routes returns just the
# manifest entries — defaults are added by the
# _routes_for_bottle wrapper.
b = _bottle([{"host": "x.example"}])
manifest = [r.host for r in egress_proxy_manifest_routes(b)]
manifest = [r.host for r in egress_manifest_routes(b)]
self.assertEqual(["x.example"], manifest)
@@ -125,12 +125,12 @@ class TestTokenEnvMap(unittest.TestCase):
"auth": {"scheme": "Bearer", "token_ref": "T1"}},
{"host": "passthrough.example"},
])
routes = egress_proxy_manifest_routes(b)
m = egress_proxy_token_env_map(routes)
self.assertEqual({"EGRESS_PROXY_TOKEN_0": "T1"}, m)
routes = egress_manifest_routes(b)
m = egress_token_env_map(routes)
self.assertEqual({"EGRESS_TOKEN_0": "T1"}, m)
def test_no_routes_empty(self):
self.assertEqual({}, egress_proxy_token_env_map(()))
self.assertEqual({}, egress_token_env_map(()))
class TestRenderRoutes(unittest.TestCase):
@@ -140,14 +140,14 @@ class TestRenderRoutes(unittest.TestCase):
"auth": {"scheme": "Bearer", "token_ref": "GH_PAT"},
"path_allowlist": ["/repos/x/"],
}])
routes = egress_proxy_manifest_routes(b)
payload = json.loads(egress_proxy_render_routes(routes))
routes = egress_manifest_routes(b)
payload = json.loads(egress_render_routes(routes))
self.assertEqual(
[{
"host": "api.github.com",
"path_allowlist": ["/repos/x/"],
"auth_scheme": "Bearer",
"token_env": "EGRESS_PROXY_TOKEN_0",
"token_env": "EGRESS_TOKEN_0",
}],
payload["routes"],
)
@@ -158,8 +158,8 @@ class TestRenderRoutes(unittest.TestCase):
# enforces both-or-neither, so emitting empty strings would
# round-trip as a partial pair and crash.
b = _bottle([{"host": "github.com", "path_allowlist": ["/x/"]}])
routes = egress_proxy_manifest_routes(b)
payload = json.loads(egress_proxy_render_routes(routes))
routes = egress_manifest_routes(b)
payload = json.loads(egress_render_routes(routes))
entry = payload["routes"][0]
self.assertNotIn("auth_scheme", entry)
self.assertNotIn("token_env", entry)
@@ -169,14 +169,14 @@ class TestRenderRoutes(unittest.TestCase):
"host": "api.anthropic.com",
"auth": {"scheme": "Bearer", "token_ref": "CL"},
}])
routes = egress_proxy_manifest_routes(b)
payload = json.loads(egress_proxy_render_routes(routes))
routes = egress_manifest_routes(b)
payload = json.loads(egress_render_routes(routes))
self.assertNotIn("path_allowlist", payload["routes"][0])
def test_round_trip_through_addon_core(self):
# Render here → parse in the addon must succeed for every
# combination the manifest can produce.
from claude_bottle.egress_proxy_addon_core import load_routes
from claude_bottle.egress_addon_core import load_routes
b = _bottle([
{"host": "api.github.com",
"auth": {"scheme": "Bearer", "token_ref": "GH_PAT"},
@@ -184,34 +184,34 @@ class TestRenderRoutes(unittest.TestCase):
{"host": "github.com", "path_allowlist": ["/x/"]},
{"host": "api.anthropic.com"},
])
routes = egress_proxy_manifest_routes(b)
addon_routes = load_routes(egress_proxy_render_routes(routes))
routes = egress_manifest_routes(b)
addon_routes = load_routes(egress_render_routes(routes))
self.assertEqual(3, len(addon_routes))
self.assertEqual("Bearer", addon_routes[0].auth_scheme)
self.assertEqual("EGRESS_PROXY_TOKEN_0", addon_routes[0].token_env)
self.assertEqual("EGRESS_TOKEN_0", addon_routes[0].token_env)
self.assertEqual("", addon_routes[1].auth_scheme)
self.assertEqual("", addon_routes[2].auth_scheme)
class TestResolveTokenValues(unittest.TestCase):
def test_reads_host_env(self):
out = egress_proxy_resolve_token_values(
{"EGRESS_PROXY_TOKEN_0": "GH_PAT"},
out = egress_resolve_token_values(
{"EGRESS_TOKEN_0": "GH_PAT"},
{"GH_PAT": "the-value"},
)
self.assertEqual({"EGRESS_PROXY_TOKEN_0": "the-value"}, out)
self.assertEqual({"EGRESS_TOKEN_0": "the-value"}, out)
def test_missing_token_ref_dies(self):
with self.assertRaises(Die):
egress_proxy_resolve_token_values(
{"EGRESS_PROXY_TOKEN_0": "GH_PAT"},
egress_resolve_token_values(
{"EGRESS_TOKEN_0": "GH_PAT"},
{},
)
def test_empty_token_ref_dies(self):
with self.assertRaises(Die):
egress_proxy_resolve_token_values(
{"EGRESS_PROXY_TOKEN_0": "GH_PAT"},
egress_resolve_token_values(
{"EGRESS_TOKEN_0": "GH_PAT"},
{"GH_PAT": ""},
)
@@ -1,12 +1,12 @@
"""Unit: pure-logic core of the egress-proxy mitmproxy addon (PRD 0017).
"""Unit: pure-logic core of the egress mitmproxy addon (PRD 0017).
These tests target `egress_proxy_addon_core` the host-importable
These tests target `egress_addon_core` the host-importable
half of the addon. The mitmproxy hook wrapper in
`egress_proxy_addon.py` is container-only and is not exercised here."""
`egress_addon.py` is container-only and is not exercised here."""
import unittest
from claude_bottle.egress_proxy_addon_core import (
from claude_bottle.egress_addon_core import (
Decision,
Route,
decide,
@@ -34,12 +34,12 @@ class TestParseRoutes(unittest.TestCase):
"host": "api.github.com",
"path_allowlist": ["/repos/x/", "/users/x"],
"auth_scheme": "Bearer",
"token_env": "EGRESS_PROXY_TOKEN_0",
"token_env": "EGRESS_TOKEN_0",
}]})
r = routes[0]
self.assertEqual(("/repos/x/", "/users/x"), r.path_allowlist)
self.assertEqual("Bearer", r.auth_scheme)
self.assertEqual("EGRESS_PROXY_TOKEN_0", r.token_env)
self.assertEqual("EGRESS_TOKEN_0", r.token_env)
def test_order_preserved(self):
# Host match is exact (not longest-prefix), but the file order
@@ -69,7 +69,7 @@ class TestParseRoutes(unittest.TestCase):
with self.assertRaises(ValueError) as cm:
parse_routes({"routes": [{
"host": "x.example",
"token_env": "EGRESS_PROXY_TOKEN_0",
"token_env": "EGRESS_TOKEN_0",
}]})
self.assertIn("both set or both empty", str(cm.exception))
@@ -161,9 +161,9 @@ class TestMatchRoute(unittest.TestCase):
class TestDecide(unittest.TestCase):
def test_no_matching_route_blocks(self):
# Defense-in-depth: egress-proxy gates the bottle's allowlist
# Defense-in-depth: egress gates the bottle's allowlist
# too, not just pipelock. Any host the operator didn't declare
# in egress_proxy.routes is 403'd at egress-proxy before it
# in egress.routes is 403'd at egress before it
# ever reaches pipelock.
d = decide((), "elsewhere.example", "/anything", {})
self.assertEqual("block", d.action)
@@ -197,8 +197,8 @@ class TestDecide(unittest.TestCase):
def test_auth_injection_uses_environ_value(self):
d = decide(
(Route(host="api.github.com", auth_scheme="Bearer",
token_env="EGRESS_PROXY_TOKEN_0"),),
"api.github.com", "/repos/x", {"EGRESS_PROXY_TOKEN_0": "the-token"},
token_env="EGRESS_TOKEN_0"),),
"api.github.com", "/repos/x", {"EGRESS_TOKEN_0": "the-token"},
)
self.assertEqual("forward", d.action)
self.assertEqual("Bearer the-token", d.inject_authorization)
@@ -210,11 +210,11 @@ class TestDecide(unittest.TestCase):
# request the upstream would reject.
d = decide(
(Route(host="api.github.com", auth_scheme="Bearer",
token_env="EGRESS_PROXY_TOKEN_0"),),
token_env="EGRESS_TOKEN_0"),),
"api.github.com", "/repos/x", {},
)
self.assertEqual("block", d.action)
self.assertIn("EGRESS_PROXY_TOKEN_0", d.reason)
self.assertIn("EGRESS_TOKEN_0", d.reason)
def test_auth_with_empty_token_env_blocks(self):
# Empty env var is treated the same as unset — we don't inject
@@ -222,8 +222,8 @@ class TestDecide(unittest.TestCase):
# upstream rate limit with a 401.
d = decide(
(Route(host="api.github.com", auth_scheme="Bearer",
token_env="EGRESS_PROXY_TOKEN_0"),),
"api.github.com", "/repos/x", {"EGRESS_PROXY_TOKEN_0": ""},
token_env="EGRESS_TOKEN_0"),),
"api.github.com", "/repos/x", {"EGRESS_TOKEN_0": ""},
)
self.assertEqual("block", d.action)
@@ -240,8 +240,8 @@ class TestDecide(unittest.TestCase):
# go-gitea/gitea#16734). The addon is scheme-agnostic.
d = decide(
(Route(host="git.example", auth_scheme="token",
token_env="EGRESS_PROXY_TOKEN_0"),),
"git.example", "/api/v1/repos", {"EGRESS_PROXY_TOKEN_0": "abc"},
token_env="EGRESS_TOKEN_0"),),
"git.example", "/api/v1/repos", {"EGRESS_TOKEN_0": "abc"},
)
self.assertEqual("token abc", d.inject_authorization)
@@ -6,8 +6,8 @@ import unittest
import json
from claude_bottle.backend.docker.egress_proxy_apply import (
EgressProxyApplyError,
from claude_bottle.backend.docker.egress_apply import (
EgressApplyError,
_hosts_in_routes,
_merge_single_route,
_pipelock_safe_hosts,
@@ -27,30 +27,30 @@ class TestValidateRoutesContent(unittest.TestCase):
'{"routes": [{"host": "api.github.com",'
' "path_allowlist": ["/repos/x/"],'
' "auth_scheme": "Bearer",'
' "token_env": "EGRESS_PROXY_TOKEN_0"}]}'
' "token_env": "EGRESS_TOKEN_0"}]}'
)
def test_rejects_bad_json(self):
with self.assertRaises(EgressProxyApplyError) as cm:
with self.assertRaises(EgressApplyError) as cm:
validate_routes_content("{not json")
self.assertIn("not valid", str(cm.exception))
def test_rejects_non_object_top_level(self):
with self.assertRaises(EgressProxyApplyError):
with self.assertRaises(EgressApplyError):
validate_routes_content("[]")
def test_rejects_missing_routes_key(self):
with self.assertRaises(EgressProxyApplyError):
with self.assertRaises(EgressApplyError):
validate_routes_content('{"other": []}')
def test_rejects_non_list_routes(self):
with self.assertRaises(EgressProxyApplyError):
with self.assertRaises(EgressApplyError):
validate_routes_content('{"routes": "not a list"}')
def test_rejects_partial_auth_pair(self):
# The addon-core parser enforces both-or-neither — the apply
# path picks this up before SIGHUP'ing the sidecar.
with self.assertRaises(EgressProxyApplyError):
with self.assertRaises(EgressApplyError):
validate_routes_content(
'{"routes": [{"host": "x.example",'
' "auth_scheme": "Bearer"}]}'
@@ -83,7 +83,7 @@ class TestHostsInRoutes(unittest.TestCase):
def test_invalid_routes_raises(self):
# The mirror helper relies on parsing succeeding; bad input
# should error before pipelock is touched.
with self.assertRaises(EgressProxyApplyError):
with self.assertRaises(EgressApplyError):
_hosts_in_routes('{"routes": [{"path": "/no-host/"}]}')
@@ -115,20 +115,20 @@ class TestMergeSingleRoute(unittest.TestCase):
new_route = json.loads(merged)["routes"][-1]
self.assertEqual("Bearer", new_route["auth_scheme"])
# First auth slot when no prior auth routes exist.
self.assertEqual("EGRESS_PROXY_TOKEN_0", new_route["token_env"])
self.assertEqual("EGRESS_TOKEN_0", new_route["token_env"])
def test_auth_slot_increments_past_existing(self):
base = json.dumps({"routes": [
{"host": "api.anthropic.com",
"auth_scheme": "Bearer",
"token_env": "EGRESS_PROXY_TOKEN_0"},
"token_env": "EGRESS_TOKEN_0"},
]})
merged = _merge_single_route(base, {
"host": "api.github.com",
"auth": {"scheme": "Bearer", "token_ref": "GH"},
})
new_route = json.loads(merged)["routes"][-1]
self.assertEqual("EGRESS_PROXY_TOKEN_1", new_route["token_env"])
self.assertEqual("EGRESS_TOKEN_1", new_route["token_env"])
def test_existing_host_merges_path_allowlist_as_union(self):
base = json.dumps({"routes": [
@@ -161,7 +161,7 @@ class TestMergeSingleRoute(unittest.TestCase):
base = json.dumps({"routes": [
{"host": "api.github.com",
"auth_scheme": "Bearer",
"token_env": "EGRESS_PROXY_TOKEN_0"},
"token_env": "EGRESS_TOKEN_0"},
]})
merged = _merge_single_route(base, {
"host": "api.github.com",
@@ -169,7 +169,7 @@ class TestMergeSingleRoute(unittest.TestCase):
})
route = json.loads(merged)["routes"][0]
self.assertEqual("Bearer", route["auth_scheme"])
self.assertEqual("EGRESS_PROXY_TOKEN_0", route["token_env"])
self.assertEqual("EGRESS_TOKEN_0", route["token_env"])
def test_host_match_is_case_insensitive(self):
base = json.dumps({"routes": [{"host": "GitHub.com"}]})
@@ -182,11 +182,11 @@ class TestMergeSingleRoute(unittest.TestCase):
self.assertEqual(["/x/"], routes[0]["path_allowlist"])
def test_missing_host_raises(self):
with self.assertRaises(EgressProxyApplyError):
with self.assertRaises(EgressApplyError):
_merge_single_route(self.BASE, {})
def test_invalid_current_yaml_raises(self):
with self.assertRaises(EgressProxyApplyError):
with self.assertRaises(EgressApplyError):
_merge_single_route("{not json", {"host": "x.example"})
@@ -198,7 +198,7 @@ class TestPipelockSafeHosts(unittest.TestCase):
)
def test_drops_wildcards(self):
# Wildcard host matching was removed from egress-proxy too,
# 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.
@@ -8,7 +8,7 @@ auth omission means unauthenticated."""
import unittest
from claude_bottle.log import Die
from claude_bottle.manifest import EgressProxyRoute, Manifest
from claude_bottle.manifest import EgressRoute, Manifest
def _bottle(routes):
@@ -201,8 +201,8 @@ class TestRouteValidation(unittest.TestCase):
b = _bottle([])
self.assertEqual((), b.egress.routes)
def test_no_egress_proxy_block_means_empty(self):
# The bottle dataclass defaults to an empty EgressProxyConfig.
def test_no_egress_block_means_empty(self):
# The bottle dataclass defaults to an empty EgressConfig.
b = Manifest.from_json_obj({
"bottles": {"dev": {}},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
@@ -211,7 +211,7 @@ class TestRouteValidation(unittest.TestCase):
class TestConfigShape(unittest.TestCase):
def test_unknown_egress_proxy_key_rejected(self):
def test_unknown_egress_key_rejected(self):
with self.assertRaises(Die):
Manifest.from_json_obj({
"bottles": {"dev": {"egress": {"wat": []}}},
+1 -1
View File
@@ -31,7 +31,7 @@ _BOTTLE_DEV = """
- host: example.com
---
The dev bottle. Anthropic OAuth via egress-proxy.
The dev bottle. Anthropic OAuth via egress.
"""
_AGENT_IMPL = """
+10 -10
View File
@@ -1,5 +1,5 @@
"""Unit: pipelock_effective_allowlist — pipelock's allowlist
mirrors `egress_proxy_routes_for_bottle` (which folds in
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)."""
@@ -25,9 +25,9 @@ def _routes(routes):
class TestEffectiveAllowlist(unittest.TestCase):
def test_default_allowlist_present_without_any_manifest_routes(self):
# No egress_proxy routes declared → pipelock allowlist is
# No egress routes declared → pipelock allowlist is
# just the baked DEFAULT_ALLOWLIST (folded in by
# egress_proxy_routes_for_bottle).
# egress_routes_for_bottle).
eff = pipelock_effective_allowlist(_bottle({}))
self.assertIn("api.anthropic.com", eff)
self.assertIn("sentry.io", eff)
@@ -62,16 +62,16 @@ class TestAllowlistWithRoutes(unittest.TestCase):
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
def test_egress_hostname_NOT_in_pipelock_allowlist(self):
# The agent never dials egress via the proxy mechanism
# — it IS the proxy. Pipelock receives upstream hostnames
# from egress-proxy's CONNECT requests, not the
# `egress-proxy` hostname itself.
# from egress's CONNECT requests, not the
# `egress` hostname itself.
eff = pipelock_effective_allowlist(_bottle(_routes([
{"host": "x.example",
"auth": {"scheme": "Bearer", "token_ref": "T"}},
])))
self.assertNotIn("egress-proxy", eff)
self.assertNotIn("egress", eff)
def test_supervise_hostname_auto_added_when_supervise_enabled(self):
eff = pipelock_effective_allowlist(_bottle({"supervise": True}))
@@ -84,9 +84,9 @@ class TestAllowlistWithRoutes(unittest.TestCase):
self.assertNotIn("supervise", eff_explicit)
def test_path_allowlist_does_not_affect_pipelock_allowlist(self):
# path_allowlist is enforced by egress-proxy, not pipelock.
# path_allowlist is enforced by egress, not pipelock.
# Pipelock only sees the upstream hostname; the path filter
# has already passed (or 403'd) at egress-proxy.
# has already passed (or 403'd) at egress.
eff = pipelock_effective_allowlist(_bottle(_routes([
{"host": "github.com", "path_allowlist": ["/x/", "/y/"]},
])))
+9 -9
View File
@@ -17,7 +17,7 @@ from claude_bottle.supervise import (
STATUS_MODIFIED,
STATUS_REJECTED,
TOOL_CAPABILITY_BLOCK,
TOOL_EGRESS_PROXY_BLOCK,
TOOL_EGRESS_BLOCK,
TOOL_PIPELOCK_BLOCK,
archive_proposal,
audit_log_path,
@@ -37,7 +37,7 @@ from claude_bottle.supervise import (
FIXED_TS = datetime(2026, 5, 25, 12, 0, 0, tzinfo=timezone.utc)
def _proposal(tool: str = TOOL_EGRESS_PROXY_BLOCK, proposed: str = "{}", justification: str = "need a route") -> Proposal:
def _proposal(tool: str = TOOL_EGRESS_BLOCK, proposed: str = "{}", justification: str = "need a route") -> Proposal:
return Proposal.new(
bottle_slug="dev",
tool=tool,
@@ -54,7 +54,7 @@ class TestProposalRoundtrip(unittest.TestCase):
self.assertTrue(p.id)
self.assertEqual("2026-05-25T12:00:00+00:00", p.arrival_timestamp)
self.assertEqual("dev", p.bottle_slug)
self.assertEqual(TOOL_EGRESS_PROXY_BLOCK, p.tool)
self.assertEqual(TOOL_EGRESS_BLOCK, p.tool)
def test_to_from_dict_roundtrip(self):
p = _proposal()
@@ -139,13 +139,13 @@ class TestQueueIO(unittest.TestCase):
def test_list_pending_sorted_by_arrival(self):
# Fabricate two with explicit timestamps.
a = Proposal.new(
bottle_slug="dev", tool=TOOL_EGRESS_PROXY_BLOCK,
bottle_slug="dev", tool=TOOL_EGRESS_BLOCK,
proposed_file="{}", justification="early",
current_file_hash="x",
now=datetime(2026, 5, 25, 10, 0, 0, tzinfo=timezone.utc),
)
b = Proposal.new(
bottle_slug="dev", tool=TOOL_EGRESS_PROXY_BLOCK,
bottle_slug="dev", tool=TOOL_EGRESS_BLOCK,
proposed_file="{}", justification="late",
current_file_hash="x",
now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc),
@@ -315,16 +315,16 @@ class TestToolConstants(unittest.TestCase):
def test_tools_tuple_matches_individual_constants(self):
self.assertEqual(
(
TOOL_EGRESS_PROXY_BLOCK,
TOOL_EGRESS_BLOCK,
TOOL_PIPELOCK_BLOCK,
TOOL_CAPABILITY_BLOCK,
supervise.TOOL_LIST_EGRESS_PROXY_ROUTES,
supervise.TOOL_LIST_EGRESS_ROUTES,
),
supervise.TOOLS,
)
def test_component_map_covers_two_remediation_tools_only(self):
self.assertIn(TOOL_EGRESS_PROXY_BLOCK, supervise.COMPONENT_FOR_TOOL)
self.assertIn(TOOL_EGRESS_BLOCK, supervise.COMPONENT_FOR_TOOL)
self.assertIn(TOOL_PIPELOCK_BLOCK, supervise.COMPONENT_FOR_TOOL)
self.assertNotIn(TOOL_CAPABILITY_BLOCK, supervise.COMPONENT_FOR_TOOL)
@@ -375,7 +375,7 @@ class TestSupervisePrepare(unittest.TestCase):
def test_prepare_only_writes_dockerfile_to_current_config(self):
# routes.yaml + allowlist live behind the
# `list-egress-proxy-routes` MCP tool now (PRD 0017 chunk 3).
# `list-egress-routes` MCP tool now (PRD 0017 chunk 3).
plan = _StubSupervise().prepare("dev", self.stage_dir)
files = sorted(p.name for p in plan.current_config_dir.iterdir())
self.assertEqual(["Dockerfile"], files)
+10 -10
View File
@@ -74,9 +74,9 @@ class TestValidation(unittest.TestCase):
)
def test_empty_proposed_file_rejected_for_tools_with_file_field(self):
# egress-proxy-block has structured input (validated in
# egress-block has structured input (validated in
# _validate_and_bundle_egress_route, not here) and
# list-egress-proxy-routes takes no input. Only the other
# list-egress-routes takes no input. Only the other
# two go through `validate_proposed_file`.
for tool in (_sv.TOOL_PIPELOCK_BLOCK, _sv.TOOL_CAPABILITY_BLOCK):
with self.subTest(tool=tool):
@@ -163,10 +163,10 @@ class TestHandleToolsList(unittest.TestCase):
names = [t["name"] for t in result["tools"]] # type: ignore[index]
self.assertEqual(
sorted([
_sv.TOOL_EGRESS_PROXY_BLOCK,
_sv.TOOL_EGRESS_BLOCK,
_sv.TOOL_PIPELOCK_BLOCK,
_sv.TOOL_CAPABILITY_BLOCK,
_sv.TOOL_LIST_EGRESS_PROXY_ROUTES,
_sv.TOOL_LIST_EGRESS_ROUTES,
]),
sorted(names),
)
@@ -186,10 +186,10 @@ class TestHandleToolsList(unittest.TestCase):
self.assertIn("justification", required)
self.assertIn(PROPOSED_FILE_FIELD[name], required) # type: ignore[index]
def test_list_egress_proxy_routes_takes_no_input(self):
def test_list_egress_routes_takes_no_input(self):
tool = next(
t for t in TOOL_DEFINITIONS
if t["name"] == _sv.TOOL_LIST_EGRESS_PROXY_ROUTES
if t["name"] == _sv.TOOL_LIST_EGRESS_ROUTES
)
schema = tool["inputSchema"]
self.assertEqual({}, schema.get("properties")) # type: ignore[union-attr]
@@ -229,7 +229,7 @@ class TestHandleToolsCall(unittest.TestCase):
try:
result = handle_tools_call(
{
"name": _sv.TOOL_EGRESS_PROXY_BLOCK,
"name": _sv.TOOL_EGRESS_BLOCK,
"arguments": {
"host": "example.com",
"justification": "need a route",
@@ -273,7 +273,7 @@ class TestHandleToolsCall(unittest.TestCase):
with self.assertRaises(_RpcError):
handle_tools_call(
{
"name": _sv.TOOL_EGRESS_PROXY_BLOCK,
"name": _sv.TOOL_EGRESS_BLOCK,
"arguments": {"host": "example.com"},
},
self.config,
@@ -284,7 +284,7 @@ class TestHandleToolsCall(unittest.TestCase):
try:
handle_tools_call(
{
"name": _sv.TOOL_EGRESS_PROXY_BLOCK,
"name": _sv.TOOL_EGRESS_BLOCK,
"arguments": {
"host": "example.com",
"justification": "x",
@@ -371,7 +371,7 @@ class TestHttpEndToEnd(unittest.TestCase):
self.assertEqual("2.0", result["jsonrpc"])
self.assertEqual(1, result["id"])
names = [t["name"] for t in result["result"]["tools"]] # type: ignore[index]
self.assertIn(_sv.TOOL_EGRESS_PROXY_BLOCK, names)
self.assertIn(_sv.TOOL_EGRESS_BLOCK, names)
def test_unknown_method_returns_jsonrpc_error(self):
result = self._post_jsonrpc(