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

refactor: provision egress routes via AgentProvisionPlan

Browse Source 
Remove provider-specific branching from egress.py and pipelock.py.
Previously, `egress_routes_for_bottle` and `pipelock_effective_tls_passthrough`
both contained `template == "codex"` checks — the same pattern the rest
of the PR moved out of the backends.

Root cause: `EgressRoute` had no `tls_passthrough` field, so pipelock
couldn't learn from the synthesised Codex routes that they needed
passthrough. Fix:

- Add `EgressRoute.tls_passthrough: bool`. `egress_manifest_routes` lifts
  the existing `pipelock.tls_passthrough` manifest flag here; provider
  routes set it directly.
- Add `AgentProvisionPlan.egress_routes`. `agent_provision_plan` populates
  it for Codex + `forward_host_credentials`, including `tls_passthrough=True`.
- Replace Codex-specific `egress_routes_for_bottle` logic with a generic
  `_merge_provider_route` helper. Backends call `egress_routes_for_bottle(bottle,
  plan.egress_routes)`; no provider type checks inside egress or pipelock.
- Rewrite `pipelock_effective_tls_passthrough` to read `route.tls_passthrough`
  from the merged route set instead of re-implementing the provider check.
- Both backends now call `agent_provision_plan` before `Egress.prepare` and
  `PipelockProxy.prepare`, threading `plan.egress_routes` to both. `has_provider_auth`
  is derived from `egress_manifest_routes` (manifest routes only — provider
  routes carry no auth roles, so the result is identical).

Assisted-by: Claude Code
This commit is contained in:
didericis (claude)
2026-06-01 23:27:27 +00:00
committed by didericis
parent 76a7921ae6
commit 884cedc160
8 changed files with 350 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_agent_provider.py
+47 -1
View File
@@ -8,7 +8,12 @@ import tempfile
import unittest
from pathlib import Path
from bot_bottle.agent_provider import agent_provision_plan, runtime_for
from bot_bottle.agent_provider import (
CODEX_HOST_CREDENTIAL_HOSTS,
agent_provision_plan,
runtime_for,
)
from bot_bottle.egress import CODEX_HOST_CREDENTIAL_TOKEN_REF
def _jwt(exp: int) -> str:
@@ -90,6 +95,47 @@ class TestAgentProviderRuntime(unittest.TestCase):
)
self.assertEqual("1", plan.env_vars["DISABLE_ERROR_REPORTING"])
def test_codex_forward_host_credentials_populates_egress_routes(self):
with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
home = Path(tmp) / "host-codex"
home.mkdir()
(home / "auth.json").write_text(json.dumps({
"auth_mode": "chatgpt",
"tokens": {"access_token": _jwt(2000000000)},
}))
plan = agent_provision_plan(
template="codex",
dockerfile="",
state_dir=Path(tmp),
forward_host_credentials=True,
host_env={"CODEX_HOME": str(home)},
)
hosts = [r.host for r in plan.egress_routes]
self.assertEqual(sorted(CODEX_HOST_CREDENTIAL_HOSTS), sorted(hosts))
for r in plan.egress_routes:
self.assertEqual("Bearer", r.auth_scheme)
self.assertEqual(CODEX_HOST_CREDENTIAL_TOKEN_REF, r.token_ref)
self.assertTrue(r.tls_passthrough)
def test_codex_without_forward_host_credentials_has_no_egress_routes(self):
with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
plan = agent_provision_plan(
template="codex",
dockerfile="",
state_dir=Path(tmp),
forward_host_credentials=False,
)
self.assertEqual((), plan.egress_routes)
def test_claude_plan_has_no_egress_routes(self):
with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
plan = agent_provision_plan(
template="claude",
dockerfile="",
state_dir=Path(tmp),
)
self.assertEqual((), plan.egress_routes)
if __name__ == "__main__":
unittest.main()
tests/unit/test_egress.py
+87 -54
View File
@@ -5,6 +5,7 @@ import unittest
from bot_bottle.egress import (
CODEX_HOST_CREDENTIAL_TOKEN_REF,
EgressRoute,
egress_manifest_routes,
egress_render_routes,
egress_resolve_token_values,
@@ -23,19 +24,13 @@ def _bottle(routes):
}).bottles["dev"]
def _codex_bottle(*, forward_host_credentials: bool, routes):
return Manifest.from_json_obj({
"bottles": {
"dev": {
"agent_provider": {
"template": "codex",
"forward_host_credentials": forward_host_credentials,
},
"egress": {"routes": routes},
}
},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
}).bottles["dev"]
def _provider_route(host: str, token_ref: str, *, tls_passthrough: bool = False) -> EgressRoute:
return EgressRoute(
host=host,
auth_scheme="Bearer",
token_ref=token_ref,
tls_passthrough=tls_passthrough,
)
class TestRoutesForBottle(unittest.TestCase):
@@ -100,9 +95,8 @@ class TestRoutesForBottle(unittest.TestCase):
self.assertEqual("", routes[1].token_env)
class TestRoutesForBottleUsesManifestOnly(unittest.TestCase):
"""The effective route table is exactly the manifest-declared
routes. Provider defaults are not injected implicitly."""
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([])
@@ -123,58 +117,97 @@ class TestRoutesForBottleUsesManifestOnly(unittest.TestCase):
effective = [r.host for r in egress_routes_for_bottle(b)]
self.assertEqual(["x.example"], effective)
def test_codex_forward_host_credentials_adds_codex_routes(self):
b = _codex_bottle(forward_host_credentials=True, routes=[])
def test_tls_passthrough_lifted_from_manifest(self):
b = _bottle([{
"host": "api.openai.com",
"auth": {"scheme": "Bearer", "token_ref": "T"},
"pipelock": {"tls_passthrough": True},
}])
routes = egress_routes_for_bottle(b)
self.assertEqual(["api.openai.com", "chatgpt.com"], [r.host for r in routes])
self.assertTrue(routes[0].tls_passthrough)
def test_tls_passthrough_false_by_default(self):
b = _bottle([{"host": "api.github.com"}])
routes = egress_routes_for_bottle(b)
self.assertFalse(routes[0].tls_passthrough)
class TestProviderRouteMerge(unittest.TestCase):
"""Provider routes are merged into manifest routes generically."""
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(CODEX_HOST_CREDENTIAL_TOKEN_REF, routes[0].token_ref)
self.assertEqual("Bearer", routes[1].auth_scheme)
self.assertEqual("EGRESS_TOKEN_0", routes[1].token_env)
self.assertEqual(CODEX_HOST_CREDENTIAL_TOKEN_REF, routes[1].token_ref)
self.assertEqual("TOK", routes[0].token_ref)
def test_codex_forward_host_credentials_upgrades_bare_chatgpt_route(self):
b = _codex_bottle(
forward_host_credentials=True,
routes=[{"host": "chatgpt.com", "path_allowlist": ["/backend-api/"]}],
)
routes = egress_routes_for_bottle(b)
self.assertEqual(2, len(routes))
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_upgrades_bare_manifest_route(self):
b = _bottle([{"host": "chatgpt.com", "path_allowlist": ["/backend-api/"]}])
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(("/backend-api/",), routes[0].path_allowlist)
self.assertEqual("api.openai.com", routes[1].host)
self.assertEqual("EGRESS_TOKEN_0", routes[1].token_env)
def test_codex_forward_host_credentials_accepts_explicit_synthetic_route(self):
b = _codex_bottle(
forward_host_credentials=True,
routes=[{
"host": "api.openai.com",
"auth": {
"scheme": "Bearer",
"token_ref": CODEX_HOST_CREDENTIAL_TOKEN_REF,
},
}],
)
routes = egress_routes_for_bottle(b)
self.assertEqual(["api.openai.com", "chatgpt.com"], [r.host for r in routes])
def test_provider_route_noop_when_same_auth_already_in_manifest(self):
b = _bottle([{
"host": "api.openai.com",
"auth": {"scheme": "Bearer", "token_ref": CODEX_HOST_CREDENTIAL_TOKEN_REF},
}])
pr = _provider_route("api.openai.com", CODEX_HOST_CREDENTIAL_TOKEN_REF)
routes = egress_routes_for_bottle(b, (pr,))
self.assertEqual(1, len(routes))
self.assertEqual("EGRESS_TOKEN_0", routes[0].token_env)
self.assertEqual("EGRESS_TOKEN_0", routes[1].token_env)
def test_codex_forward_host_credentials_conflicts_with_authed_route(self):
b = _codex_bottle(
forward_host_credentials=True,
routes=[{
"host": "chatgpt.com",
"auth": {"scheme": "Bearer", "token_ref": "OTHER"},
}],
def test_provider_route_upgrades_tls_passthrough_on_existing_same_auth(self):
b = _bottle([{
"host": "api.openai.com",
"auth": {"scheme": "Bearer", "token_ref": CODEX_HOST_CREDENTIAL_TOKEN_REF},
}])
pr = _provider_route(
"api.openai.com", CODEX_HOST_CREDENTIAL_TOKEN_REF, tls_passthrough=True,
)
routes = egress_routes_for_bottle(b, (pr,))
self.assertEqual(1, len(routes))
self.assertTrue(routes[0].tls_passthrough)
def test_provider_route_conflicts_with_different_authed_manifest_route(self):
b = _bottle([{
"host": "chatgpt.com",
"auth": {"scheme": "Bearer", "token_ref": "OTHER"},
}])
pr = _provider_route("chatgpt.com", CODEX_HOST_CREDENTIAL_TOKEN_REF)
with self.assertRaises(Die):
egress_routes_for_bottle(b)
egress_routes_for_bottle(b, (pr,))
def test_provider_route_tls_passthrough_set_on_appended_route(self):
b = _bottle([])
pr = _provider_route("api.openai.com", "TOK", tls_passthrough=True)
routes = egress_routes_for_bottle(b, (pr,))
self.assertTrue(routes[0].tls_passthrough)
def test_provider_route_tls_passthrough_set_on_upgraded_bare_route(self):
b = _bottle([{"host": "api.openai.com"}])
pr = _provider_route("api.openai.com", "TOK", tls_passthrough=True)
routes = egress_routes_for_bottle(b, (pr,))
self.assertTrue(routes[0].tls_passthrough)
class TestTokenEnvMap(unittest.TestCase):
tests/unit/test_pipelock_allowlist.py
+17 -9
View File
@@ -5,6 +5,8 @@ git-gate (PRD 0008)."""
import unittest
from bot_bottle.agent_provider import CODEX_HOST_CREDENTIAL_HOSTS
from bot_bottle.egress import CODEX_HOST_CREDENTIAL_TOKEN_REF, EgressRoute
from bot_bottle.manifest import Manifest
from bot_bottle.pipelock import (
pipelock_effective_allowlist,
@@ -116,17 +118,23 @@ class TestTlsPassthrough(unittest.TestCase):
def test_forward_host_credentials_passes_through_codex_hosts(self):
# Egress injects the host bearer on the Codex API hosts; pipelock
# must pass them through or its header DLP blocks the injected JWT
# ("request header contains secret"). These routes are auto-added
# (not in bottle.egress.routes), so passthrough is host-derived.
passthrough = pipelock_effective_tls_passthrough(_bottle({
"agent_provider": {
"template": "codex",
"forward_host_credentials": True,
},
}))
# ("request header contains secret"). Provider routes carry
# tls_passthrough=True; pipelock reads this via egress_routes_for_bottle.
provider_routes = tuple(
EgressRoute(
host=host,
auth_scheme="Bearer",
token_ref=CODEX_HOST_CREDENTIAL_TOKEN_REF,
tls_passthrough=True,
)
for host in CODEX_HOST_CREDENTIAL_HOSTS
)
passthrough = pipelock_effective_tls_passthrough(
_bottle({}), provider_routes,
)
self.assertEqual(["api.openai.com", "chatgpt.com"], passthrough)
def test_no_codex_passthrough_without_forward_host_credentials(self):
def test_no_codex_passthrough_without_provider_routes(self):
passthrough = pipelock_effective_tls_passthrough(_bottle({
"agent_provider": {"template": "codex"},
}))