fix(egress+orchestrator): inject per-bottle auth tokens in the shared gateway

The cut-over dropped the per-bottle token flow, so an authed egress route on
the shared gateway failed with 'env var EGRESS_TOKEN_0 is unset' — the gateway
reads the token from its env, but a shared gateway has no per-bottle env.

Now the bottle's egress auth tokens travel to the gateway over /resolve and
the addon injects from them, mirroring what the per-bottle sidecar's env did:
- launch resolves the token values from the host env and hands them to the
  orchestrator, which holds them IN MEMORY (keyed by bottle_id, never written
  to the registry DB) and serves them on /resolve;
- PolicyResolver.resolve_policy_and_bottle_id + resolve_client_context now
  return the token map alongside policy + bottle_id (one round-trip);
- the egress addon overlays the process env with the bottle's tokens per
  request and uses that env for auth injection AND DLP — the agent never sees
  the credential.

Secrets stay off disk (validated: /resolve returns the token, the registry DB
does not contain it). SecretProvider (#355) is the future hardening.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck

tests/unit/test_egress_addon_request_flow.py

@@ -233,16 +233,19 @@ class _CtxResolver:
     """Fake orchestrator resolver: maps source IP -> bottle id, and grants the
     same allow-list to any attributed bottle (unattributed -> deny)."""
 
-    def __init__(self, ip_to_bottle: dict[str, str]) -> None:
+    def __init__(
+        self, ip_to_bottle: dict[str, str], tokens: dict[str, str] | None = None,
+    ) -> None:
         self._map = ip_to_bottle
+        self._tokens = tokens or {}
 
     def resolve_policy_and_bottle_id(
         self, source_ip: str, identity_token: str = "",
-    ) -> tuple[str | None, str | None]:
+    ) -> tuple[str | None, str | None, dict[str, str]]:
         del identity_token
         bottle_id = self._map.get(source_ip)
         policy = "routes:\n  - host: api.example.com\n" if bottle_id else None
-        return policy, bottle_id
+        return policy, bottle_id, (self._tokens if bottle_id else {})
 
 
 # ---------------------------------------------------------------------------
@@ -803,6 +806,46 @@ class TestSuperviseMultiTenant(unittest.TestCase):
             _run_request(addon, flow)
         self.assertEqual(["bottle-b"], seen)  # proposal keyed by the resolved bottle
 
+    def test_auth_token_injected_from_resolved_tokens(self) -> None:
+        # The bottle's upstream token comes from /resolve (in-memory on the
+        # orchestrator), NOT the gateway's env — the request is forwarded with
+        # Authorization injected. This is the token-injection cut-over fix.
+        policy = (
+            'routes:\n  - host: api.example.com\n'
+            '    auth_scheme: "Bearer"\n    token_env: "EGRESS_TOKEN_0"\n'
+        )
+
+        class _AuthResolver:
+            def resolve_policy_and_bottle_id(self, ip: str, identity_token: str = ""):
+                del ip, identity_token
+                return policy, "bottle-a", {"EGRESS_TOKEN_0": "sk-secret"}
+
+        addon = _addon(Config(routes=()))
+        addon._resolver = cast(Any, _AuthResolver())
+        flow = _with_client_ip(_Flow(_Request(host="api.example.com", method="GET")), "10.0.0.1")
+        _run_request(addon, flow)
+        self.assertIsNone(flow.response)  # forwarded, not blocked
+        self.assertEqual("Bearer sk-secret", flow.request.headers.get("authorization"))
+
+    def test_authed_route_without_token_is_blocked(self) -> None:
+        # No token resolved for the bottle → the authed route can't inject, so
+        # the request is blocked (the error the cut-over originally produced).
+        policy = (
+            'routes:\n  - host: api.example.com\n'
+            '    auth_scheme: "Bearer"\n    token_env: "EGRESS_TOKEN_0"\n'
+        )
+
+        class _NoTokenResolver:
+            def resolve_policy_and_bottle_id(self, ip: str, identity_token: str = ""):
+                del ip, identity_token
+                return policy, "bottle-a", {}  # no tokens
+
+        addon = _addon(Config(routes=()))
+        addon._resolver = cast(Any, _NoTokenResolver())
+        flow = _with_client_ip(_Flow(_Request(host="api.example.com", method="GET")), "10.0.0.1")
+        _run_request(addon, flow)
+        self.assertIsNotNone(flow.response)  # blocked — token unset
+
     def test_unattributed_source_ip_cannot_supervise(self) -> None:
         addon = self._consolidated_addon()
         # 10.9.9.9 is not in the resolver map -> deny-all config, empty slug.

tests/unit/test_egress_multitenant.py

@@ -51,45 +51,55 @@ class TestResolveClientConfig(unittest.TestCase):
 
 class _FakeContextResolver:
     def __init__(
-        self, policy: str | None = None, bottle_id: str | None = None, raises: bool = False,
+        self, policy: str | None = None, bottle_id: str | None = None,
+        raises: bool = False, tokens: dict[str, str] | None = None,
     ) -> None:
         self._policy = policy
         self._bottle_id = bottle_id
         self._raises = raises
+        self._tokens = tokens or {}
 
     def resolve_policy_and_bottle_id(
         self, source_ip: str, identity_token: str = "",
-    ) -> tuple[str | None, str | None]:
+    ) -> tuple[str | None, str | None, dict[str, str]]:
         if self._raises:
             raise PolicyResolveError("orchestrator down")
-        return self._policy, self._bottle_id
+        return self._policy, self._bottle_id, self._tokens
 
 
 class TestResolveClientContext(unittest.TestCase):
-    def test_returns_config_and_bottle_id(self) -> None:
-        cfg, slug = resolve_client_context(
-            _FakeContextResolver(policy="routes:\n  - host: example.com\n", bottle_id="b1"),
+    def test_returns_config_bottle_id_and_tokens(self) -> None:
+        cfg, slug, tokens = resolve_client_context(
+            _FakeContextResolver(
+                policy="routes:\n  - host: example.com\n", bottle_id="b1",
+                tokens={"EGRESS_TOKEN_0": "sekret"},
+            ),
             "10.243.0.1",
         )
         self.assertEqual(("example.com",), tuple(r.host for r in cfg.routes))
         self.assertEqual("b1", slug)
+        self.assertEqual({"EGRESS_TOKEN_0": "sekret"}, tokens)  # for auth injection
 
     def test_unattributed_denies_and_empty_slug(self) -> None:
-        cfg, slug = resolve_client_context(
+        cfg, slug, tokens = resolve_client_context(
             _FakeContextResolver(policy=None, bottle_id=None), "10.243.0.9",
         )
         self.assertEqual((), cfg.routes)
         self.assertEqual("", slug)  # no bottle → supervise unavailable
+        self.assertEqual({}, tokens)
 
-    def test_resolver_error_denies_and_empty_slug(self) -> None:
-        cfg, slug = resolve_client_context(_FakeContextResolver(raises=True), "10.243.0.1")
+    def test_resolver_error_denies_empty_slug_no_tokens(self) -> None:
+        cfg, slug, tokens = resolve_client_context(
+            _FakeContextResolver(raises=True), "10.243.0.1",
+        )
         self.assertEqual((), cfg.routes)
         self.assertEqual("", slug)
+        self.assertEqual({}, tokens)
 
     def test_unparseable_policy_denies_but_keeps_slug(self) -> None:
         # A bad policy denies egress, but the bottle is still attributed (its
         # supervise queue is keyed by the id, independent of route parsing).
-        cfg, slug = resolve_client_context(
+        cfg, slug, _tokens = resolve_client_context(
             _FakeContextResolver(policy="routes: notalist\n", bottle_id="b2"), "10.243.0.1",
         )
         self.assertEqual((), cfg.routes)

tests/unit/test_orchestrator_service.py

@@ -92,6 +92,18 @@ class TestOrchestrator(unittest.TestCase):
         assert got is not None
         self.assertEqual('{"routes":[]}', got.policy)
 
+    def test_tokens_held_in_memory_and_cleared_on_teardown(self) -> None:
+        rec = self.orch.launch_bottle("10.243.0.5", tokens={"EGRESS_TOKEN_0": "sk"})
+        self.assertEqual({"EGRESS_TOKEN_0": "sk"}, self.orch.tokens_for(rec.bottle_id))
+        # not persisted in the registry (redacted record has no tokens field)
+        self.assertNotIn("tokens", rec.redacted())
+        self.orch.teardown_bottle(rec.bottle_id)
+        self.assertEqual({}, self.orch.tokens_for(rec.bottle_id))
+
+    def test_tokens_default_empty(self) -> None:
+        rec = self.orch.launch_bottle("10.243.0.6")
+        self.assertEqual({}, self.orch.tokens_for(rec.bottle_id))
+
     def test_set_policy_live_reload(self) -> None:
         rec = self.orch.launch_bottle("10.243.0.3")
         self.assertTrue(self.orch.set_policy(rec.bottle_id, '{"x":1}'))

tests/unit/test_policy_resolver.py

@@ -89,13 +89,17 @@ class TestPolicyResolver(unittest.TestCase):
                 self.r.resolve_bottle_id("10.243.0.1", "tok")
 
     def test_resolve_policy_and_bottle_id_one_call(self) -> None:
-        with patch(_URLOPEN, return_value=_resp({"bottle_id": "b1", "policy": "P"})) as m:
-            self.assertEqual(("P", "b1"), self.r.resolve_policy_and_bottle_id("10.243.0.1", "t"))
-        self.assertEqual(1, m.call_count)  # both from a single /resolve
+        payload = {"bottle_id": "b1", "policy": "P", "tokens": {"EGRESS_TOKEN_0": "s"}}
+        with patch(_URLOPEN, return_value=_resp(payload)) as m:
+            self.assertEqual(
+                ("P", "b1", {"EGRESS_TOKEN_0": "s"}),
+                self.r.resolve_policy_and_bottle_id("10.243.0.1", "t"),
+            )
+        self.assertEqual(1, m.call_count)  # policy + id + tokens from a single /resolve
 
-    def test_resolve_policy_and_bottle_id_403_is_none_none(self) -> None:
+    def test_resolve_policy_and_bottle_id_403_is_none_none_empty(self) -> None:
         with patch(_URLOPEN, side_effect=_http_error(403)):
-            self.assertEqual((None, None), self.r.resolve_policy_and_bottle_id("10.243.0.9"))
+            self.assertEqual((None, None, {}), self.r.resolve_policy_and_bottle_id("10.243.0.9"))
 
     def test_resolve_policy_and_bottle_id_error_raises(self) -> None:
         with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):