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

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

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()