refactor(orchestrator): drop the PolicyResolver cache (#352)
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
This commit is contained in:
@@ -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
|
request it must apply the *calling* bottle's policy, selected by the source
|
||||||
IP the attribution invariant makes unspoofable. This resolves that policy
|
IP the attribution invariant makes unspoofable. This resolves that policy
|
||||||
from the orchestrator's control plane (`POST /resolve`), keyed on the
|
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
|
`(source_ip, identity_token)` pair.
|
||||||
round-trip per request.
|
|
||||||
|
**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`)
|
**Fail-closed:** an unattributed client (the orchestrator answers `403`)
|
||||||
resolves to `None`, and the caller (the egress addon, git-gate) must then
|
resolves to `None`, and the caller (the egress addon, git-gate) must then
|
||||||
@@ -22,11 +29,9 @@ the sidecar bundle.
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import json
|
import json
|
||||||
import time
|
|
||||||
import urllib.error
|
import urllib.error
|
||||||
import urllib.request
|
import urllib.request
|
||||||
|
|
||||||
DEFAULT_TTL_SECONDS = 5.0
|
|
||||||
DEFAULT_TIMEOUT_SECONDS = 2.0
|
DEFAULT_TIMEOUT_SECONDS = 2.0
|
||||||
|
|
||||||
|
|
||||||
@@ -36,39 +41,17 @@ class PolicyResolveError(RuntimeError):
|
|||||||
|
|
||||||
|
|
||||||
class PolicyResolver:
|
class PolicyResolver:
|
||||||
"""Resolves + caches each client's policy from the orchestrator."""
|
"""Resolves each client's policy from the orchestrator, fresh per call."""
|
||||||
|
|
||||||
def __init__(
|
def __init__(self, base_url: str, *, timeout: float = DEFAULT_TIMEOUT_SECONDS) -> None:
|
||||||
self,
|
|
||||||
base_url: str,
|
|
||||||
*,
|
|
||||||
ttl: float = DEFAULT_TTL_SECONDS,
|
|
||||||
timeout: float = DEFAULT_TIMEOUT_SECONDS,
|
|
||||||
) -> None:
|
|
||||||
self._base = base_url.rstrip("/")
|
self._base = base_url.rstrip("/")
|
||||||
self._ttl = ttl
|
|
||||||
self._timeout = timeout
|
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:
|
def resolve(self, source_ip: str, identity_token: str) -> str | None:
|
||||||
"""The calling bottle's policy blob, or None if unattributed. Cached
|
"""The calling bottle's policy blob, or None if unattributed. Always
|
||||||
per source IP for `ttl` seconds. Raises `PolicyResolveError` if the
|
fetches from the orchestrator so revocations / changes / teardowns
|
||||||
|
are honored immediately. Raises `PolicyResolveError` if the
|
||||||
orchestrator can't be reached / errors."""
|
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(
|
body = json.dumps(
|
||||||
{"source_ip": source_ip, "identity_token": identity_token}
|
{"source_ip": source_ip, "identity_token": identity_token}
|
||||||
).encode()
|
).encode()
|
||||||
@@ -89,4 +72,4 @@ class PolicyResolver:
|
|||||||
return policy if isinstance(policy, str) else ""
|
return policy if isinstance(policy, str) else ""
|
||||||
|
|
||||||
|
|
||||||
__all__ = ["PolicyResolver", "PolicyResolveError", "DEFAULT_TTL_SECONDS"]
|
__all__ = ["PolicyResolver", "PolicyResolveError"]
|
||||||
|
|||||||
@@ -25,29 +25,17 @@ def _http_error(code: int) -> urllib.error.HTTPError:
|
|||||||
|
|
||||||
class TestPolicyResolver(unittest.TestCase):
|
class TestPolicyResolver(unittest.TestCase):
|
||||||
def setUp(self) -> None:
|
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:
|
def test_resolve_returns_policy(self) -> None:
|
||||||
with patch(_URLOPEN, return_value=_resp({"bottle_id": "b1", "policy": "P"})):
|
with patch(_URLOPEN, return_value=_resp({"bottle_id": "b1", "policy": "P"})):
|
||||||
self.assertEqual("P", self.r.resolve("10.243.0.1", "tok"))
|
self.assertEqual("P", self.r.resolve("10.243.0.1", "tok"))
|
||||||
|
|
||||||
def test_resolve_caches_within_ttl(self) -> None:
|
def test_resolve_always_fetches_fresh(self) -> None:
|
||||||
with patch(_URLOPEN, return_value=_resp({"policy": "P"})) as m:
|
# No cache — every resolve hits the orchestrator so revocations /
|
||||||
self.assertEqual("P", self.r.resolve("10.243.0.1", "tok"))
|
# policy changes are honored immediately.
|
||||||
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:
|
with patch(_URLOPEN, return_value=_resp({"policy": "P"})) as m:
|
||||||
self.r.resolve("10.243.0.1", "tok")
|
self.r.resolve("10.243.0.1", "tok")
|
||||||
self.r.invalidate("10.243.0.1")
|
|
||||||
self.r.resolve("10.243.0.1", "tok")
|
self.r.resolve("10.243.0.1", "tok")
|
||||||
self.assertEqual(2, m.call_count)
|
self.assertEqual(2, m.call_count)
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user