didericis-claude
9f9aa2e762
test / unit (pull_request) Successful in 48s
test / integration (pull_request) Successful in 26s
lint / lint (push) Successful in 1m45s
test / unit (push) Successful in 32s
test / integration (push) Successful in 17s
Update Quality Badges / update-badges (push) Successful in 1m21s
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
568 lines
22 KiB
Python
568 lines
22 KiB
Python
"""Unit: Egress route lift + routes.yaml render + token
|
|
resolution (PRD 0017, PRD 0053)."""
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from bot_bottle.egress import (
|
|
CODEX_HOST_CREDENTIAL_TOKEN_REF,
|
|
Egress,
|
|
EgressPlan,
|
|
EgressRoute,
|
|
egress_agent_env_entries,
|
|
egress_manifest_routes,
|
|
egress_render_routes,
|
|
egress_resolve_token_values,
|
|
egress_routes_for_bottle,
|
|
egress_sidecar_env_entries,
|
|
egress_token_env_map,
|
|
)
|
|
from bot_bottle.log import Die
|
|
from bot_bottle.manifest import ManifestIndex
|
|
from bot_bottle.yaml_subset import parse_yaml_subset
|
|
|
|
|
|
def _bottle(routes): # type: ignore
|
|
return ManifestIndex.from_json_obj({
|
|
"bottles": {"dev": {"egress": {"routes": routes}}},
|
|
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
|
|
}).bottles["dev"]
|
|
|
|
|
|
def _provider_route(host: str, token_ref: str) -> EgressRoute:
|
|
return EgressRoute(
|
|
host=host,
|
|
auth_scheme="Bearer",
|
|
token_ref=token_ref,
|
|
)
|
|
|
|
|
|
class TestManifestRouteLift(unittest.TestCase):
|
|
"""egress_manifest_routes is a pure lifter — no slot assignment."""
|
|
|
|
def test_authenticated_route_lifted_without_slot(self):
|
|
b = _bottle([{
|
|
"host": "api.github.com",
|
|
"auth": {"scheme": "Bearer", "token_ref": "GH_PAT"},
|
|
}])
|
|
routes = egress_manifest_routes(b)
|
|
self.assertEqual(1, len(routes))
|
|
r = routes[0]
|
|
self.assertEqual("api.github.com", r.host)
|
|
self.assertEqual("Bearer", r.auth_scheme)
|
|
self.assertEqual("GH_PAT", r.token_ref)
|
|
self.assertEqual("", r.token_env)
|
|
self.assertEqual((), r.matches)
|
|
|
|
def test_unauthenticated_route_has_empty_auth_fields(self):
|
|
b = _bottle([{"host": "github.com", "matches": [
|
|
{"paths": [{"value": "/x/"}]}
|
|
]}])
|
|
routes = egress_manifest_routes(b)
|
|
r = routes[0]
|
|
self.assertEqual("", r.auth_scheme)
|
|
self.assertEqual("", r.token_env)
|
|
self.assertEqual("", r.token_ref)
|
|
self.assertEqual(1, len(r.matches))
|
|
self.assertEqual(1, len(r.matches[0].paths))
|
|
self.assertEqual("/x/", r.matches[0].paths[0].value)
|
|
|
|
def test_matches_with_methods_and_headers(self):
|
|
b = _bottle([{"host": "api.example.com", "matches": [
|
|
{
|
|
"paths": [{"value": "/api/"}],
|
|
"methods": ["GET", "POST"],
|
|
"headers": [{"name": "content-type", "value": "application/json"}],
|
|
}
|
|
]}])
|
|
routes = egress_manifest_routes(b)
|
|
m = routes[0].matches[0]
|
|
self.assertEqual(("GET", "POST"), m.methods)
|
|
self.assertEqual(1, len(m.headers))
|
|
self.assertEqual("content-type", m.headers[0].name)
|
|
|
|
def test_dlp_detectors_lifted(self):
|
|
b = _bottle([{"host": "x.example", "dlp": {
|
|
"outbound_detectors": ["token_patterns"],
|
|
"inbound_detectors": False,
|
|
}}])
|
|
routes = egress_manifest_routes(b)
|
|
r = routes[0]
|
|
self.assertEqual(("token_patterns",), r.outbound_detectors)
|
|
self.assertEqual((), r.inbound_detectors)
|
|
|
|
def test_git_fetch_policy_lifted(self):
|
|
b = _bottle([{"host": "github.com", "git": {"fetch": True}}])
|
|
routes = egress_manifest_routes(b)
|
|
self.assertTrue(routes[0].git_fetch)
|
|
|
|
|
|
class TestSlotAssignment(unittest.TestCase):
|
|
"""Slot assignment happens in egress_routes_for_bottle."""
|
|
|
|
def test_authenticated_route_gets_slot(self):
|
|
b = _bottle([{
|
|
"host": "api.github.com",
|
|
"auth": {"scheme": "Bearer", "token_ref": "GH_PAT"},
|
|
}])
|
|
routes = egress_routes_for_bottle(b)
|
|
r = routes[0]
|
|
self.assertEqual("EGRESS_TOKEN_0", r.token_env)
|
|
self.assertEqual("GH_PAT", r.token_ref)
|
|
|
|
def test_shared_token_ref_collapses_to_one_slot(self):
|
|
b = _bottle([
|
|
{"host": "api.github.com",
|
|
"auth": {"scheme": "Bearer", "token_ref": "GH_PAT"}},
|
|
{"host": "github.com",
|
|
"auth": {"scheme": "Bearer", "token_ref": "GH_PAT"}},
|
|
])
|
|
routes = egress_routes_for_bottle(b)
|
|
slots = {r.token_env for r in routes}
|
|
self.assertEqual({"EGRESS_TOKEN_0"}, slots)
|
|
|
|
def test_distinct_token_refs_get_distinct_slots(self):
|
|
b = _bottle([
|
|
{"host": "a.example",
|
|
"auth": {"scheme": "Bearer", "token_ref": "T1"}},
|
|
{"host": "b.example",
|
|
"auth": {"scheme": "Bearer", "token_ref": "T2"}},
|
|
])
|
|
routes = egress_routes_for_bottle(b)
|
|
slots = [r.token_env for r in routes]
|
|
self.assertEqual(["EGRESS_TOKEN_0", "EGRESS_TOKEN_1"], slots)
|
|
|
|
def test_unauthenticated_routes_dont_consume_slots(self):
|
|
b = _bottle([
|
|
{"host": "a.example",
|
|
"auth": {"scheme": "Bearer", "token_ref": "T1"}},
|
|
{"host": "passthrough.example"},
|
|
{"host": "b.example",
|
|
"auth": {"scheme": "Bearer", "token_ref": "T2"}},
|
|
])
|
|
routes = egress_routes_for_bottle(b)
|
|
authed = [r.token_env for r in routes if r.token_env]
|
|
self.assertEqual(["EGRESS_TOKEN_0", "EGRESS_TOKEN_1"], authed)
|
|
self.assertEqual("", routes[1].token_env)
|
|
|
|
|
|
class TestRoutesForBottleManifestOnly(unittest.TestCase):
|
|
"""Without provider routes the effective table is exactly the manifest."""
|
|
|
|
def test_no_manifest_routes_means_no_effective_routes(self):
|
|
b = _bottle([])
|
|
self.assertEqual((), egress_routes_for_bottle(b))
|
|
|
|
def test_manifest_route_preserved_with_auth(self):
|
|
b = _bottle([{
|
|
"host": "api.anthropic.com",
|
|
"auth": {"scheme": "Bearer", "token_ref": "T"},
|
|
}])
|
|
routes = egress_routes_for_bottle(b)
|
|
self.assertEqual(1, len(routes))
|
|
self.assertEqual("api.anthropic.com", routes[0].host)
|
|
self.assertEqual("Bearer", routes[0].auth_scheme)
|
|
|
|
def test_manifest_only(self):
|
|
b = _bottle([{"host": "x.example"}])
|
|
effective = [r.host for r in egress_routes_for_bottle(b)]
|
|
self.assertEqual(["x.example"], effective)
|
|
|
|
|
|
|
|
class TestProviderRouteMerge(unittest.TestCase):
|
|
"""Provider routes win on host collision; manifest fills the rest."""
|
|
|
|
def test_provider_route_appended_when_not_in_manifest(self):
|
|
b = _bottle([])
|
|
pr = _provider_route("api.openai.com", "TOK")
|
|
routes = egress_routes_for_bottle(b, (pr,))
|
|
self.assertEqual(1, len(routes))
|
|
self.assertEqual("api.openai.com", routes[0].host)
|
|
self.assertEqual("Bearer", routes[0].auth_scheme)
|
|
self.assertEqual("EGRESS_TOKEN_0", routes[0].token_env)
|
|
self.assertEqual("TOK", routes[0].token_ref)
|
|
|
|
def test_unauthenticated_provider_route_appends_without_token_slot(self):
|
|
b = _bottle([])
|
|
pr = EgressRoute(host="api.openai.com")
|
|
routes = egress_routes_for_bottle(b, (pr,))
|
|
self.assertEqual(1, len(routes))
|
|
self.assertEqual("api.openai.com", routes[0].host)
|
|
self.assertEqual("", routes[0].auth_scheme)
|
|
self.assertEqual("", routes[0].token_env)
|
|
self.assertEqual("", routes[0].token_ref)
|
|
self.assertEqual({}, egress_token_env_map(routes))
|
|
|
|
def test_provider_route_wins_over_bare_manifest_route(self):
|
|
b = _bottle([{"host": "api.openai.com", "matches": [
|
|
{"paths": [{"value": "/v1/"}]}
|
|
]}])
|
|
pr = EgressRoute(host="api.openai.com")
|
|
routes = egress_routes_for_bottle(b, (pr,))
|
|
self.assertEqual(1, len(routes))
|
|
self.assertEqual("", routes[0].auth_scheme)
|
|
self.assertEqual("", routes[0].token_env)
|
|
self.assertEqual("", routes[0].token_ref)
|
|
self.assertEqual((), routes[0].matches)
|
|
self.assertEqual({}, egress_token_env_map(routes))
|
|
|
|
def test_provider_route_defaults_to_redact_on_match(self):
|
|
b = _bottle([])
|
|
pr = EgressRoute(host="api.anthropic.com")
|
|
routes = egress_routes_for_bottle(b, (pr,))
|
|
self.assertEqual("redact", routes[0].outbound_on_match)
|
|
|
|
def test_provider_route_explicit_on_match_preserved(self):
|
|
b = _bottle([])
|
|
pr = EgressRoute(host="api.anthropic.com", outbound_on_match="supervise")
|
|
routes = egress_routes_for_bottle(b, (pr,))
|
|
self.assertEqual("supervise", routes[0].outbound_on_match)
|
|
|
|
def test_manifest_route_does_not_get_redact_default(self):
|
|
b = _bottle([{"host": "api.example.com"}])
|
|
routes = egress_routes_for_bottle(b)
|
|
self.assertEqual("", routes[0].outbound_on_match)
|
|
|
|
def test_two_provider_routes_with_same_token_ref_share_slot(self):
|
|
b = _bottle([])
|
|
routes = egress_routes_for_bottle(b, (
|
|
_provider_route("api.openai.com", CODEX_HOST_CREDENTIAL_TOKEN_REF),
|
|
_provider_route("chatgpt.com", CODEX_HOST_CREDENTIAL_TOKEN_REF),
|
|
))
|
|
self.assertEqual(["api.openai.com", "chatgpt.com"], [r.host for r in routes])
|
|
self.assertEqual("EGRESS_TOKEN_0", routes[0].token_env)
|
|
self.assertEqual("EGRESS_TOKEN_0", routes[1].token_env)
|
|
|
|
def test_provider_route_wins_over_authed_manifest_route(self):
|
|
b = _bottle([{"host": "chatgpt.com",
|
|
"matches": [{"paths": [{"value": "/backend-api/"}]}],
|
|
"auth": {"scheme": "Bearer", "token_ref": "OTHER"}}])
|
|
pr = _provider_route("chatgpt.com", CODEX_HOST_CREDENTIAL_TOKEN_REF)
|
|
routes = egress_routes_for_bottle(b, (pr,))
|
|
self.assertEqual(1, len(routes))
|
|
self.assertEqual("chatgpt.com", routes[0].host)
|
|
self.assertEqual("Bearer", routes[0].auth_scheme)
|
|
self.assertEqual("EGRESS_TOKEN_0", routes[0].token_env)
|
|
self.assertEqual(CODEX_HOST_CREDENTIAL_TOKEN_REF, routes[0].token_ref)
|
|
self.assertEqual((), routes[0].matches)
|
|
|
|
def test_manifest_route_preserved_for_non_provisioned_host(self):
|
|
b = _bottle([
|
|
{"host": "api.openai.com"},
|
|
{"host": "api.github.com",
|
|
"auth": {"scheme": "Bearer", "token_ref": "GH_PAT"}},
|
|
])
|
|
pr = _provider_route("api.openai.com", CODEX_HOST_CREDENTIAL_TOKEN_REF)
|
|
routes = egress_routes_for_bottle(b, (pr,))
|
|
hosts = [r.host for r in routes]
|
|
self.assertEqual(["api.openai.com", "api.github.com"], hosts)
|
|
self.assertEqual(CODEX_HOST_CREDENTIAL_TOKEN_REF, routes[0].token_ref)
|
|
self.assertEqual("GH_PAT", routes[1].token_ref)
|
|
|
|
class TestTokenEnvMap(unittest.TestCase):
|
|
def test_only_authenticated_routes_contribute(self):
|
|
b = _bottle([
|
|
{"host": "a.example",
|
|
"auth": {"scheme": "Bearer", "token_ref": "T1"}},
|
|
{"host": "passthrough.example"},
|
|
])
|
|
routes = egress_routes_for_bottle(b)
|
|
m = egress_token_env_map(routes)
|
|
self.assertEqual({"EGRESS_TOKEN_0": "T1"}, m)
|
|
|
|
def test_no_routes_empty(self):
|
|
self.assertEqual({}, egress_token_env_map(()))
|
|
|
|
|
|
class TestRenderRoutes(unittest.TestCase):
|
|
"""Render is YAML now (PRD 0017 follow-up). Tests parse the
|
|
output via `yaml_subset` — the same parser the addon uses —
|
|
so the assertions check the actual semantic shape the addon
|
|
will see, not the textual layout."""
|
|
|
|
@staticmethod
|
|
def _parsed(routes) -> list[dict]: # type: ignore
|
|
return parse_yaml_subset(egress_render_routes(routes))["routes"] # type: ignore
|
|
|
|
def test_authenticated_route_serialised_with_auth_fields(self):
|
|
b = _bottle([{
|
|
"host": "api.github.com",
|
|
"auth": {"scheme": "Bearer", "token_ref": "GH_PAT"},
|
|
"matches": [{"paths": [{"value": "/repos/x/"}]}],
|
|
}])
|
|
routes = egress_routes_for_bottle(b)
|
|
parsed = self._parsed(routes)
|
|
self.assertEqual(1, len(parsed))
|
|
self.assertEqual("api.github.com", parsed[0]["host"])
|
|
self.assertEqual("Bearer", parsed[0]["auth_scheme"])
|
|
self.assertEqual("EGRESS_TOKEN_0", parsed[0]["token_env"])
|
|
self.assertIn("matches", parsed[0])
|
|
|
|
def test_unauthenticated_route_omits_auth_fields(self):
|
|
b = _bottle([{"host": "github.com", "matches": [
|
|
{"paths": [{"value": "/x/"}]}
|
|
]}])
|
|
routes = egress_routes_for_bottle(b)
|
|
entry = self._parsed(routes)[0]
|
|
self.assertNotIn("auth_scheme", entry)
|
|
self.assertNotIn("token_env", entry)
|
|
|
|
def test_no_matches_omits_field(self):
|
|
b = _bottle([{
|
|
"host": "api.anthropic.com",
|
|
"auth": {"scheme": "Bearer", "token_ref": "CL"},
|
|
}])
|
|
routes = egress_routes_for_bottle(b)
|
|
self.assertNotIn("matches", self._parsed(routes)[0])
|
|
|
|
def test_empty_routes_round_trips(self):
|
|
rendered = egress_render_routes(())
|
|
self.assertEqual([], parse_yaml_subset(rendered)["routes"])
|
|
|
|
def test_round_trip_through_addon_core(self):
|
|
from bot_bottle.egress_addon_core import load_config
|
|
b = _bottle([
|
|
{"host": "api.github.com",
|
|
"auth": {"scheme": "Bearer", "token_ref": "GH_PAT"},
|
|
"matches": [{"paths": [{"value": "/repos/x/"}]}]},
|
|
{"host": "github.com", "matches": [
|
|
{"paths": [{"value": "/x/"}]}
|
|
]},
|
|
{"host": "api.anthropic.com"},
|
|
])
|
|
routes = egress_routes_for_bottle(b)
|
|
addon_routes = load_config(egress_render_routes(routes)).routes
|
|
self.assertEqual(3, len(addon_routes))
|
|
self.assertEqual("Bearer", addon_routes[0].auth_scheme)
|
|
self.assertEqual("EGRESS_TOKEN_0", addon_routes[0].token_env)
|
|
self.assertEqual("", addon_routes[1].auth_scheme)
|
|
self.assertEqual("", addon_routes[2].auth_scheme)
|
|
|
|
def test_dlp_round_trips(self):
|
|
from bot_bottle.egress_addon_core import load_config
|
|
b = _bottle([{"host": "x.example", "dlp": {
|
|
"outbound_detectors": ["token_patterns"],
|
|
"inbound_detectors": False,
|
|
}}])
|
|
routes = egress_routes_for_bottle(b)
|
|
rendered = egress_render_routes(routes)
|
|
addon_routes = load_config(rendered).routes
|
|
self.assertEqual(("token_patterns",), addon_routes[0].outbound_detectors)
|
|
self.assertEqual((), addon_routes[0].inbound_detectors)
|
|
|
|
def test_outbound_on_match_round_trips(self):
|
|
from bot_bottle.egress_addon_core import load_config
|
|
b = _bottle([{"host": "logs.example", "dlp": {
|
|
"outbound_on_match": "redact",
|
|
}}])
|
|
routes = egress_routes_for_bottle(b)
|
|
rendered = egress_render_routes(routes)
|
|
self.assertIn('outbound_on_match: "redact"', rendered)
|
|
addon_routes = load_config(rendered).routes
|
|
self.assertEqual("redact", addon_routes[0].outbound_on_match)
|
|
|
|
def test_outbound_on_match_default_omitted_from_render(self):
|
|
b = _bottle([{"host": "x.example"}])
|
|
routes = egress_routes_for_bottle(b)
|
|
rendered = egress_render_routes(routes)
|
|
self.assertNotIn("outbound_on_match", rendered)
|
|
|
|
def test_git_fetch_policy_round_trips(self):
|
|
from bot_bottle.egress_addon_core import load_config
|
|
b = _bottle([{"host": "github.com", "git": {"fetch": True}}])
|
|
routes = egress_routes_for_bottle(b)
|
|
rendered = egress_render_routes(routes)
|
|
self.assertEqual({"fetch": True}, self._parsed(routes)[0]["git"])
|
|
addon_routes = load_config(rendered).routes
|
|
self.assertTrue(addon_routes[0].git_fetch)
|
|
|
|
def test_log_zero_omitted_from_render(self):
|
|
b = _bottle([{"host": "x.example"}])
|
|
routes = egress_routes_for_bottle(b)
|
|
rendered = egress_render_routes(routes, log=0)
|
|
self.assertNotIn("log:", rendered)
|
|
|
|
def test_log_level_emitted_at_top_level(self):
|
|
b = _bottle([{"host": "x.example"}])
|
|
routes = egress_routes_for_bottle(b)
|
|
for level in (1, 2):
|
|
with self.subTest(level=level):
|
|
rendered = egress_render_routes(routes, log=level)
|
|
self.assertTrue(rendered.startswith(f"log: {level}\n"))
|
|
|
|
def test_log_level_round_trips_to_addon_core(self):
|
|
from bot_bottle.egress_addon_core import load_config, LOG_FULL
|
|
b = _bottle([{"host": "x.example"}])
|
|
routes = egress_routes_for_bottle(b)
|
|
rendered = egress_render_routes(routes, log=LOG_FULL)
|
|
cfg = load_config(rendered)
|
|
self.assertEqual(LOG_FULL, cfg.log)
|
|
self.assertEqual("x.example", cfg.routes[0].host)
|
|
|
|
def test_log_via_manifest_flows_to_render(self):
|
|
from bot_bottle.manifest import ManifestIndex
|
|
from bot_bottle.egress_addon_core import load_config, LOG_BLOCKS
|
|
m = ManifestIndex.from_json_obj({
|
|
"bottles": {"dev": {"egress": {
|
|
"log": 1,
|
|
"routes": [{"host": "x.example"}],
|
|
}}},
|
|
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
|
|
})
|
|
bottle = m.bottles["dev"]
|
|
self.assertEqual(LOG_BLOCKS, bottle.egress.Log)
|
|
routes = egress_routes_for_bottle(bottle)
|
|
rendered = egress_render_routes(routes, log=bottle.egress.Log)
|
|
cfg = load_config(rendered)
|
|
self.assertEqual(LOG_BLOCKS, cfg.log)
|
|
|
|
|
|
class TestResolveTokenValues(unittest.TestCase):
|
|
def test_reads_host_env(self):
|
|
out = egress_resolve_token_values(
|
|
{"EGRESS_TOKEN_0": "GH_PAT"},
|
|
{"GH_PAT": "the-value"},
|
|
)
|
|
self.assertEqual({"EGRESS_TOKEN_0": "the-value"}, out)
|
|
|
|
def test_missing_token_ref_dies(self):
|
|
with self.assertRaises(Die):
|
|
egress_resolve_token_values(
|
|
{"EGRESS_TOKEN_0": "GH_PAT"},
|
|
{},
|
|
)
|
|
|
|
def test_empty_token_ref_dies(self):
|
|
with self.assertRaises(Die):
|
|
egress_resolve_token_values(
|
|
{"EGRESS_TOKEN_0": "GH_PAT"},
|
|
{"GH_PAT": ""},
|
|
)
|
|
|
|
def test_codex_host_credential_ref_resolved_via_provisioned_env(self):
|
|
out = egress_resolve_token_values(
|
|
{"EGRESS_TOKEN_0": CODEX_HOST_CREDENTIAL_TOKEN_REF},
|
|
{CODEX_HOST_CREDENTIAL_TOKEN_REF: "codex-access-token"},
|
|
)
|
|
self.assertEqual({"EGRESS_TOKEN_0": "codex-access-token"}, out)
|
|
|
|
|
|
class TestCanaryGeneration(unittest.TestCase):
|
|
"""Egress.prepare() generates a unique canary token per session."""
|
|
|
|
def _bottle_obj(self):
|
|
return ManifestIndex.from_json_obj({
|
|
"bottles": {"dev": {"egress": {"routes": []}}},
|
|
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
|
|
}).bottles["dev"]
|
|
|
|
def _make_plan(self) -> EgressPlan:
|
|
# Use a concrete no-op subclass so we can call prepare() without
|
|
# a real backend.
|
|
class _TestEgress(Egress):
|
|
pass
|
|
|
|
e = _TestEgress()
|
|
with tempfile.TemporaryDirectory() as td:
|
|
return e.prepare(self._bottle_obj(), "test-slug", Path(td))
|
|
|
|
def test_canary_is_non_empty(self):
|
|
plan = self._make_plan()
|
|
self.assertIsInstance(plan.canary, str)
|
|
self.assertGreater(len(plan.canary), 0)
|
|
self.assertRegex(plan.canary_env, r"^[A-Z]+_[A-Z]+_SECRET$")
|
|
|
|
def test_canary_is_unique_per_session(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
bottle = self._bottle_obj()
|
|
|
|
class _TestEgress(Egress):
|
|
pass
|
|
|
|
e = _TestEgress()
|
|
plan_a = e.prepare(bottle, "slug-a", Path(td))
|
|
plan_b = e.prepare(bottle, "slug-b", Path(td))
|
|
self.assertNotEqual(plan_a.canary, plan_b.canary)
|
|
|
|
def test_canary_detected_by_scan_known_secrets(self):
|
|
from bot_bottle.dlp_detectors import scan_known_secrets
|
|
|
|
plan = self._make_plan()
|
|
env = {plan.canary_env: plan.canary}
|
|
result = scan_known_secrets(
|
|
f"exfil={plan.canary}",
|
|
env=env,
|
|
sensitive_prefixes=(plan.canary_env,),
|
|
)
|
|
self.assertIsNotNone(result)
|
|
assert result is not None
|
|
self.assertEqual("block", result.severity)
|
|
self.assertIn(plan.canary_env, result.reason)
|
|
|
|
def test_egress_plan_canary_field_default_empty(self):
|
|
# Verify EgressPlan can be constructed with an empty canary (backward compat).
|
|
from pathlib import Path
|
|
plan = EgressPlan(
|
|
slug="s",
|
|
routes_path=Path("/tmp/r.yaml"),
|
|
routes=(),
|
|
token_env_map={},
|
|
)
|
|
self.assertEqual("", plan.canary)
|
|
self.assertEqual("", plan.canary_env)
|
|
|
|
|
|
class TestEgressEnvEntries(unittest.TestCase):
|
|
def test_sidecar_entries_include_route_tokens_and_canary_scan_prefix(self):
|
|
plan = EgressPlan(
|
|
slug="s",
|
|
routes_path=Path("/tmp/r.yaml"),
|
|
routes=(EgressRoute(host="api.example"),),
|
|
token_env_map={"EGRESS_TOKEN_1": "T1", "EGRESS_TOKEN_0": "T0"},
|
|
canary="fake-canary-value",
|
|
canary_env="CANON_ALPHA_SECRET",
|
|
)
|
|
|
|
self.assertEqual(
|
|
(
|
|
"EGRESS_TOKEN_0",
|
|
"EGRESS_TOKEN_1",
|
|
"CANON_ALPHA_SECRET=fake-canary-value",
|
|
"BOT_BOTTLE_SENSITIVE_PREFIXES=CANON_ALPHA_SECRET",
|
|
),
|
|
egress_sidecar_env_entries(plan),
|
|
)
|
|
|
|
def test_agent_entries_include_only_canary_bait(self):
|
|
plan = EgressPlan(
|
|
slug="s",
|
|
routes_path=Path("/tmp/r.yaml"),
|
|
routes=(),
|
|
token_env_map={},
|
|
canary="fake-canary-value",
|
|
canary_env="CANON_ALPHA_SECRET",
|
|
)
|
|
|
|
self.assertEqual(
|
|
("CANON_ALPHA_SECRET=fake-canary-value",),
|
|
egress_agent_env_entries(plan),
|
|
)
|
|
|
|
def test_canary_entries_omitted_when_name_missing(self):
|
|
plan = EgressPlan(
|
|
slug="s",
|
|
routes_path=Path("/tmp/r.yaml"),
|
|
routes=(),
|
|
token_env_map={},
|
|
canary="fake-canary-value",
|
|
)
|
|
|
|
self.assertEqual((), egress_sidecar_env_entries(plan))
|
|
self.assertEqual((), egress_agent_env_entries(plan))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|