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
21 changed files with 1119 additions and 12 deletions
+18 -2
View File
@@ -352,10 +352,15 @@ auth through egress and gitea.dideric.is over SSH.
For a Codex-backed base bottle, set `agent_provider.template: codex`.
The Codex template expects ChatGPT/device login state instead of an
`OPENAI_API_KEY` env var; no API-key placeholder is forwarded into the
agent. To let headless device-code login request a user code, add an
unauthenticated egress route for the device-auth endpoint:
agent. To let bot-bottle read the host's current Codex ChatGPT access
token and inject it from egress only for Codex's API calls, opt in
explicitly:
```yaml
agent_provider:
template: codex
forward_host_credentials: true
egress:
routes:
- host: auth.openai.com
@@ -363,6 +368,17 @@ egress:
- /api/accounts/deviceauth/
```
Run `codex login --device-auth` on the host before launch. The
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, 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
while keeping the bot-bottle sidecars in place.
+6
View File
@@ -286,6 +286,7 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
intercepted without per-tool reconfiguration."""
self.provision_ca(plan, target)
prompt_path = self.provision_prompt(plan, target)
self.provision_provider_auth(plan, target)
self.provision_skills(plan, target)
self.provision_git(plan, target)
self.provision_supervise(plan, target)
@@ -300,6 +301,11 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
backend overrides to docker-cp the cert in and run
`update-ca-certificates`."""
def provision_provider_auth(self, plan: PlanT, target: str) -> None:
"""Install non-secret provider auth marker files into the agent
home when a provider needs them to select the right auth mode.
The default is no-op."""
@abstractmethod
def provision_prompt(self, plan: PlanT, target: str) -> str | None:
"""Copy the prompt file into the running bottle. Returns the
+4
View File
@@ -29,6 +29,7 @@ from .bottle_plan import DockerBottlePlan
from .provision import ca as _ca
from .provision import git as _git
from .provision import prompt as _prompt
from .provision import provider_auth as _provider_auth
from .provision import skills as _skills
from .provision import supervise as _supervise_prov
@@ -62,6 +63,9 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
def provision_prompt(self, plan: DockerBottlePlan, target: str) -> str | None:
return _prompt.provision_prompt(plan, target)
def provision_provider_auth(self, plan: DockerBottlePlan, target: str) -> None:
_provider_auth.provision_provider_auth(plan, target)
def provision_skills(self, plan: DockerBottlePlan, target: str) -> None:
_skills.provision_skills(plan, target)
+1
View File
@@ -55,6 +55,7 @@ class DockerBottlePlan(BottlePlan):
agent_command: str = "claude"
agent_prompt_mode: PromptMode = "append_file"
agent_provider_template: str = "claude"
codex_auth_file: Path | None = None
def print(self, *, remote_control: bool) -> None:
"""Render the y/N preflight summary to stderr — compact form
+12 -1
View File
@@ -42,7 +42,11 @@ from contextlib import ExitStack, contextmanager
from pathlib import Path
from typing import Callable, Generator
from ...egress import egress_resolve_token_values
from ...codex_auth import codex_host_access_token
from ...egress import (
CODEX_HOST_CREDENTIAL_TOKEN_REF,
egress_resolve_token_values,
)
from ...log import info
from . import network as network_mod
from . import util as docker_mod
@@ -181,6 +185,13 @@ def launch(
token_values = egress_resolve_token_values(
plan.egress_plan.token_env_map, dict(os.environ),
)
if plan.spec.manifest.bottle_for(
plan.spec.agent_name,
).agent_provider.forward_host_credentials:
access_token = codex_host_access_token(dict(os.environ))
for token_env, token_ref in plan.egress_plan.token_env_map.items():
if token_ref == CODEX_HOST_CREDENTIAL_TOKEN_REF:
token_values[token_env] = access_token
compose_env: dict[str, str] = {
**os.environ,
**plan.forwarded_env,
+5
View File
@@ -15,6 +15,7 @@ from datetime import datetime, timezone
from pathlib import Path
from ...agent_provider import runtime_for
from ...codex_auth import write_codex_dummy_auth_file
from ...egress import Egress
from ...env import ResolvedEnv, resolve_env
from ...git_gate import GitGate
@@ -155,6 +156,7 @@ def resolve_plan(
agent_dir.mkdir(parents=True, exist_ok=True)
env_file = agent_dir / "agent.env"
prompt_file = agent_dir / "prompt.txt"
codex_auth_file = agent_dir / "codex-auth.json"
prompt_file.write_text("")
prompt_file.chmod(0o600)
@@ -219,6 +221,8 @@ def resolve_plan(
# error reporting) that egress can't gate by auth.
forwarded_env.setdefault("CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC", "1")
forwarded_env.setdefault("DISABLE_ERROR_REPORTING", "1")
if provider.forward_host_credentials:
write_codex_dummy_auth_file(codex_auth_file, dict(os.environ))
_write_env_file(resolved, env_file)
prompt_file.write_text(agent.prompt)
@@ -245,6 +249,7 @@ def resolve_plan(
agent_command=provider_runtime.command,
agent_prompt_mode=provider_runtime.prompt_mode,
agent_provider_template=provider.template,
codex_auth_file=codex_auth_file if provider.forward_host_credentials else None,
)
@@ -0,0 +1,43 @@
"""Provision non-secret provider auth markers into a Docker bottle."""
from __future__ import annotations
import os
import subprocess
from ..bottle_plan import DockerBottlePlan
def provision_provider_auth(plan: DockerBottlePlan, target: str) -> None:
"""Copy a dummy Codex auth marker when host credentials are
forwarded through egress.
The file contains no real access or refresh token values; it only
nudges Codex into the same user/device auth branch as the host.
"""
if not plan.codex_auth_file:
return
container_home = os.environ.get("BOT_BOTTLE_CONTAINER_HOME", "/home/node")
auth_dir = f"{container_home}/.codex"
auth_path = f"{auth_dir}/auth.json"
subprocess.run(
["docker", "exec", "-u", "0", target, "mkdir", "-p", auth_dir],
stdout=subprocess.DEVNULL,
check=True,
)
subprocess.run(
["docker", "cp", str(plan.codex_auth_file), f"{target}:{auth_path}"],
stdout=subprocess.DEVNULL,
check=True,
)
subprocess.run(
["docker", "exec", "-u", "0", target, "chown", "node:node", auth_path],
stdout=subprocess.DEVNULL,
check=True,
)
subprocess.run(
["docker", "exec", "-u", "0", target, "chmod", "600", auth_path],
stdout=subprocess.DEVNULL,
check=True,
)
@@ -19,6 +19,7 @@ from .bottle_plan import SmolmachinesBottlePlan
from .provision import ca as _ca
from .provision import git as _git
from .provision import prompt as _prompt
from .provision import provider_auth as _provider_auth
from .provision import skills as _skills
from .provision import supervise as _supervise
@@ -61,6 +62,11 @@ class SmolmachinesBottleBackend(
) -> str | None:
return _prompt.provision_prompt(plan, target)
def provision_provider_auth(
self, plan: SmolmachinesBottlePlan, target: str
) -> None:
_provider_auth.provision_provider_auth(plan, target)
def provision_skills(
self, plan: SmolmachinesBottlePlan, target: str
) -> None:
@@ -97,6 +97,7 @@ class SmolmachinesBottlePlan(BottlePlan):
agent_prompt_mode: PromptMode = "append_file"
agent_provider_template: str = "claude"
agent_dockerfile_path: str = ""
codex_auth_file: Path | None = None
def print(self, *, remote_control: bool) -> None:
"""Compact y/N preflight. Same shape as the Docker
+16 -2
View File
@@ -26,7 +26,12 @@ from contextlib import ExitStack, contextmanager
from pathlib import Path
from typing import Callable, Generator
from ...egress import EGRESS_ROUTES_IN_CONTAINER, egress_resolve_token_values
from ...codex_auth import codex_host_access_token
from ...egress import (
CODEX_HOST_CREDENTIAL_TOKEN_REF,
EGRESS_ROUTES_IN_CONTAINER,
egress_resolve_token_values,
)
from ...pipelock import (
PIPELOCK_CA_CERT_IN_CONTAINER,
PIPELOCK_CA_KEY_IN_CONTAINER,
@@ -422,7 +427,16 @@ def _resolve_token_env(
ep = plan.egress_plan
if not ep.routes:
return {}
return egress_resolve_token_values(ep.token_env_map, dict(host_env))
env = dict(host_env)
token_values = egress_resolve_token_values(ep.token_env_map, env)
if plan.spec.manifest.bottle_for(
plan.spec.agent_name,
).agent_provider.forward_host_credentials:
access_token = codex_host_access_token(env)
for token_env, token_ref in ep.token_env_map.items():
if token_ref == CODEX_HOST_CREDENTIAL_TOKEN_REF:
token_values[token_env] = access_token
return token_values
def _ensure_smolmachine(image_ref: str, *, dockerfile: str = "") -> Path:
@@ -16,6 +16,7 @@ from pathlib import Path
from ...agent_provider import runtime_for
from ...backend import BottleSpec
from ...codex_auth import write_codex_dummy_auth_file
from ...backend.docker.bottle_state import (
BottleMetadata,
agent_state_dir,
@@ -144,9 +145,12 @@ def resolve_plan(
agent_dir = agent_state_dir(slug)
agent_dir.mkdir(parents=True, exist_ok=True)
prompt_file = agent_dir / "prompt.txt"
codex_auth_file = agent_dir / "codex-auth.json"
agent = manifest.agents[spec.agent_name]
prompt_file.write_text(agent.prompt or "")
prompt_file.chmod(0o600)
if provider.forward_host_credentials:
write_codex_dummy_auth_file(codex_auth_file, dict(os.environ))
machine_name = f"bot-bottle-{slug}"
# Stash the agent image ref — `launch.launch` runs the
@@ -182,6 +186,7 @@ def resolve_plan(
agent_prompt_mode=provider_runtime.prompt_mode,
agent_provider_template=provider.template,
agent_dockerfile_path=agent_dockerfile_path,
codex_auth_file=codex_auth_file if provider.forward_host_credentials else None,
)
@@ -0,0 +1,30 @@
"""Provision non-secret provider auth markers into a smolmachines bottle."""
from __future__ import annotations
import os
from .. import smolvm as _smolvm
from ..bottle_plan import SmolmachinesBottlePlan
_DEFAULT_GUEST_HOME = "/home/node"
def provision_provider_auth(plan: SmolmachinesBottlePlan, target: str) -> None:
"""Copy a dummy Codex auth marker when host credentials are
forwarded through egress.
The real host access token remains in the egress bundle env; this
file only selects Codex's user/device auth code path.
"""
if not plan.codex_auth_file:
return
guest_home = os.environ.get("BOT_BOTTLE_GUEST_HOME", _DEFAULT_GUEST_HOME)
auth_dir = f"{guest_home}/.codex"
auth_path = f"{auth_dir}/auth.json"
_smolvm.machine_exec(target, ["mkdir", "-p", auth_dir])
_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])
+264
View File
@@ -0,0 +1,264 @@
"""Host Codex auth helpers.
Reads the host's Codex ChatGPT/device-login auth state and returns only
the short-lived access token needed by egress. This module deliberately
does not expose refresh tokens or raw auth payloads.
"""
from __future__ import annotations
import base64
import json
import os
from copy import deepcopy
from datetime import datetime, timezone
from pathlib import Path
from .log import die
from .util import expand_tilde
def codex_auth_path(host_env: dict[str, str] | None = None) -> Path:
env = os.environ if host_env is None else host_env
home = env.get("CODEX_HOME")
if home:
return Path(expand_tilde(home)) / "auth.json"
return Path.home() / ".codex" / "auth.json"
def codex_host_access_token(
host_env: dict[str, str] | None = None,
*,
now: datetime | None = None,
) -> str:
path = codex_auth_path(host_env)
if not path.is_file():
die(
f"codex host credentials: auth file missing at {path}. "
"Run `codex login --device-auth` on the host or disable "
"agent_provider.forward_host_credentials."
)
raw = _read_auth_object(path)
auth_mode = raw.get("auth_mode")
if not isinstance(auth_mode, str) or auth_mode == "api_key":
die(
"codex host credentials: host Codex auth is not user/device "
"auth. Run `codex login --device-auth` on the host."
)
tokens = raw.get("tokens")
if not isinstance(tokens, dict):
die(f"codex host credentials: {path} is missing tokens")
access = tokens.get("access_token")
if not isinstance(access, str) or not access:
die(
f"codex host credentials: {path} is missing tokens.access_token. "
"Run `codex login --device-auth` on the host."
)
exp = _jwt_exp(access)
if exp is None:
die("codex host credentials: tokens.access_token is not a JWT with exp")
check_now = now or datetime.now(timezone.utc)
if exp <= check_now:
die(
"codex host credentials: host Codex access token is expired. "
"Run `codex login --device-auth` on the host and restart the bottle."
)
return access
def codex_dummy_auth_json(
host_env: dict[str, str] | None = None,
*,
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."""
path = codex_auth_path(host_env)
codex_host_access_token(host_env, now=now)
raw = _read_auth_object(path)
dummy = _redact_codex_auth(deepcopy(raw), now=now)
return json.dumps(dummy, indent=2, sort_keys=True) + "\n"
def write_codex_dummy_auth_file(
path: Path,
host_env: dict[str, str] | None = None,
*,
now: datetime | None = None,
) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(codex_dummy_auth_json(host_env, now=now))
path.chmod(0o600)
def _read_auth_object(path: Path) -> dict:
try:
raw = json.loads(path.read_text())
except (OSError, json.JSONDecodeError) as e:
die(f"codex host credentials: could not read valid JSON at {path}: {e}")
if not isinstance(raw, dict):
die(f"codex host credentials: {path} must contain a JSON object")
return raw
def _dummy_jwt(now: datetime | None = None) -> str:
check_now = now or datetime.now(timezone.utc)
exp = int(check_now.timestamp()) + 3600
return _encode_dummy_jwt({
"exp": exp,
"sub": "bot-bottle-placeholder",
})
def _dummy_jwt_from_host(value: object, *, now: datetime | None = None) -> str:
if not isinstance(value, str):
return _dummy_jwt(now)
parts = value.split(".")
if len(parts) < 2:
return _dummy_jwt(now)
try:
payload = json.loads(_b64url_decode(parts[1]))
except (ValueError, json.JSONDecodeError):
return _dummy_jwt(now)
if not isinstance(payload, dict):
return _dummy_jwt(now)
return _encode_dummy_jwt(_redact_jwt_payload(payload, now=now))
def _encode_dummy_jwt(payload: dict) -> str:
def enc(obj: dict) -> str:
raw = json.dumps(obj, separators=(",", ":")).encode()
return base64.urlsafe_b64encode(raw).decode().rstrip("=")
return f"{enc({'alg': 'none', 'typ': 'JWT'})}.{enc(payload)}.placeholder"
def _redact_jwt_payload(
payload: dict,
*,
now: datetime | None = None,
) -> dict:
check_now = now or datetime.now(timezone.utc)
out = _redact_claims(payload)
if not isinstance(out, dict):
out = {}
out["exp"] = int(check_now.timestamp()) + 3600
out.setdefault("sub", "bot-bottle-placeholder")
return out
def _redact_claims(value: object) -> object:
if isinstance(value, dict):
out: dict[str, object] = {}
for key, inner in value.items():
lower = key.lower()
if key == "https://api.openai.com/profile":
out[key] = _redact_profile_claim(inner)
elif key == "https://api.openai.com/auth":
out[key] = _redact_auth_claim(inner)
elif lower == "email":
out[key] = "bot-bottle@example.invalid"
elif lower == "email_verified":
out[key] = True
elif lower in {"exp", "iat", "nbf", "auth_time", "pwd_auth_time"}:
out[key] = inner if isinstance(inner, (int, float)) else 0
elif lower in {"aud", "scp", "amr"}:
out[key] = inner if isinstance(inner, list) else []
elif isinstance(inner, bool):
out[key] = inner
elif isinstance(inner, (dict, list)):
out[key] = _redact_claims(inner)
else:
out[key] = "bot-bottle-placeholder"
return out
if isinstance(value, list):
return []
return "bot-bottle-placeholder"
def _redact_profile_claim(value: object) -> dict:
profile = value if isinstance(value, dict) else {}
return {
"email": "bot-bottle@example.invalid",
"email_verified": bool(profile.get("email_verified", True)),
}
def _redact_auth_claim(value: object) -> dict:
auth = value if isinstance(value, dict) else {}
out: dict[str, object] = {}
for key, inner in auth.items():
lower = key.lower()
if lower == "chatgpt_plan_type" and isinstance(inner, str) and inner:
out[key] = inner
elif lower == "localhost" and isinstance(inner, bool):
out[key] = inner
elif isinstance(inner, bool):
out[key] = inner
elif isinstance(inner, list):
out[key] = []
elif isinstance(inner, dict):
out[key] = {}
else:
out[key] = "bot-bottle-placeholder"
out.setdefault("chatgpt_plan_type", "unknown")
out.setdefault("user_id", "bot-bottle-placeholder")
out.setdefault("chatgpt_user_id", "bot-bottle-placeholder")
out.setdefault("chatgpt_account_id", "bot-bottle-placeholder")
return out
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():
lower = key.lower()
if lower == "openai_api_key":
out[key] = None
elif lower == "tokens":
out[key] = _redact_codex_auth(inner, now=now)
elif lower in {"access_token", "id_token"}:
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 in {"account_id", "user_id", "email"}:
out[key] = "bot-bottle-placeholder"
else:
out[key] = _redact_codex_auth(inner, now=now)
return out
if isinstance(value, list):
return [_redact_codex_auth(v, now=now) for v in value]
return value
def _jwt_exp(token: str) -> datetime | None:
parts = token.split(".")
if len(parts) < 2:
return None
try:
payload = json.loads(_b64url_decode(parts[1]))
except (ValueError, json.JSONDecodeError):
return None
if not isinstance(payload, dict):
return None
exp = payload.get("exp")
if not isinstance(exp, (int, float)):
return None
return datetime.fromtimestamp(exp, timezone.utc)
def _b64url_decode(value: str) -> str:
padded = value + ("=" * (-len(value) % 4))
return base64.urlsafe_b64decode(padded.encode("ascii")).decode("utf-8")
__all__ = [
"codex_auth_path",
"codex_dummy_auth_json",
"codex_host_access_token",
"write_codex_dummy_auth_file",
]
+68 -4
View File
@@ -31,6 +31,9 @@ from pathlib import Path
from .log import die
from .manifest import Bottle
CODEX_HOST_CREDENTIAL_HOSTS = ("api.openai.com", "chatgpt.com")
CODEX_HOST_CREDENTIAL_TOKEN_REF = "BOT_BOTTLE_CODEX_HOST_ACCESS_TOKEN"
# DNS name agents will dial for the per-bottle egress sidecar.
# Backend-agnostic by contract: every concrete backend (Docker today,
@@ -174,11 +177,70 @@ def egress_routes_for_bottle(
"""Effective egress routes. This is what gets rendered into
routes.yaml + what the addon enforces.
Operators that want to allow a host declare it directly in
Operators that want to allow a host usually declare it directly in
`bottle.egress.routes` as an authenticated route or bare-pass entry
(`- host: <name>`). The legacy `bottle.egress.allowlist`
folding is gone egress is the single allowlist surface."""
return egress_manifest_routes(bottle)
(`- host: <name>`). Codex host-credential forwarding is the
provider-owned exception: when explicitly enabled, it adds or
upgrades the Codex API hosts to egress-owned authenticated routes. The
legacy `bottle.egress.allowlist` folding is gone egress is the
single allowlist surface."""
routes = list(egress_manifest_routes(bottle))
if not bottle.agent_provider.forward_host_credentials:
return tuple(routes)
if bottle.agent_provider.template != "codex":
return tuple(routes)
for host in CODEX_HOST_CREDENTIAL_HOSTS:
routes = _ensure_codex_host_credential_route(routes, host)
return tuple(routes)
def _next_token_env(routes: list[EgressRoute]) -> str:
return f"EGRESS_TOKEN_{len({r.token_env for r in routes if r.token_env})}"
def _codex_host_credential_token_env(routes: list[EgressRoute]) -> str:
for route in routes:
if route.token_ref == CODEX_HOST_CREDENTIAL_TOKEN_REF:
return route.token_env
return _next_token_env(routes)
def _ensure_codex_host_credential_route(
routes: list[EgressRoute], host: str,
) -> list[EgressRoute]:
for idx, route in enumerate(routes):
if route.host.lower() != host:
continue
if route.auth_scheme or route.token_ref:
if (
route.auth_scheme == "Bearer"
and route.token_ref == CODEX_HOST_CREDENTIAL_TOKEN_REF
):
return routes
die(
"codex host credential forwarding conflicts with an "
f"authenticated egress route for {host}. Remove that "
"route auth block or disable agent_provider.forward_host_credentials."
)
routes[idx] = EgressRoute(
host=route.host,
path_allowlist=route.path_allowlist,
auth_scheme="Bearer",
token_env=_codex_host_credential_token_env(routes),
token_ref=CODEX_HOST_CREDENTIAL_TOKEN_REF,
roles=route.roles,
)
return routes
routes.append(EgressRoute(
host=host,
auth_scheme="Bearer",
token_env=_codex_host_credential_token_env(routes),
token_ref=CODEX_HOST_CREDENTIAL_TOKEN_REF,
))
return routes
def egress_token_env_map(
@@ -251,6 +313,8 @@ def egress_resolve_token_values(
a sealed mapping without touching `os.environ`."""
out: dict[str, str] = {}
for token_env, token_ref in token_env_map.items():
if token_ref == CODEX_HOST_CREDENTIAL_TOKEN_REF:
continue
value = host_env.get(token_ref)
if value is None:
die(
+19 -3
View File
@@ -228,15 +228,16 @@ class AgentProvider:
template: str = "claude"
dockerfile: str = ""
forward_host_credentials: bool = False
@classmethod
def from_dict(cls, bottle_name: str, raw: object) -> "AgentProvider":
d = _as_json_object(raw, f"bottle '{bottle_name}' agent_provider")
for k in d:
if k not in {"template", "dockerfile"}:
if k not in {"template", "dockerfile", "forward_host_credentials"}:
raise ManifestError(
f"bottle '{bottle_name}' agent_provider has unknown key {k!r}; "
f"allowed: template, dockerfile"
f"allowed: template, dockerfile, forward_host_credentials"
)
template = d.get("template", "claude")
if not isinstance(template, str) or not template:
@@ -255,7 +256,22 @@ class AgentProvider:
f"bottle '{bottle_name}' agent_provider.dockerfile must be a "
f"string (was {type(dockerfile).__name__})"
)
return cls(template=template, dockerfile=dockerfile)
forward_host_credentials = d.get("forward_host_credentials", False)
if not isinstance(forward_host_credentials, bool):
raise ManifestError(
f"bottle '{bottle_name}' agent_provider.forward_host_credentials "
f"must be a boolean (was {type(forward_host_credentials).__name__})"
)
if forward_host_credentials and template != "codex":
raise ManifestError(
f"bottle '{bottle_name}' agent_provider.forward_host_credentials "
"is currently only supported for template 'codex'"
)
return cls(
template=template,
dockerfile=dockerfile,
forward_host_credentials=forward_host_credentials,
)
@dataclass(frozen=True)
@@ -0,0 +1,186 @@
# PRD 0029: Codex host credentials through egress
- **Status:** Draft
- **Author:** didericis-codex
- **Created:** 2026-05-29
- **Issue:** #109
## Summary
Allow Codex bottles to use a host-authorized ChatGPT/device-login
access token by forwarding it only into the egress sidecar, gated by an
explicit `agent_provider.forward_host_credentials` manifest flag.
## Problem
Codex bottles can reach OpenAI hosts after they are added to egress, but
requests to Codex's ChatGPT-backed API endpoints still fail with HTTP
403 when the egress route is unauthenticated. The egress proxy strips
agent-originated `Authorization` headers and only re-injects auth for
routes that declare an egress-owned token. Bare `api.openai.com` or
`chatgpt.com` routes therefore forward Codex requests without the
ChatGPT bearer token.
Copying the host `~/.codex/auth.json` into the agent would solve auth
mode detection but would also put access and refresh material inside the
agent sandbox. That cuts against bot-bottle's credential minimization
model: provider credentials should live in the sidecar boundary when
possible, not in the agent.
## Goals / Success Criteria
- A Codex bottle with host ChatGPT auth can call Codex's
`api.openai.com` and `chatgpt.com` endpoints through egress.
- Host credential forwarding happens only when the bottle declares
`agent_provider.forward_host_credentials: true`.
- The agent container does not receive `OPENAI_API_KEY`,
`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 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
launch with a clear operator-facing message.
- Existing Claude OAuth placeholder behavior remains unchanged.
## Non-goals
- Refreshing Codex tokens in the sidecar. The first cut reads the host's
current access token at launch; operators can restart after host Codex
refreshes auth.
- Copying host `~/.codex/auth.json` credentials into the agent.
- Allowing arbitrary host credential forwarding. This PRD covers Codex
ChatGPT/device-login credentials only.
- Hot-applying new authenticated Codex routes to an existing running
sidecar. The current hot-apply path cannot safely populate new token
env slots in an already-running container.
## Scope
### In scope
- Add `agent_provider.forward_host_credentials` to the bottle manifest
schema, defaulting to `false`.
- Support the flag for `agent_provider.template: codex`.
- Read host Codex auth from `$CODEX_HOME/auth.json` when `CODEX_HOME` is
set, otherwise from `~/.codex/auth.json`.
- Extract only `tokens.access_token` for egress injection.
- Generate a dummy agent-side `auth.json` from the host auth file's
mode and key shape, without copying real token values.
- Validate that host auth is not API-key mode and the access token is
present, JWT-shaped, and not expired.
- Add or upgrade `api.openai.com` and `chatgpt.com` egress routes to
inject that access token via a shared `EGRESS_TOKEN_N` sidecar env
slot.
- Pass the extracted token only into the sidecar compose/run
environment, alongside other egress token values.
### Out of scope
- Sidecar-owned refresh using `tokens.refresh_token`.
- Sharing full Codex auth state with the agent.
- Supporting host credential forwarding for non-Codex providers.
## Design
### Manifest
Extend `agent_provider`:
```yaml
agent_provider:
template: codex
forward_host_credentials: true
```
The field defaults to `false`. If set on a non-Codex provider, manifest
validation should reject it until that provider has a concrete,
credential-minimizing implementation.
### Host auth extraction
At prepare/launch time, when the flag is enabled for Codex:
1. Resolve the host Codex home directory from `$CODEX_HOME`, falling
back to `~/.codex`.
2. Parse `auth.json`.
3. Require user/device auth mode rather than API-key auth.
4. Require a non-empty `tokens.access_token`.
5. Parse the JWT payload enough to require an `exp` claim in the future.
6. Return only the access token value to the launch path.
Errors should name the missing or invalid condition and point the
operator at `codex login --device-auth`, without printing token values.
### Egress route
When forwarding host Codex credentials, the effective egress route table
should contain authenticated `api.openai.com` and `chatgpt.com` routes.
If the bottle already declares either host as a bare-pass route, upgrade
it in the effective route table rather than requiring a duplicate
manifest entry. If the bottle already declares an authenticated route for
either host, fail rather than guessing whether to override
operator-provided auth, unless that route already uses the synthetic
Codex host credential token reference.
The rendered route should look like any other egress-owned auth route:
```yaml
routes:
- host: "api.openai.com"
auth_scheme: "Bearer"
token_env: "EGRESS_TOKEN_N"
- host: "chatgpt.com"
auth_scheme: "Bearer"
token_env: "EGRESS_TOKEN_N"
```
The access token value is supplied through the sidecar process
environment for that `EGRESS_TOKEN_N` slot. It must not be written to
`routes.yaml`, compose files, env files, logs, or user-facing output.
### Data flow
```mermaid
flowchart LR
H["Host ~/.codex/auth.json"] --> L["bot-bottle launch"]
L -->|access token only| S["egress sidecar env"]
L -->|dummy auth.json only| A
A["Codex agent"] -->|HTTPS via proxy, auth stripped| E["egress"]
E -->|Bearer injected from env| C["api.openai.com / chatgpt.com"]
```
## Implementation chunks
1. **PRD first.** Land this document as the first commit on the feature
branch.
2. **Manifest schema.** Add `forward_host_credentials`, validation, and
unit tests.
3. **Host Codex auth reader.** Add a small stdlib-only helper for
parsing and validating host Codex auth without printing values.
4. **Effective egress route.** Add/upgrade the Codex API routes when the
flag is enabled, and add tests for bare route upgrade,
missing-route insertion, and authenticated-route conflict.
5. **Agent auth marker.** Provision a dummy Codex `auth.json` into the
agent home so Codex selects the host's user/device auth branch while
real credentials stay in egress.
6. **Launch wiring.** Pass the host access token into the egress sidecar
env for Docker and smolmachines without exposing it to the agent.
7. **Docs and tests.** Update README examples and run the unit suite.
## Open questions
- Should a later version support sidecar refresh using the host refresh
token, or should restart-on-expiry remain the policy?
- Should telemetry hosts such as `ab.chatgpt.com` stay blocked by
default even when Codex ChatGPT auth is enabled?
## References
- Gitea issue #109: Codex ChatGPT auth should inject host access token
via egress.
- PRD 0017: Egress-proxy — universal MITM with path filtering + auth
injection.
- PRD 0026: Agent provider templates.
+182
View File
@@ -0,0 +1,182 @@
"""Unit: host Codex auth extraction."""
from __future__ import annotations
import base64
import json
import tempfile
import unittest
from datetime import datetime, timezone
from pathlib import Path
from bot_bottle.codex_auth import (
codex_auth_path,
codex_dummy_auth_json,
codex_host_access_token,
)
from bot_bottle.log import Die
def _jwt(exp: int) -> str:
def enc(obj: dict) -> str:
raw = json.dumps(obj, separators=(",", ":")).encode()
return base64.urlsafe_b64encode(raw).decode().rstrip("=")
return f"{enc({'alg': 'none'})}.{enc({'exp': exp})}.sig"
def _jwt_payload(token: str) -> dict:
payload = token.split(".")[1]
payload += "=" * (-len(payload) % 4)
return json.loads(base64.urlsafe_b64decode(payload.encode()).decode())
class TestCodexHostAccessToken(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory(prefix="cb-codex-auth.")
self.home = Path(self.tmp.name)
self.auth_path = self.home / "auth.json"
def tearDown(self):
self.tmp.cleanup()
def _write(self, payload: dict) -> None:
self.auth_path.write_text(json.dumps(payload))
def test_auth_path_uses_codex_home(self):
self.assertEqual(
self.auth_path,
codex_auth_path({"CODEX_HOME": str(self.home)}),
)
def test_returns_fresh_chatgpt_access_token(self):
token = _jwt(2000000000)
self._write({
"auth_mode": "chatgpt",
"tokens": {"access_token": token, "refresh_token": "hidden"},
})
out = codex_host_access_token(
{"CODEX_HOME": str(self.home)},
now=datetime(2026, 1, 1, tzinfo=timezone.utc),
)
self.assertEqual(token, out)
def test_missing_auth_file_dies(self):
with self.assertRaises(Die):
codex_host_access_token({"CODEX_HOME": str(self.home)})
def test_non_chatgpt_auth_dies(self):
self._write({"auth_mode": "api_key", "tokens": {"access_token": _jwt(2)}})
with self.assertRaises(Die):
codex_host_access_token({"CODEX_HOME": str(self.home)})
def test_user_auth_mode_is_allowed(self):
token = _jwt(2000000000)
self._write({"auth_mode": "user", "tokens": {"access_token": token}})
out = codex_host_access_token(
{"CODEX_HOME": str(self.home)},
now=datetime(2026, 1, 1, tzinfo=timezone.utc),
)
self.assertEqual(token, out)
def test_expired_token_dies(self):
self._write({
"auth_mode": "chatgpt",
"tokens": {"access_token": _jwt(1000)},
})
with self.assertRaises(Die):
codex_host_access_token(
{"CODEX_HOME": str(self.home)},
now=datetime(2026, 1, 1, tzinfo=timezone.utc),
)
def test_non_jwt_token_dies(self):
self._write({
"auth_mode": "chatgpt",
"tokens": {"access_token": "not-a-jwt"},
})
with self.assertRaises(Die):
codex_host_access_token({"CODEX_HOME": str(self.home)})
def test_dummy_auth_preserves_mode_and_redacts_tokens(self):
access = _jwt(2000000000)
refresh = "host-refresh-token"
self._write({
"auth_mode": "chatgpt",
"OPENAI_API_KEY": None,
"tokens": {
"access_token": access,
"id_token": _jwt(2000000000),
"refresh_token": refresh,
"account_id": "acct-host",
},
"last_refresh": "2026-05-29T00:00:00.000Z",
})
dummy = json.loads(codex_dummy_auth_json(
{"CODEX_HOME": str(self.home)},
now=datetime(2026, 1, 1, tzinfo=timezone.utc),
))
self.assertEqual("chatgpt", dummy["auth_mode"])
self.assertIsNone(dummy["OPENAI_API_KEY"])
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("bot-bottle-placeholder", dummy["tokens"]["account_id"])
self.assertIsNotNone(
codex_host_access_token(
{"CODEX_HOME": str(self.home)},
now=datetime(2026, 1, 1, tzinfo=timezone.utc),
)
)
def test_dummy_auth_keeps_required_account_claim_shape(self):
def jwt(payload: dict) -> str:
def enc(obj: dict) -> str:
raw = json.dumps(obj, separators=(",", ":")).encode()
return base64.urlsafe_b64encode(raw).decode().rstrip("=")
return f"{enc({'alg': 'none'})}.{enc(payload)}.sig"
self._write({
"auth_mode": "chatgpt",
"tokens": {
"access_token": jwt({
"exp": 2000000000,
"https://api.openai.com/auth": {
"chatgpt_plan_type": "plus",
"chatgpt_account_id": "acct-real",
"chatgpt_user_id": "user-real",
"user_id": "auth-user-real",
"localhost": True,
},
"https://api.openai.com/profile": {
"email": "real@example.invalid",
"email_verified": True,
},
}),
"id_token": jwt({
"exp": 2000000000,
"email": "real@example.invalid",
"email_verified": True,
"https://api.openai.com/auth": {
"chatgpt_plan_type": "plus",
"chatgpt_account_id": "acct-real",
},
}),
"refresh_token": "hidden",
},
})
dummy = json.loads(codex_dummy_auth_json(
{"CODEX_HOME": str(self.home)},
now=datetime(2026, 1, 1, tzinfo=timezone.utc),
))
access_payload = _jwt_payload(dummy["tokens"]["access_token"])
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("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"])
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,104 @@
"""Unit: docker provider auth marker provisioning."""
from __future__ import annotations
import unittest
from pathlib import Path
from unittest.mock import patch
from bot_bottle.backend import BottleSpec
from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
from bot_bottle.backend.docker.provision import provider_auth as _provider_auth
from bot_bottle.egress import EgressPlan
from bot_bottle.git_gate import GitGatePlan
from bot_bottle.manifest import Manifest
from bot_bottle.pipelock import PipelockProxyPlan
def _plan(*, codex_auth_file: Path | None = None) -> DockerBottlePlan:
manifest = Manifest.from_json_obj({
"bottles": {"dev": {"agent_provider": {"template": "codex"}}},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
})
return DockerBottlePlan(
spec=BottleSpec(
manifest=manifest,
agent_name="demo",
copy_cwd=False,
user_cwd="/tmp/x",
),
stage_dir=Path("/tmp/stage"),
slug="demo-abc12",
container_name="bot-bottle-demo-abc12",
container_name_pinned=False,
image="bot-bottle-codex:latest",
derived_image="",
runtime_image="bot-bottle-codex:latest",
dockerfile_path="",
env_file=Path("/tmp/agent.env"),
forwarded_env={},
prompt_file=Path("/tmp/prompt.txt"),
proxy_plan=PipelockProxyPlan(
yaml_path=Path("/tmp/pipelock.yaml"),
slug="demo-abc12",
),
git_gate_plan=GitGatePlan(
slug="demo-abc12",
entrypoint_script=Path("/tmp/git-gate-entrypoint.sh"),
hook_script=Path("/tmp/git-gate-hook"),
access_hook_script=Path("/tmp/git-gate-access-hook"),
upstreams=(),
),
egress_plan=EgressPlan(
slug="demo-abc12",
routes_path=Path("/tmp/routes.yaml"),
routes=(),
token_env_map={},
),
supervise_plan=None,
use_runsc=False,
agent_command="codex",
agent_provider_template="codex",
codex_auth_file=codex_auth_file,
)
class TestProvisionProviderAuth(unittest.TestCase):
def test_noop_without_codex_auth_file(self):
with patch.object(_provider_auth.subprocess, "run") as run:
_provider_auth.provision_provider_auth(
_plan(), "bot-bottle-demo-abc12",
)
self.assertEqual(0, run.call_count)
def test_copies_dummy_auth_json_to_codex_home(self):
with patch.object(_provider_auth.subprocess, "run") as run:
_provider_auth.provision_provider_auth(
_plan(codex_auth_file=Path("/tmp/codex-auth.json")),
"bot-bottle-demo-abc12",
)
argvs = [call.args[0] for call in run.call_args_list]
self.assertIn(
["docker", "exec", "-u", "0", "bot-bottle-demo-abc12",
"mkdir", "-p", "/home/node/.codex"],
argvs,
)
self.assertIn(
["docker", "cp", "/tmp/codex-auth.json",
"bot-bottle-demo-abc12:/home/node/.codex/auth.json"],
argvs,
)
self.assertIn(
["docker", "exec", "-u", "0", "bot-bottle-demo-abc12",
"chown", "node:node", "/home/node/.codex/auth.json"],
argvs,
)
self.assertIn(
["docker", "exec", "-u", "0", "bot-bottle-demo-abc12",
"chmod", "600", "/home/node/.codex/auth.json"],
argvs,
)
if __name__ == "__main__":
unittest.main()
+76
View File
@@ -4,6 +4,7 @@ resolution (PRD 0017)."""
import unittest
from bot_bottle.egress import (
CODEX_HOST_CREDENTIAL_TOKEN_REF,
egress_manifest_routes,
egress_render_routes,
egress_resolve_token_values,
@@ -22,6 +23,21 @@ def _bottle(routes):
}).bottles["dev"]
def _codex_bottle(*, forward_host_credentials: bool, routes):
return Manifest.from_json_obj({
"bottles": {
"dev": {
"agent_provider": {
"template": "codex",
"forward_host_credentials": forward_host_credentials,
},
"egress": {"routes": routes},
}
},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
}).bottles["dev"]
class TestRoutesForBottle(unittest.TestCase):
def test_authenticated_route_gets_slot(self):
b = _bottle([{
@@ -107,6 +123,59 @@ class TestRoutesForBottleUsesManifestOnly(unittest.TestCase):
effective = [r.host for r in egress_routes_for_bottle(b)]
self.assertEqual(["x.example"], effective)
def test_codex_forward_host_credentials_adds_codex_routes(self):
b = _codex_bottle(forward_host_credentials=True, routes=[])
routes = egress_routes_for_bottle(b)
self.assertEqual(["api.openai.com", "chatgpt.com"], [r.host for r in routes])
self.assertEqual("Bearer", routes[0].auth_scheme)
self.assertEqual("EGRESS_TOKEN_0", routes[0].token_env)
self.assertEqual(CODEX_HOST_CREDENTIAL_TOKEN_REF, routes[0].token_ref)
self.assertEqual("Bearer", routes[1].auth_scheme)
self.assertEqual("EGRESS_TOKEN_0", routes[1].token_env)
self.assertEqual(CODEX_HOST_CREDENTIAL_TOKEN_REF, routes[1].token_ref)
def test_codex_forward_host_credentials_upgrades_bare_chatgpt_route(self):
b = _codex_bottle(
forward_host_credentials=True,
routes=[{"host": "chatgpt.com", "path_allowlist": ["/backend-api/"]}],
)
routes = egress_routes_for_bottle(b)
self.assertEqual(2, len(routes))
self.assertEqual("chatgpt.com", routes[0].host)
self.assertEqual("Bearer", routes[0].auth_scheme)
self.assertEqual("EGRESS_TOKEN_0", routes[0].token_env)
self.assertEqual(CODEX_HOST_CREDENTIAL_TOKEN_REF, routes[0].token_ref)
self.assertEqual(("/backend-api/",), routes[0].path_allowlist)
self.assertEqual("api.openai.com", routes[1].host)
self.assertEqual("EGRESS_TOKEN_0", routes[1].token_env)
def test_codex_forward_host_credentials_accepts_explicit_synthetic_route(self):
b = _codex_bottle(
forward_host_credentials=True,
routes=[{
"host": "api.openai.com",
"auth": {
"scheme": "Bearer",
"token_ref": CODEX_HOST_CREDENTIAL_TOKEN_REF,
},
}],
)
routes = egress_routes_for_bottle(b)
self.assertEqual(["api.openai.com", "chatgpt.com"], [r.host for r in routes])
self.assertEqual("EGRESS_TOKEN_0", routes[0].token_env)
self.assertEqual("EGRESS_TOKEN_0", routes[1].token_env)
def test_codex_forward_host_credentials_conflicts_with_authed_route(self):
b = _codex_bottle(
forward_host_credentials=True,
routes=[{
"host": "chatgpt.com",
"auth": {"scheme": "Bearer", "token_ref": "OTHER"},
}],
)
with self.assertRaises(Die):
egress_routes_for_bottle(b)
class TestTokenEnvMap(unittest.TestCase):
def test_only_authenticated_routes_contribute(self):
@@ -217,6 +286,13 @@ class TestResolveTokenValues(unittest.TestCase):
{"GH_PAT": ""},
)
def test_codex_host_credential_ref_is_resolved_by_launch(self):
out = egress_resolve_token_values(
{"EGRESS_TOKEN_0": CODEX_HOST_CREDENTIAL_TOKEN_REF},
{},
)
self.assertEqual({}, out)
if __name__ == "__main__":
unittest.main()
+34
View File
@@ -29,6 +29,13 @@ def _provider_bottle(provider, routes):
}).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"}])
@@ -52,6 +59,33 @@ class TestMinimalRoute(unittest.TestCase):
_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,
})
class TestPathAllowlist(unittest.TestCase):
def test_optional(self):
b = _bottle([{"host": "x.example"}])
+39
View File
@@ -20,6 +20,7 @@ from bot_bottle.backend.smolmachines.provision import (
ca as _ca,
git as _git,
prompt as _prompt,
provider_auth as _provider_auth,
skills as _skills,
supervise as _supervise,
)
@@ -53,6 +54,7 @@ def _plan(
bundle_ip: str = "192.168.50.2",
agent_git_gate_host: str = "127.0.0.1:55555",
agent_supervise_url: str = "http://127.0.0.1:55556/",
codex_auth_file: Path | None = None,
) -> SmolmachinesBottlePlan:
bottle_json: dict = {}
git_json: dict = {}
@@ -127,6 +129,7 @@ def _plan(
supervise_plan=supervise_plan,
agent_git_gate_host=agent_git_gate_host,
agent_supervise_url=agent_supervise_url,
codex_auth_file=codex_auth_file,
)
@@ -187,6 +190,42 @@ class TestProvisionPrompt(unittest.TestCase):
)
class TestProvisionProviderAuth(unittest.TestCase):
def test_noop_without_codex_auth_file(self):
with patch(
"bot_bottle.backend.smolmachines.provision.provider_auth._smolvm.machine_cp"
) as cp, patch(
"bot_bottle.backend.smolmachines.provision.provider_auth._smolvm.machine_exec"
) as ex:
_provider_auth.provision_provider_auth(
_plan(), "bot-bottle-demo-abc12",
)
self.assertEqual(0, cp.call_count)
self.assertEqual(0, ex.call_count)
def test_copies_dummy_auth_json_to_codex_home(self):
with patch(
"bot_bottle.backend.smolmachines.provision.provider_auth._smolvm.machine_cp"
) as cp, patch(
"bot_bottle.backend.smolmachines.provision.provider_auth._smolvm.machine_exec"
) as ex:
_provider_auth.provision_provider_auth(
_plan(codex_auth_file=Path("/tmp/codex-auth.json")),
"bot-bottle-demo-abc12",
)
cp.assert_called_once_with(
"/tmp/codex-auth.json",
"bot-bottle-demo-abc12:/home/node/.codex/auth.json",
)
argv_seen = [call.args[1] for call in ex.call_args_list]
self.assertIn(["mkdir", "-p", "/home/node/.codex"], argv_seen)
self.assertIn(
["chown", "node:node", "/home/node/.codex/auth.json"],
argv_seen,
)
self.assertIn(["chmod", "600", "/home/node/.codex/auth.json"], argv_seen)
class TestProvisionSkills(unittest.TestCase):
def _patch_host_skill_dir(self, returns: dict[str, str]):
return patch(