didericis/bot-bottle
1
0
Fork 0
Code Issues 25 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

feat(orchestrator): slice 7 — sidecar-side PolicyResolver (#352)

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
This commit is contained in:
didericis
2026-07-13 17:21:27 -04:00
parent 186b905dff
commit c5ae93c8ba
2 changed files with 175 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bot_bottle/policy_resolver.py
+92
View File
@@ -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"]
tests/unit/test_policy_resolver.py
+83
View File
@@ -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()