From c5ae93c8ba1cda084640fc1aeda0f7ca1dcdc121 Mon Sep 17 00:00:00 2001 From: didericis Date: Mon, 13 Jul 2026 17:21:27 -0400 Subject: [PATCH] =?UTF-8?q?feat(orchestrator):=20slice=207=20=E2=80=94=20s?= =?UTF-8?q?idecar-side=20PolicyResolver=20(#352)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The data-plane bridge that lets the consolidated sidecar apply each bottle's policy per request. `PolicyResolver` resolves a client's policy from the orchestrator's `POST /resolve` keyed on (source_ip, identity token) and caches it briefly (short TTL) so it isn't a round-trip per request; `invalidate()` drops an entry on teardown / live reload. Fail-closed: an unattributed client (orchestrator answers 403) resolves to None so the caller denies; unreachable / unexpected status raises so the caller can fail closed too rather than serve stale/empty policy. Stdlib only and free of bot-bottle imports, so it can be COPYed flat into the sidecar bundle. Scope note: this is the sidecar-side *client*. Wiring it into the live egress mitmproxy addon (select `Config` per client IP in the request path) and git-gate, plus routing all bottles' egress to the one shared sidecar, are the remaining data-plane pieces — a heavier change to the sidecar bundle's adversarial-input code, taken next. Tests: resolve returns/caches/expires/invalidates; 403 -> None (fail closed); other HTTP status + unreachable raise; missing policy -> empty; posts source_ip + identity_token. Full suite green. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck --- bot_bottle/policy_resolver.py | 92 ++++++++++++++++++++++++++++++ tests/unit/test_policy_resolver.py | 83 +++++++++++++++++++++++++++ 2 files changed, 175 insertions(+) create mode 100644 bot_bottle/policy_resolver.py create mode 100644 tests/unit/test_policy_resolver.py diff --git a/bot_bottle/policy_resolver.py b/bot_bottle/policy_resolver.py new file mode 100644 index 0000000..f480148 --- /dev/null +++ b/bot_bottle/policy_resolver.py @@ -0,0 +1,92 @@ +"""Sidecar-side per-client policy resolver (PRD 0070). + +The consolidated sidecar serves every bottle from one process, so for each +request it must apply the *calling* bottle's policy, selected by the source +IP the attribution invariant makes unspoofable. This resolves that policy +from the orchestrator's control plane (`POST /resolve`), keyed on the +`(source_ip, identity_token)` pair, and caches it briefly so it isn't a +round-trip per request. + +**Fail-closed:** an unattributed client (the orchestrator answers `403`) +resolves to `None`, and the caller (the egress addon, git-gate) must then +deny — exactly as an unknown bottle should be treated. Orchestrator +*errors* (unreachable / unexpected status) raise, so the caller can fail +closed too rather than silently serving stale or empty policy. + +The resolved value is the policy blob the orchestrator stores verbatim; the +consumer parses it (e.g. the egress addon's `load_config`). This module is +stdlib-only and free of bot-bottle imports so it can be COPYed flat into +the sidecar bundle. +""" + +from __future__ import annotations + +import json +import time +import urllib.error +import urllib.request + +DEFAULT_TTL_SECONDS = 5.0 +DEFAULT_TIMEOUT_SECONDS = 2.0 + + +class PolicyResolveError(RuntimeError): + """The orchestrator was unreachable or returned an unexpected status — + distinct from a clean `403` (unattributed), which returns None.""" + + +class PolicyResolver: + """Resolves + caches each client's policy from the orchestrator.""" + + def __init__( + self, + base_url: str, + *, + ttl: float = DEFAULT_TTL_SECONDS, + timeout: float = DEFAULT_TIMEOUT_SECONDS, + ) -> None: + self._base = base_url.rstrip("/") + self._ttl = ttl + self._timeout = timeout + # source_ip -> (fetched_at_monotonic, policy | None) + self._cache: dict[str, tuple[float, str | None]] = {} + + def resolve(self, source_ip: str, identity_token: str) -> str | None: + """The calling bottle's policy blob, or None if unattributed. Cached + per source IP for `ttl` seconds. Raises `PolicyResolveError` if the + orchestrator can't be reached / errors.""" + now = time.monotonic() + hit = self._cache.get(source_ip) + if hit is not None and now - hit[0] < self._ttl: + return hit[1] + policy = self._fetch(source_ip, identity_token) + self._cache[source_ip] = (now, policy) + return policy + + def invalidate(self, source_ip: str) -> None: + """Drop a cached entry — e.g. on teardown or a policy live-reload, + so the next request re-resolves immediately.""" + self._cache.pop(source_ip, None) + + def _fetch(self, source_ip: str, identity_token: str) -> str | None: + body = json.dumps( + {"source_ip": source_ip, "identity_token": identity_token} + ).encode() + req = urllib.request.Request( + f"{self._base}/resolve", data=body, method="POST", + headers={"Content-Type": "application/json"}, + ) + try: + with urllib.request.urlopen(req, timeout=self._timeout) as resp: + payload = json.loads(resp.read()) + except urllib.error.HTTPError as e: + if e.code == 403: + return None # unattributed → fail closed (caller denies) + 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 policy if isinstance(policy, str) else "" + + +__all__ = ["PolicyResolver", "PolicyResolveError", "DEFAULT_TTL_SECONDS"] diff --git a/tests/unit/test_policy_resolver.py b/tests/unit/test_policy_resolver.py new file mode 100644 index 0000000..818e4d0 --- /dev/null +++ b/tests/unit/test_policy_resolver.py @@ -0,0 +1,83 @@ +"""Unit tests for the sidecar-side PolicyResolver (PRD 0070). HTTP mocked.""" + +from __future__ import annotations + +import json +import unittest +import urllib.error +from unittest.mock import MagicMock, patch + +from bot_bottle.policy_resolver import PolicyResolveError, PolicyResolver + +_URLOPEN = "bot_bottle.policy_resolver.urllib.request.urlopen" + + +def _resp(payload: object) -> MagicMock: + """A urlopen() return value: a context manager whose read() yields JSON.""" + m = MagicMock() + m.__enter__.return_value.read.return_value = json.dumps(payload).encode() + return m + + +def _http_error(code: int) -> urllib.error.HTTPError: + return urllib.error.HTTPError("http://x/resolve", code, "err", {}, None) # type: ignore[arg-type] + + +class TestPolicyResolver(unittest.TestCase): + def setUp(self) -> None: + self.r = PolicyResolver("http://orch:8080", ttl=5.0) + + def test_resolve_returns_policy(self) -> None: + with patch(_URLOPEN, return_value=_resp({"bottle_id": "b1", "policy": "P"})): + self.assertEqual("P", self.r.resolve("10.243.0.1", "tok")) + + def test_resolve_caches_within_ttl(self) -> None: + with patch(_URLOPEN, return_value=_resp({"policy": "P"})) as m: + self.assertEqual("P", self.r.resolve("10.243.0.1", "tok")) + self.assertEqual("P", self.r.resolve("10.243.0.1", "tok")) + m.assert_called_once() # second hit came from the cache + + def test_cache_expires(self) -> None: + r = PolicyResolver("http://orch:8080", ttl=0.0) + with patch(_URLOPEN, return_value=_resp({"policy": "P"})) as m: + r.resolve("10.243.0.1", "tok") + r.resolve("10.243.0.1", "tok") + self.assertEqual(2, m.call_count) # ttl=0 → always refetch + + def test_invalidate_forces_refetch(self) -> None: + with patch(_URLOPEN, return_value=_resp({"policy": "P"})) as m: + self.r.resolve("10.243.0.1", "tok") + self.r.invalidate("10.243.0.1") + self.r.resolve("10.243.0.1", "tok") + self.assertEqual(2, m.call_count) + + def test_unattributed_403_is_none_fail_closed(self) -> None: + with patch(_URLOPEN, side_effect=_http_error(403)): + self.assertIsNone(self.r.resolve("10.243.0.9", "tok")) + + def test_other_http_error_raises(self) -> None: + with patch(_URLOPEN, side_effect=_http_error(500)): + with self.assertRaises(PolicyResolveError): + self.r.resolve("10.243.0.1", "tok") + + def test_unreachable_raises(self) -> None: + with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")): + with self.assertRaises(PolicyResolveError): + self.r.resolve("10.243.0.1", "tok") + + def test_missing_policy_field_is_empty(self) -> None: + with patch(_URLOPEN, return_value=_resp({"bottle_id": "b1"})): + self.assertEqual("", self.r.resolve("10.243.0.1", "tok")) + + def test_posts_source_ip_and_token(self) -> None: + with patch(_URLOPEN, return_value=_resp({"policy": "P"})) as m: + self.r.resolve("10.243.0.7", "the-token") + req = m.call_args.args[0] + self.assertTrue(req.full_url.endswith("/resolve")) + sent = json.loads(req.data) + self.assertEqual("10.243.0.7", sent["source_ip"]) + self.assertEqual("the-token", sent["identity_token"]) + + +if __name__ == "__main__": + unittest.main()