feat(egress-proxy): block HTTPS git push + restore role provisioner
test / unit (pull_request) Successful in 17s
test / integration (pull_request) Successful in 1m1s

Two related fixes on top of PR #29's chunk-2 cutover:

1. Universal HTTPS git-push block in the egress-proxy addon
   (`is_git_push_request` in egress_proxy_addon_core, called from the
   mitmproxy request hook before route matching). 403s any
   `/git-receive-pack` or `info/refs?service=git-receive-pack` —
   defense in depth so git-gate (PRD 0008) remains the only outbound
   path for writes, gitleaks-scanned by its pre-receive. Replicates
   cred-proxy's `is_git_push_request` behavior.

2. Restored agent-side role provisioner. Brings back `Role` on
   EgressProxyRoute (manifest + runtime) with three roles —
   `anthropic-base-url`, `npm-registry`, `tea-login`. Singleton
   constraint on the first two carries over from cred-proxy.
   `git-insteadof` is intentionally absent (option 1 above handles
   the push-bypass concern, and the canonical-URL rewrite has no
   function when egress-proxy is on HTTPS_PROXY).

   The provisioner (`backend/docker/provision/egress_proxy.py`):
     - `~/.npmrc` registry= the canonical upstream URL.
     - `~/.config/tea/config.yml` logins[] entry per tea-login route.
     - `ANTHROPIC_BASE_URL` env set in prepare.py based on the
       anthropic-base-url role (was a token_ref="CLAUDE_CODE_OAUTH_TOKEN"
       check in this PR's earlier draft — the role marker is cleaner
       and matches the cred-proxy precedent the user wants kept).

   All three dotfile values point at canonical upstream URLs; the
   agent's HTTPS_PROXY=egress-proxy routes them through the proxy
   automatically.

Tests: 11 new role-validation tests, 11 new provisioner-render tests,
the chunk-1 manifest fixture exercise role=anthropic-base-url. 400
tests pass (was 376).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-25 14:48:13 -04:00
parent 70f773ac61
commit fa06a3a0ab
12 changed files with 552 additions and 26 deletions
+18 -9
View File
@@ -219,20 +219,23 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
argv. argv.
Default orchestration: ca → prompt → skills → git → Default orchestration: ca → prompt → skills → git →
supervise. CA install runs first so the agent's trust store egress_proxy → supervise. CA install runs first so the
is rebuilt before anything inside the agent makes a TLS call. agent's trust store is rebuilt before anything inside the
Subclasses typically don't override this; they implement the agent makes a TLS call. egress_proxy runs after git because
sub-methods below. its provisioner may layer on top of `~/.gitconfig` entries
provision_git writes. Subclasses typically don't override
this; they implement the sub-methods below.
PRD 0017: cred-proxy's agent-side dotfile rewrites (~/.npmrc, PRD 0017: most agent-side rewrites (HTTPS routing) are
~/.gitconfig insteadOf, tea config) are gone. Egress-proxy is obsolete because egress-proxy sits on the HTTPS_PROXY path.
on the agent's HTTP_PROXY path so every tool that respects The remaining rewrites — npm registry, tea config,
HTTPS_PROXY (claude-code, git over HTTPS, npm, curl) is ANTHROPIC_BASE_URL — exist for tools that need an explicit
intercepted without per-tool reconfiguration.""" URL config rather than just respecting HTTPS_PROXY."""
self.provision_ca(plan, target) self.provision_ca(plan, target)
prompt_path = self.provision_prompt(plan, target) prompt_path = self.provision_prompt(plan, target)
self.provision_skills(plan, target) self.provision_skills(plan, target)
self.provision_git(plan, target) self.provision_git(plan, target)
self.provision_egress_proxy(plan, target)
self.provision_supervise(plan, target) self.provision_supervise(plan, target)
return prompt_path return prompt_path
@@ -262,6 +265,12 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
"""Copy the host's cwd `.git` directory into the running """Copy the host's cwd `.git` directory into the running
bottle if the user requested --cwd. No-op otherwise.""" bottle if the user requested --cwd. No-op otherwise."""
def provision_egress_proxy(self, plan: PlanT, target: str) -> None:
"""Drop the agent-side dotfiles driven by `egress_proxy.routes[].role`
per PRD 0017 (~/.npmrc, ~/.config/tea/config.yml). Default
impl is a no-op for backends that don't yet support the
egress-proxy sidecar; the Docker backend overrides."""
def provision_supervise(self, plan: PlanT, target: str) -> None: def provision_supervise(self, plan: PlanT, target: str) -> None:
"""Write the in-bottle Claude Code MCP config so the agent """Write the in-bottle Claude Code MCP config so the agent
discovers the per-bottle supervise sidecar (PRD 0013). discovers the per-bottle supervise sidecar (PRD 0013).
+4
View File
@@ -27,6 +27,7 @@ from .egress_proxy import DockerEgressProxy
from .git_gate import DockerGitGate from .git_gate import DockerGitGate
from .pipelock import DockerPipelockProxy from .pipelock import DockerPipelockProxy
from .provision import ca as _ca from .provision import ca as _ca
from .provision import egress_proxy as _egress_proxy_prov
from .provision import git as _git from .provision import git as _git
from .provision import prompt as _prompt from .provision import prompt as _prompt
from .provision import skills as _skills from .provision import skills as _skills
@@ -80,6 +81,9 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
def provision_git(self, plan: DockerBottlePlan, target: str) -> None: def provision_git(self, plan: DockerBottlePlan, target: str) -> None:
_git.provision_git(plan, target) _git.provision_git(plan, target)
def provision_egress_proxy(self, plan: DockerBottlePlan, target: str) -> None:
_egress_proxy_prov.provision_egress_proxy(plan, target)
def provision_supervise(self, plan: DockerBottlePlan, target: str) -> None: def provision_supervise(self, plan: DockerBottlePlan, target: str) -> None:
_supervise_prov.provision_supervise(plan, target) _supervise_prov.provision_supervise(plan, target)
+22 -12
View File
@@ -175,21 +175,31 @@ def resolve_plan(
# never lands on argv or in env_file) goes into one dict. Nothing # never lands on argv or in env_file) goes into one dict. Nothing
# mutates the host os.environ. # mutates the host os.environ.
forwarded_env: dict[str, str] = dict(resolved.forwarded) forwarded_env: dict[str, str] = dict(resolved.forwarded)
# When the bottle declares an egress-proxy route for the Anthropic # Find the (at most one) egress-proxy route claiming the
# OAuth flow, claude-code's outbound Authorization gets stripped + # anthropic-base-url role. Manifest validation enforces the
# re-injected by egress-proxy. The agent's environ still needs # singleton constraint. The role flips on claude-code's
# *something* claude-code recognises as a credential or it refuses # placeholder OAuth token + telemetry-off env vars and pins
# to start; ship a non-secret placeholder. The placeholder is not # ANTHROPIC_BASE_URL at the route's host. Egress-proxy then
# any real `auth.token_ref` value, so leaking it would tell an # strips inbound Authorization on every request and injects
# attacker only that egress-proxy is in front. # the real one from the route's `auth.token_ref` env var.
has_anthropic_auth = any( anthropic_route = next(
r.token_ref == "CLAUDE_CODE_OAUTH_TOKEN" (r for r in egress_proxy_plan.routes if "anthropic-base-url" in r.roles),
for r in egress_proxy_plan.routes None,
) )
if has_anthropic_auth: if anthropic_route is not None:
# Point claude-code at the canonical Anthropic URL. HTTPS_PROXY
# routes the request through egress-proxy, which injects the
# real OAuth header from the host env named by the route's
# auth.token_ref.
forwarded_env["ANTHROPIC_BASE_URL"] = f"https://{anthropic_route.host}"
# claude-code refuses to start without *some* credential in
# its env. The proxy strips inbound Authorization on every
# request and injects the real one — so a non-secret
# placeholder is sufficient. The agent cannot exfiltrate
# this string because it carries no meaning to upstream.
forwarded_env["CLAUDE_CODE_OAUTH_TOKEN"] = "egress-proxy-placeholder" forwarded_env["CLAUDE_CODE_OAUTH_TOKEN"] = "egress-proxy-placeholder"
# Belt-and-braces: turn off telemetry endpoints (statsig, # Belt-and-braces: turn off telemetry endpoints (statsig,
# error reporting) that egress-proxy can't gate by auth. # error reporting) that don't route through ANTHROPIC_BASE_URL.
forwarded_env.setdefault("CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC", "1") forwarded_env.setdefault("CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC", "1")
forwarded_env.setdefault("DISABLE_ERROR_REPORTING", "1") forwarded_env.setdefault("DISABLE_ERROR_REPORTING", "1")
_write_env_file(resolved, env_file) _write_env_file(resolved, env_file)
@@ -0,0 +1,142 @@
"""Egress-proxy agent-side provisioning (PRD 0017).
Writes the dotfiles / env-var nudges that point tools needing an
explicit URL config at the canonical upstream. The agent's
`HTTPS_PROXY=egress-proxy` already catches HTTPS traffic — these
provisioners exist for tools that:
- need an explicit base-URL setting beyond the proxy (claude-code
via `ANTHROPIC_BASE_URL`), or
- read a per-tool config to know which upstream to talk to (npm's
`~/.npmrc`, tea's `~/.config/tea/config.yml`).
The `ANTHROPIC_BASE_URL` env is set at `docker run -e` time by the
backend's launch step, not here — it has to be in the agent's environ
before claude starts, and there is no point in writing it to a
dotfile the agent would have to source. See `prepare.py` for that.
This module handles the rest: ~/.npmrc and ~/.config/tea/config.yml.
"""
from __future__ import annotations
import os
import subprocess
from pathlib import Path
from ....egress_proxy import EgressProxyRoute
from ....log import info
from .. import util as docker_mod
from ..bottle_plan import DockerBottlePlan
def provision_egress_proxy(plan: DockerBottlePlan, target: str) -> None:
"""Drop the agent-side dotfiles for each declared egress-proxy
route role. No-op when the bottle has no roles to provision."""
routes = plan.egress_proxy_plan.routes
if not routes:
return
_provision_npmrc(plan, target, routes)
_provision_tea_config(plan, target, routes)
# --- npm --------------------------------------------------------------------
def render_npmrc(routes: tuple[EgressProxyRoute, ...]) -> str:
"""Render `~/.npmrc` content. Driven by the `npm-registry` role:
finds the (single) route that claims it and writes a registry=
line pointing at the canonical upstream URL. Empty string when
no such route exists, so callers can branch on emptiness.
Egress-proxy is on the agent's HTTPS_PROXY, so the canonical
URL routes through the proxy automatically and gets DLP +
path_allowlist + auth-injection. The npmrc deliberately carries
no `_authToken` — the proxy strips inbound Authorization and
injects the upstream one from the bottle's egress-proxy auth
config. Manifest validation enforces that the role is a
singleton, so the first match is the only match."""
for r in routes:
if "npm-registry" in r.roles:
return f"registry=https://{r.host}/\n"
return ""
def _provision_npmrc(
plan: DockerBottlePlan,
target: str,
routes: tuple[EgressProxyRoute, ...],
) -> None:
content = render_npmrc(routes)
if not content:
return
container_home = os.environ.get("CLAUDE_BOTTLE_CONTAINER_HOME", "/home/node")
container_npmrc = f"{container_home}/.npmrc"
npmrc = plan.stage_dir / "agent_npmrc"
npmrc.write_text(content)
npmrc.chmod(0o600)
info(f"writing {container_npmrc} (egress-proxy npm registry)")
subprocess.run(
["docker", "cp", str(npmrc), f"{target}:{container_npmrc}"],
stdout=subprocess.DEVNULL,
check=True,
)
docker_mod.docker_exec_root(target, ["chown", "node:node", container_npmrc])
docker_mod.docker_exec_root(target, ["chmod", "644", container_npmrc])
# --- tea --------------------------------------------------------------------
def render_tea_config(routes: tuple[EgressProxyRoute, ...]) -> str:
"""Render `~/.config/tea/config.yml`. Driven by the `tea-login`
role: each route that claims it produces one `logins:` entry
pointing at the canonical Gitea URL. The egress-proxy injects
the real token at request time; the value in `token:` here is a
placeholder. `tea` refuses to make calls without a non-empty
token field, so the placeholder is necessary."""
tea_routes = [r for r in routes if "tea-login" in r.roles]
if not tea_routes:
return ""
lines = ["logins:"]
for r in tea_routes:
lines.extend([
f"- name: {r.host}",
f" url: https://{r.host}",
" token: egress-proxy-placeholder",
" default: false",
" ssh_host: \"\"",
" ssh_key: \"\"",
" insecure: false",
])
return "\n".join(lines) + "\n"
def _provision_tea_config(
plan: DockerBottlePlan,
target: str,
routes: tuple[EgressProxyRoute, ...],
) -> None:
content = render_tea_config(routes)
if not content:
return
container_home = os.environ.get("CLAUDE_BOTTLE_CONTAINER_HOME", "/home/node")
container_tea = f"{container_home}/.config/tea/config.yml"
cfg = plan.stage_dir / "agent_tea_config.yml"
cfg.write_text(content)
cfg.chmod(0o600)
info(
f"writing {container_tea} "
f"({len([r for r in routes if 'tea-login' in r.roles])} tea login(s))"
)
docker_mod.docker_exec_root(
target, ["mkdir", "-p", str(Path(container_tea).parent)]
)
subprocess.run(
["docker", "cp", str(cfg), f"{target}:{container_tea}"],
stdout=subprocess.DEVNULL,
check=True,
)
docker_mod.docker_exec_root(target, [
"chown", "-R", "node:node", str(Path(container_tea).parent),
])
docker_mod.docker_exec_root(target, ["chmod", "600", container_tea])
+9 -1
View File
@@ -62,13 +62,19 @@ class EgressProxyRoute:
(e.g. `EGRESS_PROXY_TOKEN_0`); `token_ref` is the host env var (e.g. `EGRESS_PROXY_TOKEN_0`); `token_ref` is the host env var
the CLI reads at launch and forwards into the container's environ the CLI reads at launch and forwards into the container's environ
under `token_env`. Routes that share a `token_ref` coalesce to under `token_env`. Routes that share a `token_ref` coalesce to
one `token_env` slot.""" one `token_env` slot.
`roles` are the provisioner tags from the manifest route (see
`manifest.EGRESS_PROXY_ROLES`). Each tag drives one agent-side
dotfile / env rewrite at bottle bring-up. The addon never reads
these — they flow through the plan to the host-side provisioner."""
host: str host: str
path_allowlist: tuple[str, ...] = () path_allowlist: tuple[str, ...] = ()
auth_scheme: str = "" auth_scheme: str = ""
token_env: str = "" token_env: str = ""
token_ref: str = "" token_ref: str = ""
roles: tuple[str, ...] = ()
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -148,11 +154,13 @@ def egress_proxy_routes_for_bottle(
auth_scheme=r.AuthScheme, auth_scheme=r.AuthScheme,
token_env=token_env, token_env=token_env,
token_ref=r.TokenRef, token_ref=r.TokenRef,
roles=r.Role,
)) ))
else: else:
out.append(EgressProxyRoute( out.append(EgressProxyRoute(
host=r.Host, host=r.Host,
path_allowlist=r.PathAllowlist, path_allowlist=r.PathAllowlist,
roles=r.Role,
)) ))
return tuple(out) return tuple(out)
+20 -2
View File
@@ -36,7 +36,7 @@ from mitmproxy import http # type: ignore[import-not-found]
# Absolute import (NOT `from .egress_proxy_addon_core`) — the # Absolute import (NOT `from .egress_proxy_addon_core`) — the
# container drops both files flat into /app/ so they are sibling # container drops both files flat into /app/ so they are sibling
# top-level modules to mitmdump's loader, not a package. # top-level modules to mitmdump's loader, not a package.
from egress_proxy_addon_core import Route, decide, load_routes # type: ignore[import-not-found] from egress_proxy_addon_core import Route, decide, is_git_push_request, load_routes # type: ignore[import-not-found]
DEFAULT_ROUTES_PATH = "/etc/egress-proxy/routes.yaml" DEFAULT_ROUTES_PATH = "/etc/egress-proxy/routes.yaml"
@@ -93,7 +93,25 @@ class EgressProxyAddon:
# below. # below.
flow.request.headers.pop("authorization", None) flow.request.headers.pop("authorization", None)
request_path = flow.request.path.split("?", 1)[0] request_path, _, query = flow.request.path.partition("?")
# Universal HTTPS git-push block. Defense-in-depth: git-gate
# (PRD 0008) is the only sanctioned outbound path for git
# writes — its pre-receive runs gitleaks. Letting HTTPS push
# through egress-proxy + auth injection would route around
# that scan, so we 403 before any route logic.
if is_git_push_request(request_path, query):
flow.response = http.Response.make(
403,
(
b"egress-proxy: git push over HTTPS is not supported; "
b"use the bottle.git SSH path (gitleaks-scanned by "
b"git-gate's pre-receive hook)."
),
{"Content-Type": "text/plain; charset=utf-8"},
)
return
decision = decide( decision = decide(
self.routes, self.routes,
flow.request.pretty_host, flow.request.pretty_host,
+31
View File
@@ -135,6 +135,36 @@ def load_routes(text: str) -> tuple[Route, ...]:
return parse_routes(payload) return parse_routes(payload)
def is_git_push_request(path: str, query: str) -> bool:
"""Return True if the request is a git smart-HTTP push.
git push over HTTPS hits two endpoints:
GET <repo>/info/refs?service=git-receive-pack (capabilities)
POST <repo>/git-receive-pack (the push)
Fetches use `service=git-upload-pack` / `/git-upload-pack` and
are unaffected. Egress-proxy refuses HTTPS push because git-gate's
pre-receive gitleaks scan is the gate for outbound git data;
routing push through egress-proxy would bypass that. Use the
bottle.git SSH path if you need to push.
Universal across routes — the block fires even when no
egress_proxy route matches the host. A bare-pass route (host with
no auth, no path_allowlist) would otherwise let push through to
pipelock + upstream untouched.
"""
if path.endswith("/git-receive-pack"):
return True
if path.endswith("/info/refs"):
# Query string is parsed leniently — `service=git-receive-pack`
# may appear with other params in any order.
for pair in query.split("&"):
k, _, v = pair.partition("=")
if k == "service" and v == "git-receive-pack":
return True
return False
def match_route( def match_route(
routes: typing.Sequence[Route], routes: typing.Sequence[Route],
request_host: str, request_host: str,
@@ -207,6 +237,7 @@ __all__ = [
"Decision", "Decision",
"Route", "Route",
"decide", "decide",
"is_git_push_request",
"load_routes", "load_routes",
"match_route", "match_route",
"parse_routes", "parse_routes",
+89 -2
View File
@@ -129,6 +129,44 @@ class GitEntry:
# token-not-Bearer quirk (go-gitea/gitea#16734). # token-not-Bearer quirk (go-gitea/gitea#16734).
EGRESS_PROXY_AUTH_SCHEMES = ("Bearer", "token") EGRESS_PROXY_AUTH_SCHEMES = ("Bearer", "token")
# Agent-side provisioner role tags a route may carry. Each tag drives
# one dotfile / env rewrite at bottle bring-up so tools that need an
# explicit URL config (rather than just respecting HTTPS_PROXY) point
# at the canonical upstream. Egress-proxy is on the agent's HTTP_PROXY
# path, so the canonical URL routes through the proxy automatically —
# the dotfile values are upstream URLs, not proxy URLs.
#
# anthropic-base-url: set ANTHROPIC_BASE_URL=https://<host> in the
# agent's environ (signals claude-code to use
# a non-default Anthropic endpoint; in practice
# the host is api.anthropic.com, so the value
# matches claude-code's default — the marker
# is what drives the placeholder-token +
# telemetry-off env vars).
# npm-registry: write ~/.npmrc `registry=https://<host>/`.
# tea-login: add an entry to ~/.config/tea/config.yml
# (url = https://<host>) so `tea` knows which
# Gitea host to talk to.
#
# Routes without a `role` are pure proxy entries: egress-proxy
# enforces path_allowlist + injects auth, but no agent-side dotfile
# is written. (`git-insteadof` is intentionally absent — egress-proxy
# already 403s HTTPS git push universally; PRD 0017's git story is
# `bottle.git` + git-gate for SSH push.)
EGRESS_PROXY_ROLES = frozenset({
"anthropic-base-url",
"npm-registry",
"tea-login",
})
# Roles whose semantics imply a single route can carry them. A second
# route claiming the same role would make the provisioner's choice
# ambiguous (which host goes into ANTHROPIC_BASE_URL?).
EGRESS_PROXY_SINGLETON_ROLES = frozenset({
"anthropic-base-url",
"npm-registry",
})
@dataclass(frozen=True) @dataclass(frozen=True)
class EgressProxyRoute: class EgressProxyRoute:
@@ -143,6 +181,10 @@ class EgressProxyRoute:
manifest's `auth` block is omitted both fields are empty strings — manifest's `auth` block is omitted both fields are empty strings —
no Authorization is written, no token forwarded. no Authorization is written, no token forwarded.
`Role` carries optional provisioner tags (see EGRESS_PROXY_ROLES).
Each tag drives one agent-side dotfile / env rewrite when the
sidecar comes up.
Validation rules (enforced in `from_dict`): Validation rules (enforced in `from_dict`):
- `host` required, non-empty. - `host` required, non-empty.
- `path_allowlist` optional, list of absolute path prefixes. - `path_allowlist` optional, list of absolute path prefixes.
@@ -150,12 +192,17 @@ class EgressProxyRoute:
`token_ref` as non-empty strings; an empty `auth: {}` is an `token_ref` as non-empty strings; an empty `auth: {}` is an
error rather than a synonym for "no auth" (omit `auth` for error rather than a synonym for "no auth" (omit `auth` for
that case). that case).
- `role` optional. String or list of strings drawn from
EGRESS_PROXY_ROLES. Singleton roles (see
EGRESS_PROXY_SINGLETON_ROLES) may appear on at most one
route per bottle.
""" """
Host: str Host: str
PathAllowlist: tuple[str, ...] = () PathAllowlist: tuple[str, ...] = ()
AuthScheme: str = "" AuthScheme: str = ""
TokenRef: str = "" TokenRef: str = ""
Role: tuple[str, ...] = ()
@classmethod @classmethod
def from_dict(cls, bottle_name: str, idx: int, raw: object) -> "EgressProxyRoute": def from_dict(cls, bottle_name: str, idx: int, raw: object) -> "EgressProxyRoute":
@@ -226,11 +273,37 @@ class EgressProxyRoute:
auth_scheme = auth_scheme_raw auth_scheme = auth_scheme_raw
token_ref = token_ref_raw token_ref = token_ref_raw
role_raw = d.get("role")
roles: tuple[str, ...] = ()
if role_raw is None:
roles = ()
elif isinstance(role_raw, str):
roles = (role_raw,)
elif isinstance(role_raw, list):
role_list = cast(list[object], role_raw)
collected_roles: list[str] = []
for r in role_list:
if not isinstance(r, str):
die(f"{label} role items must be strings (got {type(r).__name__})")
collected_roles.append(r)
roles = tuple(collected_roles)
else:
die(
f"{label} role must be a string or a list of strings "
f"(was {type(role_raw).__name__})"
)
for r in roles:
if r not in EGRESS_PROXY_ROLES:
die(
f"{label} role {r!r} is not one of "
f"{', '.join(sorted(EGRESS_PROXY_ROLES))}"
)
for k in d: for k in d:
if k not in ("host", "path_allowlist", "auth"): if k not in ("host", "path_allowlist", "auth", "role"):
die( die(
f"{label} has unknown key {k!r}; accepted keys are " f"{label} has unknown key {k!r}; accepted keys are "
f"'host', 'path_allowlist', 'auth'" f"'host', 'path_allowlist', 'auth', 'role'"
) )
return cls( return cls(
@@ -238,6 +311,7 @@ class EgressProxyRoute:
PathAllowlist=prefixes, PathAllowlist=prefixes,
AuthScheme=auth_scheme, AuthScheme=auth_scheme,
TokenRef=token_ref, TokenRef=token_ref,
Role=roles,
) )
@@ -715,6 +789,10 @@ def _validate_egress_proxy_routes(
- Hosts must be unique within the bottle. The proxy matches by - Hosts must be unique within the bottle. The proxy matches by
exact-host (v1, prefix matching is on path_allowlist only); exact-host (v1, prefix matching is on path_allowlist only);
duplicate hosts leave the route choice ambiguous. duplicate hosts leave the route choice ambiguous.
- Singleton roles (`anthropic-base-url`, `npm-registry`) may
appear on at most one route — each drives a single agent-side
dotfile/env entry, so two routes claiming the role would make
the choice ambiguous.
No cross-validation against `bottle.git` is performed. git-gate No cross-validation against `bottle.git` is performed. git-gate
(SSH push/fetch) and egress-proxy (HTTPS) broker different (SSH push/fetch) and egress-proxy (HTTPS) broker different
@@ -729,6 +807,15 @@ def _validate_egress_proxy_routes(
f"{r.Host!r}; each host must be unique on the proxy." f"{r.Host!r}; each host must be unique on the proxy."
) )
seen_hosts[key] = None seen_hosts[key] = None
for role in EGRESS_PROXY_SINGLETON_ROLES:
with_role = [r for r in routes if role in r.Role]
if len(with_role) > 1:
hosts = ", ".join(r.Host for r in with_role)
die(
f"bottle '{bottle_name}' egress_proxy.routes has {len(with_role)} "
f"routes with role {role!r} (hosts: {hosts}); this role drives a "
f"single agent-side rewrite — pick one."
)
def _validate_unique_git_names(bottle_name: str, git: tuple[GitEntry, ...]) -> None: def _validate_unique_git_names(bottle_name: str, git: tuple[GitEntry, ...]) -> None:
@@ -10,6 +10,7 @@ from claude_bottle.egress_proxy_addon_core import (
Decision, Decision,
Route, Route,
decide, decide,
is_git_push_request,
load_routes, load_routes,
match_route, match_route,
parse_routes, parse_routes,
@@ -245,5 +246,52 @@ class TestDecisionDefaults(unittest.TestCase):
self.assertIsNone(d.inject_authorization) self.assertIsNone(d.inject_authorization)
# --- is_git_push_request ------------------------------------------------
class TestIsGitPushRequest(unittest.TestCase):
def test_post_git_receive_pack_endpoint(self):
# The POST that carries the actual push payload.
self.assertTrue(is_git_push_request("/owner/repo.git/git-receive-pack", ""))
def test_info_refs_with_receive_pack_service(self):
# The capability advertisement GET that precedes a push.
self.assertTrue(is_git_push_request(
"/owner/repo.git/info/refs",
"service=git-receive-pack",
))
def test_info_refs_with_extra_query_params(self):
# service= may appear with other params in any order.
self.assertTrue(is_git_push_request(
"/owner/repo.git/info/refs",
"foo=bar&service=git-receive-pack&z=1",
))
self.assertTrue(is_git_push_request(
"/owner/repo.git/info/refs",
"service=git-receive-pack&foo=bar",
))
def test_fetch_endpoints_not_blocked(self):
# `service=git-upload-pack` is fetch; never blocked.
self.assertFalse(is_git_push_request(
"/owner/repo.git/info/refs",
"service=git-upload-pack",
))
self.assertFalse(is_git_push_request(
"/owner/repo.git/git-upload-pack", "",
))
def test_info_refs_without_service_not_blocked(self):
# Bare info/refs (no query) defaults to git-upload-pack on
# the server side; not push.
self.assertFalse(is_git_push_request("/x/info/refs", ""))
def test_unrelated_paths_not_blocked(self):
self.assertFalse(is_git_push_request("/repos/owner/repo", ""))
self.assertFalse(is_git_push_request("/v1/messages", ""))
self.assertFalse(is_git_push_request("/", ""))
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+58
View File
@@ -128,6 +128,64 @@ class TestAuth(unittest.TestCase):
}]) }])
class TestRole(unittest.TestCase):
def test_omitted_means_no_roles(self):
b = _bottle([{"host": "x.example"}])
self.assertEqual((), b.egress_proxy.routes[0].Role)
def test_string_normalizes_to_tuple(self):
b = _bottle([{"host": "api.anthropic.com",
"role": "anthropic-base-url",
"auth": {"scheme": "Bearer", "token_ref": "T"}}])
self.assertEqual(("anthropic-base-url",),
b.egress_proxy.routes[0].Role)
def test_list_supported(self):
b = _bottle([{"host": "registry.npmjs.org",
"role": ["npm-registry"]}])
self.assertEqual(("npm-registry",), b.egress_proxy.routes[0].Role)
def test_unknown_role_rejected(self):
with self.assertRaises(Die):
_bottle([{"host": "x.example", "role": "git-insteadof"}])
def test_non_string_role_rejected(self):
with self.assertRaises(Die):
_bottle([{"host": "x.example", "role": 42}])
def test_list_with_non_string_item_rejected(self):
with self.assertRaises(Die):
_bottle([{"host": "x.example", "role": ["npm-registry", 42]}])
def test_singleton_anthropic_base_url_enforced(self):
with self.assertRaises(Die):
_bottle([
{"host": "api.anthropic.com", "role": "anthropic-base-url",
"auth": {"scheme": "Bearer", "token_ref": "T1"}},
{"host": "api2.anthropic.example",
"role": "anthropic-base-url",
"auth": {"scheme": "Bearer", "token_ref": "T2"}},
])
def test_singleton_npm_registry_enforced(self):
with self.assertRaises(Die):
_bottle([
{"host": "registry.npmjs.org", "role": "npm-registry"},
{"host": "npm.example", "role": "npm-registry"},
])
def test_tea_login_is_not_singleton(self):
# Multiple Gitea instances on one bottle is a legitimate
# dev setup; tea-login lays one logins[] entry per route.
b = _bottle([
{"host": "gitea.example", "role": "tea-login",
"auth": {"scheme": "token", "token_ref": "T1"}},
{"host": "other-gitea.example", "role": "tea-login",
"auth": {"scheme": "token", "token_ref": "T2"}},
])
self.assertEqual(2, len(b.egress_proxy.routes))
class TestRouteValidation(unittest.TestCase): class TestRouteValidation(unittest.TestCase):
def test_duplicate_hosts_rejected(self): def test_duplicate_hosts_rejected(self):
# Routes match by exact host; duplicates leave the choice # Routes match by exact host; duplicates leave the choice
+2
View File
@@ -25,6 +25,7 @@ _BOTTLE_DEV = """
egress_proxy: egress_proxy:
routes: routes:
- host: api.anthropic.com - host: api.anthropic.com
role: anthropic-base-url
auth: auth:
scheme: Bearer scheme: Bearer
token_ref: CLAUDE_CODE_OAUTH_TOKEN token_ref: CLAUDE_CODE_OAUTH_TOKEN
@@ -92,6 +93,7 @@ class TestBottleFileParses(_ResolveCase):
self.assertEqual("api.anthropic.com", routes[0].Host) self.assertEqual("api.anthropic.com", routes[0].Host)
self.assertEqual("Bearer", routes[0].AuthScheme) self.assertEqual("Bearer", routes[0].AuthScheme)
self.assertEqual("CLAUDE_CODE_OAUTH_TOKEN", routes[0].TokenRef) self.assertEqual("CLAUDE_CODE_OAUTH_TOKEN", routes[0].TokenRef)
self.assertEqual(("anthropic-base-url",), routes[0].Role)
self.assertEqual(["example.com"], list(m.bottles["dev"].egress.allowlist)) self.assertEqual(["example.com"], list(m.bottles["dev"].egress.allowlist))
+109
View File
@@ -0,0 +1,109 @@
"""Unit: agent-side provisioning for egress-proxy roles (PRD 0017).
Each role drives one dotfile / env-var rewrite at bottle bring-up.
HTTPS_PROXY routes the canonical URL through egress-proxy, which
injects auth and DLP-scans on the upstream leg."""
import unittest
from claude_bottle.backend.docker.provision.egress_proxy import (
render_npmrc,
render_tea_config,
)
from claude_bottle.egress_proxy import egress_proxy_routes_for_bottle
from claude_bottle.manifest import Manifest
def _routes(manifest_routes):
m = Manifest.from_json_obj({
"bottles": {"dev": {"egress_proxy": {"routes": manifest_routes}}},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
})
return egress_proxy_routes_for_bottle(m.bottles["dev"])
# --- npmrc -----------------------------------------------------------
class TestRenderNpmrc(unittest.TestCase):
def test_canonical_upstream_url(self):
routes = _routes([
{"host": "registry.npmjs.org", "role": "npm-registry",
"auth": {"scheme": "Bearer", "token_ref": "NPM_TOKEN"}},
])
self.assertEqual(
"registry=https://registry.npmjs.org/\n",
render_npmrc(routes),
)
def test_empty_when_no_npm_role(self):
routes = _routes([
{"host": "api.github.com",
"auth": {"scheme": "Bearer", "token_ref": "GH"}},
])
self.assertEqual("", render_npmrc(routes))
def test_no_routes_empty(self):
self.assertEqual("", render_npmrc(()))
def test_no_auth_token_in_npmrc(self):
# The proxy injects auth; the npmrc must carry no secret —
# not even `:always-auth=true` lines that would prompt npm
# to wait for credentials. Just the registry URL.
routes = _routes([
{"host": "registry.npmjs.org", "role": "npm-registry",
"auth": {"scheme": "Bearer", "token_ref": "NPM_TOKEN"}},
])
out = render_npmrc(routes)
self.assertNotIn("_authToken", out)
self.assertNotIn("NPM_TOKEN", out)
# --- tea config ------------------------------------------------------
class TestRenderTeaConfig(unittest.TestCase):
def test_single_login(self):
routes = _routes([
{"host": "gitea.dideric.is", "role": "tea-login",
"auth": {"scheme": "token", "token_ref": "GITEA_TOKEN"}},
])
out = render_tea_config(routes)
self.assertIn("- name: gitea.dideric.is", out)
self.assertIn("url: https://gitea.dideric.is", out)
self.assertIn("token: egress-proxy-placeholder", out)
def test_multiple_logins_each_get_own_entry(self):
routes = _routes([
{"host": "gitea.a.example", "role": "tea-login",
"auth": {"scheme": "token", "token_ref": "T_A"}},
{"host": "gitea.b.example", "role": "tea-login",
"auth": {"scheme": "token", "token_ref": "T_B"}},
])
out = render_tea_config(routes)
self.assertIn("- name: gitea.a.example", out)
self.assertIn("- name: gitea.b.example", out)
def test_empty_when_no_tea_role(self):
routes = _routes([
{"host": "api.github.com",
"auth": {"scheme": "Bearer", "token_ref": "GH"}},
])
self.assertEqual("", render_tea_config(routes))
def test_no_routes_empty(self):
self.assertEqual("", render_tea_config(()))
def test_no_real_token_in_config(self):
routes = _routes([
{"host": "gitea.dideric.is", "role": "tea-login",
"auth": {"scheme": "token", "token_ref": "GITEA_TOKEN"}},
])
out = render_tea_config(routes)
# GITEA_TOKEN is just the env var name, not the value —
# placeholder-only is the SC.
self.assertIn("egress-proxy-placeholder", out)
if __name__ == "__main__":
unittest.main()