hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 32 Packages Projects Releases Wiki Activity

refactor(hermes_cli): auth_nous/codex — unify refresh_nous_oauth_pure into from_state, codex login POST helper

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:18:03 -07:00
parent 1ecf5e96f9
commit 31e6c6fab4
4 changed files with 71 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
hermes_cli/auth_codex.py
View File

@@ -4,9 +4,8 @@ Tokens live in ~/.hermes/auth.json, NOT ~/.codex/: Hermes keeps its own Codex OA
separate from the Codex CLI / VS Code extension so one app's refresh-token rotation cannot
invalidate the other's session.
Split out of ``hermes_cli/auth.py``; every name is re-exported there so ``hermes_cli.auth.<name>``
keeps resolving (and monkeypatching). Origin-internal helpers are imported lazily inside each
function (no import cycle; patches on ``hermes_cli.auth.<helper>`` still intercept).
Split out of ``hermes_cli/auth.py`` and re-exported there; origin helpers are imported lazily
inside each function so ``hermes_cli.auth.<name>`` patches still intercept (and no import cycle).
"""
from __future__ import annotations
@@ -220,6 +219,15 @@ def _refresh_payload_access_token(
return payload, access
def _codex_login_post(url: str, *, failure: Tuple[str, str], **kwargs: Any) -> "httpx.Response":
"""One 15s POST for the device-login flow; transport errors become ``_codex_err(*failure)``."""
try:
with _codex_http_client(timeout=httpx.Timeout(15.0)) as client:
return client.post(url, **kwargs)
except Exception as exc:
raise _codex_err(f"{failure[0]}: {exc}", failure[1])
def _codex_http_client(**kwargs: Any) -> "httpx.Client":
"""Build an ``httpx.Client`` for Codex OAuth/probe endpoints with Happy-Eyeballs racing.
@@ -708,13 +716,10 @@ def _codex_request_device_code(issuer: str, client_id: str) -> Dict[str, Any]:
resp = None
max_attempts = 4
for attempt in range(1, max_attempts + 1):
try:
with _codex_http_client(timeout=httpx.Timeout(15.0)) as client:
resp = client.post(
f"{issuer}/api/accounts/deviceauth/usercode", json={"client_id": client_id},
headers={"Content-Type": "application/json"})
except Exception as exc:
raise _codex_err(f"Failed to request device code: {exc}", "device_code_request_failed")
resp = _codex_login_post(
f"{issuer}/api/accounts/deviceauth/usercode", json={"client_id": client_id},
headers={"Content-Type": "application/json"},
failure=("Failed to request device code", "device_code_request_failed"))
if resp.status_code != 429:
break
if attempt < max_attempts:
@@ -777,17 +782,14 @@ def _codex_exchange_authorization_code(
raise _codex_err(
"Device auth response missing authorization_code or code_verifier.",
"device_code_incomplete_exchange")
try:
with _codex_http_client(timeout=httpx.Timeout(15.0)) as client:
token_resp = client.post(
CODEX_OAUTH_TOKEN_URL,
data={
"grant_type": "authorization_code", "code": authorization_code,
"redirect_uri": f"{issuer}/deviceauth/callback", "client_id": client_id,
"code_verifier": code_verifier},
headers={"Content-Type": "application/x-www-form-urlencoded"})
except Exception as exc:
raise _codex_err(f"Token exchange failed: {exc}", "token_exchange_failed")
token_resp = _codex_login_post(
CODEX_OAUTH_TOKEN_URL,
data={
"grant_type": "authorization_code", "code": authorization_code,
"redirect_uri": f"{issuer}/deviceauth/callback", "client_id": client_id,
"code_verifier": code_verifier},
headers={"Content-Type": "application/x-www-form-urlencoded"},
failure=("Token exchange failed", "token_exchange_failed"))
if token_resp.status_code == 429:
raise _codex_login_rate_limited_error(token_resp, during=" during token exchange")
if token_resp.status_code != 200:

5
hermes_cli/auth_device_flow.py
View File

@@ -1,8 +1,7 @@
"""Shared device-code / browser / TLS helpers for interactive OAuth logins.
Split out of ``hermes_cli/auth.py``; every name is re-exported there so ``hermes_cli.auth.<name>``
keeps resolving (and monkeypatching). Origin-internal helpers are imported lazily inside each
function (no import cycle; patches on ``hermes_cli.auth.<helper>`` still intercept).
Split out of ``hermes_cli/auth.py`` and re-exported there; origin helpers are imported lazily
inside each function so ``hermes_cli.auth.<name>`` patches still intercept (and no import cycle).
"""
from __future__ import annotations

89
hermes_cli/auth_nous.py
View File

@@ -1,8 +1,7 @@
"""Nous Portal OAuth: device-code login, refresh, shared-store mirroring, JWT selection, status.
Split out of ``hermes_cli/auth.py``; every name is re-exported there so ``hermes_cli.auth.<name>``
keeps resolving (and monkeypatching). Origin-internal helpers are imported lazily inside each
function (no import cycle; patches on ``hermes_cli.auth.<helper>`` still intercept).
Split out of ``hermes_cli/auth.py`` and re-exported there; origin helpers are imported lazily
inside each function so ``hermes_cli.auth.<name>`` patches still intercept (and no import cycle).
"""
from __future__ import annotations
@@ -732,17 +731,37 @@ def refresh_nous_oauth_pure(
``on_state_update`` fires after a successful access-token refresh so callers owning persistent
state can save the rotated refresh token before later validation can fail.
"""
return refresh_nous_oauth_from_state(
{
"access_token": access_token, "refresh_token": refresh_token, "client_id": client_id,
"portal_base_url": portal_base_url, "inference_base_url": inference_base_url,
"token_type": token_type, "scope": scope, "obtained_at": obtained_at,
"expires_at": expires_at, "agent_key": agent_key,
"agent_key_expires_at": agent_key_expires_at,
"tls": {"insecure": insecure, "ca_bundle": ca_bundle}},
timeout_seconds=timeout_seconds, force_refresh=force_refresh,
on_state_update=on_state_update)
def refresh_nous_oauth_from_state(
src: Dict[str, Any], *, timeout_seconds: float = 15.0, force_refresh: bool = False,
on_state_update: Optional[Callable[[Dict[str, Any], str], None]] = None) -> Dict[str, Any]:
"""Refresh Nous OAuth from a state dict (defaults filled in) without mutating auth.json."""
from hermes_cli.auth import (
_assert_nous_inference_jwt_usable, _refresh_access_token, _resolve_verify,
_select_nous_invoke_jwt)
tls = src.get("tls") or {}
insecure, ca_bundle = tls.get("insecure"), tls.get("ca_bundle")
state: Dict[str, Any] = {
"access_token": access_token, "refresh_token": refresh_token,
"client_id": client_id or DEFAULT_NOUS_CLIENT_ID,
"portal_base_url": (portal_base_url or DEFAULT_NOUS_PORTAL_URL).rstrip("/"),
"inference_base_url": (inference_base_url or DEFAULT_NOUS_INFERENCE_URL).rstrip("/"),
"token_type": token_type or "Bearer", "scope": scope or DEFAULT_NOUS_SCOPE,
"obtained_at": obtained_at, "expires_at": expires_at, "agent_key": agent_key,
"agent_key_expires_at": agent_key_expires_at,
"access_token": src.get("access_token", ""), "refresh_token": src.get("refresh_token", ""),
"client_id": src.get("client_id") or DEFAULT_NOUS_CLIENT_ID,
"portal_base_url": (src.get("portal_base_url") or DEFAULT_NOUS_PORTAL_URL).rstrip("/"),
"inference_base_url": (
src.get("inference_base_url") or DEFAULT_NOUS_INFERENCE_URL).rstrip("/"),
"token_type": src.get("token_type") or "Bearer",
"scope": src.get("scope") or DEFAULT_NOUS_SCOPE,
"obtained_at": src.get("obtained_at"), "expires_at": src.get("expires_at"),
"agent_key": src.get("agent_key"), "agent_key_expires_at": src.get("agent_key_expires_at"),
"tls": {"insecure": bool(insecure), "ca_bundle": ca_bundle}}
verify = _resolve_verify(insecure=insecure, ca_bundle=ca_bundle, auth_state=state)
with _nous_http_client(timeout_seconds or 15.0, verify) as client:
@@ -767,23 +786,6 @@ def refresh_nous_oauth_pure(
return state
def refresh_nous_oauth_from_state(
state: Dict[str, Any], *, timeout_seconds: float = 15.0, force_refresh: bool = False,
on_state_update: Optional[Callable[[Dict[str, Any], str], None]] = None) -> Dict[str, Any]:
"""Refresh Nous OAuth from a state dict. Thin wrapper around refresh_nous_oauth_pure."""
tls = state.get("tls") or {}
return refresh_nous_oauth_pure(
state.get("access_token", ""), state.get("refresh_token", ""),
state.get("client_id", "hermes-cli"), state.get("portal_base_url", DEFAULT_NOUS_PORTAL_URL),
state.get("inference_base_url", DEFAULT_NOUS_INFERENCE_URL),
token_type=state.get("token_type", "Bearer"), scope=state.get("scope", DEFAULT_NOUS_SCOPE),
obtained_at=state.get("obtained_at"), expires_at=state.get("expires_at"),
agent_key=state.get("agent_key"), agent_key_expires_at=state.get("agent_key_expires_at"),
timeout_seconds=timeout_seconds, insecure=tls.get("insecure"),
ca_bundle=tls.get("ca_bundle"),
force_refresh=force_refresh, on_state_update=on_state_update)
def persist_nous_credentials(creds: Dict[str, Any], *, label: Optional[str] = None):
"""Persist Nous OAuth credentials as the singleton provider state.
@@ -1082,24 +1084,23 @@ def _snapshot_nous_pool_status() -> Dict[str, Any]:
return (agent_exp, access_exp, -int(getattr(entry, "priority", 0) or 0))
entry = max(entries, key=_entry_sort_key)
if not getattr(entry, "runtime_api_key", None):
attr = lambda name, default=None: getattr(entry, name, default) # noqa: E731
if not attr("runtime_api_key"):
return _empty_nous_auth_status()
access_token = getattr(entry, "access_token", None)
auth_type = str(getattr(entry, "auth_type", "") or "").strip().lower()
refresh_token = getattr(entry, "refresh_token", None)
access_token, refresh_token = attr("access_token"), attr("refresh_token")
auth_type = str(attr("auth_type", "") or "").strip().lower()
is_portal_oauth = bool(access_token) and (
auth_type.startswith("oauth") or bool(refresh_token))
label = getattr(entry, "label", "unknown")
portal_status_url = (
(getattr(entry, "portal_base_url", None) or DEFAULT_NOUS_PORTAL_URL) if is_portal_oauth
else None)
label = attr("label", "unknown")
return {
"logged_in": is_portal_oauth, "portal_base_url": portal_status_url,
"inference_base_url": getattr(entry, "inference_base_url", None)
or getattr(entry, "runtime_base_url", None) or getattr(entry, "base_url", None),
"logged_in": is_portal_oauth,
"portal_base_url": (
(attr("portal_base_url") or DEFAULT_NOUS_PORTAL_URL) if is_portal_oauth else None),
"inference_base_url": (
attr("inference_base_url") or attr("runtime_base_url") or attr("base_url")),
"access_token": access_token if is_portal_oauth else None,
"access_expires_at": getattr(entry, "expires_at", None),
"agent_key_expires_at": getattr(entry, "agent_key_expires_at", None),
"access_expires_at": attr("expires_at"),
"agent_key_expires_at": attr("agent_key_expires_at"),
"has_refresh_token": bool(refresh_token), "inference_credential_present": True,
"credential_source": f"pool:{label}", "source": f"pool:{label}"}
except Exception:
@@ -1417,11 +1418,9 @@ def _pick_nous_model_after_login(
raise _nous_err("No runtime API key available to fetch models", "invalid_token")
from hermes_cli.models import (
get_curated_nous_model_ids, get_pricing_for_provider,
check_nous_free_tier, partition_nous_models_by_tier,
nous_policy_allowed_ids, restrict_to_nous_policy,
union_with_portal_free_recommendations,
union_with_portal_paid_recommendations)
get_curated_nous_model_ids, get_pricing_for_provider, check_nous_free_tier,
partition_nous_models_by_tier, nous_policy_allowed_ids, restrict_to_nous_policy,
union_with_portal_free_recommendations, union_with_portal_paid_recommendations)
model_ids = get_curated_nous_model_ids()
_portal = auth_state.get("portal_base_url", "")

5
hermes_cli/auth_oauth_grants.py
View File

@@ -1,8 +1,7 @@
"""Single-use OAuth grant hygiene: strip cloned grants from profiles, heal forked grants.
Split out of ``hermes_cli/auth.py``; every name is re-exported there so ``hermes_cli.auth.<name>``
keeps resolving (and monkeypatching). Origin-internal helpers are imported lazily inside each
function (no import cycle; patches on ``hermes_cli.auth.<helper>`` still intercept).
Split out of ``hermes_cli/auth.py`` and re-exported there; origin helpers are imported lazily
inside each function so ``hermes_cli.auth.<name>`` patches still intercept (and no import cycle).
"""
from __future__ import annotations