Compare commits
6 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 8e5262b539 | |||
| 237afce844 | |||
| 91d6f5d8d2 | |||
| 55117c5ff4 | |||
| 1a5ea3b713 | |||
| 6e7f0756cf |
@@ -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,18 @@ 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. 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.
|
||||
|
||||
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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -45,19 +45,11 @@ _HOME_FOR = {
|
||||
}
|
||||
|
||||
|
||||
def _env_flags_for(user: str) -> list[str]:
|
||||
def _env_assignments_for(user: str, env: Mapping[str, str]) -> list[str]:
|
||||
home = _HOME_FOR.get(user, f"/home/{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] = []
|
||||
out = [f"HOME={home}", f"USER={user}"]
|
||||
for k, v in env.items():
|
||||
out += ["-e", f"{k}={v}"]
|
||||
out.append(f"{k}={v}")
|
||||
return out
|
||||
|
||||
|
||||
@@ -98,9 +90,8 @@ class SmolmachinesBottle(Bottle):
|
||||
flags = ["smolvm", "machine", "exec", "--name", self.name]
|
||||
if tty:
|
||||
flags += ["-i", "-t"]
|
||||
flags += _env_flags_for("node")
|
||||
flags += _guest_env_flags(self._guest_env)
|
||||
agent_tail = [self.agent_command]
|
||||
agent_tail = ["env", *_env_assignments_for("node", self._guest_env),
|
||||
self.agent_command]
|
||||
provider_prompt_args = prompt_args(
|
||||
self._agent_prompt_mode, self._prompt_path, argv=argv,
|
||||
)
|
||||
@@ -148,16 +139,16 @@ class SmolmachinesBottle(Bottle):
|
||||
on both backends. Pass `user="root"` for tests that need
|
||||
root.
|
||||
|
||||
`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.
|
||||
`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.
|
||||
r = subprocess.run(
|
||||
["smolvm", "machine", "exec", "--name", self.name] + argv,
|
||||
capture_output=True, text=True, check=False,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
@@ -423,7 +428,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,
|
||||
@@ -128,6 +129,24 @@ 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:
|
||||
@@ -144,9 +163,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 +204,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,49 @@
|
||||
"""Provision non-secret provider auth markers into a smolmachines bottle."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
from ....log import die
|
||||
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])
|
||||
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}"
|
||||
)
|
||||
@@ -0,0 +1,290 @@
|
||||
"""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.
|
||||
|
||||
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."""
|
||||
path = codex_auth_path(host_env)
|
||||
access = 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)
|
||||
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_exp(now: datetime | None, exp_ts: int | None) -> int:
|
||||
if exp_ts is not None:
|
||||
return exp_ts
|
||||
check_now = now or datetime.now(timezone.utc)
|
||||
return 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),
|
||||
"sub": "bot-bottle-placeholder",
|
||||
})
|
||||
|
||||
|
||||
def _dummy_jwt_from_host(
|
||||
value: object, *, now: datetime | None = None, exp_ts: int | None = None,
|
||||
) -> str:
|
||||
if not isinstance(value, str):
|
||||
return _dummy_jwt(now, exp_ts=exp_ts)
|
||||
parts = value.split(".")
|
||||
if len(parts) < 2:
|
||||
return _dummy_jwt(now, exp_ts=exp_ts)
|
||||
try:
|
||||
payload = json.loads(_b64url_decode(parts[1]))
|
||||
except (ValueError, json.JSONDecodeError):
|
||||
return _dummy_jwt(now, exp_ts=exp_ts)
|
||||
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))
|
||||
|
||||
|
||||
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,
|
||||
exp_ts: int | None = None,
|
||||
) -> dict:
|
||||
out = _redact_claims(payload)
|
||||
if not isinstance(out, dict):
|
||||
out = {}
|
||||
out["exp"] = _dummy_exp(now, exp_ts)
|
||||
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 == "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):
|
||||
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, exp_ts: int | 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, exp_ts=exp_ts)
|
||||
elif lower in {"access_token", "id_token"}:
|
||||
out[key] = _dummy_jwt_from_host(inner, now=now, exp_ts=exp_ts)
|
||||
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)
|
||||
return out
|
||||
if isinstance(value, list):
|
||||
return [_redact_codex_auth(v, now=now, exp_ts=exp_ts) 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
@@ -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
@@ -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)
|
||||
|
||||
+18
-1
@@ -21,7 +21,11 @@ from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import cast
|
||||
|
||||
from .egress import EGRESS_HOSTNAME, egress_routes_for_bottle
|
||||
from .egress import (
|
||||
CODEX_HOST_CREDENTIAL_HOSTS,
|
||||
EGRESS_HOSTNAME,
|
||||
egress_routes_for_bottle,
|
||||
)
|
||||
from .supervise import SUPERVISE_HOSTNAME
|
||||
from .manifest import Bottle
|
||||
|
||||
@@ -111,6 +115,19 @@ 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())
|
||||
|
||||
|
||||
|
||||
@@ -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, keeps the selected ChatGPT
|
||||
account id, 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.
|
||||
@@ -0,0 +1,207 @@
|
||||
"""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("acct-host", 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_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:
|
||||
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("acct-real", 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()
|
||||
@@ -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()
|
||||
|
||||
@@ -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"}])
|
||||
|
||||
@@ -113,6 +113,25 @@ 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):
|
||||
|
||||
@@ -59,10 +59,9 @@ 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,
|
||||
@@ -107,7 +106,7 @@ class TestClaudeArgvWrapped(unittest.TestCase):
|
||||
HTTPS_PROXY="http://127.0.0.1:1234",
|
||||
NO_PROXY="localhost",
|
||||
).agent_argv([]))
|
||||
self.assertIn("-e", argv)
|
||||
self.assertIn("env", argv)
|
||||
self.assertIn("HTTPS_PROXY=http://127.0.0.1:1234", argv)
|
||||
self.assertIn("NO_PROXY=localhost", argv)
|
||||
|
||||
@@ -119,8 +118,8 @@ class TestClaudeArgvWrapped(unittest.TestCase):
|
||||
argv = _bottle().agent_argv([])
|
||||
agent_idx = argv.index("claude")
|
||||
self.assertEqual(
|
||||
["runuser", "-u", "node", "--"],
|
||||
argv[agent_idx - 4:agent_idx],
|
||||
["runuser", "-u", "node", "--", "env"],
|
||||
argv[agent_idx - 7:agent_idx - 2],
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -21,6 +21,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,
|
||||
)
|
||||
@@ -55,6 +56,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 = {}
|
||||
@@ -129,6 +131,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,
|
||||
)
|
||||
|
||||
|
||||
@@ -189,6 +192,66 @@ 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:
|
||||
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",
|
||||
)
|
||||
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)
|
||||
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):
|
||||
def _patch_host_skill_dir(self, returns: dict[str, str]):
|
||||
return patch(
|
||||
|
||||
Reference in New Issue
Block a user