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

Dockerfile.sidecars

@@ -64,6 +64,7 @@ COPY --from=gitleaks-src /usr/bin/gitleaks /usr/bin/gitleaks
 COPY bot_bottle/egress_addon_core.py /app/egress_addon_core.py
 COPY bot_bottle/egress_dlp_config.py /app/egress_dlp_config.py
 COPY bot_bottle/egress_addon.py      /app/egress_addon.py
+COPY bot_bottle/policy_resolver.py   /app/policy_resolver.py
 COPY bot_bottle/dlp_detectors.py     /app/dlp_detectors.py
 COPY bot_bottle/yaml_subset.py       /app/yaml_subset.py
 COPY bot_bottle/paths.py             /app/paths.py

bot_bottle/egress_addon.py

@@ -32,6 +32,7 @@ from egress_addon_core import (  # type: ignore[import-not-found]  # pylint: dis
     is_git_push_request,
     load_config,
     match_route,
+    resolve_client_config,
     outbound_scan_headers,
     route_to_yaml_dict,
     scan_inbound,
@@ -51,11 +52,26 @@ try:
 except ImportError:  # pragma: no cover - host-side path
     from bot_bottle import supervise as _sv  # type: ignore[import-not-found]
 
+try:
+    from policy_resolver import PolicyResolver  # type: ignore[import-not-found]
+except ImportError:  # pragma: no cover - host-side path
+    from bot_bottle.policy_resolver import PolicyResolver
+
 
 DEFAULT_ROUTES_PATH = "/etc/egress/routes.yaml"
 
 INTROSPECT_HOST = "_egress.local"
 
+# Consolidated (multi-tenant) mode: when this points at the per-host
+# orchestrator's control plane, the addon resolves each client's Config by
+# source IP per request instead of using a single static routes file. Unset
+# → legacy per-bottle single-tenant mode (unchanged).
+ORCHESTRATOR_URL_ENV = "BOT_BOTTLE_ORCHESTRATOR_URL"
+
+# App-layer identity token (defense-in-depth over the source-IP invariant);
+# the agent injects it, the addon strips it so it never leaks upstream.
+IDENTITY_HEADER = "x-bot-bottle-identity"
+
 # Seconds the egress proxy holds a token-blocked request open waiting for the
 # operator's supervisor decision (PRD 0062), overridable via env.
 DEFAULT_TOKEN_ALLOW_TIMEOUT_SECONDS = 300.0
@@ -72,9 +88,17 @@ _TOKEN_ALLOW_JUSTIFICATION = (
 
 
 class EgressAddon:
+    # Class default so addons built via __new__ (e.g. in tests) default to
+    # single-tenant; __init__ sets the instance attribute for real runs.
+    _resolver: "PolicyResolver | None" = None
+
     def __init__(self) -> None:
         self.routes_path = os.environ.get("EGRESS_ROUTES", DEFAULT_ROUTES_PATH)
         self.config: Config = Config(routes=())
+        # Consolidated mode: resolve per-client Config from the orchestrator.
+        # Absent → single-tenant (static routes file); behaviour unchanged.
+        orch_url = os.environ.get(ORCHESTRATOR_URL_ENV, "").strip()
+        self._resolver = PolicyResolver(orch_url) if orch_url else None
         # Tokens the operator has approved this session (PRD 0062). In-memory
         # only — a restart re-prompts. Mutated only from the asyncio loop that
         # runs the addon hooks, so no lock is needed.
@@ -194,6 +218,20 @@ class EgressAddon:
             + "\n"
         )
 
+    def _active_config(self, flow: http.HTTPFlow) -> Config:
+        """The Config to apply to this request. Single-tenant → the static
+        `self.config`. Consolidated → the calling bottle's Config, resolved
+        by source IP (fail-closed to deny-all if unattributed). The identity
+        token, if the agent injected one, is read then stripped so it never
+        leaks upstream."""
+        if self._resolver is None:
+            return self.config
+        conn = flow.client_conn
+        client_ip = conn.peername[0] if conn and conn.peername else ""
+        token = flow.request.headers.get(IDENTITY_HEADER, "")
+        flow.request.headers.pop(IDENTITY_HEADER, None)
+        return resolve_client_config(self._resolver, client_ip, token)
+
     async def request(self, flow: http.HTTPFlow) -> None:
         request_path, _, query = flow.request.path.partition("?")
 
@@ -201,10 +239,12 @@ class EgressAddon:
             self._serve_introspection(flow, request_path)
             return
 
+        config = self._active_config(flow)
+
         # DLP outbound scan BEFORE stripping auth — catches tokens the
         # agent tried to smuggle in any header, path, query param, or body.
         # Hostname is included to catch DNS-tunnelling exfiltration attempts.
-        route = match_route(self.config.routes, flow.request.pretty_host)
+        route = match_route(config.routes, flow.request.pretty_host)
         if route is not None:
             if not await self._handle_outbound_dlp(flow, route):
                 return
@@ -224,7 +264,7 @@ class EgressAddon:
 
         if is_git_fetch_request(request_path, query):
             git_decision = decide_git_fetch(
-                self.config.routes, flow.request.pretty_host,
+                config.routes, flow.request.pretty_host,
             )
             if git_decision.action == "block":
                 self._block(
@@ -242,7 +282,7 @@ class EgressAddon:
         req_headers = {k.lower(): v for k, v in flow.request.headers.items()}
 
         decision = decide(
-            self.config.routes,
+            config.routes,
             flow.request.pretty_host,
             request_path,
             os.environ,
@@ -257,7 +297,7 @@ class EgressAddon:
         if decision.inject_authorization is not None:
             flow.request.headers["authorization"] = decision.inject_authorization
 
-        if self.config.log >= LOG_FULL:
+        if config.log >= LOG_FULL:
             self._log_request(flow)
 
     def _block_dlp(self, flow: http.HTTPFlow, result: ScanResult) -> None:

bot_bottle/egress_addon_core.py

@@ -413,6 +413,34 @@ def load_config(text: str) -> "Config":
     return parse_config(payload)
 
 
+class PolicyResolverLike(typing.Protocol):
+    """The bit of `policy_resolver.PolicyResolver` this module needs — kept a
+    Protocol so egress_addon_core stays free of that import."""
+
+    def resolve(self, source_ip: str, identity_token: str = ...) -> "str | None":
+        ...
+
+
+def resolve_client_config(
+    resolver: PolicyResolverLike, client_ip: str, identity_token: str = ""
+) -> "Config":
+    """The calling client's egress Config, resolved from the orchestrator via
+    `resolver` and parsed — **fail-closed**. An unattributed client (None), a
+    resolver error, or an unparseable policy all yield a deny-all Config (no
+    routes → every request blocked). A compromised, absent, or confused
+    orchestrator must never *widen* a bottle's egress."""
+    try:
+        policy = resolver.resolve(client_ip, identity_token)
+    except Exception:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+        return Config(routes=())  # orchestrator unreachable/errored → deny
+    if not policy:
+        return Config(routes=())  # unattributed or empty → deny-all
+    try:
+        return load_config(policy)
+    except ValueError:
+        return Config(routes=())  # unparseable policy → deny
+
+
 # ---------------------------------------------------------------------------
 # Match evaluation
 # ---------------------------------------------------------------------------
@@ -804,6 +832,8 @@ __all__ = [
     "is_git_push_request",
     "is_git_fetch_request",
     "load_config",
+    "resolve_client_config",
+    "PolicyResolverLike",
     "match_route",
     "outbound_scan_headers",
     "parse_config",

bot_bottle/orchestrator/control_plane.py

@@ -107,7 +107,7 @@ def dispatch(  # pylint: disable=too-many-return-statements,too-many-branches
             return 200, {"torn_down": True}
         return 404, {"error": "no such bottle"}
 
-    if method == "POST" and route in ("/attribute", "/resolve"):
+    if method == "POST" and route == "/attribute":
         try:
             data = _parse_json_object(body)
         except ValueError as e:
@@ -119,13 +119,25 @@ def dispatch(  # pylint: disable=too-many-return-statements,too-many-branches
         rec = orch.attribute(source_ip, token)
         if rec is None:
             return 403, {"error": "unattributed"}
-        # /attribute is the lightweight identity check; /resolve additionally
-        # returns the bottle's policy — the source-IP-keyed lookup the
-        # multi-tenant sidecar makes per request.
-        if route == "/resolve":
-            return 200, {"bottle_id": rec.bottle_id, "policy": rec.policy}
         return 200, {"bottle_id": rec.bottle_id}
 
+    if method == "POST" and route == "/resolve":
+        # The per-request lookup the multi-tenant sidecar makes: returns the
+        # bottle's policy. identity_token is OPTIONAL — absent means resolve
+        # by source IP alone (network-layer attribution).
+        try:
+            data = _parse_json_object(body)
+        except ValueError as e:
+            return 400, {"error": f"invalid JSON: {e}"}
+        source_ip = data.get("source_ip")
+        token = data.get("identity_token")
+        if not isinstance(source_ip, str) or not source_ip:
+            return 400, {"error": "source_ip (string) is required"}
+        rec = orch.resolve(source_ip, token if isinstance(token, str) else "")
+        if rec is None:
+            return 403, {"error": "unattributed"}
+        return 200, {"bottle_id": rec.bottle_id, "policy": rec.policy}
+
     return 404, {"error": "not found"}
 
 

bot_bottle/orchestrator/registry.py

@@ -212,14 +212,13 @@ class RegistryStore(DbStore):
             ).fetchall()
         return [_row_to_record(r) for r in rows]
 
-    def attribute(self, source_ip: str, identity_token: str) -> BottleRecord | None:
-        """Fail-closed attribution. Returns the bottle only when exactly one
-        active record has this source IP AND its identity token matches
-        (constant-time). Either signal alone is insufficient: an unknown IP,
-        an ambiguous IP (more than one active bottle — a misconfiguration),
-        an empty token, or a token mismatch all deny."""
-        if not identity_token:
-            return None
+    def by_source_ip(self, source_ip: str) -> BottleRecord | None:
+        """Network-layer attribution: the single active bottle at this source
+        IP, or None if unknown or ambiguous (more than one — a
+        misconfiguration). Safe as the *sole* attributor only where the
+        source IP is unspoofable (Firecracker `/31` + nft) and the control
+        plane is reachable only by the trusted sidecar; pair with the
+        identity token (`attribute`) elsewhere."""
         with self._connect() as conn:
             rows = conn.execute(
                 "SELECT * FROM orchestrator_bottles "
@@ -228,7 +227,17 @@ class RegistryStore(DbStore):
             ).fetchall()
         if len(rows) != 1:
             return None
-        rec = _row_to_record(rows[0])
+        return _row_to_record(rows[0])
+
+    def attribute(self, source_ip: str, identity_token: str) -> BottleRecord | None:
+        """Fail-closed attribution: `by_source_ip` AND a matching identity
+        token (constant-time). Either signal alone is insufficient here — an
+        unknown/ambiguous IP, an empty token, or a token mismatch all deny."""
+        if not identity_token:
+            return None
+        rec = self.by_source_ip(source_ip)
+        if rec is None:
+            return None
         if not hmac.compare_digest(rec.identity_token, identity_token):
             return None
         return rec

bot_bottle/orchestrator/service.py

@@ -79,11 +79,20 @@ class Orchestrator:
         return True
 
     def attribute(self, source_ip: str, identity_token: str) -> BottleRecord | None:
-        """Fail-closed attribution (delegates to the registry). The returned
-        record carries the bottle's `policy` — this is the source-IP-keyed
-        resolution the multi-tenant sidecar makes per request."""
+        """Fail-closed attribution (delegates to the registry)."""
         return self.registry.attribute(source_ip, identity_token)
 
+    def resolve(self, source_ip: str, identity_token: str = "") -> BottleRecord | None:
+        """Resolve the bottle behind a request — the source-IP-keyed lookup
+        the multi-tenant sidecar makes per request; the returned record
+        carries its `policy`. With a token, full attribution (source IP +
+        token); without, network-layer attribution by source IP alone
+        (valid where the IP is unspoofable and the control plane is
+        sidecar-only)."""
+        if identity_token:
+            return self.registry.attribute(source_ip, identity_token)
+        return self.registry.by_source_ip(source_ip)
+
     def set_policy(self, bottle_id: str, policy: str) -> bool:
         """Update a bottle's sidecar policy in place (live reload). False if
         the bottle is unknown."""

bot_bottle/policy_resolver.py

@@ -47,11 +47,12 @@ class PolicyResolver:
         self._base = base_url.rstrip("/")
         self._timeout = timeout
 
-    def resolve(self, source_ip: str, identity_token: str) -> str | None:
+    def resolve(self, source_ip: str, identity_token: str = "") -> str | None:
         """The calling bottle's policy blob, or None if unattributed. Always
         fetches from the orchestrator so revocations / changes / teardowns
-        are honored immediately. Raises `PolicyResolveError` if the
-        orchestrator can't be reached / errors."""
+        are honored immediately. `identity_token` is optional — omit it to
+        resolve by source IP alone (the network-layer attribution).
+        Raises `PolicyResolveError` if the orchestrator can't be reached."""
         body = json.dumps(
             {"source_ip": source_ip, "identity_token": identity_token}
         ).encode()

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