Compare commits

..

5 Commits

Author SHA1 Message Date
didericis-codex f89ae45f29 fix(codex): include account claims in dummy auth
test / unit (pull_request) Successful in 29s
test / integration (pull_request) Successful in 41s
2026-05-29 04:01:17 -04:00
didericis-codex 6c52a70078 fix(codex): provision dummy user auth state
test / unit (pull_request) Successful in 29s
test / integration (pull_request) Successful in 42s
2026-05-29 03:46:15 -04:00
didericis-codex 915ee3d144 fix(codex): forward host credentials to api route
test / unit (pull_request) Successful in 33s
test / integration (pull_request) Successful in 46s
2026-05-29 03:34:11 -04:00
didericis-codex 39afafc05a feat(codex): inject host credentials via egress
test / unit (pull_request) Successful in 35s
test / integration (pull_request) Successful in 55s
2026-05-29 03:21:43 -04:00
didericis-codex 80da66fd5d docs(prd): add Codex host credentials egress plan
test / unit (pull_request) Successful in 32s
test / integration (pull_request) Successful in 56s
2026-05-29 03:14:13 -04:00
23 changed files with 88 additions and 692 deletions
+1 -3
View File
@@ -31,7 +31,6 @@
# 9099 egress (mitmproxy, pipelock's upstream — not externally
# addressed by the agent)
# 9418 git-gate (git-daemon)
# 9420 git-gate smart HTTP (smolmachines agent-facing transport)
# 9100 supervise (MCP HTTP)
# Stage 1: pipelock binary. The upstream pipelock image is a
@@ -82,7 +81,6 @@ COPY bot_bottle/yaml_subset.py /app/yaml_subset.py
COPY bot_bottle/supervise.py /app/supervise.py
COPY bot_bottle/supervise_server.py /app/supervise_server.py
COPY bot_bottle/sidecar_init.py /app/sidecar_init.py
COPY bot_bottle/git_http_backend.py /app/git_http_backend.py
COPY bot_bottle/egress_entrypoint.sh /app/egress-entrypoint.sh
RUN chmod +x /app/egress-entrypoint.sh
@@ -99,7 +97,7 @@ RUN mkdir -p \
# Documentation only — the compose renderer publishes whichever
# subset the bottle uses.
EXPOSE 8888 9099 9418 9420 9100
EXPOSE 8888 9099 9418 9100
# WORKDIR matches Dockerfile.supervise's prior layout so the
# in-app same-dir import in supervise_server.py stays deterministic.
+5 -6
View File
@@ -373,12 +373,11 @@ launcher reads `tokens.access_token` from the host's
`~/.codex/auth.json`, verifies it is fresh user/device auth, and passes
it to the sidecar's `EGRESS_TOKEN_N` env slot. The agent container gets
a dummy `~/.codex/auth.json` that preserves the host auth-mode shape
but replaces credential values with placeholders. It keeps the selected
ChatGPT account id so Codex sends requests for the same account while
egress owns the real bearer token. The agent never receives real access
tokens, refresh tokens, or `OPENAI_API_KEY`. The effective egress table
automatically adds or upgrades `api.openai.com` and `chatgpt.com` to
authenticated routes when `forward_host_credentials` is true.
but replaces credential values with placeholders, so Codex chooses the
user/device auth path without receiving real access tokens, refresh
tokens, or `OPENAI_API_KEY`. The effective egress table automatically
adds or upgrades `api.openai.com` and `chatgpt.com` to authenticated
routes when `forward_host_credentials` is true.
The built-in Codex template uses `Dockerfile.codex`; set
`agent_provider.dockerfile` to build the agent from a custom Dockerfile
+24 -15
View File
@@ -45,11 +45,19 @@ _HOME_FOR = {
}
def _env_assignments_for(user: str, env: Mapping[str, str]) -> list[str]:
def _env_flags_for(user: str) -> list[str]:
home = _HOME_FOR.get(user, f"/home/{user}")
out = [f"HOME={home}", f"USER={user}"]
return ["-e", f"HOME={home}", "-e", f"USER={user}"]
def _guest_env_flags(env: Mapping[str, str]) -> list[str]:
"""Render `{K: V}` into a flat `-e K=V` argv slice for
`smolvm machine exec`. `smolvm machine create -e` set env
on PID 1 but it doesn't propagate to fresh exec process
trees, so we have to re-pass them every call."""
out: list[str] = []
for k, v in env.items():
out.append(f"{k}={v}")
out += ["-e", f"{k}={v}"]
return out
@@ -90,8 +98,9 @@ class SmolmachinesBottle(Bottle):
flags = ["smolvm", "machine", "exec", "--name", self.name]
if tty:
flags += ["-i", "-t"]
agent_tail = ["env", *_env_assignments_for("node", self._guest_env),
self.agent_command]
flags += _env_flags_for("node")
flags += _guest_env_flags(self._guest_env)
agent_tail = [self.agent_command]
provider_prompt_args = prompt_args(
self._agent_prompt_mode, self._prompt_path, argv=argv,
)
@@ -139,16 +148,16 @@ class SmolmachinesBottle(Bottle):
on both backends. Pass `user="root"` for tests that need
root.
`runuser -u <user> -- env ... /bin/sh -c <script>` switches UID
without invoking a login shell, then sets HOME / USER and the
bottle env in the child process."""
argv = [
"--", "runuser", "-u", user, "--",
"env", *_env_assignments_for(user, self._guest_env),
"/bin/sh", "-c", script,
]
# Call smolvm directly because this path needs the host-side
# subprocess capture shape used by the Docker backend.
`runuser -u <user> -- /bin/sh -c <script>` switches UID
without invoking a login shell; HOME / USER are set via
`smolvm -e` (see `_env_flags_for`)."""
argv = (
_env_flags_for(user)
+ _guest_env_flags(self._guest_env)
+ ["--", "runuser", "-u", user, "--", "/bin/sh", "-c", script]
)
# _smolvm.machine_exec expects argv (the bit after `--`);
# the -e flags go before, so call smolvm directly.
r = subprocess.run(
["smolvm", "machine", "exec", "--name", self.name] + argv,
capture_output=True, text=True, check=False,
@@ -68,7 +68,7 @@ class SmolmachinesBottlePlan(BottlePlan):
# empty when the agent has no prompt — claude-code reads it
# via --append-system-prompt-file only when non-empty.
prompt_file: Path
# Inner Plans for the sidecar bundle daemons. The same shape the
# Inner Plans for the four bundle daemons. The same shape the
# docker backend uses — same `.prepare()` calls produced
# them — but our launch step doesn't populate the
# docker-specific network fields (internal_network,
+11 -12
View File
@@ -50,6 +50,7 @@ from ..docker.git_gate import (
GIT_GATE_CREDS_DIR_IN_CONTAINER,
GIT_GATE_ENTRYPOINT_IN_CONTAINER,
GIT_GATE_HOOK_IN_CONTAINER,
GIT_GATE_PORT as _GIT_GATE_PORT,
)
from ..docker.pipelock import (
BUNDLE_LOCAL_PIPELOCK_URL,
@@ -81,7 +82,6 @@ _SMOLMACHINE_CACHE_DIR = Path.home() / ".cache" / "bot-bottle" / "smolmachines"
# them up post-start. Pipelock's port is an env-overridable string
# in docker.pipelock; coerce to int here.
_PIPELOCK_PORT = int(_PIPELOCK_PORT_STR)
_GIT_HTTP_PORT = 9420
_SUPERVISE_PORT = SUPERVISE_PORT
@@ -177,7 +177,7 @@ def launch(
agent_git_gate_host = ""
if plan.git_gate_plan.upstreams:
git_gate_host_port = _bundle.bundle_host_port(
plan.slug, _GIT_HTTP_PORT, host_ip=loopback_ip,
plan.slug, _GIT_GATE_PORT, host_ip=loopback_ip,
)
agent_git_gate_host = f"{loopback_ip}:{git_gate_host_port}"
agent_supervise_url = ""
@@ -195,11 +195,10 @@ def launch(
# otherwise claude's HTTPS_PROXY catches direct calls to
# the supervise URL (`http://<alias>:<port>/`) and proxies
# them through egress, which has no route for the alias
# and rejects with "Failed to connect". The smolmachines
# git-gate URL uses smart HTTP, so it also has to bypass
# the agent's HTTP_PROXY and go straight to the host-
# published git HTTP endpoint. Append rather than overwrite
# so prepare.py's
# and rejects with "Failed to connect". The git-gate URL
# uses git://, not affected by HTTP_PROXY, so the alias
# only has to be in NO_PROXY for the MCP / supervise
# path. Append rather than overwrite so prepare.py's
# `localhost,127.0.0.1` baseline stays in place.
existing_no_proxy = plan.guest_env.get("NO_PROXY", "localhost,127.0.0.1")
guest_env = {
@@ -209,7 +208,7 @@ def launch(
"NO_PROXY": f"{existing_no_proxy},{loopback_ip}",
}
if agent_git_gate_host:
guest_env["GIT_GATE_URL"] = f"http://{agent_git_gate_host}"
guest_env["GIT_GATE_URL"] = f"git://{agent_git_gate_host}"
if agent_supervise_url:
guest_env["MCP_SUPERVISE_URL"] = agent_supervise_url
plan = dataclasses.replace(
@@ -311,10 +310,10 @@ def _bundle_launch_spec(
Daemons in the CSV:
- egress + pipelock are always present (pipelock is the
agent's first hop; egress is its upstream).
- git-gate + git-http are conditional on plan.git_gate_plan.upstreams.
- git-gate is conditional on plan.git_gate_plan.upstreams.
- supervise is conditional on plan.supervise_plan.
Env + volumes are the union of the sidecar daemons' needs, with
Env + volumes are the union of the four daemons' needs, with
daemon-private values only (HTTPS_PROXY is scoped to the
egress process by egress_entrypoint.sh — see PRD 0024's bundle
bind-address PR)."""
@@ -359,7 +358,7 @@ def _bundle_launch_spec(
extra_hosts: list[str] = []
gp = plan.git_gate_plan
if gp.upstreams:
daemons += ["git-gate", "git-http"]
daemons.append("git-gate")
volumes += [
(str(gp.entrypoint_script), GIT_GATE_ENTRYPOINT_IN_CONTAINER, True),
(str(gp.hook_script), GIT_GATE_HOOK_IN_CONTAINER, True),
@@ -401,7 +400,7 @@ def _bundle_launch_spec(
else:
ports_to_publish = [_PIPELOCK_PORT]
if gp.upstreams:
ports_to_publish.append(_GIT_HTTP_PORT)
ports_to_publish.append(_GIT_GATE_PORT)
if sp is not None:
ports_to_publish.append(_SUPERVISE_PORT)
@@ -129,24 +129,6 @@ def resolve_plan(
if provider.template == "claude" and has_provider_auth:
guest_env.setdefault("CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC", "1")
guest_env.setdefault("DISABLE_ERROR_REPORTING", "1")
if provider.template == "codex":
# Codex is a Rust/rustls client: unlike the Node agents it does
# NOT consult the system trust store or honor NODE_EXTRA_CA_CERTS.
# It reads CODEX_CA_CERTIFICATE (falling back to SSL_CERT_FILE)
# for custom roots, across HTTPS *and* the wss responses channel.
# Point it at the bundle update-ca-certificates rebuilt with the
# egress MITM CA so Codex trusts the proxy and egress can inject
# the host bearer — without this, codex bottles need
# pipelock tls_passthrough, which disables auth injection.
guest_env["CODEX_CA_CERTIFICATE"] = (
"/etc/ssl/certs/ca-certificates.crt"
)
if provider.template == "codex" and provider.forward_host_credentials:
# Smolvm exec process trees do not reliably inherit the image
# user's login environment. Pin CODEX_HOME to the same path
# provision_provider_auth writes so Codex never falls back to a
# root or unset home and shows the sign-in picker.
guest_env["CODEX_HOME"] = "/home/node/.codex"
supervise_plan = None
if bottle.supervise:
@@ -18,7 +18,7 @@ Three concerns, all about git in the agent:
Differs from `backend.docker.provision.git` in one address detail:
the TSI-allowlisted guest can only reach the bundle's pinned IP
(no DNS resolver in the /32 allowlist), so the insteadOf URLs
are `http://<bundle_ip>:<port>/<name>.git` rather than the
are `git://<bundle_ip>:<port>/<name>.git` rather than the
docker backend's `git://git-gate/<name>.git`. The render itself
is the shared `git_gate_render_gitconfig` on the platform-neutral
git_gate module."""
@@ -82,14 +82,12 @@ def _provision_git_gate_config(plan: SmolmachinesBottlePlan, target: str) -> Non
if not bottle.git:
return
# `<loopback alias>:<host port>` form: the bundle's git-gate
# HTTP port is published on host loopback at launch time so
# the smolvm guest (which can only reach macOS networking via
# `127.0.0.1:<host port>` form: the bundle's git-gate port
# is published on host loopback at launch time so the
# smolvm guest (which can only reach macOS networking via
# TSI, not the docker bridge IP) can dial it. launch.py
# populates `plan.agent_git_gate_host` after bundle bringup.
content = git_gate_render_gitconfig(
bottle.git, plan.agent_git_gate_host, scheme="http",
)
content = git_gate_render_gitconfig(bottle.git, plan.agent_git_gate_host)
guest_gitconfig = f"{_guest_home()}/.gitconfig"
# Stage the file under the plan's stage_dir so `machine cp`
@@ -4,7 +4,6 @@ from __future__ import annotations
import os
from ....log import die
from .. import smolvm as _smolvm
from ..bottle_plan import SmolmachinesBottlePlan
@@ -29,21 +28,3 @@ def provision_provider_auth(plan: SmolmachinesBottlePlan, target: str) -> None:
_smolvm.machine_cp(str(plan.codex_auth_file), f"{target}:{auth_path}")
_smolvm.machine_exec(target, ["chown", "node:node", auth_path])
_smolvm.machine_exec(target, ["chmod", "600", auth_path])
result = _smolvm.machine_exec(
target,
[
"runuser", "-u", "node", "--",
"env",
f"HOME={guest_home}",
f"CODEX_HOME={auth_dir}",
"codex", "login", "status",
],
)
if result.returncode != 0:
detail = (result.stderr or result.stdout).strip()
if detail:
detail = f": {detail}"
die(
"codex host credentials: dummy auth was copied into the "
f"smolmachine, but Codex did not accept it{detail}"
)
+19 -45
View File
@@ -75,22 +75,11 @@ def codex_dummy_auth_json(
now: datetime | None = None,
) -> str:
"""Return a non-secret `auth.json` that keeps Codex in the host's
auth branch while egress owns the real bearer token.
The dummy access/id tokens carry the *host* token's real `exp` so
Codex's proactive refresh lifecycle (it refreshes when its local
access token is at/past expiry) tracks the real token instead of
firing after an artificial TTL. Codex cannot refresh inside the
bottle the refresh token is a placeholder and the OpenAI token
endpoint is off-route so a shorter dummy exp would drop Codex to
the sign-in screen the moment it lapsed, even while egress still
holds a valid bearer."""
auth branch while egress owns the real bearer token."""
path = codex_auth_path(host_env)
access = codex_host_access_token(host_env, now=now)
codex_host_access_token(host_env, now=now)
raw = _read_auth_object(path)
host_exp = _jwt_exp(access)
exp_ts = int(host_exp.timestamp()) if host_exp is not None else None
dummy = _redact_codex_auth(deepcopy(raw), now=now, exp_ts=exp_ts)
dummy = _redact_codex_auth(deepcopy(raw), now=now)
return json.dumps(dummy, indent=2, sort_keys=True) + "\n"
@@ -115,35 +104,29 @@ def _read_auth_object(path: Path) -> dict:
return raw
def _dummy_exp(now: datetime | None, exp_ts: int | None) -> int:
if exp_ts is not None:
return exp_ts
def _dummy_jwt(now: datetime | None = None) -> str:
check_now = now or datetime.now(timezone.utc)
return int(check_now.timestamp()) + 3600
exp = int(check_now.timestamp()) + 3600
def _dummy_jwt(now: datetime | None = None, *, exp_ts: int | None = None) -> str:
return _encode_dummy_jwt({
"exp": _dummy_exp(now, exp_ts),
"exp": exp,
"sub": "bot-bottle-placeholder",
})
def _dummy_jwt_from_host(
value: object, *, now: datetime | None = None, exp_ts: int | None = None,
) -> str:
def _dummy_jwt_from_host(value: object, *, now: datetime | None = None) -> str:
if not isinstance(value, str):
return _dummy_jwt(now, exp_ts=exp_ts)
return _dummy_jwt(now)
parts = value.split(".")
if len(parts) < 2:
return _dummy_jwt(now, exp_ts=exp_ts)
return _dummy_jwt(now)
try:
payload = json.loads(_b64url_decode(parts[1]))
except (ValueError, json.JSONDecodeError):
return _dummy_jwt(now, exp_ts=exp_ts)
return _dummy_jwt(now)
if not isinstance(payload, dict):
return _dummy_jwt(now, exp_ts=exp_ts)
return _encode_dummy_jwt(_redact_jwt_payload(payload, now=now, exp_ts=exp_ts))
return _dummy_jwt(now)
return _encode_dummy_jwt(_redact_jwt_payload(payload, now=now))
def _encode_dummy_jwt(payload: dict) -> str:
@@ -158,12 +141,12 @@ def _redact_jwt_payload(
payload: dict,
*,
now: datetime | None = None,
exp_ts: int | None = None,
) -> dict:
check_now = now or datetime.now(timezone.utc)
out = _redact_claims(payload)
if not isinstance(out, dict):
out = {}
out["exp"] = _dummy_exp(now, exp_ts)
out["exp"] = int(check_now.timestamp()) + 3600
out.setdefault("sub", "bot-bottle-placeholder")
return out
@@ -212,11 +195,6 @@ def _redact_auth_claim(value: object) -> dict:
lower = key.lower()
if lower == "chatgpt_plan_type" and isinstance(inner, str) and inner:
out[key] = inner
elif lower == "chatgpt_account_id" and isinstance(inner, str) and inner:
# Current Codex uses the selected account id when building
# ChatGPT requests. Keep that non-secret identifier aligned
# with the host while egress owns the real bearer token.
out[key] = inner
elif lower == "localhost" and isinstance(inner, bool):
out[key] = inner
elif isinstance(inner, bool):
@@ -234,9 +212,7 @@ def _redact_auth_claim(value: object) -> dict:
return out
def _redact_codex_auth(
value: object, *, now: datetime | None = None, exp_ts: int | None = None,
) -> object:
def _redact_codex_auth(value: object, *, now: datetime | None = None) -> object:
if isinstance(value, dict):
out: dict[str, object] = {}
for key, inner in value.items():
@@ -244,20 +220,18 @@ def _redact_codex_auth(
if lower == "openai_api_key":
out[key] = None
elif lower == "tokens":
out[key] = _redact_codex_auth(inner, now=now, exp_ts=exp_ts)
out[key] = _redact_codex_auth(inner, now=now)
elif lower in {"access_token", "id_token"}:
out[key] = _dummy_jwt_from_host(inner, now=now, exp_ts=exp_ts)
out[key] = _dummy_jwt_from_host(inner, now=now)
elif "token" in lower or "secret" in lower or lower.endswith("_key"):
out[key] = "bot-bottle-placeholder"
elif lower == "account_id" and isinstance(inner, str) and inner:
out[key] = inner
elif lower in {"account_id", "user_id", "email"}:
out[key] = "bot-bottle-placeholder"
else:
out[key] = _redact_codex_auth(inner, now=now, exp_ts=exp_ts)
out[key] = _redact_codex_auth(inner, now=now)
return out
if isinstance(value, list):
return [_redact_codex_auth(v, now=now, exp_ts=exp_ts) for v in value]
return [_redact_codex_auth(v, now=now) for v in value]
return value
+3 -10
View File
@@ -41,10 +41,6 @@ from .manifest import Bottle, GitEntry
# Short network alias for git-gate inside the sidecar bundle. The
# agent's `.gitconfig` insteadOf rewrites resolve through this name.
GIT_GATE_HOSTNAME = "git-gate"
# Bound half-open git client sessions. If an agent/tool runner is
# interrupted during push, git daemon should reap the receive-pack
# child instead of keeping the gate wedged indefinitely.
GIT_GATE_DAEMON_TIMEOUT_SECS = 15
def _empty_str_map() -> dict[str, str]:
@@ -146,13 +142,13 @@ def git_gate_aggregate_extra_hosts(
def git_gate_render_gitconfig(
entries: tuple[GitEntry, ...], gate_host: str, *, scheme: str = "git",
entries: tuple[GitEntry, ...], gate_host: str
) -> str:
"""Render the agent's ~/.gitconfig content for git-gate
`insteadOf` rewrites. Pure host-side, no docker / smolvm;
exposed for tests + reuse across backends.
`gate_host` is the part of the URL between `<scheme>://` and the
`gate_host` is the part of the URL between `git://` and the
repo path backends differ here:
- docker: `git-gate` (the short network alias)
- smolmachines: `<bundle_ip>:<port>` (no DNS in the
@@ -169,7 +165,7 @@ def git_gate_render_gitconfig(
"# fetch-from-upstream-before-every-upload-pack via access-hook).\n",
]
for entry in entries:
out.append(f'[url "{scheme}://{gate_host}/{entry.Name}.git"]\n')
out.append(f'[url "git://{gate_host}/{entry.Name}.git"]\n')
out.append(f"\tinsteadOf = {entry.Upstream}\n")
if entry.RemoteKey and entry.RemoteKey != entry.UpstreamHost:
port = (
@@ -237,7 +233,6 @@ def git_gate_render_entrypoint(upstreams: tuple[GitGateUpstream, ...]) -> str:
" git -C \"$repo\" config git-gate.identityFile \"$keyfile\"",
" git -C \"$repo\" config git-gate.knownHosts \"$hostsfile\"",
" git -C \"$repo\" config receive.denyCurrentBranch ignore",
" git -C \"$repo\" config http.receivepack true",
" install -m 755 /etc/git-gate/pre-receive \"$repo/hooks/pre-receive\"",
"}",
"",
@@ -252,8 +247,6 @@ def git_gate_render_entrypoint(upstreams: tuple[GitGateUpstream, ...]) -> str:
"",
"exec git daemon \\",
" --reuseaddr \\",
f" --timeout={GIT_GATE_DAEMON_TIMEOUT_SECS} \\",
f" --init-timeout={GIT_GATE_DAEMON_TIMEOUT_SECS} \\",
" --base-path=/git \\",
" --export-all \\",
" --enable=receive-pack \\",
-149
View File
@@ -1,149 +0,0 @@
"""Tiny smart-HTTP wrapper for git-gate repos.
Used by the smolmachines backend where `git://` push traffic over the
host-published Docker port can hang before receive-pack reaches hooks.
The wrapper serves the same `/git/*.git` bare repos through
`git http-backend`, so pre-receive and upstream forwarding remain the
git-gate enforcement point.
"""
from __future__ import annotations
import os
import subprocess
import sys
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlsplit
DEFAULT_PORT = 9420
class GitHttpHandler(BaseHTTPRequestHandler):
server_version = "bot-bottle-git-http/1"
def do_GET(self) -> None:
self._run_backend()
def do_POST(self) -> None:
self._run_backend()
def _run_backend(self) -> None:
parsed = urlsplit(self.path)
if self._is_upload_pack(parsed.path, parsed.query):
repo_dir = self._repo_dir(parsed.path)
if repo_dir is None:
self.send_error(404)
return
hook_path = os.environ.get(
"GIT_GATE_ACCESS_HOOK", "/etc/git-gate/access-hook",
)
hook = subprocess.run(
[hook_path, "upload-pack",
str(repo_dir), self.client_address[0], self.client_address[0]],
capture_output=True,
check=False,
)
if hook.returncode != 0:
self.send_response(403)
self.send_header("Content-Type", "text/plain; charset=utf-8")
self.end_headers()
self.wfile.write(hook.stderr or hook.stdout)
return
env = os.environ.copy()
env.update({
"GIT_PROJECT_ROOT": os.environ.get("GIT_PROJECT_ROOT", "/git"),
"GIT_HTTP_EXPORT_ALL": "1",
"REQUEST_METHOD": self.command,
"PATH_INFO": parsed.path,
"QUERY_STRING": parsed.query,
"CONTENT_TYPE": self.headers.get("content-type", ""),
"CONTENT_LENGTH": self.headers.get("content-length", "0"),
"REMOTE_ADDR": self.client_address[0],
"REMOTE_PORT": str(self.client_address[1]),
"REMOTE_USER": "",
"SERVER_NAME": self.server.server_name,
"SERVER_PORT": str(self.server.server_port),
"SERVER_PROTOCOL": self.request_version,
})
for header, variable in (
("accept", "HTTP_ACCEPT"),
("content-encoding", "HTTP_CONTENT_ENCODING"),
("git-protocol", "HTTP_GIT_PROTOCOL"),
("user-agent", "HTTP_USER_AGENT"),
):
value = self.headers.get(header)
if value:
env[variable] = value
length = int(self.headers.get("content-length", "0") or "0")
body = self.rfile.read(length) if length else b""
proc = subprocess.run(
["git", "http-backend"],
input=body,
env=env,
capture_output=True,
check=False,
)
self._write_cgi_response(proc.stdout)
def _repo_dir(self, path: str) -> Path | None:
root = Path(os.environ.get("GIT_PROJECT_ROOT", "/git")).resolve()
relative = path.lstrip("/").split(".git", 1)[0] + ".git"
candidate = (root / relative).resolve()
if root not in (candidate, *candidate.parents):
return None
if not candidate.is_dir():
return None
return candidate
@staticmethod
def _is_upload_pack(path: str, query: str) -> bool:
if path.endswith("/git-upload-pack"):
return True
if path.endswith("/info/refs"):
return any(
pair == "service=git-upload-pack"
for pair in query.split("&")
)
return False
def _write_cgi_response(self, raw: bytes) -> None:
head, sep, body = raw.partition(b"\r\n\r\n")
line_sep = b"\r\n"
if not sep:
head, sep, body = raw.partition(b"\n\n")
line_sep = b"\n"
status = 200
headers: list[tuple[str, str]] = []
for line in head.split(line_sep):
if not line:
continue
key, _, value = line.decode("latin1").partition(":")
value = value.strip()
if key.lower() == "status":
status = int(value.split()[0])
else:
headers.append((key, value))
self.send_response(status)
for key, value in headers:
self.send_header(key, value)
self.end_headers()
self.wfile.write(body)
def log_message(self, fmt: str, *args: object) -> None:
sys.stdout.write(fmt % args + "\n")
sys.stdout.flush()
def main() -> int:
port = int(os.environ.get("GIT_HTTP_PORT", str(DEFAULT_PORT)))
server = ThreadingHTTPServer(("0.0.0.0", port), GitHttpHandler)
sys.stdout.write(f"git-http listening on 0.0.0.0:{port}\n")
sys.stdout.flush()
server.serve_forever()
return 0
if __name__ == "__main__":
raise SystemExit(main())
+1 -18
View File
@@ -21,11 +21,7 @@ from dataclasses import dataclass
from pathlib import Path
from typing import cast
from .egress import (
CODEX_HOST_CREDENTIAL_HOSTS,
EGRESS_HOSTNAME,
egress_routes_for_bottle,
)
from .egress import EGRESS_HOSTNAME, egress_routes_for_bottle
from .supervise import SUPERVISE_HOSTNAME
from .manifest import Bottle
@@ -115,19 +111,6 @@ def pipelock_effective_tls_passthrough(bottle: Bottle) -> list[str]:
for route in bottle.egress.routes:
if route.Pipelock.TlsPassthrough:
seen.setdefault(route.Host, None)
# forward_host_credentials makes egress inject the host ChatGPT bearer
# on the Codex API hosts AFTER the agent boundary. Pipelock sits
# downstream of egress and DLP-scans request headers; left to MITM
# these routes it flags the injected JWT as a leaked secret
# ("request header contains secret") and blocks. Pass them through so
# pipelock still enforces the host allowlist on CONNECT but does not
# decrypt + rescan egress-owned auth. The auto-added routes live in
# egress_routes_for_bottle, not bottle.egress.routes, so add the
# hosts explicitly here.
provider = bottle.agent_provider
if provider.forward_host_credentials and provider.template == "codex":
for host in CODEX_HOST_CREDENTIAL_HOSTS:
seen.setdefault(host, None)
return sorted(seen.keys())
+1 -2
View File
@@ -20,7 +20,7 @@ sick daemon."
Daemon subset is env-driven. The compose renderer narrows it via
`BOT_BOTTLE_SIDECAR_DAEMONS=egress,pipelock` for bottles that
don't use git-gate or supervise. Default: all daemons.
don't use git-gate or supervise. Default: all four.
Stdlib-only by design adding supervisord/s6/runit for four
daemons is heavier than this script.
@@ -98,7 +98,6 @@ _DAEMONS: tuple[_DaemonSpec, ...] = (
"--listen", "0.0.0.0:8888"),
),
_DaemonSpec("git-gate", ("/bin/sh", "/git-gate-entrypoint.sh")),
_DaemonSpec("git-http", ("python3", "/app/git_http_backend.py")),
_DaemonSpec("supervise", ("python3", "/app/supervise_server.py")),
)
@@ -37,8 +37,8 @@ possible, not in the agent.
`CODEX_ACCESS_TOKEN`, or real `tokens.access_token` /
`tokens.refresh_token` values.
- The agent container receives only a dummy Codex `auth.json` that
preserves the host auth-mode shape, keeps the selected ChatGPT
account id, and replaces credential values with placeholders.
preserves the host auth-mode shape and replaces credential values
with placeholders.
- Egress route files remain non-secret: they contain only host/path/auth
slot metadata, never token values.
- Missing, API-key, malformed, or expired host Codex auth fails
+2 -27
View File
@@ -120,7 +120,7 @@ class TestCodexHostAccessToken(unittest.TestCase):
self.assertNotEqual(access, dummy["tokens"]["access_token"])
self.assertNotEqual(refresh, dummy["tokens"]["refresh_token"])
self.assertEqual("bot-bottle-placeholder", dummy["tokens"]["refresh_token"])
self.assertEqual("acct-host", dummy["tokens"]["account_id"])
self.assertEqual("bot-bottle-placeholder", dummy["tokens"]["account_id"])
self.assertIsNotNone(
codex_host_access_token(
{"CODEX_HOME": str(self.home)},
@@ -128,31 +128,6 @@ class TestCodexHostAccessToken(unittest.TestCase):
)
)
def test_dummy_auth_tokens_inherit_host_token_exp(self):
# Codex refreshes when its local access token is at/past exp;
# the dummy must carry the host token's real exp so Codex does
# not drop to the sign-in screen after an artificial TTL while
# egress still holds a valid bearer.
host_exp = 2000000000
self._write({
"auth_mode": "chatgpt",
"tokens": {
"access_token": _jwt(host_exp),
"id_token": _jwt(host_exp),
"refresh_token": "hidden",
},
})
dummy = json.loads(codex_dummy_auth_json(
{"CODEX_HOME": str(self.home)},
now=datetime(2026, 1, 1, tzinfo=timezone.utc),
))
self.assertEqual(
host_exp, _jwt_payload(dummy["tokens"]["access_token"])["exp"],
)
self.assertEqual(
host_exp, _jwt_payload(dummy["tokens"]["id_token"])["exp"],
)
def test_dummy_auth_keeps_required_account_claim_shape(self):
def jwt(payload: dict) -> str:
def enc(obj: dict) -> str:
@@ -197,7 +172,7 @@ class TestCodexHostAccessToken(unittest.TestCase):
auth = access_payload["https://api.openai.com/auth"]
profile = access_payload["https://api.openai.com/profile"]
self.assertEqual("plus", auth["chatgpt_plan_type"])
self.assertEqual("acct-real", auth["chatgpt_account_id"])
self.assertEqual("bot-bottle-placeholder", auth["chatgpt_account_id"])
self.assertEqual("bot-bottle-placeholder", auth["chatgpt_user_id"])
self.assertEqual("bot-bottle@example.invalid", profile["email"])
self.assertTrue(profile["email_verified"])
-90
View File
@@ -4,14 +4,7 @@ These tests target `egress_addon_core` — the host-importable
half of the addon. The mitmproxy hook wrapper in
`egress_addon.py` is container-only and is not exercised here."""
import http.server
import subprocess
import tempfile
import threading
import time
import unittest
from pathlib import Path
from urllib.parse import urlsplit
from bot_bottle.egress_addon_core import (
Decision,
@@ -333,88 +326,5 @@ class TestIsGitPushRequest(unittest.TestCase):
self.assertFalse(is_git_push_request("/", ""))
class TestGitPushBlockFailFast(unittest.TestCase):
def test_real_git_push_fails_fast_when_egress_blocks_receive_pack(self):
"""A real git client should see egress's HTTPS-push 403 and exit.
The local server stands in for the egress proxy response after
CONNECT/TLS interception; git smart-HTTP uses the same paths over
plain HTTP here, which keeps this regression test hermetic.
"""
seen_paths: list[str] = []
class Handler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
self._handle()
def do_POST(self):
self._handle()
def _handle(self):
parsed = urlsplit(self.path)
seen_paths.append(self.path)
if is_git_push_request(parsed.path, parsed.query):
body = (
b"egress: git push over HTTPS is not supported; "
b"use the bottle.git SSH path (gitleaks-scanned by "
b"git-gate's pre-receive hook)."
)
self.send_response(403)
self.send_header("Content-Type", "text/plain; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
return
self.send_response(404)
self.send_header("Content-Length", "0")
self.end_headers()
def log_message(self, _fmt, *_args):
pass
server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
self.addCleanup(server.shutdown)
self.addCleanup(server.server_close)
with tempfile.TemporaryDirectory() as tmp:
repo = Path(tmp) / "repo"
repo.mkdir()
subprocess.run(["git", "init"], cwd=repo, check=True,
capture_output=True, text=True)
subprocess.run(["git", "config", "user.name", "test"],
cwd=repo, check=True)
subprocess.run(["git", "config", "user.email", "test@example.invalid"],
cwd=repo, check=True)
(repo / "README.md").write_text("test\n")
subprocess.run(["git", "add", "README.md"], cwd=repo, check=True)
subprocess.run(["git", "commit", "-m", "test"],
cwd=repo, check=True, capture_output=True, text=True)
remote = f"http://127.0.0.1:{server.server_port}/owner/repo.git"
subprocess.run(["git", "remote", "add", "origin", remote],
cwd=repo, check=True)
started = time.monotonic()
result = subprocess.run(
["git", "push", "origin", "HEAD:refs/heads/main"],
cwd=repo,
capture_output=True,
text=True,
timeout=5,
check=False,
)
elapsed = time.monotonic() - started
self.assertNotEqual(0, result.returncode)
self.assertLess(elapsed, 5)
self.assertTrue(
any("service=git-receive-pack" in p for p in seen_paths),
f"git did not request receive-pack capabilities; saw {seen_paths!r}",
)
self.assertIn("403", result.stderr)
if __name__ == "__main__":
unittest.main()
-5
View File
@@ -170,12 +170,7 @@ class TestEntrypointRender(unittest.TestCase):
# Daemon line is what keeps PID 1 alive.
self.assertIn("exec git daemon", script)
self.assertIn("--enable=receive-pack", script)
self.assertIn("--timeout=15", script)
self.assertIn("--init-timeout=15", script)
self.assertIn("--base-path=/git", script)
# Smart HTTP receive-pack uses the same bare repos and hooks
# as git-daemon, so repos must opt in to HTTP pushes too.
self.assertIn("http.receivepack true", script)
# The access-hook is what makes fetch a mirror operation
# against the upstream (PRD 0008 v1.1).
self.assertIn("--access-hook=/etc/git-gate/access-hook", script)
-169
View File
@@ -1,169 +0,0 @@
"""Unit: smart-HTTP git-gate wrapper."""
import os
import subprocess
import tempfile
import threading
import unittest
import urllib.request
from pathlib import Path
from unittest import mock
from bot_bottle.git_http_backend import GitHttpHandler
class TestGitHttpBackend(unittest.TestCase):
def test_real_git_push_reaches_bare_repo(self):
from http.server import ThreadingHTTPServer
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
bare = root / "repo.git"
subprocess.run(["git", "init", "--bare", str(bare)],
check=True, capture_output=True, text=True)
subprocess.run(
["git", "-C", str(bare), "config", "http.receivepack", "true"],
check=True,
)
old_root = os.environ.get("GIT_PROJECT_ROOT")
os.environ["GIT_PROJECT_ROOT"] = str(root)
self.addCleanup(self._restore_env, old_root)
old_hook = os.environ.get("GIT_GATE_ACCESS_HOOK")
hook = root / "access-hook"
hook.write_text("#!/bin/sh\nexit 0\n")
hook.chmod(0o700)
os.environ["GIT_GATE_ACCESS_HOOK"] = str(hook)
self.addCleanup(self._restore_hook, old_hook)
server = ThreadingHTTPServer(("127.0.0.1", 0), GitHttpHandler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
self.addCleanup(server.shutdown)
self.addCleanup(server.server_close)
work = root / "work"
work.mkdir()
subprocess.run(["git", "init"], cwd=work, check=True,
capture_output=True, text=True)
subprocess.run(["git", "config", "user.name", "test"],
cwd=work, check=True)
subprocess.run(["git", "config", "user.email", "test@example.invalid"],
cwd=work, check=True)
(work / "README.md").write_text("test\n")
subprocess.run(["git", "add", "README.md"], cwd=work, check=True)
subprocess.run(["git", "commit", "-m", "init"], cwd=work,
check=True, capture_output=True, text=True)
url = f"http://127.0.0.1:{server.server_port}/repo.git"
subprocess.run(
["git", "push", url, "HEAD:refs/heads/main"],
cwd=work,
check=True,
capture_output=True,
text=True,
timeout=5,
)
pushed = subprocess.check_output(
["git", "-C", str(bare), "rev-parse", "refs/heads/main"],
text=True,
).strip()
head = subprocess.check_output(
["git", "-C", str(work), "rev-parse", "HEAD"],
text=True,
).strip()
self.assertEqual(head, pushed)
subprocess.run(
["git", "-C", str(bare), "symbolic-ref", "HEAD", "refs/heads/main"],
check=True,
)
clone = root / "clone"
subprocess.run(
["git", "clone", url, str(clone)],
check=True,
capture_output=True,
text=True,
timeout=5,
)
cloned = subprocess.check_output(
["git", "-C", str(clone), "rev-parse", "HEAD"],
text=True,
).strip()
self.assertEqual(head, cloned)
def test_post_forwards_git_cgi_headers(self):
from http.server import ThreadingHTTPServer
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
(root / "repo.git").mkdir()
old_root = os.environ.get("GIT_PROJECT_ROOT")
os.environ["GIT_PROJECT_ROOT"] = str(root)
self.addCleanup(self._restore_env, old_root)
server = ThreadingHTTPServer(("127.0.0.1", 0), GitHttpHandler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
self.addCleanup(server.shutdown)
self.addCleanup(server.server_close)
backend_response = (
b"Status: 200 OK\r\n"
b"Content-Type: application/x-git-upload-pack-result\r\n"
b"\r\n"
b"0000"
)
calls = [
subprocess.CompletedProcess(["hook"], 0, b"", b""),
subprocess.CompletedProcess(["git"], 0, backend_response, b""),
]
with mock.patch(
"bot_bottle.git_http_backend.subprocess.run",
side_effect=calls,
) as run:
request = urllib.request.Request(
f"http://127.0.0.1:{server.server_port}"
"/repo.git/git-upload-pack",
data=b"compressed",
headers={
"Accept": "application/x-git-upload-pack-result",
"Content-Encoding": "gzip",
"Content-Type": "application/x-git-upload-pack-request",
"Git-Protocol": "version=2",
"User-Agent": "git/test",
},
method="POST",
)
with urllib.request.urlopen(request, timeout=5) as response:
self.assertEqual(200, response.status)
self.assertEqual(b"0000", response.read())
env = run.call_args_list[1].kwargs["env"]
self.assertEqual("gzip", env["HTTP_CONTENT_ENCODING"])
self.assertEqual("version=2", env["HTTP_GIT_PROTOCOL"])
self.assertEqual(
"application/x-git-upload-pack-result",
env["HTTP_ACCEPT"],
)
self.assertEqual("git/test", env["HTTP_USER_AGENT"])
@staticmethod
def _restore_env(value: str | None) -> None:
if value is None:
os.environ.pop("GIT_PROJECT_ROOT", None)
else:
os.environ["GIT_PROJECT_ROOT"] = value
@staticmethod
def _restore_hook(value: str | None) -> None:
if value is None:
os.environ.pop("GIT_GATE_ACCESS_HOOK", None)
else:
os.environ["GIT_GATE_ACCESS_HOOK"] = value
if __name__ == "__main__":
unittest.main()
-19
View File
@@ -113,25 +113,6 @@ class TestTlsPassthrough(unittest.TestCase):
])))
self.assertEqual(["api.openai.com"], passthrough)
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,
},
}))
self.assertEqual(["api.openai.com", "chatgpt.com"], passthrough)
def test_no_codex_passthrough_without_forward_host_credentials(self):
passthrough = pipelock_effective_tls_passthrough(_bottle({
"agent_provider": {"template": "codex"},
}))
self.assertEqual([], passthrough)
class TestSsrfIpAllowlist(unittest.TestCase):
def test_default_empty(self):
-9
View File
@@ -60,15 +60,6 @@ class TestGitGateGitconfigRender(unittest.TestCase):
'[url "git://192.168.20.2:9418/bot-bottle.git"]', out,
)
def test_scheme_can_be_http_for_smolmachines(self):
bottle = fixture_with_git().bottles["dev"]
out = git_gate_render_gitconfig(
bottle.git, "127.0.0.16:57001", scheme="http",
)
self.assertIn(
'[url "http://127.0.0.16:57001/bot-bottle.git"]', out,
)
def test_ip_upstream_also_rewrites_logical_remote_key(self):
m = Manifest.from_json_obj({
"bottles": {"dev": {"git": {"remotes": {
+3 -3
View File
@@ -50,15 +50,15 @@ class TestEnvForDaemon(unittest.TestCase):
env = _env_for_daemon("pipelock", self._BASE)
self.assertNotIn("EGRESS_TOKEN_0", env)
self.assertNotIn("EGRESS_TOKEN_1", env)
# Non-token bundle env stays — supervise / git-gate / git-http / the
# Non-token bundle env stays — supervise / git-gate / the
# upstream proxy URL are all load-bearing for other
# daemons.
self.assertEqual("/usr/bin", env["PATH"])
self.assertEqual("http://127.0.0.1:8888", env["EGRESS_UPSTREAM_PROXY"])
self.assertEqual("9100", env["SUPERVISE_PORT"])
def test_git_daemons_and_supervise_also_lose_egress_tokens(self):
for name in ("git-gate", "git-http", "supervise"):
def test_git_gate_and_supervise_also_lose_egress_tokens(self):
for name in ("git-gate", "supervise"):
env = _env_for_daemon(name, self._BASE)
self.assertNotIn("EGRESS_TOKEN_0", env)
self.assertNotIn("EGRESS_TOKEN_1", env)
+5 -4
View File
@@ -59,9 +59,10 @@ class TestClaudeArgvWrapped(unittest.TestCase):
"smolvm", "machine", "exec", "--name",
"bot-bottle-dev-abc",
"-i", "-t",
"-e", "HOME=/home/node",
"-e", "USER=node",
"--",
"runuser", "-u", "node", "--",
"env", "HOME=/home/node", "USER=node",
"claude",
],
argv,
@@ -106,7 +107,7 @@ class TestClaudeArgvWrapped(unittest.TestCase):
HTTPS_PROXY="http://127.0.0.1:1234",
NO_PROXY="localhost",
).agent_argv([]))
self.assertIn("env", argv)
self.assertIn("-e", argv)
self.assertIn("HTTPS_PROXY=http://127.0.0.1:1234", argv)
self.assertIn("NO_PROXY=localhost", argv)
@@ -118,8 +119,8 @@ class TestClaudeArgvWrapped(unittest.TestCase):
argv = _bottle().agent_argv([])
agent_idx = argv.index("claude")
self.assertEqual(
["runuser", "-u", "node", "--", "env"],
argv[agent_idx - 7:agent_idx - 2],
["runuser", "-u", "node", "--"],
argv[agent_idx - 4:agent_idx],
)
+5 -59
View File
@@ -9,7 +9,6 @@ from __future__ import annotations
import subprocess
import tempfile
import unittest
from dataclasses import replace
from pathlib import Path
from unittest.mock import patch
@@ -25,10 +24,9 @@ from bot_bottle.backend.smolmachines.provision import (
skills as _skills,
supervise as _supervise,
)
from bot_bottle.backend.smolmachines.launch import _bundle_launch_spec
from bot_bottle.backend.smolmachines.smolvm import SmolvmRunResult
from bot_bottle.egress import EgressPlan, EgressRoute
from bot_bottle.git_gate import GitGatePlan, GitGateUpstream
from bot_bottle.git_gate import GitGatePlan
from bot_bottle.manifest import GitEntry, Manifest
from bot_bottle.pipelock import PipelockProxyPlan
from bot_bottle.supervise import SupervisePlan
@@ -211,7 +209,6 @@ class TestProvisionProviderAuth(unittest.TestCase):
) as cp, patch(
"bot_bottle.backend.smolmachines.provision.provider_auth._smolvm.machine_exec"
) as ex:
ex.return_value = SmolvmRunResult(0, "Logged in using ChatGPT\n", "")
_provider_auth.provision_provider_auth(
_plan(codex_auth_file=Path("/tmp/codex-auth.json")),
"bot-bottle-demo-abc12",
@@ -227,29 +224,6 @@ class TestProvisionProviderAuth(unittest.TestCase):
argv_seen,
)
self.assertIn(["chmod", "600", "/home/node/.codex/auth.json"], argv_seen)
self.assertIn(
[
"runuser", "-u", "node", "--",
"env",
"HOME=/home/node",
"CODEX_HOME=/home/node/.codex",
"codex", "login", "status",
],
argv_seen,
)
def test_dies_when_codex_rejects_dummy_auth(self):
with patch(
"bot_bottle.backend.smolmachines.provision.provider_auth._smolvm.machine_cp"
), patch(
"bot_bottle.backend.smolmachines.provision.provider_auth._smolvm.machine_exec"
) as ex:
ex.return_value = SmolvmRunResult(1, "Not logged in\n", "")
with self.assertRaises(SystemExit):
_provider_auth.provision_provider_auth(
_plan(codex_auth_file=Path("/tmp/codex-auth.json")),
"bot-bottle-demo-abc12",
)
class TestProvisionSkills(unittest.TestCase):
@@ -512,9 +486,9 @@ class TestProvisionGit(unittest.TestCase):
def test_writes_gitconfig_with_ip_port_form_for_smolmachines(self):
# Smolmachines's TSI-allowlisted guest dials git-gate via
# smart HTTP at `127.0.0.1:<host port>` — the bundle's
# git HTTP port is published on host loopback at launch
# time, and the plan carries the discovered host port.
# `127.0.0.1:<host port>` — the bundle's git-gate port is
# published on host loopback at launch time, and the plan
# carries the discovered host port (here mocked to 9418).
plan = _plan(
git=[GitEntry(
Name="bot-bottle",
@@ -537,41 +511,13 @@ class TestProvisionGit(unittest.TestCase):
self.assertEqual(self.stage, staged_path.parent)
content = staged_path.read_text()
self.assertIn(
'[url "http://127.0.0.1:9418/bot-bottle.git"]', content,
'[url "git://127.0.0.1:9418/bot-bottle.git"]', content,
)
self.assertIn(
"\tinsteadOf = ssh://git@host/repo.git", content,
)
class TestBundleLaunchSpec(unittest.TestCase):
def test_git_gate_uses_http_daemon_for_smolmachines(self):
plan = _plan()
plan = replace(
plan,
git_gate_plan=replace(
plan.git_gate_plan,
upstreams=(GitGateUpstream(
name="bot-bottle",
upstream_url="ssh://git@host/repo.git",
upstream_host="host",
upstream_port="22",
identity_file="/tmp/key",
known_host_key="",
),),
),
)
spec = _bundle_launch_spec(plan, "net", "127.0.0.16")
self.assertEqual(
"egress,pipelock,git-gate,git-http",
spec.daemons_csv,
)
self.assertIn(9420, spec.ports_to_publish)
self.assertNotIn(9418, spec.ports_to_publish)
class TestProvisionGitUser(unittest.TestCase):
"""`_provision_git_user` runs `git config --global` inside the
guest as the node user with HOME forced via `smolvm -e`