didericis-claude
de9bd7eb83
Operators can now declare:
agent_provider:
template: claude
auth_token: BOT_BOTTLE_CLAUDE_OAUTH_TOKEN
and the provisioner injects a provider-owned api.anthropic.com egress
route (Bearer, tls_passthrough) rather than requiring a manually
declared route with the former claude_code_oauth role.
Changes:
- Add auth_token field to AgentProvider; validate claude-only.
- Remove claude_code_oauth from EGRESS_ROLES / PROVIDER_EGRESS_ROLES.
Manifests that declare the role now fail at parse time with "unknown
role" — the provisioner owns the route.
- agent_provision_plan: replace manifest_egress_routes/has_provider_auth
with auth_token; Claude branch injects the api.anthropic.com route,
placeholder env, and nonessential-traffic flags when auth_token is set.
- Add hidden_env_names: frozenset[str] to AgentProvisionPlan; Claude
branch populates it with CLAUDE_CODE_OAUTH_TOKEN.
- Remove auth_role from AgentProviderRuntime and placeholder_env_for().
- print_util.visible_agent_env_names: accept hidden_env_names from the
plan instead of dispatching on agent_provider_template.
- Both backends: drop manifest_egress_routes call, pass auth_token.
- PRD 0029 rescoped to cover both Codex and Claude provider auth.
Assisted-by: Claude Code
358 lines
12 KiB
Python
358 lines
12 KiB
Python
"""Unit: manifest parsing for `bottle.egress.routes[]` (PRD 0017).
|
|
|
|
The route shape is new: `host` (required), optional `path_allowlist`,
|
|
optional nested `auth: { scheme, token_ref }`. Validation rules per
|
|
the PRD: empty `auth: {}` is an error, partial `auth` is an error,
|
|
auth omission means unauthenticated."""
|
|
|
|
import unittest
|
|
|
|
from bot_bottle.manifest import ManifestError, EgressRoute, Manifest
|
|
|
|
|
|
def _bottle(routes):
|
|
return Manifest.from_json_obj({
|
|
"bottles": {"dev": {"egress": {"routes": routes}}},
|
|
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
|
|
}).bottles["dev"]
|
|
|
|
|
|
def _provider_bottle(provider, routes):
|
|
return Manifest.from_json_obj({
|
|
"bottles": {
|
|
"dev": {
|
|
"agent_provider": {"template": provider},
|
|
"egress": {"routes": routes},
|
|
}
|
|
},
|
|
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
|
|
}).bottles["dev"]
|
|
|
|
|
|
def _provider_config_bottle(agent_provider):
|
|
return Manifest.from_json_obj({
|
|
"bottles": {"dev": {"agent_provider": agent_provider}},
|
|
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
|
|
}).bottles["dev"]
|
|
|
|
|
|
class TestMinimalRoute(unittest.TestCase):
|
|
def test_host_only(self):
|
|
b = _bottle([{"host": "api.example.com"}])
|
|
self.assertEqual(1, len(b.egress.routes))
|
|
r = b.egress.routes[0]
|
|
self.assertEqual("api.example.com", r.Host)
|
|
self.assertEqual((), r.PathAllowlist)
|
|
self.assertEqual("", r.AuthScheme)
|
|
self.assertEqual("", r.TokenRef)
|
|
|
|
def test_host_required(self):
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{}])
|
|
|
|
def test_host_must_be_non_empty(self):
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{"host": ""}])
|
|
|
|
def test_unknown_top_level_key_dies(self):
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{"host": "x.example", "wat": "yes"}])
|
|
|
|
|
|
class TestAgentProviderHostCredentials(unittest.TestCase):
|
|
def test_forward_host_credentials_defaults_false(self):
|
|
b = _provider_config_bottle({"template": "codex"})
|
|
self.assertFalse(b.agent_provider.forward_host_credentials)
|
|
|
|
def test_forward_host_credentials_allowed_for_codex(self):
|
|
b = _provider_config_bottle({
|
|
"template": "codex",
|
|
"forward_host_credentials": True,
|
|
})
|
|
self.assertTrue(b.agent_provider.forward_host_credentials)
|
|
|
|
def test_forward_host_credentials_must_be_boolean(self):
|
|
with self.assertRaises(ManifestError):
|
|
_provider_config_bottle({
|
|
"template": "codex",
|
|
"forward_host_credentials": "yes",
|
|
})
|
|
|
|
def test_forward_host_credentials_rejected_for_claude(self):
|
|
with self.assertRaises(ManifestError):
|
|
_provider_config_bottle({
|
|
"template": "claude",
|
|
"forward_host_credentials": True,
|
|
})
|
|
|
|
def test_auth_token_defaults_empty(self):
|
|
b = _provider_config_bottle({"template": "claude"})
|
|
self.assertEqual("", b.agent_provider.auth_token)
|
|
|
|
def test_auth_token_allowed_for_claude(self):
|
|
b = _provider_config_bottle({
|
|
"template": "claude",
|
|
"auth_token": "BOT_BOTTLE_CLAUDE_OAUTH_TOKEN",
|
|
})
|
|
self.assertEqual("BOT_BOTTLE_CLAUDE_OAUTH_TOKEN", b.agent_provider.auth_token)
|
|
|
|
def test_auth_token_must_be_string(self):
|
|
with self.assertRaises(ManifestError):
|
|
_provider_config_bottle({
|
|
"template": "claude",
|
|
"auth_token": 42,
|
|
})
|
|
|
|
def test_auth_token_rejected_for_codex(self):
|
|
with self.assertRaises(ManifestError):
|
|
_provider_config_bottle({
|
|
"template": "codex",
|
|
"auth_token": "SOME_TOKEN",
|
|
})
|
|
|
|
|
|
class TestPathAllowlist(unittest.TestCase):
|
|
def test_optional(self):
|
|
b = _bottle([{"host": "x.example"}])
|
|
self.assertEqual((), b.egress.routes[0].PathAllowlist)
|
|
|
|
def test_must_be_array(self):
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{"host": "x.example", "path_allowlist": "/x/"}])
|
|
|
|
def test_items_must_be_strings(self):
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{"host": "x.example", "path_allowlist": [42]}])
|
|
|
|
def test_items_must_be_absolute_paths(self):
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{"host": "x.example", "path_allowlist": ["nope/"]}])
|
|
|
|
def test_full_list(self):
|
|
b = _bottle([{
|
|
"host": "github.com",
|
|
"path_allowlist": ["/didericis/", "/users/didericis"],
|
|
}])
|
|
self.assertEqual(
|
|
("/didericis/", "/users/didericis"),
|
|
b.egress.routes[0].PathAllowlist,
|
|
)
|
|
|
|
|
|
class TestAuth(unittest.TestCase):
|
|
def test_omitted_means_no_auth(self):
|
|
b = _bottle([{"host": "github.com"}])
|
|
r = b.egress.routes[0]
|
|
self.assertEqual("", r.AuthScheme)
|
|
self.assertEqual("", r.TokenRef)
|
|
|
|
def test_full_auth(self):
|
|
b = _bottle([{
|
|
"host": "api.github.com",
|
|
"auth": {"scheme": "Bearer", "token_ref": "GH_PAT"},
|
|
}])
|
|
r = b.egress.routes[0]
|
|
self.assertEqual("Bearer", r.AuthScheme)
|
|
self.assertEqual("GH_PAT", r.TokenRef)
|
|
|
|
def test_empty_auth_block_rejected(self):
|
|
# Per PRD 0017: `auth: {}` is an error, not a synonym for
|
|
# "no auth" — that's what omission is for.
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{"host": "x.example", "auth": {}}])
|
|
|
|
def test_missing_scheme_rejected(self):
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{
|
|
"host": "x.example",
|
|
"auth": {"token_ref": "T"},
|
|
}])
|
|
|
|
def test_missing_token_ref_rejected(self):
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{
|
|
"host": "x.example",
|
|
"auth": {"scheme": "Bearer"},
|
|
}])
|
|
|
|
def test_unknown_scheme_rejected(self):
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{
|
|
"host": "x.example",
|
|
"auth": {"scheme": "Basic", "token_ref": "T"},
|
|
}])
|
|
|
|
def test_token_scheme_allowed(self):
|
|
# Gitea quirk: `Authorization: token <pat>` (not Bearer).
|
|
b = _bottle([{
|
|
"host": "git.example",
|
|
"auth": {"scheme": "token", "token_ref": "GITEA_PAT"},
|
|
}])
|
|
self.assertEqual("token", b.egress.routes[0].AuthScheme)
|
|
|
|
def test_unknown_auth_key_rejected(self):
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{
|
|
"host": "x.example",
|
|
"auth": {"scheme": "Bearer", "token_ref": "T", "extra": "no"},
|
|
}])
|
|
|
|
|
|
class TestRole(unittest.TestCase):
|
|
def test_omitted_means_no_roles(self):
|
|
b = _bottle([{"host": "x.example"}])
|
|
self.assertEqual((), b.egress.routes[0].Role)
|
|
|
|
def test_string_normalizes_to_tuple(self):
|
|
b = _provider_bottle("codex", [{
|
|
"host": "api.openai.com",
|
|
"role": "codex_auth",
|
|
"auth": {"scheme": "Bearer", "token_ref": "T"},
|
|
}])
|
|
self.assertEqual(("codex_auth",), b.egress.routes[0].Role)
|
|
|
|
def test_list_supported(self):
|
|
b = _provider_bottle("codex", [{
|
|
"host": "api.openai.com",
|
|
"role": ["codex_auth"],
|
|
"auth": {"scheme": "Bearer", "token_ref": "T"},
|
|
}])
|
|
self.assertEqual(("codex_auth",), b.egress.routes[0].Role)
|
|
|
|
def test_unknown_role_rejected(self):
|
|
# The role enum is locked down — typos shouldn't silently
|
|
# become no-op markers.
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{"host": "x.example", "role": "totally-made-up"}])
|
|
|
|
def test_claude_code_oauth_role_rejected(self):
|
|
# claude_code_oauth was removed; provisioner injects the route
|
|
# automatically via agent_provider.auth_token.
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{"host": "api.anthropic.com",
|
|
"role": "claude_code_oauth",
|
|
"auth": {"scheme": "Bearer", "token_ref": "T"}}])
|
|
|
|
def test_non_string_role_rejected(self):
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{"host": "x.example", "role": 42}])
|
|
|
|
def test_list_with_non_string_item_rejected(self):
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{"host": "x.example",
|
|
"role": ["codex_auth", 42]}])
|
|
|
|
def test_codex_auth_role_allowed_for_codex_provider(self):
|
|
b = _provider_bottle("codex", [{
|
|
"host": "api.openai.com",
|
|
"role": "codex_auth",
|
|
"auth": {"scheme": "Bearer", "token_ref": "OPENAI_TOKEN"},
|
|
}])
|
|
self.assertEqual(("codex_auth",), b.egress.routes[0].Role)
|
|
|
|
def test_codex_role_rejected_for_default_claude_provider(self):
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{
|
|
"host": "api.openai.com",
|
|
"role": "codex_auth",
|
|
"auth": {"scheme": "Bearer", "token_ref": "T"},
|
|
}])
|
|
|
|
|
|
class TestPipelockPolicy(unittest.TestCase):
|
|
def test_tls_passthrough_route_policy(self):
|
|
b = _bottle([{
|
|
"host": "api.openai.com",
|
|
"pipelock": {"tls_passthrough": True},
|
|
}])
|
|
self.assertTrue(b.egress.routes[0].Pipelock.TlsPassthrough)
|
|
|
|
def test_ssrf_ip_allowlist_route_policy(self):
|
|
b = _bottle([{
|
|
"host": "gitea.dideric.is",
|
|
"pipelock": {"ssrf_ip_allowlist": ["100.78.141.42/32"]},
|
|
}])
|
|
self.assertEqual(
|
|
("100.78.141.42/32",),
|
|
b.egress.routes[0].Pipelock.SsrfIpAllowlist,
|
|
)
|
|
|
|
def test_tls_passthrough_defaults_false(self):
|
|
b = _bottle([{"host": "api.openai.com"}])
|
|
self.assertFalse(b.egress.routes[0].Pipelock.TlsPassthrough)
|
|
self.assertEqual((), b.egress.routes[0].Pipelock.SsrfIpAllowlist)
|
|
|
|
def test_pipelock_policy_must_be_object(self):
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{"host": "x.example", "pipelock": True}])
|
|
|
|
def test_tls_passthrough_must_be_bool(self):
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{
|
|
"host": "x.example",
|
|
"pipelock": {"tls_passthrough": "yes"},
|
|
}])
|
|
|
|
def test_ssrf_ip_allowlist_must_be_array(self):
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{
|
|
"host": "x.example",
|
|
"pipelock": {"ssrf_ip_allowlist": "100.78.141.42/32"},
|
|
}])
|
|
|
|
def test_ssrf_ip_allowlist_items_must_be_cidr_or_ip(self):
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{
|
|
"host": "x.example",
|
|
"pipelock": {"ssrf_ip_allowlist": ["not-an-ip"]},
|
|
}])
|
|
|
|
def test_unknown_pipelock_key_rejected(self):
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([{"host": "x.example", "pipelock": {"wat": True}}])
|
|
|
|
|
|
class TestRouteValidation(unittest.TestCase):
|
|
def test_duplicate_hosts_rejected(self):
|
|
# Routes match by exact host; duplicates leave the choice
|
|
# ambiguous, so we reject them up front rather than picking
|
|
# the first/last silently.
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([
|
|
{"host": "github.com"},
|
|
{"host": "github.com", "path_allowlist": ["/x/"]},
|
|
])
|
|
|
|
def test_duplicate_host_case_insensitive(self):
|
|
with self.assertRaises(ManifestError):
|
|
_bottle([
|
|
{"host": "GitHub.com"},
|
|
{"host": "github.com"},
|
|
])
|
|
|
|
def test_empty_routes_allowed(self):
|
|
b = _bottle([])
|
|
self.assertEqual((), b.egress.routes)
|
|
|
|
def test_no_egress_block_means_empty(self):
|
|
# The bottle dataclass defaults to an empty EgressConfig.
|
|
b = Manifest.from_json_obj({
|
|
"bottles": {"dev": {}},
|
|
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
|
|
}).bottles["dev"]
|
|
self.assertEqual((), b.egress.routes)
|
|
|
|
|
|
class TestConfigShape(unittest.TestCase):
|
|
def test_unknown_egress_key_rejected(self):
|
|
with self.assertRaises(ManifestError):
|
|
Manifest.from_json_obj({
|
|
"bottles": {"dev": {"egress": {"wat": []}}},
|
|
"agents": {"demo": {"skills": [], "prompt": "",
|
|
"bottle": "dev"}},
|
|
})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|