feat(orchestrator+egress): slice 8 — multi-tenant egress via the resolver (#352)

The egress addon now selects each request's Config by the calling bottle's
source IP, so one shared sidecar serves every bottle. Opt-in and fail-closed;
single-tenant behaviour is unchanged.

Orchestrator side (source-IP-primary attribution, per the PRD invariant):
  * registry: `by_source_ip` (the single active bottle at a source IP —
    network-layer attribution); `attribute` now composes it + the token.
  * service: `resolve(source_ip, token="")` — with a token, strict
    attribution; without, source IP alone.
  * control_plane: `POST /resolve`'s identity_token is now OPTIONAL (absent
    → source-IP-only); split cleanly from the token-required `/attribute`.
  * policy_resolver: `resolve` token now optional.

Egress side:
  * egress_addon_core: `resolve_client_config(resolver, client_ip, token)` —
    fetches + parses the client's Config, **fail-closed**: unattributed, a
    resolver error, or an unparseable policy all yield deny-all (no routes).
    Host-testable; `PolicyResolverLike` Protocol keeps it import-free.
  * egress_addon: consolidated mode when `BOT_BOTTLE_ORCHESTRATOR_URL` is
    set → `_active_config(flow)` resolves per client IP (reads + strips the
    `x-bot-bottle-identity` header); `request()` uses it. Unset → the static
    routes file, exactly as before. `PolicyResolver` added to the bundle.

Security note: source-IP-only resolution is safe where the IP is unspoofable
(Firecracker /31 + nft) AND the control plane is reachable only by the
trusted sidecar; the identity token, when the agent injects it, strengthens
it on weaker backends.

Scope note: the egress data plane is now multi-tenant. Remaining to be fully
live: the network topology routing every bottle's proxy to the one shared
sidecar, git-gate multitenancy, and agent-side identity-token injection.

Tests: registry by_source_ip; orchestrator resolve (with/without token);
control-plane /resolve token-optional; resolver token-optional;
resolve_client_config fail-closed matrix. All 182 egress tests still pass
(single-tenant unchanged). Full suite green.

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

tests/unit/test_egress_multitenant.py

@@ -0,0 +1,53 @@
+"""Unit: resolve_client_config — fail-closed per-client egress config (PRD 0070)."""
+
+from __future__ import annotations
+
+import unittest
+
+from bot_bottle.egress_addon_core import resolve_client_config
+from bot_bottle.policy_resolver import PolicyResolveError
+
+
+class _FakeResolver:
+    def __init__(self, result: str | None = None, raises: bool = False) -> None:
+        self._result = result
+        self._raises = raises
+        self.calls: list[tuple[str, str]] = []
+
+    def resolve(self, source_ip: str, identity_token: str = "") -> str | None:
+        self.calls.append((source_ip, identity_token))
+        if self._raises:
+            raise PolicyResolveError("orchestrator down")
+        return self._result
+
+
+class TestResolveClientConfig(unittest.TestCase):
+    def test_valid_policy_is_parsed(self) -> None:
+        cfg = resolve_client_config(
+            _FakeResolver(result="routes:\n  - host: example.com\n"), "10.243.0.1"
+        )
+        self.assertEqual(("example.com",), tuple(r.host for r in cfg.routes))
+
+    def test_unattributed_none_denies_all(self) -> None:
+        cfg = resolve_client_config(_FakeResolver(result=None), "10.243.0.9")
+        self.assertEqual((), cfg.routes)  # no routes → default-deny
+
+    def test_empty_policy_denies_all(self) -> None:
+        self.assertEqual((), resolve_client_config(_FakeResolver(result=""), "10.243.0.1").routes)
+
+    def test_resolver_error_denies_all(self) -> None:
+        # Orchestrator unreachable/errored must never widen egress.
+        self.assertEqual((), resolve_client_config(_FakeResolver(raises=True), "10.243.0.1").routes)
+
+    def test_unparseable_policy_denies_all(self) -> None:
+        cfg = resolve_client_config(_FakeResolver(result="routes: notalist\n"), "10.243.0.1")
+        self.assertEqual((), cfg.routes)
+
+    def test_forwards_source_ip_and_token(self) -> None:
+        r = _FakeResolver(result=None)
+        resolve_client_config(r, "10.243.0.1", "tok")
+        self.assertEqual(("10.243.0.1", "tok"), r.calls[0])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_orchestrator_control_plane.py

@@ -171,6 +171,22 @@ class TestDispatch(unittest.TestCase):
         )
         self.assertEqual(400, status)
 
+    def test_resolve_without_token_by_source_ip(self) -> None:
+        _, reg = dispatch(
+            self.orch, "POST", "/bottles",
+            _body({"source_ip": "10.243.0.5", "policy": "P"}),
+        )
+        status, payload = dispatch(
+            self.orch, "POST", "/resolve", _body({"source_ip": "10.243.0.5"})
+        )
+        self.assertEqual(200, status)
+        self.assertEqual(reg["bottle_id"], payload["bottle_id"])
+        self.assertEqual("P", payload["policy"])
+
+    def test_resolve_requires_source_ip(self) -> None:
+        status, _ = dispatch(self.orch, "POST", "/resolve", _body({}))
+        self.assertEqual(400, status)
+
 
 class TestServerRoundTrip(unittest.TestCase):
     def test_http_register_health_attribute(self) -> None:

tests/unit/test_orchestrator_registry.py

@@ -121,6 +121,18 @@ class TestRegistryStore(unittest.TestCase):
         assert got is not None
         self.assertEqual("", got.policy)
 
+    def test_by_source_ip_returns_single_active(self) -> None:
+        rec = self.store.register("10.243.0.1")
+        got = self.store.by_source_ip("10.243.0.1")
+        assert got is not None
+        self.assertEqual(rec.bottle_id, got.bottle_id)
+
+    def test_by_source_ip_unknown_or_ambiguous_denied(self) -> None:
+        self.assertIsNone(self.store.by_source_ip("10.243.9.9"))  # unknown
+        self.store.register("10.243.0.1", bottle_id="a")
+        self.store.register("10.243.0.1", bottle_id="b")
+        self.assertIsNone(self.store.by_source_ip("10.243.0.1"))  # ambiguous
+
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_orchestrator_service.py

@@ -102,6 +102,18 @@ class TestOrchestrator(unittest.TestCase):
     def test_set_policy_unknown_is_false(self) -> None:
         self.assertFalse(self.orch.set_policy("ghost", "{}"))
 
+    def test_resolve_by_source_ip_without_token(self) -> None:
+        rec = self.orch.launch_bottle("10.243.0.1", policy="P")
+        got = self.orch.resolve("10.243.0.1")  # network-layer, no token
+        assert got is not None
+        self.assertEqual(rec.bottle_id, got.bottle_id)
+        self.assertEqual("P", got.policy)
+
+    def test_resolve_with_token_stays_strict(self) -> None:
+        rec = self.orch.launch_bottle("10.243.0.3")
+        self.assertIsNotNone(self.orch.resolve("10.243.0.3", rec.identity_token))
+        self.assertIsNone(self.orch.resolve("10.243.0.3", "wrong-token"))
+
     def test_launch_rolls_back_registry_on_broker_failure(self) -> None:
         orch = Orchestrator(self.store, _FailingBroker(self.secret), self.secret)
         with self.assertRaises(RuntimeError):

tests/unit/test_policy_resolver.py

@@ -66,6 +66,11 @@ class TestPolicyResolver(unittest.TestCase):
         self.assertEqual("10.243.0.7", sent["source_ip"])
         self.assertEqual("the-token", sent["identity_token"])
 
+    def test_resolve_without_token_sends_empty(self) -> None:
+        with patch(_URLOPEN, return_value=_resp({"policy": "P"})) as m:
+            self.r.resolve("10.243.0.7")  # token optional
+        self.assertEqual("", json.loads(m.call_args.args[0].data)["identity_token"])
+
 
 if __name__ == "__main__":
     unittest.main()