didericis
39c823d3c0
Consolidated egress ran every bottle through one process but keyed the supervise proposal queue off a single SUPERVISE_BOTTLE_SLUG env and kept one *global* DLP safelist — so in the shared gateway an operator's token approval for bottle A would (a) be attributed to the wrong bottle and (b) leak into bottle B's DLP scan (A's approved secret passes B's egress). This slice keys both per bottle, resolved by source IP. - policy_resolver: add `resolve_policy_and_bottle_id` — policy + bottle id in one `/resolve`, so egress keys routing *and* the supervise queue/safelist from a single round-trip. Fail-closed (403 -> (None,None)). - egress_addon_core: add `resolve_client_context` (+ `ContextResolverLike`) returning `(Config, bottle_id)`, sharing the fail-closed parse with `resolve_client_config` via `_config_from_policy`. - egress_addon: `_active_config` -> `_resolve_flow` returns `(Config, slug)`; `safe_tokens` set -> per-bottle `_safe_tokens_for(slug)`; the token-allow write/await/archive + the approved-token add all use the resolved slug. Single-tenant (no resolver) unchanged — slug = the env SUPERVISE_BOTTLE_SLUG. New tests cover the resolver, the fail-closed context matrix, and the cross-tenant isolation (an approval lands only in the calling bottle's safelist; the proposal is keyed by the source-IP-attributed bottle; unattributed IPs can't supervise). Out of scope (noted): the git-gate gitleaks-allow hook + supervise_server agent-proposal paths, and websocket DLP (still self.config-only, inert in consolidated mode) — follow-up slices. pyright 0 errors; pylint 9.83/10; unit suite green (1700 tests; the 13 test_sidecar_init /bin/sleep errors are pre-existing NixOS-local noise). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
108 lines
4.7 KiB
Python
108 lines
4.7 KiB
Python
"""Unit tests for the gateway-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")
|
|
|
|
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_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.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"])
|
|
|
|
def test_resolve_without_token_sends_empty(self) -> None:
|
|
with patch(_URLOPEN, return_value=_resp({"policy": "P"})) as m:
|
|
self.r.resolve("10.243.0.7") # token optional
|
|
self.assertEqual("", json.loads(m.call_args.args[0].data)["identity_token"])
|
|
|
|
def test_resolve_bottle_id_returns_id(self) -> None:
|
|
with patch(_URLOPEN, return_value=_resp({"bottle_id": "b1", "policy": "P"})):
|
|
self.assertEqual("b1", self.r.resolve_bottle_id("10.243.0.1", "tok"))
|
|
|
|
def test_resolve_bottle_id_403_is_none_fail_closed(self) -> None:
|
|
with patch(_URLOPEN, side_effect=_http_error(403)):
|
|
self.assertIsNone(self.r.resolve_bottle_id("10.243.0.9", "tok"))
|
|
|
|
def test_resolve_bottle_id_missing_field_is_none(self) -> None:
|
|
with patch(_URLOPEN, return_value=_resp({"policy": "P"})):
|
|
self.assertIsNone(self.r.resolve_bottle_id("10.243.0.1", "tok"))
|
|
|
|
def test_resolve_bottle_id_error_raises(self) -> None:
|
|
with patch(_URLOPEN, side_effect=_http_error(500)):
|
|
with self.assertRaises(PolicyResolveError):
|
|
self.r.resolve_bottle_id("10.243.0.1", "tok")
|
|
|
|
def test_resolve_policy_and_bottle_id_one_call(self) -> None:
|
|
with patch(_URLOPEN, return_value=_resp({"bottle_id": "b1", "policy": "P"})) as m:
|
|
self.assertEqual(("P", "b1"), self.r.resolve_policy_and_bottle_id("10.243.0.1", "t"))
|
|
self.assertEqual(1, m.call_count) # both from a single /resolve
|
|
|
|
def test_resolve_policy_and_bottle_id_403_is_none_none(self) -> None:
|
|
with patch(_URLOPEN, side_effect=_http_error(403)):
|
|
self.assertEqual((None, None), self.r.resolve_policy_and_bottle_id("10.243.0.9"))
|
|
|
|
def test_resolve_policy_and_bottle_id_error_raises(self) -> None:
|
|
with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
|
|
with self.assertRaises(PolicyResolveError):
|
|
self.r.resolve_policy_and_bottle_id("10.243.0.1")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|