refactor(egress): centralize launch env entries

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,

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)
 
 

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

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",
 ]

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()