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

bot_bottle/egress_addon.py

@@ -32,7 +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,
+    resolve_client_context,
     outbound_scan_headers,
     route_to_yaml_dict,
     scan_inbound,
@@ -99,17 +99,29 @@ class EgressAddon:
         # 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.
-        self.safe_tokens: set[str] = set()
+        # Tokens the operator has approved this session (PRD 0062), keyed by
+        # bottle so the shared gateway keeps each bottle's safelist separate —
+        # a global set would let bottle A's approved secret pass bottle B's DLP
+        # scan. In-memory only (a restart re-prompts); mutated only from the
+        # asyncio loop that runs the addon hooks, so no lock is needed.
+        self._safe_tokens: dict[str, set[str]] = {}
         self._supervise_slug = os.environ.get("SUPERVISE_BOTTLE_SLUG", "").strip()
         self._token_allow_timeout = _token_allow_timeout_from_env(os.environ)
         self._reload(initial=True)
         self._install_sighup()
 
-    def _supervise_available(self) -> bool:
-        return bool(self._supervise_slug)
+    @staticmethod
+    def _supervise_available(slug: str) -> bool:
+        """Supervise is reachable for this request iff we resolved a bottle to
+        attribute its proposals to (single-tenant env slug, or a source-IP
+        -attributed bottle id). Empty → fail closed (no queue to write to)."""
+        return bool(slug)
+
+    def _safe_tokens_for(self, slug: str) -> set[str]:
+        """This bottle's operator-approved DLP safelist (PRD 0062), created on
+        first use. Keyed by bottle so the shared gateway never leaks one
+        bottle's approved token into another's scan."""
+        return self._safe_tokens.setdefault(slug, set())
 
     def _reload(self, *, initial: bool = False) -> None:
         try:
@@ -218,19 +230,21 @@ 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."""
+    def _resolve_flow(self, flow: http.HTTPFlow) -> "tuple[Config, str]":
+        """The `(Config, supervise slug)` to apply to this request. Single-tenant
+        → the static `self.config` and the env slug. Consolidated → the calling
+        bottle's Config and its bottle id, resolved by source IP in one
+        round-trip (fail-closed to deny-all + empty slug if unattributed). The
+        identity token, if the agent injected one, is read then stripped so it
+        never leaks upstream. The slug keys both the proposal queue and the
+        per-bottle safelist, so an approval only ever affects its own bottle."""
         if self._resolver is None:
-            return self.config
+            return self.config, self._supervise_slug
         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)
+        return resolve_client_context(self._resolver, client_ip, token)
 
     async def request(self, flow: http.HTTPFlow) -> None:
         request_path, _, query = flow.request.path.partition("?")
@@ -239,14 +253,14 @@ class EgressAddon:
             self._serve_introspection(flow, request_path)
             return
 
-        config = self._active_config(flow)
+        config, slug = self._resolve_flow(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(config.routes, flow.request.pretty_host)
         if route is not None:
-            if not await self._handle_outbound_dlp(flow, route):
+            if not await self._handle_outbound_dlp(flow, route, slug):
                 return
             # The redact policy may have rewritten the request line; recompute
             # the path/query the git checks below rely on.
@@ -310,6 +324,7 @@ class EgressAddon:
         self,
         flow: http.HTTPFlow,
         route: Route,
+        slug: str,
     ) -> bool:
         """Scan the outbound request and apply the route's on-match policy
         (PRD 0062). Returns True if the request may be forwarded, False if a
@@ -331,7 +346,7 @@ class EgressAddon:
             )
             result = scan_outbound(
                 route, scan_text, os.environ,
-                safe_tokens=self.safe_tokens, crlf_text=crlf_text,
+                safe_tokens=self._safe_tokens_for(slug), crlf_text=crlf_text,
             )
             if result is None or result.severity != "block":
                 return True
@@ -366,10 +381,10 @@ class EgressAddon:
 
             # supervise (default): hold the request for operator approval.
             # Fall back to a hard 403 when supervise isn't wired for the bottle.
-            if not self._supervise_available():
+            if not self._supervise_available(slug):
                 self._block_dlp(flow, result)
                 return False
-            approved = await self._supervise_token_block(flow, request_path, result)
+            approved = await self._supervise_token_block(flow, request_path, result, slug)
             if not approved:
                 return False  # _supervise_token_block wrote the 403 response
             # loop: the approved value is now in safe_tokens; re-scan.
@@ -412,12 +427,16 @@ class EgressAddon:
         flow: http.HTTPFlow,
         request_path: str,
         result: ScanResult,
+        slug: str,
     ) -> bool:
         """Route a token DLP block to the operator's supervisor queue and wait.
 
-        Returns True if the operator approved (the matched value is added to
-        `self.safe_tokens` and the caller re-scans); False if the request must
-        be blocked (a 403 response has been written to `flow`)."""
+        `slug` attributes the proposal to the calling bottle (its own queue +
+        safelist) — in the shared gateway this is what keeps one bottle's
+        approval from unblocking another's request. Returns True if the operator
+        approved (the matched value is added to that bottle's safelist and the
+        caller re-scans); False if the request must be blocked (a 403 response
+        has been written to `flow`)."""
         host = flow.request.pretty_host
         payload = build_token_allow_payload(
             redact_tokens(host, env=os.environ),
@@ -426,7 +445,7 @@ class EgressAddon:
             result,
         )
         proposal = _sv.Proposal.new(
-            bottle_slug=self._supervise_slug,
+            bottle_slug=slug,
             tool=_sv.TOOL_EGRESS_TOKEN_ALLOW,
             proposed_file=payload,
             justification=_TOKEN_ALLOW_JUSTIFICATION,
@@ -449,13 +468,13 @@ class EgressAddon:
             **self._req_ctx(flow),
         }) + "\n")
 
-        response = await self._await_token_response(proposal.id)
-        _sv.archive_proposal(self._supervise_slug, proposal.id)
+        response = await self._await_token_response(proposal.id, slug)
+        _sv.archive_proposal(slug, proposal.id)
 
         if response is not None and response.status in (
             _sv.STATUS_APPROVED, _sv.STATUS_MODIFIED,
         ):
-            self.safe_tokens.add(result.matched)
+            self._safe_tokens_for(slug).add(result.matched)
             if self.config.log >= LOG_BLOCKS:
                 sys.stderr.write(json.dumps({
                     "event": "egress_token_allowed",
@@ -478,6 +497,7 @@ class EgressAddon:
     async def _await_token_response(
         self,
         proposal_id: str,
+        slug: str,
     ) -> "_sv.Response | None":
         """Poll the DB for the operator's response without blocking the
         proxy event loop. Returns the Response, or None on timeout."""
@@ -485,7 +505,7 @@ class EgressAddon:
         deadline = loop.time() + self._token_allow_timeout
         while True:
             try:
-                return _sv.read_response(self._supervise_slug, proposal_id)
+                return _sv.read_response(slug, proposal_id)
             except (OSError, ValueError, KeyError):
                 # Not written yet, or a partial/malformed write — retry until
                 # the deadline, then fail closed.
@@ -539,6 +559,9 @@ class EgressAddon:
         """
         if flow.websocket is None:  # type: ignore[union-attr]
             return
+        # WebSocket DLP runs against the static config only (single-tenant); in
+        # the consolidated gateway self.config has no routes, so this is inert
+        # until websocket routing is made source-IP-aware (a separate slice).
         route = match_route(self.config.routes, flow.request.pretty_host)
         if route is None:
             return
@@ -549,7 +572,7 @@ class EgressAddon:
             # not an injection vector here — scan only for credential leakage.
             result = scan_outbound(
                 route, content, os.environ,
-                safe_tokens=self.safe_tokens, crlf_text="",
+                safe_tokens=self._safe_tokens_for(self._supervise_slug), crlf_text="",
             )
             if result is not None and result.severity == "block":
                 sys.stderr.write(f"egress DLP: {result.reason}\n")

bot_bottle/egress_addon_core.py

@@ -421,6 +421,18 @@ class PolicyResolverLike(typing.Protocol):
         ...
 
 
+def _config_from_policy(policy: "str | None") -> "Config":
+    """Parse a resolved policy blob into a Config, fail-closed: None / empty /
+    unparseable all become a deny-all Config (no routes → every request
+    blocked)."""
+    if not policy:
+        return Config(routes=())  # unattributed or empty → deny-all
+    try:
+        return load_config(policy)
+    except ValueError:
+        return Config(routes=())  # unparseable policy → deny
+
+
 def resolve_client_config(
     resolver: PolicyResolverLike, client_ip: str, identity_token: str = ""
 ) -> "Config":
@@ -433,12 +445,33 @@ def resolve_client_config(
         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
+    return _config_from_policy(policy)
+
+
+class ContextResolverLike(typing.Protocol):
+    """The bit of `policy_resolver.PolicyResolver` `resolve_client_context`
+    needs — one round-trip returning both policy and bottle id."""
+
+    def resolve_policy_and_bottle_id(
+        self, source_ip: str, identity_token: str = ...,
+    ) -> "tuple[str | None, str | None]":
+        ...
+
+
+def resolve_client_context(
+    resolver: ContextResolverLike, client_ip: str, identity_token: str = "",
+) -> "tuple[Config, str]":
+    """The calling client's `(Config, bottle_id)` in one round-trip —
+    **fail-closed**. The Config follows `resolve_client_config`'s deny-all
+    rules; the bottle id is `""` whenever unattributed or the orchestrator
+    errored, which the caller treats as "supervise unavailable for this
+    bottle" (never another bottle's queue). One `/resolve` keys both the
+    per-request egress policy and the per-bottle supervise queue + safelist."""
     try:
-        return load_config(policy)
-    except ValueError:
-        return Config(routes=())  # unparseable policy → deny
+        policy, bottle_id = resolver.resolve_policy_and_bottle_id(client_ip, identity_token)
+    except Exception:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+        return Config(routes=()), ""  # orchestrator unreachable/errored → deny
+    return _config_from_policy(policy), (bottle_id or "")
 
 
 # ---------------------------------------------------------------------------
@@ -833,7 +866,9 @@ __all__ = [
     "is_git_fetch_request",
     "load_config",
     "resolve_client_config",
+    "resolve_client_context",
     "PolicyResolverLike",
+    "ContextResolverLike",
     "match_route",
     "outbound_scan_headers",
     "parse_config",

bot_bottle/git_http_backend.py

@@ -6,6 +6,15 @@ where the guest reaches the sidecar over the point-to-point TAP). The
 wrapper serves the same `/git/*.git` bare repos through
 `git http-backend`, so pre-receive and upstream forwarding remain the
 git-gate enforcement point.
+
+Consolidated (PRD 0070): when `BOT_BOTTLE_ORCHESTRATOR_URL` is set, one
+shared gateway serves every bottle, and each request is served from the
+calling bottle's repo namespace (`<root>/<bottle_id>`), attributed from
+the unspoofable source IP via the orchestrator. Per-repo credentials +
+hooks scope by repo directory, so isolating the *root* per bottle isolates
+its creds too. Unattributed clients fail closed (404). Unset → the legacy
+per-bottle single-tenant flat root, unchanged — a transitional path that
+gets stripped out once every backend runs the consolidated gateway.
 """
 
 from __future__ import annotations
@@ -13,13 +22,86 @@ from __future__ import annotations
 import os
 import subprocess
 import sys
+import typing
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from urllib.parse import urlsplit
 
+# policy_resolver ships flat alongside this file in the sidecar bundle
+# image (see Dockerfile.sidecars); the bot_bottle.* fallback is the
+# host-side / test path. Mirrors egress_addon's import shape.
+try:
+    from policy_resolver import (  # type: ignore[import-not-found]
+        PolicyResolveError,
+        PolicyResolver,
+    )
+except ImportError:  # pragma: no cover - host-side path
+    from bot_bottle.policy_resolver import PolicyResolveError, PolicyResolver
+
 
 DEFAULT_PORT = 9420
 
+# Consolidated (multi-tenant) mode: when this points at the per-host
+# orchestrator's control plane, the backend serves each request from the
+# *calling* bottle's repo namespace, selected by source IP, instead of a
+# single flat repo root. Unset → legacy per-bottle single-tenant mode
+# (unchanged). Same env the egress addon reads, so one orchestrator setting
+# flips the whole shared gateway multi-tenant.
+ORCHESTRATOR_URL_ENV = "BOT_BOTTLE_ORCHESTRATOR_URL"
+
+# App-layer identity token (defense-in-depth over the source-IP invariant);
+# the agent injects it, the backend reads it for attribution and never
+# forwards it to `git http-backend`. Mirrors egress_addon.IDENTITY_HEADER
+# (duplicated, not imported: egress_addon pulls in mitmproxy).
+IDENTITY_HEADER = "x-bot-bottle-identity"
+
+# Default flat repo root (single-tenant, and the base under which
+# consolidated mode nests each sandbox's namespace).
+DEFAULT_REPO_ROOT = "/git"
+
+
+class ResolverLike(typing.Protocol):
+    """Structural type for the resolver `resolve_sandbox_root` needs — just
+    `resolve_bottle_id`. A Protocol so tests can pass a fake without
+    importing PolicyResolver (mirrors egress_addon_core.PolicyResolverLike)."""
+
+    def resolve_bottle_id(
+        self, source_ip: str, identity_token: str = ...,
+    ) -> "str | None": ...
+
+
+def resolve_sandbox_root(
+    resolver: "ResolverLike | None",
+    base_root: Path,
+    source_ip: str,
+    identity_token: str = "",
+) -> Path | None:
+    """The per-sandbox repo root to serve this request from, or None to
+    deny (404).
+
+    Single-tenant (`resolver is None`): the flat `base_root`, unchanged.
+    NOTE: this legacy per-bottle single-tenant path is transitional — it
+    will be stripped out once every backend runs the consolidated gateway
+    (PRD 0070), leaving only the source-IP-attributed path below.
+
+    Consolidated: `base_root/<bottle_id>`, where the sandbox is attributed
+    from the source IP via the orchestrator. Fail-closed — an unattributed
+    client, a resolver error, or a namespace that would escape `base_root`
+    all deny, so one sandbox can never reach another's repos."""
+    if resolver is None:
+        return base_root
+    try:
+        bottle_id = resolver.resolve_bottle_id(source_ip, identity_token)
+    except PolicyResolveError:
+        return None  # orchestrator unreachable/errored → deny
+    if not bottle_id:
+        return None  # unattributed → deny
+    base = base_root.resolve()
+    namespace = (base / bottle_id).resolve()
+    if base not in namespace.parents:
+        return None  # bottle_id tried to escape the root → deny
+    return namespace
+
 # Mirrors git_gate_render.GIT_GATE_TIMEOUT_SECS. Duplicated rather than
 # imported: this module ships as a flat top-level sibling in the sidecar
 # bundle image (see Dockerfile.sidecars), not as part of the bot_bottle
@@ -40,10 +122,25 @@ class GitHttpHandler(BaseHTTPRequestHandler):
     def do_POST(self) -> None:
         self._run_backend()
 
+    def _sandbox_root(self) -> Path | None:
+        """This request's per-sandbox repo root, or None to deny. Single-tenant
+        unless the server was started with a resolver (consolidated mode), in
+        which case the root is the calling sandbox's source-IP-selected
+        namespace. `GIT_PROJECT_ROOT` keeps git's own env-var name."""
+        base = Path(os.environ.get("GIT_PROJECT_ROOT", DEFAULT_REPO_ROOT))
+        resolver = getattr(self.server, "policy_resolver", None)
+        token = self.headers.get(IDENTITY_HEADER, "")
+        return resolve_sandbox_root(resolver, base, self.client_address[0], token)
+
     def _run_backend(self) -> None:
+        sandbox_root = self._sandbox_root()
+        if sandbox_root is None:
+            # Unattributed / resolver error: deny before touching any repo.
+            self.send_error(404)
+            return
         parsed = urlsplit(self.path)
         if self._is_upload_pack(parsed.path, parsed.query):
-            repo_dir = self._repo_dir(parsed.path)
+            repo_dir = self._repo_dir(sandbox_root, parsed.path)
             if repo_dir is None:
                 self.send_error(404)
                 return
@@ -77,7 +174,7 @@ class GitHttpHandler(BaseHTTPRequestHandler):
                 return
         env = os.environ.copy()
         env.update({
-            "GIT_PROJECT_ROOT": os.environ.get("GIT_PROJECT_ROOT", "/git"),
+            "GIT_PROJECT_ROOT": str(sandbox_root),
             "GIT_HTTP_EXPORT_ALL": "1",
             "REQUEST_METHOD": self.command,
             "PATH_INFO": parsed.path,
@@ -123,8 +220,8 @@ class GitHttpHandler(BaseHTTPRequestHandler):
         )
         self._write_cgi_response(proc.stdout)
 
-    def _repo_dir(self, path: str) -> Path | None:
-        root = Path(os.environ.get("GIT_PROJECT_ROOT", "/git")).resolve()
+    def _repo_dir(self, sandbox_root: Path, path: str) -> Path | None:
+        root = sandbox_root.resolve()
         relative = path.lstrip("/").split(".git", 1)[0] + ".git"
         candidate = (root / relative).resolve()
         if root not in (candidate, *candidate.parents):
@@ -181,7 +278,13 @@ class GitHttpHandler(BaseHTTPRequestHandler):
 def main() -> int:
     port = int(os.environ.get("GIT_HTTP_PORT", str(DEFAULT_PORT)))
     server = ThreadingHTTPServer(("0.0.0.0", port), GitHttpHandler)
-    sys.stdout.write(f"git-http listening on 0.0.0.0:{port}\n")
+    orch_url = os.environ.get(ORCHESTRATOR_URL_ENV, "").strip()
+    # Consolidated mode: resolve each request's sandbox namespace by source
+    # IP. Absent → single-tenant (flat repo root); behaviour unchanged.
+    resolver = PolicyResolver(orch_url) if orch_url else None
+    server.policy_resolver = resolver  # type: ignore[attr-defined]
+    mode = "multi-tenant" if orch_url else "single-tenant"
+    sys.stdout.write(f"git-http listening on 0.0.0.0:{port} ({mode})\n")
     sys.stdout.flush()
     server.serve_forever()
     return 0

bot_bottle/policy_resolver.py

@@ -47,12 +47,11 @@ class PolicyResolver:
         self._base = base_url.rstrip("/")
         self._timeout = timeout
 
-    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. `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."""
+    def _post_resolve(self, source_ip: str, identity_token: str) -> dict[str, object] | None:
+        """The orchestrator's `/resolve` payload for this client, or None if
+        unattributed (a clean `403`). Raises `PolicyResolveError` on an
+        unreachable / unexpected-status / malformed response so every caller
+        can fail closed. Shared by `resolve` and `resolve_bottle_id`."""
         body = json.dumps(
             {"source_ip": source_ip, "identity_token": identity_token}
         ).encode()
@@ -69,8 +68,56 @@ class PolicyResolver:
             raise PolicyResolveError(f"/resolve returned HTTP {e.code}") from e
         except (urllib.error.URLError, TimeoutError, OSError, ValueError) as e:
             raise PolicyResolveError(f"/resolve unreachable or malformed: {e}") from e
-        policy = payload.get("policy") if isinstance(payload, dict) else None
+        return payload if isinstance(payload, dict) else {}
+
+    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.
+
+        `identity_token` is optional *transitionally* — omitting it resolves
+        by source IP alone (the network-layer attribution, sound by
+        construction on Firecracker's `/31`+nft, weaker elsewhere). The
+        consolidated end state makes the token mandatory so the app-layer
+        defense is always enforced, not silently degraded; that flips at the
+        `/resolve` boundary once identity-token *delivery* lands (a PRD 0070
+        open question — we can't require what the agent can't yet present).
+        Raises `PolicyResolveError` if the orchestrator can't be reached."""
+        payload = self._post_resolve(source_ip, identity_token)
+        if payload is None:
+            return None
+        policy = payload.get("policy")
         return policy if isinstance(policy, str) else ""
 
+    def resolve_bottle_id(self, source_ip: str, identity_token: str = "") -> str | None:
+        """The calling bottle's id, or None if unattributed. This is the
+        source-IP-keyed identity the git-gate uses to select the bottle's
+        repo namespace (there is no per-bottle policy blob to parse — the
+        bottle *is* the namespace). Same fail-closed contract as `resolve`."""
+        payload = self._post_resolve(source_ip, identity_token)
+        if payload is None:
+            return None
+        bottle_id = payload.get("bottle_id")
+        return bottle_id if isinstance(bottle_id, str) and bottle_id else None
+
+    def resolve_policy_and_bottle_id(
+        self, source_ip: str, identity_token: str = "",
+    ) -> tuple[str | None, str | None]:
+        """Both the policy blob and the bottle id in a single `/resolve` — so
+        a caller that needs each (the egress addon: policy for routing, bottle
+        id to key the calling bottle's supervise queue + safelist) makes one
+        round-trip, not two. Returns `(None, None)` when unattributed (a clean
+        `403`). Raises `PolicyResolveError` if the orchestrator can't be
+        reached, so the caller still fails closed."""
+        payload = self._post_resolve(source_ip, identity_token)
+        if payload is None:
+            return None, None
+        policy = payload.get("policy")
+        bottle_id = payload.get("bottle_id")
+        return (
+            policy if isinstance(policy, str) else "",
+            bottle_id if isinstance(bottle_id, str) and bottle_id else None,
+        )
+
 
 __all__ = ["PolicyResolver", "PolicyResolveError"]

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_git_http_multitenant.py

@@ -0,0 +1,65 @@
+"""Unit: resolve_sandbox_root — source-IP-keyed git-gate repo namespace (PRD 0070).
+
+The consolidated git-http backend serves every bottle from one process and
+selects each request's repo root by the calling bottle's source IP. This is
+the fail-closed selection logic, tested without a live server.
+"""
+
+from __future__ import annotations
+
+import unittest
+from pathlib import Path
+
+from bot_bottle.git_http_backend import resolve_sandbox_root
+from bot_bottle.policy_resolver import PolicyResolveError
+
+_BASE = Path("/git")
+
+
+class _FakeResolver:
+    def __init__(self, bottle_id: str | None = None, raises: bool = False) -> None:
+        self._bottle_id = bottle_id
+        self._raises = raises
+        self.calls: list[tuple[str, str]] = []
+
+    def resolve_bottle_id(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._bottle_id
+
+
+class TestResolveRepoRoot(unittest.TestCase):
+    def test_single_tenant_passthrough(self) -> None:
+        # No resolver → the flat base root, unchanged (legacy per-bottle mode).
+        self.assertEqual(_BASE, resolve_sandbox_root(None, _BASE, "10.243.0.1"))
+
+    def test_attributed_bottle_gets_namespaced_root(self) -> None:
+        root = resolve_sandbox_root(_FakeResolver(bottle_id="ab12cd34"), _BASE, "10.243.0.1")
+        self.assertEqual(Path("/git/ab12cd34"), root)
+
+    def test_unattributed_denies(self) -> None:
+        self.assertIsNone(resolve_sandbox_root(_FakeResolver(bottle_id=None), _BASE, "10.243.0.9"))
+
+    def test_empty_bottle_id_denies(self) -> None:
+        self.assertIsNone(resolve_sandbox_root(_FakeResolver(bottle_id=""), _BASE, "10.243.0.1"))
+
+    def test_resolver_error_denies(self) -> None:
+        # Orchestrator unreachable/errored must never serve repos.
+        self.assertIsNone(resolve_sandbox_root(_FakeResolver(raises=True), _BASE, "10.243.0.1"))
+
+    def test_namespace_escape_denies(self) -> None:
+        # A bottle_id that would climb out of the root is rejected (defense in
+        # depth — registry ids are token_hex, but never trust the namespace).
+        self.assertIsNone(
+            resolve_sandbox_root(_FakeResolver(bottle_id="../etc"), _BASE, "10.243.0.1")
+        )
+
+    def test_forwards_source_ip_and_token(self) -> None:
+        r = _FakeResolver(bottle_id="b1")
+        resolve_sandbox_root(r, _BASE, "10.243.0.1", "tok")
+        self.assertEqual(("10.243.0.1", "tok"), r.calls[0])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_policy_resolver.py

@@ -71,6 +71,37 @@ class TestPolicyResolver(unittest.TestCase):
             self.r.resolve("10.243.0.7")  # token optional
         self.assertEqual("", json.loads(m.call_args.args[0].data)["identity_token"])
 
+    def test_resolve_bottle_id_returns_id(self) -> None:
+        with patch(_URLOPEN, return_value=_resp({"bottle_id": "b1", "policy": "P"})):
+            self.assertEqual("b1", self.r.resolve_bottle_id("10.243.0.1", "tok"))
+
+    def test_resolve_bottle_id_403_is_none_fail_closed(self) -> None:
+        with patch(_URLOPEN, side_effect=_http_error(403)):
+            self.assertIsNone(self.r.resolve_bottle_id("10.243.0.9", "tok"))
+
+    def test_resolve_bottle_id_missing_field_is_none(self) -> None:
+        with patch(_URLOPEN, return_value=_resp({"policy": "P"})):
+            self.assertIsNone(self.r.resolve_bottle_id("10.243.0.1", "tok"))
+
+    def test_resolve_bottle_id_error_raises(self) -> None:
+        with patch(_URLOPEN, side_effect=_http_error(500)):
+            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()