diff --git a/Dockerfile.sidecars b/Dockerfile.sidecars index dc7e911..2d43a97 100644 --- a/Dockerfile.sidecars +++ b/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 diff --git a/bot_bottle/egress_addon.py b/bot_bottle/egress_addon.py index d0cf0c7..e0fc00b 100644 --- a/bot_bottle/egress_addon.py +++ b/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: diff --git a/bot_bottle/egress_addon_core.py b/bot_bottle/egress_addon_core.py index af4cc39..be1b6a2 100644 --- a/bot_bottle/egress_addon_core.py +++ b/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", diff --git a/bot_bottle/orchestrator/control_plane.py b/bot_bottle/orchestrator/control_plane.py index df8d8f9..bcf085a 100644 --- a/bot_bottle/orchestrator/control_plane.py +++ b/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"} diff --git a/bot_bottle/orchestrator/registry.py b/bot_bottle/orchestrator/registry.py index 01eb9b4..24055ae 100644 --- a/bot_bottle/orchestrator/registry.py +++ b/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 diff --git a/bot_bottle/orchestrator/service.py b/bot_bottle/orchestrator/service.py index edc449f..827d925 100644 --- a/bot_bottle/orchestrator/service.py +++ b/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.""" diff --git a/bot_bottle/policy_resolver.py b/bot_bottle/policy_resolver.py index 76f3a33..1504a90 100644 --- a/bot_bottle/policy_resolver.py +++ b/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() diff --git a/tests/unit/test_egress_multitenant.py b/tests/unit/test_egress_multitenant.py new file mode 100644 index 0000000..7c69fb7 --- /dev/null +++ b/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() diff --git a/tests/unit/test_orchestrator_control_plane.py b/tests/unit/test_orchestrator_control_plane.py index 6b0bed4..6e1d8da 100644 --- a/tests/unit/test_orchestrator_control_plane.py +++ b/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: diff --git a/tests/unit/test_orchestrator_registry.py b/tests/unit/test_orchestrator_registry.py index f7e8571..16f9dc2 100644 --- a/tests/unit/test_orchestrator_registry.py +++ b/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() diff --git a/tests/unit/test_orchestrator_service.py b/tests/unit/test_orchestrator_service.py index 13f034b..2135327 100644 --- a/tests/unit/test_orchestrator_service.py +++ b/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): diff --git a/tests/unit/test_policy_resolver.py b/tests/unit/test_policy_resolver.py index bde11a1..143ec18 100644 --- a/tests/unit/test_policy_resolver.py +++ b/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()