Files
bot-bottle/tests/unit/test_manifest_egress_proxy.py
T
didericis 3df54573d4
test / unit (pull_request) Successful in 18s
test / integration (pull_request) Successful in 1m39s
feat(egress-proxy): add mitmproxy-based sidecar core (PRD 0017 chunk 1)
Lands the new egress-proxy artifact alongside cred-proxy. Chunk 2
wires the agent's HTTP_PROXY to it and removes cred-proxy.

  - `Dockerfile.egress-proxy` — mitmproxy 11.1.3 base, COPY addon
    files flat to /app, mkdir routes dir at /etc/egress-proxy/.
    Digest pin deferred to chunk 2.
  - `egress_proxy_addon_core.py` — pure-logic parse + decide
    (host-importable; 21 unit tests).
  - `egress_proxy_addon.py` — mitmproxy hook wrapper, container-only
    (boot + SIGHUP reload, strip-Authorization + decide + 403/inject).
  - `egress_proxy.py` — host helpers: manifest lift, routes.yaml
    render (JSON content), token-env-map, Plan + abstract class.
  - `backend/docker/egress_proxy.py` — `DockerEgressProxy` start/stop
    mirroring `DockerCredProxy`; not yet called from launch.py.
  - `manifest.py` — new `EgressProxyRoute` + `EgressProxyConfig` types
    with the nested `auth: { scheme, token_ref }` block per PRD;
    `bottle.egress_proxy` added to the bottle key set alongside
    `cred_proxy` (chunk 2 hard-fails on the latter).

All 427 unit tests pass. Image builds; `docker run` boots mitmdump
and the addon loads routes from a mounted routes.yaml.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-25 13:58:24 -04:00

174 lines
5.8 KiB
Python

"""Unit: manifest parsing for `bottle.egress_proxy.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 claude_bottle.log import Die
from claude_bottle.manifest import EgressProxyRoute, Manifest
def _bottle(routes):
return Manifest.from_json_obj({
"bottles": {"dev": {"egress_proxy": {"routes": routes}}},
"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_proxy.routes))
r = b.egress_proxy.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(Die):
_bottle([{}])
def test_host_must_be_non_empty(self):
with self.assertRaises(Die):
_bottle([{"host": ""}])
def test_unknown_top_level_key_dies(self):
with self.assertRaises(Die):
_bottle([{"host": "x.example", "wat": "yes"}])
class TestPathAllowlist(unittest.TestCase):
def test_optional(self):
b = _bottle([{"host": "x.example"}])
self.assertEqual((), b.egress_proxy.routes[0].PathAllowlist)
def test_must_be_array(self):
with self.assertRaises(Die):
_bottle([{"host": "x.example", "path_allowlist": "/x/"}])
def test_items_must_be_strings(self):
with self.assertRaises(Die):
_bottle([{"host": "x.example", "path_allowlist": [42]}])
def test_items_must_be_absolute_paths(self):
with self.assertRaises(Die):
_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_proxy.routes[0].PathAllowlist,
)
class TestAuth(unittest.TestCase):
def test_omitted_means_no_auth(self):
b = _bottle([{"host": "github.com"}])
r = b.egress_proxy.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_proxy.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(Die):
_bottle([{"host": "x.example", "auth": {}}])
def test_missing_scheme_rejected(self):
with self.assertRaises(Die):
_bottle([{
"host": "x.example",
"auth": {"token_ref": "T"},
}])
def test_missing_token_ref_rejected(self):
with self.assertRaises(Die):
_bottle([{
"host": "x.example",
"auth": {"scheme": "Bearer"},
}])
def test_unknown_scheme_rejected(self):
with self.assertRaises(Die):
_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_proxy.routes[0].AuthScheme)
def test_unknown_auth_key_rejected(self):
with self.assertRaises(Die):
_bottle([{
"host": "x.example",
"auth": {"scheme": "Bearer", "token_ref": "T", "extra": "no"},
}])
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(Die):
_bottle([
{"host": "github.com"},
{"host": "github.com", "path_allowlist": ["/x/"]},
])
def test_duplicate_host_case_insensitive(self):
with self.assertRaises(Die):
_bottle([
{"host": "GitHub.com"},
{"host": "github.com"},
])
def test_empty_routes_allowed(self):
b = _bottle([])
self.assertEqual((), b.egress_proxy.routes)
def test_no_egress_proxy_block_means_empty(self):
# The bottle dataclass defaults to an empty EgressProxyConfig.
b = Manifest.from_json_obj({
"bottles": {"dev": {}},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
}).bottles["dev"]
self.assertEqual((), b.egress_proxy.routes)
class TestConfigShape(unittest.TestCase):
def test_unknown_egress_proxy_key_rejected(self):
with self.assertRaises(Die):
Manifest.from_json_obj({
"bottles": {"dev": {"egress_proxy": {"wat": []}}},
"agents": {"demo": {"skills": [], "prompt": "",
"bottle": "dev"}},
})
if __name__ == "__main__":
unittest.main()