From 72ea500342e76f64293de714b2fa893a2e604db1 Mon Sep 17 00:00:00 2001 From: didericis Date: Mon, 13 Jul 2026 17:21:27 -0400 Subject: [PATCH 1/2] =?UTF-8?q?feat(orchestrator):=20slice=207=20=E2=80=94?= =?UTF-8?q?=20sidecar-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() -- 2.52.0 From de9da29027932d934e8d629f486223583cec0b5f Mon Sep 17 00:00:00 2001 From: didericis Date: Mon, 13 Jul 2026 17:32:05 -0400 Subject: [PATCH 2/2] refactor(orchestrator): drop the PolicyResolver cache (#352) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review: the resolver is called rarely enough that a round-trip doesn't matter, and correctness beats speed — always fetching means a revocation, policy change, or teardown the orchestrator knows about is honored immediately instead of lingering for a cache TTL. Remove the TTL cache (and `invalidate`); `resolve` now hits the orchestrator every call. Noted in the docstring that any future caching should use orchestrator-driven invalidation, not a blind TTL. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck --- bot_bottle/policy_resolver.py | 47 ++++++++++-------------------- tests/unit/test_policy_resolver.py | 20 +++---------- 2 files changed, 19 insertions(+), 48 deletions(-) diff --git a/bot_bottle/policy_resolver.py b/bot_bottle/policy_resolver.py index f480148..76f3a33 100644 --- a/bot_bottle/policy_resolver.py +++ b/bot_bottle/policy_resolver.py @@ -4,8 +4,15 @@ 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. +`(source_ip, identity_token)` pair. + +**Always fresh — no cache.** The resolver is called rarely enough that a +round-trip doesn't matter, and correctness matters more than speed: every +resolve reflects the orchestrator's *current* view, so a revocation, a +policy change, or a bottle teardown the orchestrator knows about is honored +immediately rather than lingering for a cache TTL. (If this ever becomes a +hot path, add caching with orchestrator-driven invalidation — not a blind +TTL.) **Fail-closed:** an unattributed client (the orchestrator answers `403`) resolves to `None`, and the caller (the egress addon, git-gate) must then @@ -22,11 +29,9 @@ 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 @@ -36,39 +41,17 @@ class PolicyResolveError(RuntimeError): class PolicyResolver: - """Resolves + caches each client's policy from the orchestrator.""" + """Resolves each client's policy from the orchestrator, fresh per call.""" - def __init__( - self, - base_url: str, - *, - ttl: float = DEFAULT_TTL_SECONDS, - timeout: float = DEFAULT_TIMEOUT_SECONDS, - ) -> None: + def __init__(self, base_url: str, *, 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 + """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.""" - 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() @@ -89,4 +72,4 @@ class PolicyResolver: return policy if isinstance(policy, str) else "" -__all__ = ["PolicyResolver", "PolicyResolveError", "DEFAULT_TTL_SECONDS"] +__all__ = ["PolicyResolver", "PolicyResolveError"] diff --git a/tests/unit/test_policy_resolver.py b/tests/unit/test_policy_resolver.py index 818e4d0..bde11a1 100644 --- a/tests/unit/test_policy_resolver.py +++ b/tests/unit/test_policy_resolver.py @@ -25,29 +25,17 @@ def _http_error(code: int) -> urllib.error.HTTPError: class TestPolicyResolver(unittest.TestCase): def setUp(self) -> None: - self.r = PolicyResolver("http://orch:8080", ttl=5.0) + self.r = PolicyResolver("http://orch:8080") 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: + def test_resolve_always_fetches_fresh(self) -> None: + # No cache — every resolve hits the orchestrator so revocations / + # policy changes are honored immediately. 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) -- 2.52.0