feat(supervise+orchestrator): slice 11 — per-bottle supervise queue + DLP safelist

Consolidated egress ran every bottle through one process but keyed the
supervise proposal queue off a single SUPERVISE_BOTTLE_SLUG env and kept
one *global* DLP safelist — so in the shared gateway an operator's token
approval for bottle A would (a) be attributed to the wrong bottle and
(b) leak into bottle B's DLP scan (A's approved secret passes B's egress).
This slice keys both per bottle, resolved by source IP.

- policy_resolver: add `resolve_policy_and_bottle_id` — policy + bottle id
  in one `/resolve`, so egress keys routing *and* the supervise
  queue/safelist from a single round-trip. Fail-closed (403 -> (None,None)).
- egress_addon_core: add `resolve_client_context` (+ `ContextResolverLike`)
  returning `(Config, bottle_id)`, sharing the fail-closed parse with
  `resolve_client_config` via `_config_from_policy`.
- egress_addon: `_active_config` -> `_resolve_flow` returns `(Config, slug)`;
  `safe_tokens` set -> per-bottle `_safe_tokens_for(slug)`; the token-allow
  write/await/archive + the approved-token add all use the resolved slug.
  Single-tenant (no resolver) unchanged — slug = the env SUPERVISE_BOTTLE_SLUG.

New tests cover the resolver, the fail-closed context matrix, and the
cross-tenant isolation (an approval lands only in the calling bottle's
safelist; the proposal is keyed by the source-IP-attributed bottle;
unattributed IPs can't supervise).

Out of scope (noted): the git-gate gitleaks-allow hook + supervise_server
agent-proposal paths, and websocket DLP (still self.config-only, inert in
consolidated mode) — follow-up slices.

pyright 0 errors; pylint 9.83/10; unit suite green (1700 tests; the 13
test_sidecar_init /bin/sleep errors are pre-existing NixOS-local noise).

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

tests/unit/test_egress_addon_log_redaction.py

@@ -46,7 +46,7 @@ def _addon() -> EgressAddon:
     """Return a bare EgressAddon with LOG_FULL config and no routes file."""
     a: EgressAddon = EgressAddon.__new__(EgressAddon)
     a.config = Config(routes=(), log=LOG_FULL)
-    a.safe_tokens = set()
+    a._safe_tokens = {}
     a._supervise_slug = ""
     a._token_allow_timeout = 300.0
     return a

tests/unit/test_egress_addon_request_flow.py

@@ -211,7 +211,7 @@ def _addon(config: Config) -> EgressAddon:
     """Bare EgressAddon with a supplied config and no supervise wiring."""
     a: EgressAddon = EgressAddon.__new__(EgressAddon)
     a.config = config
-    a.safe_tokens = set()
+    a._safe_tokens = {}
     a._supervise_slug = ""
     a._token_allow_timeout = 300.0
     a.routes_path = "/nonexistent/routes.yaml"
@@ -222,6 +222,29 @@ def _run_request(addon: EgressAddon, flow: _Flow) -> None:
     asyncio.run(addon.request(flow))  # type: ignore[arg-type]
 
 
+def _with_client_ip(flow: _Flow, ip: str) -> _Flow:
+    """Attach a mitmproxy-style client_conn so consolidated resolution can read
+    the source IP (`flow.client_conn.peername[0]`)."""
+    flow.client_conn = types.SimpleNamespace(peername=(ip, 54321))  # type: ignore[attr-defined]
+    return flow
+
+
+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:
+        self._map = ip_to_bottle
+
+    def resolve_policy_and_bottle_id(
+        self, source_ip: str, identity_token: str = "",
+    ) -> tuple[str | None, str | None]:
+        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
+
+
 # ---------------------------------------------------------------------------
 # Introspection endpoint
 # ---------------------------------------------------------------------------
@@ -418,7 +441,8 @@ class TestSuperviseBranch(unittest.TestCase):
         with patch.object(_ea_mod, "_sv", _fake_sv("approved")):
             _run_request(addon, flow)
         self.assertIsNone(flow.response)  # forwarded after approval
-        self.assertIn(_OPENAI_KEY, addon.safe_tokens)
+        # Approval lands in the calling bottle's safelist (keyed by slug).
+        self.assertIn(_OPENAI_KEY, addon._safe_tokens_for("test-bottle"))
 
     def test_operator_rejection_blocks(self) -> None:
         addon = self._supervised_addon()
@@ -735,5 +759,62 @@ class TestLogFullRequest(unittest.TestCase):
         self.assertTrue(any(e.get("event") == "egress_request" for e in logged))
 
 
+class TestSuperviseMultiTenant(unittest.TestCase):
+    """Consolidated gateway: supervise proposals + the DLP safelist are keyed
+    per bottle, resolved by source IP (PRD 0070)."""
+
+    def _consolidated_addon(self) -> EgressAddon:
+        # Static config is empty; the resolver supplies each bottle's config.
+        addon = _addon(Config(routes=()))
+        addon._resolver = cast(Any, _CtxResolver({"10.0.0.1": "bottle-a", "10.0.0.2": "bottle-b"}))
+        addon._token_allow_timeout = 0.05
+        return addon
+
+    def test_approval_is_scoped_to_the_calling_bottle(self) -> None:
+        addon = self._consolidated_addon()
+        # bottle-a (10.0.0.1) sends the token; the operator approves.
+        flow = _with_client_ip(
+            _Flow(_Request(host="api.example.com", method="POST", body=f"k={_OPENAI_KEY}")),
+            "10.0.0.1",
+        )
+        with patch.object(_ea_mod, "_sv", _fake_sv("approved")):
+            _run_request(addon, flow)
+        self.assertIsNone(flow.response)  # forwarded after approval
+        # The approval lands ONLY in bottle-a's safelist — never bottle-b's.
+        # A global set here would be the cross-tenant leak this slice closes.
+        self.assertIn(_OPENAI_KEY, addon._safe_tokens_for("bottle-a"))
+        self.assertNotIn(_OPENAI_KEY, addon._safe_tokens_for("bottle-b"))
+
+    def test_proposal_is_attributed_to_the_source_ip_bottle(self) -> None:
+        addon = self._consolidated_addon()
+        seen: list[str] = []
+        fake = _fake_sv("approved")
+
+        def _capture(**kw: Any) -> Any:
+            seen.append(kw["bottle_slug"])
+            return types.SimpleNamespace(id="p")
+
+        fake.Proposal = types.SimpleNamespace(new=_capture)
+        flow = _with_client_ip(
+            _Flow(_Request(host="api.example.com", method="POST", body=f"k={_OPENAI_KEY}")),
+            "10.0.0.2",
+        )
+        with patch.object(_ea_mod, "_sv", fake):
+            _run_request(addon, flow)
+        self.assertEqual(["bottle-b"], seen)  # proposal keyed by the resolved bottle
+
+    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.
+        flow = _with_client_ip(
+            _Flow(_Request(host="api.example.com", method="POST", body=f"k={_OPENAI_KEY}")),
+            "10.9.9.9",
+        )
+        with patch.object(_ea_mod, "_sv", _fake_sv("approved")):
+            _run_request(addon, flow)
+        self.assertIsNotNone(flow.response)  # blocked (no route, no supervise)
+        self.assertNotIn(_OPENAI_KEY, addon._safe_tokens_for(""))
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_egress_multitenant.py

@@ -1,10 +1,10 @@
-"""Unit: resolve_client_config — fail-closed per-client egress config (PRD 0070)."""
+"""Unit: fail-closed per-client egress resolution — config + context (PRD 0070)."""
 
 from __future__ import annotations
 
 import unittest
 
-from bot_bottle.egress_addon_core import resolve_client_config
+from bot_bottle.egress_addon_core import resolve_client_config, resolve_client_context
 from bot_bottle.policy_resolver import PolicyResolveError
 
 
@@ -49,5 +49,52 @@ class TestResolveClientConfig(unittest.TestCase):
         self.assertEqual(("10.243.0.1", "tok"), r.calls[0])
 
 
+class _FakeContextResolver:
+    def __init__(
+        self, policy: str | None = None, bottle_id: str | None = None, raises: bool = False,
+    ) -> None:
+        self._policy = policy
+        self._bottle_id = bottle_id
+        self._raises = raises
+
+    def resolve_policy_and_bottle_id(
+        self, source_ip: str, identity_token: str = "",
+    ) -> tuple[str | None, str | None]:
+        if self._raises:
+            raise PolicyResolveError("orchestrator down")
+        return self._policy, self._bottle_id
+
+
+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"),
+            "10.243.0.1",
+        )
+        self.assertEqual(("example.com",), tuple(r.host for r in cfg.routes))
+        self.assertEqual("b1", slug)
+
+    def test_unattributed_denies_and_empty_slug(self) -> None:
+        cfg, slug = resolve_client_context(
+            _FakeContextResolver(policy=None, bottle_id=None), "10.243.0.9",
+        )
+        self.assertEqual((), cfg.routes)
+        self.assertEqual("", slug)  # no bottle → supervise unavailable
+
+    def test_resolver_error_denies_and_empty_slug(self) -> None:
+        cfg, slug = resolve_client_context(_FakeContextResolver(raises=True), "10.243.0.1")
+        self.assertEqual((), cfg.routes)
+        self.assertEqual("", slug)
+
+    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(
+            _FakeContextResolver(policy="routes: notalist\n", bottle_id="b2"), "10.243.0.1",
+        )
+        self.assertEqual((), cfg.routes)
+        self.assertEqual("b2", slug)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_policy_resolver.py

@@ -88,6 +88,20 @@ class TestPolicyResolver(unittest.TestCase):
             with self.assertRaises(PolicyResolveError):
                 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
+
+    def test_resolve_policy_and_bottle_id_403_is_none_none(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"))
+
+    def test_resolve_policy_and_bottle_id_error_raises(self) -> None:
+        with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
+            with self.assertRaises(PolicyResolveError):
+                self.r.resolve_policy_and_bottle_id("10.243.0.1")
+
 
 if __name__ == "__main__":
     unittest.main()