fix(codex): provision dummy user auth state
This commit is contained in:
@@ -10,6 +10,7 @@ from __future__ import annotations
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
from copy import deepcopy
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
@@ -37,16 +38,12 @@ def codex_host_access_token(
|
||||
"Run `codex login --device-auth` on the host or disable "
|
||||
"agent_provider.forward_host_credentials."
|
||||
)
|
||||
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")
|
||||
raw = _read_auth_object(path)
|
||||
|
||||
if raw.get("auth_mode") != "chatgpt":
|
||||
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 ChatGPT/device "
|
||||
"codex host credentials: host Codex auth is not user/device "
|
||||
"auth. Run `codex login --device-auth` on the host."
|
||||
)
|
||||
|
||||
@@ -72,6 +69,79 @@ def codex_host_access_token(
|
||||
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
|
||||
|
||||
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'})}."
|
||||
f"{enc({'exp': exp, 'sub': 'bot-bottle-placeholder'})}."
|
||||
"placeholder"
|
||||
)
|
||||
|
||||
|
||||
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(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:
|
||||
@@ -93,4 +163,9 @@ def _b64url_decode(value: str) -> str:
|
||||
return base64.urlsafe_b64decode(padded.encode("ascii")).decode("utf-8")
|
||||
|
||||
|
||||
__all__ = ["codex_auth_path", "codex_host_access_token"]
|
||||
__all__ = [
|
||||
"codex_auth_path",
|
||||
"codex_dummy_auth_json",
|
||||
"codex_host_access_token",
|
||||
"write_codex_dummy_auth_file",
|
||||
]
|
||||
|
||||
Reference in New Issue
Block a user