diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py
index c4dc691ab1..a2f44a7258 100644
--- a/agent/auxiliary_client.py
+++ b/agent/auxiliary_client.py
@@ -2110,6 +2110,39 @@ def _read_codex_access_token() -> Optional[str]:
token = _pool_runtime_api_key(entry)
if token:
return token
+ return _read_codex_singleton_token()
+
+
+def _codex_pool_entry_base_url(entry: Any) -> str:
+ """Host the chat route sends this pooled Codex entry's key to (``model.base_url`` included)."""
+ row_base = _pool_runtime_base_url(entry, _CODEX_AUX_BASE_URL)
+ try:
+ from hermes_cli.auth_codex import _codex_pool_route_base_url
+ return _codex_pool_route_base_url(row_base) or row_base
+ except Exception as exc:
+ logger.debug("Codex pool route base resolution failed: %s", exc)
+ return row_base
+
+
+def _resolve_codex_credential_and_base() -> Tuple[Optional[str], str]:
+ """``(token, base_url)`` taken from ONE authority, so a Codex key is only ever sent to the host
+ it belongs to (#121486): the profile-scoped ``HERMES_CODEX_BASE_URL`` wins; otherwise a pooled
+ key goes where that pool entry routes (row URL / ``model.base_url``) and the auth.json OAuth
+ token goes to the ChatGPT default. ``(None, )`` without a usable token."""
+ override = _codex_base_url_override()
+ pool_present, entry = _select_pool_entry("openai-codex")
+ if pool_present:
+ token = _pool_runtime_api_key(entry)
+ if token:
+ return token, override or _codex_pool_entry_base_url(entry) or _CODEX_AUX_BASE_URL
+ # A present pool whose selection gave no token: read auth.json directly — re-selecting could
+ # rotate to another pool row and pair its key with the default host.
+ token = _read_codex_singleton_token() if pool_present else _read_codex_access_token()
+ return token, override or _CODEX_AUX_BASE_URL
+
+
+def _read_codex_singleton_token() -> Optional[str]:
+ """The profile's auth.json Codex access token (expired JWTs skipped), else None."""
try:
from hermes_cli.auth import _read_codex_tokens
access_token = _read_codex_tokens().get("tokens", {}).get("access_token")
@@ -2927,16 +2960,9 @@ def _build_codex_client(model: str) -> Tuple[Optional[Any], Optional[str]]:
"pass model explicitly (auxiliary..model in config.yaml)."
)
return None, None
- pool_present, entry = _select_pool_entry("openai-codex")
- codex_token = _pool_runtime_api_key(entry) if pool_present else None
- codex_override = _codex_base_url_override()
- if codex_token:
- base_url = codex_override or _pool_runtime_base_url(entry, _CODEX_AUX_BASE_URL) or _CODEX_AUX_BASE_URL
- else:
- codex_token = _read_codex_access_token()
- if not codex_token:
- return None, None
- base_url = codex_override or _CODEX_AUX_BASE_URL
+ codex_token, base_url = _resolve_codex_credential_and_base()
+ if not codex_token:
+ return None, None
logger.debug("Auxiliary client: Codex OAuth (%s via Responses API)", model)
real_client = _create_openai_client(
api_key=codex_token, base_url=base_url,
@@ -5000,11 +5026,10 @@ def _resolve_openai_codex_branch(req: _ResolveRequest) -> _ResolveResult:
no_token_msg = "resolve_provider_client: openai-codex requested but no Codex OAuth token found (run: hermes model)"
if req.raw_codex:
# Raw OpenAI client for callers needing responses.stream() (main agent loop).
- codex_token = _read_codex_access_token()
+ codex_token, base_url = _resolve_codex_credential_and_base()
if not codex_token:
logger.warning(no_token_msg)
return None, None
- base_url = _codex_base_url_override() or _CODEX_AUX_BASE_URL
raw_client = _create_openai_client(api_key=codex_token, base_url=base_url,
default_headers=_codex_cloudflare_headers(codex_token, base_url=base_url))
return raw_client, _normalize_resolved_model(model, req.provider)
diff --git a/agent/model_metadata.py b/agent/model_metadata.py
index 6b4b803298..c9689ebe82 100644
--- a/agent/model_metadata.py
+++ b/agent/model_metadata.py
@@ -1805,6 +1805,23 @@ def _codex_catalog_urls(base_url: str = "") -> Tuple[str, ...]:
CODEX_MODELS_CATALOG_URLS = _codex_catalog_urls()
+def _codex_catalog_probe_allowed(access_token: str, base_url: str = "") -> bool:
+ """Whether a catalog probe may carry ``access_token`` to ``base_url``'s ``/models``.
+
+ The caller binds ``base_url`` to the credential's own route, so a custom gateway is asked with
+ its own key (opaque or JWT). chatgpt.com only accepts ChatGPT OAuth access tokens — JWTs — so a
+ non-JWT credential aimed there is a gateway key composed with the wrong host: refuse it
+ (defense in depth, mirroring ``_probe_codex_quota_restored``'s gate; #121486).
+ """
+ if not access_token:
+ return False
+ base = (base_url or "").strip() or CODEX_MODELS_CATALOG_ENDPOINT
+ if not base_url_host_matches(base, "chatgpt.com"):
+ return True
+ from hermes_cli.auth_constants import _decode_jwt_claims
+ return bool(_decode_jwt_claims(access_token))
+
+
def fetch_codex_catalog_entries(get: Callable[[str], Any], base_url: str = "") -> Tuple[List[Any], Optional[int]]:
"""``(models, last_status)`` from the first catalog URL that answers HTTP 200 with a non-empty
``models`` list; ``get(url)`` is any client returning an object with ``status_code``/``json()``.
@@ -1833,11 +1850,7 @@ def _fetch_codex_oauth_context_lengths_with_source(access_token: str, base_url:
fingerprint (windows vary by entitlement); ``max_context_window`` lands in
``_codex_oauth_max_context_cache`` under the same key. An in-process hit reports False: not a
fresh provider confirmation, must not drive persistent writes."""
- # Real Codex access tokens are JWTs; a gateway's API key is not one. Refusing to probe non-JWT
- # credentials keeps the key off chatgpt.com — a service it does not belong to and cannot answer
- # for (#121486), mirroring ``_probe_codex_quota_restored``'s gate — and skips a doomed request.
- from hermes_cli.auth_constants import _decode_jwt_claims
- if not access_token or not _decode_jwt_claims(access_token):
+ if not _codex_catalog_probe_allowed(access_token, base_url):
return {}, False
now = time.time()
cache_key = _codex_oauth_token_fingerprint(access_token, base_url)
diff --git a/hermes_cli/auth.py b/hermes_cli/auth.py
index bb95b4aa8c..b8eed5e9e2 100644
--- a/hermes_cli/auth.py
+++ b/hermes_cli/auth.py
@@ -1947,10 +1947,15 @@ def get_codex_auth_status() -> Dict[str, Any]:
"""Status snapshot for Codex auth (pool first, then legacy provider state).
Read-only by contract: status/doctor must never adopt, refresh or persist a credential (#68004)."""
- return _pool_first_oauth_status(
+ status = _pool_first_oauth_status(
"openai-codex", is_expiring=_codex_access_token_is_expiring, auth_mode="chatgpt",
resolve=lambda: resolve_codex_runtime_credentials(read_only=True),
on_pool_miss=_codex_pool_rate_limited_status)
+ if str(status.get("source") or "").startswith("pool:"):
+ # Pool rows keep the canonical URL; the chat route may send this key to model.base_url.
+ from hermes_cli.auth_codex import _codex_pool_route_base_url
+ status["base_url"] = _codex_pool_route_base_url(status.get("base_url"))
+ return status
def get_xai_oauth_auth_status() -> Dict[str, Any]:
diff --git a/hermes_cli/auth_codex.py b/hermes_cli/auth_codex.py
index 4896fd97c6..5134b3d12c 100644
--- a/hermes_cli/auth_codex.py
+++ b/hermes_cli/auth_codex.py
@@ -71,10 +71,27 @@ def _codex_base_url() -> str:
return os.getenv("HERMES_CODEX_BASE_URL", "").strip().rstrip("/") or DEFAULT_CODEX_BASE_URL
+def _codex_pool_route_base_url(entry_base_url: Optional[str] = "") -> str:
+ """Base URL the chat route sends a pooled Codex credential to — the same rule
+ ``runtime_provider._pool_entry_mode_and_url`` applies (``HERMES_CODEX_BASE_URL`` > ``model.base_url``
+ while the row still carries the canonical URL > the row's own URL). A pooled gateway key belongs to
+ that host only; composing it with the ambient default sends it to chatgpt.com (#121486)."""
+ base = _stripped(entry_base_url).rstrip("/")
+ try:
+ from hermes_cli.config import load_config_readonly
+ from hermes_cli.runtime_provider import _pool_entry_mode_and_url
+ model_cfg = load_config_readonly().get("model")
+ return _pool_entry_mode_and_url(
+ "openai-codex", None, model_cfg if isinstance(model_cfg, dict) else {}, "", base)[1]
+ except Exception:
+ logger.debug("Codex pool route base resolution failed", exc_info=True)
+ return _stripped(os.getenv("HERMES_CODEX_BASE_URL")).rstrip("/") or base or DEFAULT_CODEX_BASE_URL
+
+
def _codex_runtime_result(
- api_key: str, *, source: str, last_refresh: Optional[str]) -> Dict[str, Any]:
+ api_key: str, *, source: str, last_refresh: Optional[str], base_url: Optional[str] = None) -> Dict[str, Any]:
return {
- "provider": "openai-codex", "base_url": _codex_base_url(), "api_key": api_key,
+ "provider": "openai-codex", "base_url": base_url or _codex_base_url(), "api_key": api_key,
"source": source, "last_refresh": last_refresh, "auth_mode": "chatgpt"}
@@ -611,14 +628,17 @@ def resolve_codex_runtime_credentials(
if imported:
data = {"tokens": imported, "last_refresh": imported.get("last_refresh")}
if data is None:
- pool_token = _pool_codex_access_token()
+ pool_token, pool_base = _pool_codex_credential()
if pool_token and force_refresh and not read_only:
# Pool-only setup: a forced refresh must rotate the pool entry, not resend its token.
from agent.credential_pool import load_pool
refreshed = load_pool("openai-codex").try_refresh_matching(api_key_hint=pool_token)
pool_token = refreshed.runtime_api_key if refreshed is not None else ""
if pool_token:
- return _codex_runtime_result(pool_token, source="credential_pool", last_refresh=None)
+ # Report the host this row routes to, not the ambient default: a pooled gateway key
+ # paired with chatgpt.com leaks to every consumer of this result (#121486).
+ return _codex_runtime_result(pool_token, source="credential_pool", last_refresh=None,
+ base_url=_codex_pool_route_base_url(pool_base))
pool_rate_limit = _codex_pool_rate_limit_status()
if pool_rate_limit:
# Before surfacing the persisted cooldown, ask the usage endpoint whether the quota
@@ -628,10 +648,11 @@ def resolve_codex_runtime_credentials(
if not read_only and _probe_codex_pool_entry_quota_restored(pool_rate_limit):
logger.info("Codex quota restored upstream — clearing stale pool cooldown(s).")
clear_codex_pool_quota_cooldowns()
- pool_token = _pool_codex_access_token()
+ pool_token, pool_base = _pool_codex_credential()
if pool_token:
return _codex_runtime_result(
- pool_token, source="credential_pool", last_refresh=None)
+ pool_token, source="credential_pool", last_refresh=None,
+ base_url=_codex_pool_route_base_url(pool_base))
reset_at = pool_rate_limit.get("reset_at")
in_future = isinstance(reset_at, (int, float)) and reset_at > time.time()
raise _codex_quota_exhausted_error(int(reset_at - time.time()) if in_future else None)
@@ -895,6 +916,12 @@ def _pool_codex_access_token() -> str:
Fallback for ``resolve_codex_runtime_credentials`` when the singleton has no creds; reads
through ``read_credential_pool`` so a profile inherits the global-root pool (#34143).
"""
+ return _pool_codex_credential()[0]
+
+
+def _pool_codex_credential() -> Tuple[str, str]:
+ """``(access_token, row base_url)`` of the entry ``_pool_codex_access_token`` picks, so the
+ caller routes the token to the host that row belongs to; ``("", "")`` when none is usable."""
from agent.credential_pool import _parse_absolute_timestamp
from hermes_cli.auth import _nonempty_str, read_credential_pool
try:
@@ -905,10 +932,10 @@ def _pool_codex_access_token() -> str:
reset_at = _parse_absolute_timestamp(entry.get("last_error_reset_at"))
in_cooldown = reset_at is not None and reset_at > time.time()
if _nonempty_str(token) and not in_cooldown:
- return token.strip()
+ return token.strip(), _stripped(entry.get("base_url"))
except Exception:
logger.debug("Codex pool fallback lookup failed", exc_info=True)
- return ""
+ return "", ""
def _login_openai_codex(args, pconfig: ProviderConfig, *, force_new_login: bool = False) -> None:
diff --git a/hermes_cli/auth_nous.py b/hermes_cli/auth_nous.py
index 163b4d80ff..9e79f39e50 100644
--- a/hermes_cli/auth_nous.py
+++ b/hermes_cli/auth_nous.py
@@ -1327,7 +1327,11 @@ def _pool_first_oauth_status(
"logged_in": True, "auth_store": str(_auth_file_path()),
"last_refresh": getattr(entry, "last_refresh", None),
"auth_mode": auth_mode,
- "source": f"pool:{getattr(entry, 'label', 'unknown')}", "api_key": api_key}
+ "source": f"pool:{getattr(entry, 'label', 'unknown')}", "api_key": api_key,
+ # The host this entry's key belongs to, so a caller never pairs it with
+ # another provider default (#121486).
+ "base_url": str(getattr(entry, "runtime_base_url", None)
+ or getattr(entry, "base_url", None) or "").rstrip("/")}
if on_pool_miss is not None and (degraded := on_pool_miss()):
return degraded
except Exception:
@@ -1338,7 +1342,7 @@ def _pool_first_oauth_status(
"logged_in": True, "auth_store": str(_auth_file_path()),
"last_refresh": creds.get("last_refresh"),
"auth_mode": creds.get("auth_mode"), "source": creds.get("source"),
- "api_key": creds.get("api_key")}
+ "api_key": creds.get("api_key"), "base_url": creds.get("base_url") or ""}
except AuthError as exc:
return {"logged_in": False, "auth_store": str(_auth_file_path()), "error": str(exc)}
diff --git a/hermes_cli/cli_model_switch_mixin.py b/hermes_cli/cli_model_switch_mixin.py
index e297926b51..56ea3aaf2e 100644
--- a/hermes_cli/cli_model_switch_mixin.py
+++ b/hermes_cli/cli_model_switch_mixin.py
@@ -377,7 +377,9 @@ class CLIModelSwitchMixin:
fallback_model = DEFAULT_CODEX_MODELS[0]
try:
- available = get_codex_model_ids(access_token=self.api_key if self.api_key else None)
+ # self.base_url is the route resolved with self.api_key (#121486).
+ available = get_codex_model_ids(
+ access_token=self.api_key if self.api_key else None, base_url=getattr(self, "base_url", None) or None)
if available:
fallback_model = available[0]
except Exception:
diff --git a/hermes_cli/codex_models.py b/hermes_cli/codex_models.py
index c38386139d..a993cc84db 100644
--- a/hermes_cli/codex_models.py
+++ b/hermes_cli/codex_models.py
@@ -151,23 +151,26 @@ def _ranked_slugs(entries: object) -> List[str]:
return _dedupe(slug for _, slug in sortable)
-def _fetch_models_from_api(access_token: str) -> List[str]:
- """Fetch available models from the Codex API. Returns visible models sorted by priority."""
+def _fetch_models_from_api(access_token: str, base_url: Optional[str] = None) -> List[str]:
+ """Fetch available models from the Codex API. Returns visible models sorted by priority.
+
+ ``base_url`` is the host the credential is routed to (resolved together with it); the
+ catalog is fetched there, never from a host the credential does not belong to (#121486).
+ """
try:
- # Real Codex access tokens are JWTs; a custom base's gateway key is not one. Refusing to
- # probe non-JWT credentials keeps the key off chatgpt.com (#121486) and skips a request
- # that can never answer for it — mirroring the quota probe's gate in auth_codex.
- from hermes_cli.auth_constants import DEFAULT_CODEX_BASE_URL, _decode_jwt_claims
- if not access_token or not _decode_jwt_claims(access_token):
+ from agent.model_metadata import _codex_catalog_probe_allowed
+ from hermes_cli.auth_codex import _codex_base_url
+ catalog_base = (base_url or "").strip().rstrip("/") or _codex_base_url()
+ if not _codex_catalog_probe_allowed(access_token, catalog_base):
return []
- catalog_base = os.getenv("HERMES_CODEX_BASE_URL", "").strip().rstrip("/") or DEFAULT_CODEX_BASE_URL
import httpx
# The per-account catalog needs ChatGPT-Account-ID (else ``{"models":[]}`` with HTTP 200
# masquerades as "no models") and, for residency-enforced workspaces, the residency header.
from agent.codex_headers import codex_account_headers
headers = {"Authorization": f"Bearer {access_token}", **codex_account_headers(access_token)}
from agent.model_metadata import fetch_codex_catalog_entries
- entries, _status = fetch_codex_catalog_entries(lambda url: httpx.get(url, headers=headers, timeout=10), base_url=catalog_base)
+ entries, _status = fetch_codex_catalog_entries(
+ lambda url: httpx.get(url, headers=headers, timeout=10), base_url=catalog_base)
except Exception as exc:
logger.debug("Failed to fetch Codex models from API: %s", exc)
return []
@@ -202,11 +205,14 @@ def _read_cache_models(codex_home: Path) -> List[str]:
return _ranked_slugs(entries if isinstance(entries, list) else [])
-def get_codex_model_ids(access_token: Optional[str] = None) -> List[str]:
- """Available Codex model IDs: live API (if token) > config.toml default > local cache > defaults."""
+def get_codex_model_ids(access_token: Optional[str] = None, base_url: Optional[str] = None) -> List[str]:
+ """Available Codex model IDs: live API (if token) > config.toml default > local cache > defaults.
+
+ Pass the ``base_url`` resolved together with ``access_token`` (runtime/pool route) so live
+ discovery asks the credential's own host."""
codex_home = Path(os.getenv("CODEX_HOME", "").strip() or str(Path.home() / ".codex")).expanduser()
if access_token:
- api_models = _fetch_models_from_api(access_token)
+ api_models = _fetch_models_from_api(access_token, base_url=base_url)
if api_models:
return _finalize_codex_models(api_models)
default_model = _read_default_model(codex_home)
diff --git a/hermes_cli/model_setup_flows.py b/hermes_cli/model_setup_flows.py
index f461eed172..42a6b195b0 100644
--- a/hermes_cli/model_setup_flows.py
+++ b/hermes_cli/model_setup_flows.py
@@ -374,16 +374,20 @@ def _model_flow_openai_codex(config, current_model=""):
# Prefer the credential pool (where `hermes auth` stores device_code tokens),
# fall back to legacy provider state.
- _codex_token = None
+ # Token and route base travel together (#121486): a pooled gateway key must never be sent to
+ # the chatgpt.com default by the catalog probe.
+ _codex_token = _codex_base = None
with contextlib.suppress(Exception):
_codex_status = get_codex_auth_status()
- _codex_token = _codex_status.get("api_key") if _codex_status.get("logged_in") else None
+ if _codex_status.get("logged_in"):
+ _codex_token, _codex_base = _codex_status.get("api_key"), _codex_status.get("base_url")
if not _codex_token:
with contextlib.suppress(Exception):
from hermes_cli.auth import resolve_codex_runtime_credentials
- _codex_token = resolve_codex_runtime_credentials().get("api_key")
+ _creds = resolve_codex_runtime_credentials()
+ _codex_token, _codex_base = _creds.get("api_key"), _creds.get("base_url")
- codex_models = get_codex_model_ids(access_token=_codex_token)
+ codex_models = get_codex_model_ids(access_token=_codex_token, base_url=_codex_base)
selected = _prompt_model_selection(
codex_models, current_model=current_model, confirm_provider="openai-codex",
confirm_base_url=DEFAULT_CODEX_BASE_URL, confirm_api_key=_codex_token or "")
diff --git a/hermes_cli/models.py b/hermes_cli/models.py
index a18fbb2e6c..a5e35ddae2 100644
--- a/hermes_cli/models.py
+++ b/hermes_cli/models.py
@@ -1295,15 +1295,19 @@ def _codex_catalog(normalized: str, force_refresh: bool) -> list[str]:
# catalog without a token / when unreachable. Read-only (#68004): a picker never imports,
# refreshes or persists a credential, so an expired stored token means the hardcoded catalog
# until the runtime lease refreshes it.
+ # The token and the host it is routed to come from the same resolution (#121486): a pooled
+ # gateway key is only ever sent to that gateway, never to the chatgpt.com default.
+ base_url = None
try:
from hermes_cli.auth import _codex_access_token_is_expiring, resolve_codex_runtime_credentials
- access_token = resolve_codex_runtime_credentials(read_only=True).get("api_key")
+ creds = resolve_codex_runtime_credentials(read_only=True)
+ access_token, base_url = creds.get("api_key"), creds.get("base_url")
if _codex_access_token_is_expiring(access_token, 0):
access_token = None
except Exception:
access_token = None
- return get_codex_model_ids(access_token=access_token)
+ return get_codex_model_ids(access_token=access_token, base_url=base_url)
_COPILOT_ACP_SESSION_MEMO_TTL = 300.0 # 5 min; SWR disk cache handles the rest
diff --git a/plugins/image_gen/openai-codex/__init__.py b/plugins/image_gen/openai-codex/__init__.py
index 9f0db6d419..0ae03107bc 100644
--- a/plugins/image_gen/openai-codex/__init__.py
+++ b/plugins/image_gen/openai-codex/__init__.py
@@ -77,6 +77,21 @@ def _read_codex_access_token() -> Optional[str]:
return None
+def _read_codex_credential() -> Tuple[Optional[str], Optional[str]]:
+ """``(token, base_url)`` from one resolution: the image request goes to the host the token's
+ credential routes to (pool row / ``model.base_url`` / profile override), never a default the
+ credential does not belong to (#121486)."""
+ try:
+ from agent.auxiliary_client import _resolve_codex_credential_and_base
+
+ token, base_url = _resolve_codex_credential_and_base()
+ token = token.strip() if isinstance(token, str) and token.strip() else None
+ return token, (base_url or None) if token else None
+ except Exception as exc:
+ logger.debug("Could not resolve Codex credential: %s", exc)
+ return None, None
+
+
def _httpx_available() -> bool:
try:
import httpx # noqa: F401
@@ -186,17 +201,20 @@ def _build_image_request(
def _post_image_request(
- token: str, *, prompt: str, size: str, quality: str, input_images: Optional[List[Dict[str, str]]] = None
+ token: str, *, prompt: str, size: str, quality: str, input_images: Optional[List[Dict[str, str]]] = None,
+ base_url: Optional[str] = None,
) -> Dict[str, Any]:
"""POST to the native Codex images endpoint; return the decoded JSON body plus
- ``imagegen_request_id`` (backend correlation id, for support tickets)."""
+ ``imagegen_request_id`` (backend correlation id, for support tickets).
+
+ ``base_url`` should come from the same resolution as ``token`` (``_read_codex_credential``)."""
import httpx
from agent.auxiliary_client import _codex_base_url_override
from agent.codex_headers import codex_cloudflare_headers
# Match the text auxiliary route, including profile-scoped overrides. Resolve
# per request: a multiplexed process can serve different Codex gateways.
- base_url = _codex_base_url_override() or _CODEX_BASE_URL
+ base_url = (base_url or "").strip().rstrip("/") or _codex_base_url_override() or _CODEX_BASE_URL
headers = codex_cloudflare_headers(token, base_url=base_url)
headers.update({
"Authorization": f"Bearer {token}",
@@ -266,7 +284,7 @@ class OpenAICodexImageGenProvider(StaticImageGenProvider):
aspect = resolve_aspect_ratio(aspect_ratio)
if not prompt:
return prompt_required_error("openai-codex", aspect)
- token = _read_codex_access_token()
+ token, base_url = _read_codex_credential()
if not token:
return error_factory("openai-codex", aspect)(_NO_AUTH, "auth_required")
if not _httpx_available():
@@ -283,7 +301,8 @@ class OpenAICodexImageGenProvider(StaticImageGenProvider):
try:
payload = _post_image_request(
- token, prompt=prompt, size=size, quality=meta["quality"], input_images=input_images or None)
+ token, prompt=prompt, size=size, quality=meta["quality"], input_images=input_images or None,
+ base_url=base_url)
except Exception as exc:
logger.debug("Codex image generation failed", exc_info=True)
return fail(f"OpenAI image generation via Codex auth failed: {exc}", "api_error")
diff --git a/tests/agent/test_auxiliary_client.py b/tests/agent/test_auxiliary_client.py
index 533dc744f4..a203f2ac9e 100644
--- a/tests/agent/test_auxiliary_client.py
+++ b/tests/agent/test_auxiliary_client.py
@@ -690,7 +690,9 @@ class TestBuildCodexClient:
def test_pool_without_selected_entry_falls_back_to_auth_store(self):
with (
patch("agent.auxiliary_client._select_pool_entry", return_value=(True, None)),
- patch("agent.auxiliary_client._read_codex_access_token", return_value="codex-auth-token"),
+ # A present pool with no usable row reads auth.json directly (no re-selection that
+ # could pair another row's key with the default host, #121486).
+ patch("agent.auxiliary_client._read_codex_singleton_token", return_value="codex-auth-token"),
patch("agent.auxiliary_client.OpenAI") as mock_openai,
):
mock_openai.return_value = MagicMock()
diff --git a/tests/hermes_cli/test_codex_credential_host_binding.py b/tests/hermes_cli/test_codex_credential_host_binding.py
new file mode 100644
index 0000000000..2e72ab8541
--- /dev/null
+++ b/tests/hermes_cli/test_codex_credential_host_binding.py
@@ -0,0 +1,308 @@
+"""A Codex credential is only ever sent to the host it belongs to (#121486).
+
+Adversarial regressions for the catalog picker, the context probe, the auxiliary Codex client and
+the image plugin behind a custom Codex gateway. Every case records each outbound request and
+asserts that no ``Authorization`` header reaches a host other than the credential's own route —
+including the ``model.base_url``-only gateway shape (``HERMES_CODEX_BASE_URL`` unset), an env/route
+mismatch, opaque and JWT-shaped gateway keys, a pool-selected credential, and the direct-ChatGPT
+positive control.
+"""
+
+import base64
+import importlib.util
+import json
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+from urllib.parse import urlparse
+
+import pytest
+
+GW = "https://codex-gw.example/backend-api/codex"
+OTHER_GW = "https://other-gw.example/backend-api/codex"
+CHATGPT = "https://chatgpt.com/backend-api/codex"
+OPAQUE = "dummy-gateway-pool-key"
+
+
+def _b64(raw: bytes) -> str:
+ return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()
+
+
+def _jwt(account: str = "acct") -> str:
+ claims = {"sub": account, "https://api.openai.com/auth": {"chatgpt_account_id": account}}
+ header = _b64(json.dumps({"alg": "RS256"}).encode())
+ return f"{header}.{_b64(json.dumps(claims).encode())}.sig"
+
+
+JWT = _jwt()
+_REPO = Path(__file__).resolve().parents[2]
+
+
+def _home(monkeypatch) -> Path:
+ import os
+ home = Path(os.environ["HERMES_HOME"])
+ home.mkdir(parents=True, exist_ok=True)
+ monkeypatch.delenv("HERMES_CODEX_BASE_URL", raising=False)
+ return home
+
+
+def _write_config(home: Path, *, base_url: str = "") -> None:
+ model = {"provider": "openai-codex", "default": "gpt-5.5"}
+ if base_url:
+ model["base_url"] = base_url
+ import yaml
+ (home / "config.yaml").write_text(yaml.safe_dump({"model": model})) # load cache keys on stat
+
+
+def _write_pool(home: Path, key: str, *, row_base: str = CHATGPT) -> None:
+ """A pool-only setup: the gateway key lives in ``credential_pool.openai-codex`` (no singleton)."""
+ (home / "auth.json").write_text(json.dumps({
+ "version": 1, "providers": {},
+ "credential_pool": {"openai-codex": [{
+ "id": "gw", "label": "gateway", "auth_type": "api_key", "priority": 0,
+ "source": "manual", "access_token": key, "base_url": row_base,
+ }]},
+ }))
+
+
+def _write_singleton(home: Path, token: str) -> None:
+ (home / "auth.json").write_text(json.dumps({
+ "version": 1, "active_provider": "openai-codex",
+ "providers": {"openai-codex": {
+ "tokens": {"access_token": token, "refresh_token": "rt"},
+ "last_refresh": "2026-09-24T00:00:00Z", "auth_mode": "chatgpt"}},
+ }))
+
+
+def _authorized_hosts(seen) -> set:
+ return {urlparse(url).hostname for url, auth in seen if auth}
+
+
+def _catalog_recorder(seen):
+ def get(url, headers=None, **_kw):
+ seen.append((url, (headers or {}).get("Authorization", "")))
+ return SimpleNamespace(status_code=200, json=lambda: {"models": [
+ {"slug": "gpt-5.5", "visibility": "list", "priority": 1, "context_window": 272000}]})
+ return get
+
+
+@pytest.fixture
+def picker_http(monkeypatch):
+ seen = []
+ monkeypatch.setitem(sys.modules, "httpx", SimpleNamespace(get=_catalog_recorder(seen)))
+ return seen
+
+
+# ── /model picker ──────────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize("key", [OPAQUE, JWT], ids=["opaque", "jwt"])
+def test_picker_model_base_url_only_gateway_keeps_pool_key_on_gateway(monkeypatch, picker_http, key):
+ """``model.base_url`` gateway, env unset, pool-only key: the picker asks the gateway (the host
+ the chat route uses for that key), never chatgpt.com."""
+ home = _home(monkeypatch)
+ _write_config(home, base_url=GW)
+ _write_pool(home, key)
+ from hermes_cli.models import _codex_catalog
+
+ _codex_catalog("openai-codex", force_refresh=True)
+
+ assert _authorized_hosts(picker_http) == {"codex-gw.example"}
+
+
+def test_picker_pool_row_with_its_own_gateway_url(monkeypatch, picker_http):
+ """A pool row that carries its gateway URL is authoritative for its own key."""
+ home = _home(monkeypatch)
+ _write_config(home)
+ _write_pool(home, JWT, row_base=OTHER_GW)
+ from hermes_cli.models import _codex_catalog
+
+ _codex_catalog("openai-codex", force_refresh=True)
+
+ assert _authorized_hosts(picker_http) == {"other-gw.example"}
+
+
+def test_picker_uses_the_route_base_not_a_mismatched_env(monkeypatch, picker_http):
+ """env/route mismatch: the caller's resolved route wins over a stale process env — the
+ credential is composed with the base it was resolved with, not an ambient re-read."""
+ monkeypatch.setenv("HERMES_CODEX_BASE_URL", OTHER_GW)
+ from hermes_cli.codex_models import get_codex_model_ids
+
+ get_codex_model_ids(access_token=JWT, base_url=GW)
+
+ assert _authorized_hosts(picker_http) == {"codex-gw.example"}
+
+
+def test_setup_flow_status_carries_the_pool_entry_route(monkeypatch):
+ """``hermes model`` → Codex reads the token from the auth status; the status must carry the
+ host that token routes to (model.base_url gateway), not leave the caller to assume chatgpt.com."""
+ home = _home(monkeypatch)
+ _write_config(home, base_url=GW)
+ _write_pool(home, OPAQUE)
+ from hermes_cli.auth import get_codex_auth_status
+
+ status = get_codex_auth_status()
+
+ assert status["api_key"] == OPAQUE
+ assert status["base_url"] == GW
+
+
+def test_setup_flow_status_carries_a_pool_row_own_gateway_url(monkeypatch):
+ home = _home(monkeypatch)
+ _write_config(home)
+ _write_pool(home, OPAQUE, row_base=OTHER_GW)
+ from hermes_cli.auth import get_codex_auth_status
+
+ assert get_codex_auth_status()["base_url"] == OTHER_GW
+
+
+def test_model_setup_flow_catalog_goes_to_the_pool_entry_gateway(monkeypatch, picker_http):
+ """``hermes model`` → OpenAI Codex with a pooled gateway key: the model list is fetched from
+ the gateway that key routes to, not chatgpt.com."""
+ home = _home(monkeypatch)
+ _write_config(home, base_url=GW)
+ _write_pool(home, OPAQUE)
+ monkeypatch.setattr("builtins.input", lambda prompt="": "1") # reuse existing credentials
+ monkeypatch.setattr("hermes_cli.auth._prompt_model_selection", lambda *a, **kw: None)
+ from hermes_cli.model_setup_flows import _model_flow_openai_codex
+
+ _model_flow_openai_codex({}, current_model="gpt-5.5")
+
+ assert _authorized_hosts(picker_http) == {"codex-gw.example"}
+
+
+def test_cli_default_model_swap_asks_the_session_route(monkeypatch, picker_http):
+ """The CLI's untouched-default swap queries the catalog with the session's own
+ ``(api_key, base_url)`` route, even when the process env names another gateway."""
+ from types import MethodType
+
+ from hermes_cli.cli_model_switch_mixin import CLIModelSwitchMixin
+
+ monkeypatch.setenv("HERMES_CODEX_BASE_URL", OTHER_GW)
+ cli = SimpleNamespace(model="", api_key=JWT, base_url=GW, _model_is_default=True,
+ _console_print=lambda *a, **kw: None)
+ cli._normalize_model_for_provider = MethodType(CLIModelSwitchMixin._normalize_model_for_provider, cli)
+
+ cli._normalize_model_for_provider("openai-codex")
+
+ assert _authorized_hosts(picker_http) == {"codex-gw.example"}
+
+
+def test_picker_direct_chatgpt_positive_control(monkeypatch, picker_http):
+ """No gateway anywhere: a ChatGPT OAuth JWT is still sent to chatgpt.com (allowed)."""
+ home = _home(monkeypatch)
+ _write_config(home)
+ _write_singleton(home, JWT)
+ from hermes_cli.models import _codex_catalog
+
+ _codex_catalog("openai-codex", force_refresh=True)
+
+ assert _authorized_hosts(picker_http) == {"chatgpt.com"}
+
+
+def test_picker_refuses_opaque_key_aimed_at_chatgpt(picker_http):
+ """Defense in depth: a non-JWT key composed with chatgpt.com is never sent there."""
+ from hermes_cli.codex_models import get_codex_model_ids
+
+ get_codex_model_ids(access_token=OPAQUE, base_url=CHATGPT)
+
+ assert picker_http == []
+
+
+# ── context probe (every chat turn without model.context_length) ───────────────────────────
+
+
+@pytest.fixture
+def probe_http(monkeypatch):
+ from agent import model_metadata as mm
+ seen = []
+ monkeypatch.setattr(mm, "requests", SimpleNamespace(get=_catalog_recorder(seen)))
+ monkeypatch.setattr(mm, "_ensure_requests", lambda: None)
+ monkeypatch.setattr(mm, "_codex_oauth_context_cache", {})
+ return seen
+
+
+@pytest.mark.parametrize("key", [OPAQUE, JWT], ids=["opaque", "jwt"])
+def test_context_probe_asks_the_gateway_with_its_own_key(probe_http, key):
+ """The route's base is bound to its key: a gateway key (opaque or JWT) probes that gateway's
+ catalog, never chatgpt.com."""
+ from agent import model_metadata as mm
+
+ live, fresh = mm._fetch_codex_oauth_context_lengths_with_source(key, base_url=GW)
+
+ assert _authorized_hosts(probe_http) == {"codex-gw.example"}
+ assert fresh and live.get("gpt-5.5") == 272000
+
+
+def test_context_probe_refuses_opaque_key_on_the_chatgpt_default(probe_http):
+ from agent import model_metadata as mm
+
+ assert mm._fetch_codex_oauth_context_lengths_with_source(OPAQUE, base_url="") == ({}, False)
+ assert probe_http == []
+
+
+def test_context_probe_direct_chatgpt_positive_control(probe_http):
+ from agent import model_metadata as mm
+
+ mm._fetch_codex_oauth_context_lengths_with_source(JWT, base_url=CHATGPT)
+
+ assert _authorized_hosts(probe_http) == {"chatgpt.com"}
+
+
+# ── auxiliary Codex client + image plugin (pool-selected credential) ───────────────────────
+
+
+def test_aux_codex_client_binds_pool_key_to_model_base_url_gateway(monkeypatch):
+ home = _home(monkeypatch)
+ _write_config(home, base_url=GW)
+ _write_pool(home, OPAQUE)
+ from agent import auxiliary_client as aux
+
+ token, base_url = aux._resolve_codex_credential_and_base()
+
+ assert (token, base_url) == (OPAQUE, GW)
+
+
+def test_aux_codex_client_singleton_positive_control(monkeypatch):
+ home = _home(monkeypatch)
+ _write_config(home)
+ _write_singleton(home, JWT)
+ from agent import auxiliary_client as aux
+
+ assert aux._resolve_codex_credential_and_base() == (JWT, CHATGPT)
+
+
+def _load_image_plugin():
+ spec = importlib.util.spec_from_file_location(
+ "codex_img_binding_test", _REPO / "plugins/image_gen/openai-codex/__init__.py")
+ module = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(module)
+ return module
+
+
+@pytest.mark.parametrize("key", [OPAQUE, JWT], ids=["opaque", "jwt"])
+def test_image_request_goes_to_the_pool_entry_gateway(monkeypatch, key):
+ """Image generation with a pool-selected gateway key and a ``model.base_url``-only gateway:
+ zero Authorization-bearing traffic to chatgpt.com."""
+ import httpx
+
+ home = _home(monkeypatch)
+ _write_config(home, base_url=GW)
+ _write_pool(home, key)
+ img = _load_image_plugin()
+ seen = []
+
+ def handler(request):
+ seen.append((str(request.url), request.headers.get("authorization", "")))
+ return httpx.Response(200, json={"data": [{"b64_json": "aGk="}]}, request=request)
+
+ real_client = httpx.Client
+ monkeypatch.setattr(httpx, "Client", lambda *a, **kw: real_client(
+ transport=httpx.MockTransport(handler), headers=kw.get("headers"), timeout=kw.get("timeout")))
+ monkeypatch.setattr(img, "save_b64_image", lambda b64, prefix="": home / "img.png")
+
+ result = img.OpenAICodexImageGenProvider().generate("a cat")
+
+ assert result["success"] is True, result
+ assert _authorized_hosts(seen) == {"codex-gw.example"}
+ assert seen[0][1] == f"Bearer {key}"
diff --git a/tests/hermes_cli/test_codex_models.py b/tests/hermes_cli/test_codex_models.py
index c75a97301c..244cb25c57 100644
--- a/tests/hermes_cli/test_codex_models.py
+++ b/tests/hermes_cli/test_codex_models.py
@@ -39,7 +39,7 @@ def test_codex_catalog_never_offers_chatgpt_rejected_pro_slugs(monkeypatch, tmp_
# synthesis rule; none of what it adds may be -pro.
templates = list(dict.fromkeys(t for _, ts in _FORWARD_COMPAT_TEMPLATE_MODELS for t in ts))
monkeypatch.setattr(
- "hermes_cli.codex_models._fetch_models_from_api", lambda access_token: templates
+ "hermes_cli.codex_models._fetch_models_from_api", lambda access_token, **_kw: templates
)
live = get_codex_model_ids(access_token="codex-access-token")
assert {synthetic for synthetic, _ in _FORWARD_COMPAT_TEMPLATE_MODELS} <= set(live)
@@ -135,7 +135,7 @@ def test_astra_requires_live_codex_account_discovery(monkeypatch, tmp_path):
encoding="utf-8",
)
monkeypatch.setenv("CODEX_HOME", str(tmp_path))
- monkeypatch.setattr(codex_models, "_fetch_models_from_api", lambda _token: [])
+ monkeypatch.setattr(codex_models, "_fetch_models_from_api", lambda _token, **_kw: [])
assert "gpt-6-astra" not in get_codex_model_ids(access_token="stale-token")
assert "openai/gpt-6-astra" not in get_codex_model_ids(access_token="stale-token")
@@ -145,7 +145,7 @@ def test_astra_requires_live_codex_account_discovery(monkeypatch, tmp_path):
monkeypatch.setattr(
codex_models,
"_fetch_models_from_api",
- lambda _token: codex_models._finalize_codex_models(["gpt-6-astra"]),
+ lambda _token, **_kw: codex_models._finalize_codex_models(["gpt-6-astra"]),
)
entitled = get_codex_model_ids(access_token="entitled-token")
assert entitled[entitled.index("gpt-6-astra") + 1] == "gpt-6-astra-900k"
@@ -178,7 +178,7 @@ def test_model_command_prompts_to_reuse_or_reauthenticate_codex_session(monkeypa
monkeypatch.setattr("hermes_cli.auth._login_openai_codex", _fake_login)
monkeypatch.setattr(
"hermes_cli.codex_models.get_codex_model_ids",
- lambda access_token=None: ["gpt-5.4", "gpt-5.5"],
+ lambda access_token=None, **_kw: ["gpt-5.4", "gpt-5.5"],
)
monkeypatch.setattr(
"hermes_cli.auth._prompt_model_selection",
diff --git a/tests/plugins/image_gen/test_openai_codex_provider.py b/tests/plugins/image_gen/test_openai_codex_provider.py
index 00f0bd1f66..8c586af4fd 100644
--- a/tests/plugins/image_gen/test_openai_codex_provider.py
+++ b/tests/plugins/image_gen/test_openai_codex_provider.py
@@ -55,7 +55,12 @@ def provider(monkeypatch):
def codex_backend(monkeypatch):
"""Route the plugin's ``httpx.Client`` at a fake Codex images backend; returns the request log
and lets a test swap the response via ``state["respond"]``."""
+ # Seed the auth.json token below the credential/base resolver so generate() exercises the real
+ # (token, base_url) binding; no pool present.
+ from agent import auxiliary_client
monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: "codex-token")
+ monkeypatch.setattr(auxiliary_client, "_select_pool_entry", lambda provider: (False, None))
+ monkeypatch.setattr(auxiliary_client, "_read_codex_singleton_token", lambda: "codex-token")
state = {"requests": [], "respond": None}
def _default(request):
@@ -181,7 +186,10 @@ class TestGenerate:
assert request.url.host == "images.example.test"
def test_returns_auth_error_without_codex_token(self, provider, monkeypatch):
+ from agent import auxiliary_client
monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: None)
+ monkeypatch.setattr(auxiliary_client, "_select_pool_entry", lambda provider: (False, None))
+ monkeypatch.setattr(auxiliary_client, "_read_codex_singleton_token", lambda: None)
result = provider.generate("a cat")
assert result["success"] is False
assert result["error_type"] == "auth_required"