From 5204b987773542f5be9a51583bf01ffeef057209 Mon Sep 17 00:00:00 2001 From: codex Date: Thu, 25 Jun 2026 03:35:24 +0000 Subject: [PATCH] refactor(egress): centralize launch env entries --- bot_bottle/backend/docker/compose.py | 15 ++---- bot_bottle/backend/macos_container/launch.py | 17 +++---- bot_bottle/backend/smolmachines/launch.py | 16 +++--- bot_bottle/egress.py | 20 ++++++++ tests/unit/test_egress.py | 51 ++++++++++++++++++++ 5 files changed, 89 insertions(+), 30 deletions(-) diff --git a/bot_bottle/backend/docker/compose.py b/bot_bottle/backend/docker/compose.py index 2806509..1bbe708 100644 --- a/bot_bottle/backend/docker/compose.py +++ b/bot_bottle/backend/docker/compose.py @@ -28,6 +28,8 @@ from typing import Any from ...egress import ( EGRESS_HOSTNAME, EGRESS_ROUTES_IN_CONTAINER, + egress_agent_env_entries, + egress_sidecar_env_entries, ) from ...git_gate import GIT_GATE_HOSTNAME from ...log import die, warn @@ -135,13 +137,7 @@ def _sidecar_bundle_service(plan: DockerBottlePlan) -> dict[str, Any]: volumes.append(_bind(ep.mitmproxy_ca_host_path, EGRESS_CA_IN_CONTAINER)) if ep.routes: volumes.append(_bind(ep.routes_path.parent, str(Path(EGRESS_ROUTES_IN_CONTAINER).parent))) - for token_env in sorted(ep.token_env_map.keys()): - env.append(token_env) - if ep.canary and ep.canary_env: - # Inject the fake secret as a literal NAME=VALUE (not a bare name) and - # tell the sidecar to scan that exact env var as sensitive. - env.append(f"{ep.canary_env}={ep.canary}") - env.append(f"BOT_BOTTLE_SENSITIVE_PREFIXES={ep.canary_env}") + env.extend(egress_sidecar_env_entries(ep)) # --- git-gate ----------------------------------------------------- gp = plan.git_gate_plan @@ -225,10 +221,7 @@ def _agent_service(plan: DockerBottlePlan) -> dict[str, Any]: # never lands on argv or in the compose file. for name in sorted(plan.forwarded_env.keys()): env.append(name) - # Canary token: visible to the agent as a fake secret so that any - # outbound appearance of this value is a zero-FP exfil signal. - if plan.egress_plan.canary and plan.egress_plan.canary_env: - env.append(f"{plan.egress_plan.canary_env}={plan.egress_plan.canary}") + env.extend(egress_agent_env_entries(plan.egress_plan)) service: dict[str, Any] = { "image": plan.image, diff --git a/bot_bottle/backend/macos_container/launch.py b/bot_bottle/backend/macos_container/launch.py index 26c8d17..d4e75fb 100644 --- a/bot_bottle/backend/macos_container/launch.py +++ b/bot_bottle/backend/macos_container/launch.py @@ -22,7 +22,12 @@ from ...bottle_state import ( git_gate_state_dir, read_committed_image, ) -from ...egress import EGRESS_ROUTES_IN_CONTAINER, egress_resolve_token_values +from ...egress import ( + EGRESS_ROUTES_IN_CONTAINER, + egress_agent_env_entries, + egress_resolve_token_values, + egress_sidecar_env_entries, +) from ...git_gate import revoke_git_gate_provisioned_keys from ...log import die, info, warn from ...supervise import QUEUE_DIR_IN_CONTAINER, SUPERVISE_PORT @@ -350,12 +355,7 @@ def _sidecar_daemons(plan: MacosContainerBottlePlan) -> tuple[str, ...]: def _sidecar_env_entries(plan: MacosContainerBottlePlan) -> tuple[str, ...]: - env: list[str] = [] - if plan.egress_plan.routes: - env.extend(sorted(plan.egress_plan.token_env_map.keys())) - if plan.egress_plan.canary and plan.egress_plan.canary_env: - env.append(f"{plan.egress_plan.canary_env}={plan.egress_plan.canary}") - env.append(f"BOT_BOTTLE_SENSITIVE_PREFIXES={plan.egress_plan.canary_env}") + env: list[str] = list(egress_sidecar_env_entries(plan.egress_plan)) if plan.git_gate_plan.upstreams: env.append(f"BOT_BOTTLE_GIT_GATE_READY_FILE={_GIT_GATE_READY_FILE}") if plan.supervise_plan is not None: @@ -423,8 +423,7 @@ def _agent_env_entries( env.append(f"{name}={value}") for name in sorted(plan.forwarded_env.keys()): env.append(name) - if plan.egress_plan.canary and plan.egress_plan.canary_env: - env.append(f"{plan.egress_plan.canary_env}={plan.egress_plan.canary}") + env.extend(egress_agent_env_entries(plan.egress_plan)) return tuple(env) diff --git a/bot_bottle/backend/smolmachines/launch.py b/bot_bottle/backend/smolmachines/launch.py index d14a6c2..8b67e25 100644 --- a/bot_bottle/backend/smolmachines/launch.py +++ b/bot_bottle/backend/smolmachines/launch.py @@ -23,7 +23,9 @@ from typing import Callable, Generator from ...egress import ( EGRESS_ROUTES_IN_CONTAINER, + egress_agent_env_entries, egress_resolve_token_values, + egress_sidecar_env_entries, ) from ...supervise import QUEUE_DIR_IN_CONTAINER, SUPERVISE_PORT from ...util import expand_tilde @@ -228,8 +230,9 @@ def _discover_urls( guest_env["GIT_GATE_URL"] = f"http://{agent_git_gate_host}" if agent_supervise_url: guest_env["MCP_SUPERVISE_URL"] = agent_supervise_url - if plan.egress_plan.canary and plan.egress_plan.canary_env: - guest_env[plan.egress_plan.canary_env] = plan.egress_plan.canary + for entry in egress_agent_env_entries(plan.egress_plan): + name, value = entry.split("=", 1) + guest_env[name] = value return dataclasses.replace( plan, @@ -318,14 +321,7 @@ def _bundle_launch_spec( volumes.append((str(ep.mitmproxy_ca_host_path), EGRESS_CA_IN_CONTAINER, True)) if ep.routes: volumes.append((str(ep.routes_path.parent), str(Path(EGRESS_ROUTES_IN_CONTAINER).parent), True)) - # Bare-name entries for upstream-token slots. Their values - # come from the docker-run subprocess env (inherited from - # the operator's shell), never landing on argv. - for token_env in sorted(ep.token_env_map.keys()): - env.append(token_env) - if ep.canary and ep.canary_env: - env.append(f"{ep.canary_env}={ep.canary}") - env.append(f"BOT_BOTTLE_SENSITIVE_PREFIXES={ep.canary_env}") + env.extend(egress_sidecar_env_entries(ep)) # --- git-gate --------------------------------------------- gp = plan.git_gate_plan diff --git a/bot_bottle/egress.py b/bot_bottle/egress.py index cd78790..13718d6 100644 --- a/bot_bottle/egress.py +++ b/bot_bottle/egress.py @@ -62,6 +62,24 @@ def _random_canary_env() -> str: return f"{first}_{second}_SECRET" +def egress_sidecar_env_entries(plan: "EgressPlan") -> tuple[str, ...]: + """Return sidecar env entries needed by egress across all backends.""" + env: list[str] = [] + if plan.routes: + env.extend(sorted(plan.token_env_map.keys())) + if plan.canary and plan.canary_env: + env.append(f"{plan.canary_env}={plan.canary}") + env.append(f"BOT_BOTTLE_SENSITIVE_PREFIXES={plan.canary_env}") + return tuple(env) + + +def egress_agent_env_entries(plan: "EgressPlan") -> tuple[str, ...]: + """Return agent-visible egress env entries shared by all backends.""" + if plan.canary and plan.canary_env: + return (f"{plan.canary_env}={plan.canary}",) + return () + + @dataclass(frozen=True) class EgressRoute(Route): """Host-side extension of the addon's `Route`. @@ -379,5 +397,7 @@ __all__ = [ "egress_render_routes", "egress_resolve_token_values", "egress_routes_for_bottle", + "egress_agent_env_entries", + "egress_sidecar_env_entries", "egress_token_env_map", ] diff --git a/tests/unit/test_egress.py b/tests/unit/test_egress.py index ffcfe38..c2b6899 100644 --- a/tests/unit/test_egress.py +++ b/tests/unit/test_egress.py @@ -10,10 +10,12 @@ from bot_bottle.egress import ( Egress, EgressPlan, EgressRoute, + egress_agent_env_entries, egress_manifest_routes, egress_render_routes, egress_resolve_token_values, egress_routes_for_bottle, + egress_sidecar_env_entries, egress_token_env_map, ) from bot_bottle.log import Die @@ -512,5 +514,54 @@ class TestCanaryGeneration(unittest.TestCase): self.assertEqual("", plan.canary_env) +class TestEgressEnvEntries(unittest.TestCase): + def test_sidecar_entries_include_route_tokens_and_canary_scan_prefix(self): + plan = EgressPlan( + slug="s", + routes_path=Path("/tmp/r.yaml"), + routes=(EgressRoute(host="api.example"),), + token_env_map={"EGRESS_TOKEN_1": "T1", "EGRESS_TOKEN_0": "T0"}, + canary="fake-canary-value", + canary_env="CANON_ALPHA_SECRET", + ) + + self.assertEqual( + ( + "EGRESS_TOKEN_0", + "EGRESS_TOKEN_1", + "CANON_ALPHA_SECRET=fake-canary-value", + "BOT_BOTTLE_SENSITIVE_PREFIXES=CANON_ALPHA_SECRET", + ), + egress_sidecar_env_entries(plan), + ) + + def test_agent_entries_include_only_canary_bait(self): + plan = EgressPlan( + slug="s", + routes_path=Path("/tmp/r.yaml"), + routes=(), + token_env_map={}, + canary="fake-canary-value", + canary_env="CANON_ALPHA_SECRET", + ) + + self.assertEqual( + ("CANON_ALPHA_SECRET=fake-canary-value",), + egress_agent_env_entries(plan), + ) + + def test_canary_entries_omitted_when_name_missing(self): + plan = EgressPlan( + slug="s", + routes_path=Path("/tmp/r.yaml"), + routes=(), + token_env_map={}, + canary="fake-canary-value", + ) + + self.assertEqual((), egress_sidecar_env_entries(plan)) + self.assertEqual((), egress_agent_env_entries(plan)) + + if __name__ == "__main__": unittest.main()