diff --git a/bot_bottle/egress_addon.py b/bot_bottle/egress_addon.py index 428c983..d6987c3 100644 --- a/bot_bottle/egress_addon.py +++ b/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") diff --git a/bot_bottle/egress_addon_core.py b/bot_bottle/egress_addon_core.py index 9c3c322..2e9d152 100644 --- a/bot_bottle/egress_addon_core.py +++ b/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", diff --git a/bot_bottle/policy_resolver.py b/bot_bottle/policy_resolver.py index 55c81f6..cde742f 100644 --- a/bot_bottle/policy_resolver.py +++ b/bot_bottle/policy_resolver.py @@ -100,5 +100,24 @@ class PolicyResolver: 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"] diff --git a/tests/unit/test_egress_addon_log_redaction.py b/tests/unit/test_egress_addon_log_redaction.py index dbff136..6a969c5 100644 --- a/tests/unit/test_egress_addon_log_redaction.py +++ b/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 diff --git a/tests/unit/test_egress_addon_request_flow.py b/tests/unit/test_egress_addon_request_flow.py index ed8d8b2..779816d 100644 --- a/tests/unit/test_egress_addon_request_flow.py +++ b/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() diff --git a/tests/unit/test_egress_multitenant.py b/tests/unit/test_egress_multitenant.py index 7c69fb7..cff26a7 100644 --- a/tests/unit/test_egress_multitenant.py +++ b/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() diff --git a/tests/unit/test_policy_resolver.py b/tests/unit/test_policy_resolver.py index b0502d6..8110ddc 100644 --- a/tests/unit/test_policy_resolver.py +++ b/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()