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

refactor(hermes_cli): compact auth_device_flow + split forked-OAuth heal into _HealPass phases

Browse Source
This commit is contained in:
Teknium
2026-09-02 20:09:37 -07:00
parent 113f04616b
commit 40a09152f1
2 changed files with 301 additions and 377 deletions
Download Patch File Download Diff File Expand all files Collapse all files

214
hermes_cli/auth_device_flow.py
View File

@@ -1,133 +1,87 @@
"""Shared device-code / browser / TLS helpers for interactive OAuth logins.
Split out of ``hermes_cli/auth.py``; every moved name is re-imported there, so
``hermes_cli.auth.<name>`` keeps resolving (and monkeypatching) as before. 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``; 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).
"""
from __future__ import annotations
import logging
from typing import FrozenSet
import os
import ssl
import sys
import time
import webbrowser
from pathlib import Path
from typing import Any, Callable, Dict, Optional
from typing import Any, Callable, Dict, FrozenSet, Optional
from urllib.parse import urlparse
from hermes_cli.auth_constants import (
AuthError,
DEFAULT_NOUS_PORTAL_URL,
DEVICE_AUTH_POLL_INTERVAL_CAP_SECONDS,
DEVICE_CODE_GRANT_TYPE,
OAUTH_OVER_SSH_DOCS_URL,
httpx,
AuthError, DEFAULT_NOUS_PORTAL_URL, DEVICE_AUTH_POLL_INTERVAL_CAP_SECONDS,
DEVICE_CODE_GRANT_TYPE, OAUTH_OVER_SSH_DOCS_URL, httpx,
)
from utils import is_truthy_value
# Log-record parity with the origin module (caplog tests pin "hermes_cli.auth").
logger = logging.getLogger("hermes_cli.auth")
# Console/text-mode browsers that ``webbrowser`` will launch INSIDE the terminal, hijacking the
# user's TTY with an unusable text browser. When the resolved browser is one of these we refuse
# to auto-open and fall back to the print-the-URL path, same as a remote session.
_CONSOLE_BROWSER_NAMES: FrozenSet[str] = frozenset({
"w3m", "lynx", "links", "links2", "elinks", "www-browser",
"browsh", # TUI browser — still hijacks the terminal
})
# Console/text-mode browsers that ``webbrowser`` will happily launch INSIDE
# the terminal. Opening one of these is worse than not opening anything —
# it hijacks the user's TTY with an unusable text browser (the xAI OAuth
# "Account Management" page rendered in w3m, reported May 2026) instead of
# letting them copy the URL to a real browser. When the resolved browser is
# one of these we refuse to auto-open and fall back to the print-the-URL
# path, same as a remote session.
_CONSOLE_BROWSER_NAMES: FrozenSet[str] = frozenset(
{
"w3m",
"lynx",
"links",
"links2",
"elinks",
"www-browser",
"browsh", # TUI browser — still hijacks the terminal
}
# Browser-only remote IDEs / cloud shells (they don't set SSH_CLIENT / SSH_TTY). Keep this list
# narrow — well-known env vars set by the host platform — so a local shell never trips it.
_REMOTE_IDE_ENV_VARS = (
"CLOUD_SHELL", # GCP Cloud Shell
"CODESPACES", "CODESPACE_NAME", # GitHub Codespaces
"GITPOD_WORKSPACE_ID", # Gitpod
"REPL_ID", # Replit
"STACKBLITZ", # StackBlitz
)
def _is_remote_session() -> bool:
"""Detect environments where loopback OAuth can't reach the local browser.
"""Detect environments where loopback OAuth can't reach the local browser."""
return bool(
os.getenv("SSH_CLIENT") or os.getenv("SSH_TTY")
or any(os.getenv(var) for var in _REMOTE_IDE_ENV_VARS)
)
These environments typically don't set ``SSH_CLIENT`` / ``SSH_TTY``, so the SSH-only check left
them with no guidance and no fallback.
"""
if os.getenv("SSH_CLIENT") or os.getenv("SSH_TTY"):
return True
# Browser-only remote IDEs / cloud shells. Keep this list narrow
# (well-known, documented env vars set by the host platform) so
# we don't falsely trip on a developer's local shell.
for var in (
"CLOUD_SHELL", # GCP Cloud Shell
"CODESPACES", # GitHub Codespaces
"CODESPACE_NAME", # GitHub Codespaces (alt)
"GITPOD_WORKSPACE_ID", # Gitpod
"REPL_ID", # Replit
"STACKBLITZ", # StackBlitz
):
if os.getenv(var):
return True
return False
def _names_console_browser(value: str) -> bool:
token = value.strip().split()[0] if value.strip() else ""
return os.path.basename(token).lower() in _CONSOLE_BROWSER_NAMES
def _can_open_graphical_browser() -> bool:
"""Return True only when a *graphical* browser is likely to open.
``webbrowser.open()`` resolves to whatever the platform offers, and on a headless / CLI-only
Linux box with no GUI browser installed that is often a text-mode browser (w3m/lynx/links) which
launches inside the terminal and takes over the user's session.
Heuristics: * Respect ``$BROWSER`` — if it names a known console browser, refuse. * On Linux,
require a display server (``$DISPLAY`` / ``$WAYLAND_DISPLAY``) unless ``$BROWSER`` points at
something graphical; no display server almost always means no GUI browser.
On a headless Linux box ``webbrowser.open()`` often resolves to a text-mode browser that takes
over the terminal. Heuristics: a ``$BROWSER`` naming a console browser refuses; on Linux a
display server (``$DISPLAY`` / ``$WAYLAND_DISPLAY``) is required unless ``$BROWSER`` is set
(a console one already returned False, so a set ``$BROWSER`` here is graphical).
"""
import webbrowser as _webbrowser
def _names_console_browser(value: str) -> bool:
token = value.strip().split()[0] if value.strip() else ""
base = os.path.basename(token).lower()
return base in _CONSOLE_BROWSER_NAMES
browser_env = os.environ.get("BROWSER", "")
if browser_env and _names_console_browser(browser_env):
return False
if sys.platform.startswith("linux"):
has_display = bool(
os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY")
)
# An explicit graphical $BROWSER can work without $DISPLAY in odd
# setups, but a console $BROWSER already returned False above, so the
# only way to reach here with a $BROWSER set is a graphical one.
has_display = bool(os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY"))
if not has_display and not browser_env:
return False
try:
controller = _webbrowser.get()
controller = webbrowser.get()
except Exception:
# No browser resolvable at all → definitely don't auto-open.
return False
candidate = (
getattr(controller, "name", "")
or getattr(controller, "basename", "")
or ""
)
return False # No browser resolvable at all → definitely don't auto-open.
candidate = getattr(controller, "name", "") or getattr(controller, "basename", "") or ""
return not (candidate and _names_console_browser(candidate))
def _ssh_user_at_host() -> str:
"""Return best-effort 'user@hostname' for the SSH tunnel hint command.
Falls back to placeholder tokens when the values cannot be determined so the hint is always
syntactically valid even if not copy-pasteable.
"""
"""Best-effort 'user@hostname' for the SSH tunnel hint; placeholders keep it valid syntax."""
try:
import socket as _socket
hostname = _socket.gethostname() or "<this-host>"
@@ -138,11 +92,10 @@ def _ssh_user_at_host() -> str:
def _print_loopback_ssh_hint(redirect_uri: str, *, docs_url: str | None = None) -> None:
"""Print an SSH tunnel hint when running a loopback-redirect OAuth flow on a remote host. The auth
server (Spotify, MCP servers, ...) will redirect the user's browser to
``127.0.0.1:<port>/callback``. If the browser is on a different machine than the loopback
listener (the usual SSH case), the redirect can't reach the listener without a local port
forward.
"""Print an SSH tunnel hint when a loopback-redirect OAuth flow runs on a remote host.
The auth server redirects the browser to ``127.0.0.1:<port>/callback``; when the browser is
on another machine (the SSH case) the redirect needs a local port forward to reach us.
"""
from hermes_cli.auth import _is_remote_session
if not _is_remote_session():
@@ -177,10 +130,8 @@ def _print_loopback_ssh_hint(redirect_uri: str, *, docs_url: str | None = None)
def _default_verify() -> bool | ssl.SSLContext:
"""Platform-aware default SSL verify for httpx clients.
On macOS with Homebrew Python, the system OpenSSL cannot locate the system trust store and valid
public certs fail verification. When certifi is importable we pin its bundle explicitly;
elsewhere we defer to httpx's built-in default (certifi via its own dependency). Mirrors the
weixin fix in 3a0ec1d93.
On macOS with Homebrew Python the system OpenSSL cannot find the system trust store, so pin
certifi's bundle when importable; elsewhere defer to httpx's built-in default.
"""
if sys.platform == "darwin":
try:
@@ -192,35 +143,27 @@ def _default_verify() -> bool | ssl.SSLContext:
def _resolve_verify(
*,
insecure: Optional[bool] = None,
ca_bundle: Optional[str] = None,
*, insecure: Optional[bool] = None, ca_bundle: Optional[str] = None,
auth_state: Optional[Dict[str, Any]] = None,
) -> bool | ssl.SSLContext:
from hermes_cli.auth import _default_verify
tls_state = auth_state.get("tls") if isinstance(auth_state, dict) else {}
tls_state = tls_state if isinstance(tls_state, dict) else {}
effective_insecure = (
is_truthy_value(insecure, default=False) if insecure is not None
else is_truthy_value(tls_state.get("insecure", False), default=False)
)
effective_ca = (
ca_bundle
or tls_state.get("ca_bundle")
or os.getenv("HERMES_CA_BUNDLE")
or os.getenv("SSL_CERT_FILE")
or os.getenv("REQUESTS_CA_BUNDLE")
ca_bundle or tls_state.get("ca_bundle") or os.getenv("HERMES_CA_BUNDLE")
or os.getenv("SSL_CERT_FILE") or os.getenv("REQUESTS_CA_BUNDLE")
)
if effective_insecure:
return False
if effective_ca:
ca_path = str(effective_ca)
if not os.path.isfile(ca_path):
logger.warning(
"CA bundle path does not exist: %s — falling back to default certificates",
ca_path,
"CA bundle path does not exist: %s — falling back to default certificates", ca_path,
)
return _default_verify()
return ssl.create_default_context(cafile=ca_path)
@@ -228,22 +171,15 @@ def _resolve_verify(
def _request_device_code(
client: httpx.Client,
portal_base_url: str,
client_id: str,
scope: Optional[str],
client: httpx.Client, portal_base_url: str, client_id: str, scope: Optional[str],
) -> Dict[str, Any]:
"""POST to the device code endpoint. Returns device_code, user_code, etc."""
response = client.post(
f"{portal_base_url}/api/oauth/device/code",
data={
"client_id": client_id,
**({"scope": scope} if scope else {}),
},
data={"client_id": client_id, **({"scope": scope} if scope else {})},
)
response.raise_for_status()
data = response.json()
required_fields = [
"device_code", "user_code", "verification_uri",
"verification_uri_complete", "expires_in", "interval",
@@ -255,11 +191,7 @@ def _request_device_code(
def _nous_device_auth_timeout_message(portal_base_url: str) -> str:
"""Actionable timeout text for Nous device-code login failures.
A bare "timed out" gives the user nothing to act on; the usual cause is Portal sign-in failing
in the opened browser tab, so point at the Portal login page and the retry command.
"""
"""Actionable timeout text: the usual cause is Portal sign-in failing in the browser tab."""
portal = (portal_base_url or DEFAULT_NOUS_PORTAL_URL).rstrip("/")
return (
"Timed out waiting for device authorization.\n"
@@ -272,18 +204,13 @@ def _nous_device_auth_timeout_message(portal_base_url: str) -> str:
def _print_device_code_instructions(
verification_url: str,
user_code: str,
*,
open_browser: bool,
failure_dash: str = "--",
verification_url: str, user_code: str, *, open_browser: bool, failure_dash: str = "--",
swallow_open_errors: bool = False,
) -> None:
"""Print the shared "To continue" device-code block and optionally open the browser.
Callers decide *whether* to open (remote-session / graphical-browser gating differs per
provider); the wording of the fallback hint is parameterized so each provider keeps its
historical dash style.
provider); *failure_dash* keeps each provider's historical hint wording.
"""
print()
print("To continue:")
@@ -305,10 +232,7 @@ def _print_device_code_instructions(
def _poll_device_token_generic(
post: Callable[[], "httpx.Response"],
*,
expires_in: int,
poll_interval: int,
post: Callable[[], "httpx.Response"], *, expires_in: int, poll_interval: int,
validate_success: Callable[[Dict[str, Any]], None],
on_non_json_error: Callable[["httpx.Response"], Exception],
on_error: Callable[["httpx.Response", Dict[str, Any]], Exception],
@@ -317,8 +241,8 @@ def _poll_device_token_generic(
"""RFC 8628 device-code polling loop shared by the Nous and xAI flows.
``authorization_pending`` sleeps and retries; ``slow_down`` grows the interval by 1s (cap 30s).
Every other error, a non-JSON error body, and the deadline are turned into provider-specific
exceptions by the supplied factories so each caller keeps its exact error contract.
Every other error, a non-JSON error body, and the deadline become provider-specific exceptions
via the supplied factories so each caller keeps its exact error contract.
"""
deadline = time.monotonic() + max(1, expires_in)
current_interval = poll_interval
@@ -346,12 +270,8 @@ def _poll_device_token_generic(
def _poll_for_token(
client: httpx.Client,
portal_base_url: str,
client_id: str,
device_code: str,
expires_in: int,
poll_interval: int,
client: httpx.Client, portal_base_url: str, client_id: str, device_code: str,
expires_in: int, poll_interval: int,
) -> Dict[str, Any]:
"""Poll the Nous token endpoint until the user approves or the code expires."""
def _validate(payload: Dict[str, Any]) -> None:
@@ -367,8 +287,7 @@ def _poll_for_token(
lambda: client.post(
f"{portal_base_url}/api/oauth/token",
data={
"grant_type": DEVICE_CODE_GRANT_TYPE,
"client_id": client_id,
"grant_type": DEVICE_CODE_GRANT_TYPE, "client_id": client_id,
"device_code": device_code,
},
),
@@ -377,9 +296,8 @@ def _poll_for_token(
validate_success=_validate,
on_non_json_error=lambda _r: RuntimeError("Token endpoint returned a non-JSON error response"),
on_error=_error,
# Enriched at the SOURCE so every caller inherits the guidance:
# the CLI login (_nous_device_code_login) and the dashboard/desktop
# poller (web_server._nous_poller, which surfaces str(e) to the UI).
# Enriched at the SOURCE so the CLI login and the dashboard/desktop poller
# (web_server._nous_poller surfaces str(e) to the UI) both inherit the guidance.
on_timeout=lambda: TimeoutError(_nous_device_auth_timeout_message(portal_base_url)),
)
@@ -403,12 +321,8 @@ def _print_login_success(provider_id: str, config_path: Path, *, show_auth_state
def _offer_existing_oauth_credentials(
provider_id: str,
*,
resolve: Callable[[], Dict[str, Any]],
is_expiring: Callable[[str, int], bool],
display_name: str,
default_base_url: str,
provider_id: str, *, resolve: Callable[[], Dict[str, Any]],
is_expiring: Callable[[str, int], bool], display_name: str, default_base_url: str,
expired_notice: Optional[str] = None,
) -> bool:
"""Offer to reuse still-valid stored OAuth credentials. Returns True when the user accepted.

464
hermes_cli/auth_oauth_grants.py
View File

@@ -1,9 +1,8 @@
"""Single-use OAuth grant hygiene: strip cloned grants from profiles, heal forked grants.
Split out of ``hermes_cli/auth.py``; every moved name is re-imported there, so
``hermes_cli.auth.<name>`` keeps resolving (and monkeypatching) as before. 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``; 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).
"""
from __future__ import annotations
@@ -19,50 +18,46 @@ from hermes_cli.auth_constants import _decode_jwt_claims
logger = logging.getLogger("hermes_cli.auth")
# Pool providers whose OAuth refresh tokens are SINGLE-USE: redeeming the
# refresh token rotates the pair and revokes the old one. A grant forked into
# two auth.json files is therefore not two credentials but one credential with
# two owners — the first owner to refresh strands the other with
# ``invalid_grant`` / ``refresh_token_reused`` (#100339; same class as the
# ``providers.<id>`` write-through hazard in #48415 / #43589). Profiles must
# never receive a copy of these grants: ONE grant lives at the global root and
# named profiles read it through the ``read_credential_pool`` root fallback.
SINGLE_USE_REFRESH_POOL_PROVIDERS = frozenset({
"anthropic",
"openai-codex",
"xai-oauth",
})
# Pool providers whose OAuth refresh tokens are SINGLE-USE: redeeming rotates the pair and
# revokes the old one, so a grant forked into two auth.json files is ONE credential with two
# owners — the first to refresh strands the other with ``invalid_grant`` /
# ``refresh_token_reused``.
# Profiles must never receive a copy: ONE grant lives at the global root and named profiles read
# it through the ``read_credential_pool`` root fallback.
SINGLE_USE_REFRESH_POOL_PROVIDERS = frozenset({"anthropic", "openai-codex", "xai-oauth"})
# Singleton credential files that hold the same single-use grants outside
# ``auth.json``. Copying one into a profile re-seeds a forked pool row on the
# profile's next ``load_pool()``.
# Singleton credential files holding the same single-use grants outside ``auth.json``. Copying one
# into a profile re-seeds a forked pool row on the profile's next ``load_pool()``.
SINGLE_USE_OAUTH_SINGLETON_FILES = (".anthropic_oauth.json",)
# Providers whose device-code grants live under ``providers.<id>`` (not only the pool).
_DEVICE_CODE_BLOCK_PROVIDERS = ("openai-codex", "xai-oauth")
def _is_oauth_pool_payload(entry: Any) -> bool:
if not isinstance(entry, dict):
return False
auth_type = str(entry.get("auth_type") or "").strip().lower()
if auth_type == "oauth":
return True
# Legacy rows predating ``auth_type``: an Anthropic OAuth access token or
# any row carrying a refresh token is an OAuth grant.
if str(entry.get("refresh_token") or "").strip():
return True
return str(entry.get("access_token") or "").startswith("sk-ant-oat")
# Legacy rows predating ``auth_type``: an Anthropic OAuth access token or any row carrying a
# refresh token is an OAuth grant.
return (
str(entry.get("auth_type") or "").strip().lower() == "oauth"
or bool(str(entry.get("refresh_token") or "").strip())
or str(entry.get("access_token") or "").startswith("sk-ant-oat")
)
def _is_pkce_row(row: Dict[str, Any]) -> bool:
return str(row.get("source") or "").endswith("hermes_pkce")
def strip_cloned_single_use_oauth_grants(profile_dir: Path) -> Dict[str, Any]:
"""Remove forked single-use OAuth grants from a freshly cloned profile.
Called after any code path that copies credential files from one profile into another (``hermes
profile create --clone-all``, the dashboard/TUI ``mirror_credentials`` flow). API-key pool rows
are kept — a static key is safe to duplicate.
Returns a summary ``{"pool": [...provider ids], "providers": [...], "files": [...]}`` of what
was stripped (empty lists when nothing was). Never raises: a clone must not fail because
credential hygiene could not run — the caller logs the summary.
Called after any path that copies credential files between profiles (``hermes profile create
--clone-all``, the dashboard/TUI ``mirror_credentials`` flow). API-key pool rows are kept — a
static key is safe to duplicate. Returns ``{"pool": [...provider ids], "providers": [...],
"files": [...]}`` of what was stripped. Never raises: a clone must not fail because hygiene
could not run — the caller logs the summary.
"""
from hermes_cli.auth import _save_auth_store
stripped: Dict[str, Any] = {"pool": [], "providers": [], "files": []}
@@ -102,15 +97,13 @@ def strip_cloned_single_use_oauth_grants(profile_dir: Path) -> Dict[str, Any]:
if kept:
pool[provider_id] = kept
else:
# No local rows at all → read_credential_pool falls back
# to the root slice for this provider.
# No local rows → read_credential_pool falls back to the root slice.
del pool[provider_id]
providers = store.get("providers")
if isinstance(providers, dict):
# Device-code grants for these providers live under providers.<id>;
# _load_provider_state has the same root fallback, so dropping the
# copy keeps the profile working while removing the fork.
for provider_id in ("openai-codex", "xai-oauth"):
# _load_provider_state has the same root fallback, so dropping the copy keeps the
# profile working while removing the fork.
for provider_id in _DEVICE_CODE_BLOCK_PROVIDERS:
block = providers.get(provider_id)
if isinstance(block, dict) and block:
del providers[provider_id]
@@ -122,35 +115,26 @@ def strip_cloned_single_use_oauth_grants(profile_dir: Path) -> Dict[str, Any]:
_save_auth_store(store, target_path=auth_path)
except Exception:
logger.debug(
"Failed to strip cloned single-use OAuth grants from %s",
auth_path,
exc_info=True,
"Failed to strip cloned single-use OAuth grants from %s", auth_path, exc_info=True,
)
return stripped
_OAUTH_TOKEN_FIELDS = (
"access_token",
"refresh_token",
"expires_at",
"expires_at_ms",
"last_refresh",
"access_token", "refresh_token", "expires_at", "expires_at_ms", "last_refresh",
)
_oauth_heal_notices: List[str] = []
# provider -> (profile auth.json path, auth.json mtime_ns, singleton mtime_ns)
# of the last store verified fork-free; lets load_pool() skip the locked scan.
# provider -> (profile auth.json path, auth.json mtime_ns, singleton mtime_ns) of the last store
# verified fork-free; lets load_pool() skip the locked scan.
_oauth_heal_clean_marks: Dict[str, Tuple[str, Optional[int], Optional[int]]] = {}
def consume_oauth_heal_notices() -> List[str]:
"""Return (and clear) human-readable notes about heals run in this process.
``hermes auth list`` / ``hermes auth status`` print them so the user sees that a forked grant
was consolidated rather than only finding it in logs.
``hermes auth list`` / ``hermes auth status`` print them so the user sees the consolidation.
"""
from hermes_cli.auth import _oauth_heal_notices
notes = list(_oauth_heal_notices)
@@ -162,8 +146,7 @@ def _oauth_identity(entry: Dict[str, Any]) -> Optional[str]:
"""Stable account identity for an OAuth row when the token carries one.
Codex / xAI access tokens are JWTs with ``sub`` / ``email`` / ``chatgpt_account_id`` claims;
Anthropic ``sk-ant-oat`` tokens carry none (returns None, so lineage rests on id / token
material).
Anthropic ``sk-ant-oat`` tokens carry none (None → lineage rests on id / token material).
"""
from hermes_cli.auth import _nonempty_str
if not isinstance(entry, dict):
@@ -184,8 +167,7 @@ def _oauth_freshness(entry: Dict[str, Any]) -> float:
"""Best-effort 'how recently was this pair issued' score (epoch seconds).
A rotation always issues a later-expiring access token, so ``expires_at`` ordering identifies
the live copy; ``last_refresh`` and the JWT ``exp`` claim are fallbacks for rows that do not
persist expiry.
the live copy; ``last_refresh`` and the JWT ``exp`` claim are fallbacks.
"""
from agent.credential_pool import _parse_absolute_timestamp
@@ -208,8 +190,8 @@ def _find_root_counterpart(
"""Index of the root OAuth row that shares a grant lineage with *profile_row*.
Fallback per the one-grant-at-root rule: same provider + same OAuth client — every Anthropic
``hermes_pkce`` grant uses one client id and carries no claims, so two Anthropic OAuth rows with
no contrary identity are one lineage.
``hermes_pkce`` grant uses one client id and carries no claims, so two Anthropic OAuth rows
with no contrary identity are one lineage.
"""
from hermes_cli.auth import _nonempty_str
candidates = [i for i, r in enumerate(root_rows) if _is_oauth_pool_payload(r)]
@@ -231,8 +213,8 @@ def _find_root_counterpart(
for i in candidates:
if root_rows[i].get(key) == p_val:
return i
# Fallback: same provider + same client. Only a contradicting identity
# (both sides carry claims and they differ from every root row) blocks it.
# Fallback: same provider + same client. Only a contradicting identity (both sides carry
# claims and they differ from every root row) blocks it.
if p_ident:
for i in candidates:
if not _oauth_identity(root_rows[i]):
@@ -270,24 +252,17 @@ def _singleton_as_row(path: Path) -> Optional[Dict[str, Any]]:
}
# ── One-time heal for installs that ALREADY forked a single-use grant ────────
#
# Fleets created before the clone-strip (``strip_cloned_single_use_oauth_grants``) / root-write-through have
# profile-local copies of the root grant. Those copies are the same credential
# with several owners: whichever profile rotated last holds the only live
# refresh token and every other copy (root included) is spent. Upgrading alone
# does not fix that — the first load in each profile would keep using its own
# doomed copy. ``heal_forked_single_use_oauth_grants`` runs at profile
# ``load_pool()`` time: it finds the profile rows that share LINEAGE with a
# root row (same pool id — clone-all and the old borrowed-persist both kept
# it — or the same account identity / token material), keeps the copy most
# likely to still be live (freshest rotation), writes that copy into ROOT when
# root's is older, and strips the profile's copy so the profile borrows root
# from then on. Idempotent (a healed profile has no matched rows), never
# touches API-key rows, never deletes a row that has no root counterpart
# (an independent ``hermes -p <p> auth add`` grant, or the only surviving
# copy), and reads only the two auth.json files the existing root fallback
# already reads — no environ / secret-scope reads.
# One-time heal for installs that ALREADY forked a single-use grant (fleets created before the
# clone-strip / root-write-through). Those copies are one credential with several owners:
# whichever profile rotated last holds the only live refresh token and every other copy (root
# included) is spent; upgrading alone would keep each profile on its own doomed copy.
# ``heal_forked_single_use_oauth_grants`` runs at profile ``load_pool()`` time: it finds profile
# rows that share LINEAGE with a root row (same pool id, or same account identity / token
# material), keeps the copy most likely live (freshest rotation), writes it into ROOT when root's
# is older, and strips the profile's copy so the profile borrows root from then on. Idempotent,
# never touches API-key rows, never deletes a row with no root counterpart (an independent
# ``hermes -p <p> auth add`` grant, or the only surviving copy), and reads only the two auth.json
# files the existing root fallback already reads — no environ / secret-scope reads.
def heal_forked_single_use_oauth_grants(provider_id: str) -> Optional[Dict[str, Any]]:
@@ -338,181 +313,216 @@ def _heal_forked_provider_block(
return adopted
def _mtime_ns(p: Optional[Path]) -> Optional[int]:
try:
return p.stat().st_mtime_ns if p is not None else None
except OSError:
return None
def _pool_rows(store: Dict[str, Any], provider_id: str) -> Tuple[Any, List[Any]]:
"""``(store["credential_pool"], its provider_id rows-or-[])`` — pool may be None/non-dict."""
pool = store.get("credential_pool")
rows = pool.get(provider_id) if isinstance(pool, dict) else None
return pool, rows if isinstance(rows, list) else []
def _adopt_if_fresher(
target: Dict[str, Any], candidate: Dict[str, Any],
) -> Optional[Dict[str, Any]]:
"""*target* carrying *candidate*'s pair when the candidate rotated later, else None."""
if _oauth_freshness(candidate) > _oauth_freshness(target):
return _adopt_oauth_material(target, candidate)
return None
class _HealPass:
"""One heal pass over a profile store vs the root store for *provider_id*."""
def __init__(self, profile_store, root_store, provider_id, root_singleton: Optional[Path]):
self.profile_store, self.root_store = profile_store, root_store
self.provider_id = provider_id
self.summary: Dict[str, Any] = {
"adopted": False, "stripped_ids": [], "files": [], "providers_block": False,
}
self.profile_changed = self.root_changed = False
self.p_pool, self.p_rows = _pool_rows(profile_store, provider_id)
self.r_pool, self.r_rows = _pool_rows(root_store, provider_id)
self.r_oauth = [r for r in self.r_rows if _is_oauth_pool_payload(r)]
self.root_singleton = root_singleton
self.root_singleton_row = (
_singleton_as_row(root_singleton)
if root_singleton is not None and root_singleton.exists() else None
)
def _adopt_root_row(self, idx: int, row: Dict[str, Any]) -> None:
merged = _adopt_if_fresher(self.r_rows[idx], row)
if merged is not None:
self.r_rows[idx] = merged
self.root_changed = self.summary["adopted"] = True
def _adopt_root_singleton(self, row: Dict[str, Any]) -> None:
merged = _adopt_if_fresher(self.root_singleton_row, row)
if merged is not None:
self.root_singleton_row = merged
self.summary["adopted"] = True
def heal_pool_rows(self) -> None:
kept_rows: List[Any] = []
for row in self.p_rows:
if not _is_oauth_pool_payload(row):
kept_rows.append(row) # API keys are safe to duplicate
continue
match_idx = _find_root_counterpart(row, self.r_rows)
if match_idx is not None:
self._adopt_root_row(match_idx, row)
# No root pool counterpart. Root's grant may live only in its .anthropic_oauth.json
# (the ``hermes auth`` PKCE shape); a profile hermes_pkce-family row is its copy.
elif _is_pkce_row(row) and self.root_singleton_row is not None and not self.r_oauth:
self._adopt_root_singleton(row)
else:
# Root holds no copy of this lineage (independent account, or root never had the
# grant): the profile's row may be the only surviving copy — leave it alone.
kept_rows.append(row)
continue
self.summary["stripped_ids"].append(row.get("id"))
self.profile_changed = True
if self.profile_changed and isinstance(self.p_pool, dict):
if kept_rows:
self.p_pool[self.provider_id] = kept_rows
else:
self.p_pool.pop(self.provider_id, None)
def heal_provider_block(self) -> None:
if self.provider_id not in _DEVICE_CODE_BLOCK_PROVIDERS:
return
block_result = _heal_forked_provider_block(
self.profile_store, self.root_store, self.provider_id,
)
if block_result is not None:
self.profile_changed = self.summary["providers_block"] = True
if block_result:
self.root_changed = self.summary["adopted"] = True
def heal_profile_singleton(self, profile_singleton: Optional[Path]) -> None:
if profile_singleton is None or not profile_singleton.exists():
return
p_single = _singleton_as_row(profile_singleton)
root_has_grant = bool(self.r_oauth) or self.root_singleton_row is not None
# Otherwise root has NO grant for this provider (or the file is not a grant): the
# profile's singleton may be the only surviving copy — never delete it.
if p_single is None or not root_has_grant:
return
if self.root_singleton_row is not None:
self._adopt_root_singleton(p_single)
else:
# Root only has pool rows: fold the singleton's pair into the freshest-matching
# root pkce row, if any.
idx = next(
(i for i, r in enumerate(self.r_rows)
if _is_oauth_pool_payload(r) and _is_pkce_row(r)),
None,
)
if idx is not None:
self._adopt_root_row(idx, p_single)
try:
profile_singleton.unlink()
self.summary["files"].append(profile_singleton.name)
except OSError:
logger.debug("could not remove %s", profile_singleton, exc_info=True)
def sync_root_singleton_with_pkce_row(self) -> None:
"""Keep root's singleton and its ``hermes_pkce``-seeded pool row in step.
Root's next load_pool() re-seeds that row FROM the singleton file, so a stale file would
resurrect the spent pair (and a stale row would be overwritten by a fresh file).
"""
if not (
self.summary["adopted"] and self.root_singleton is not None
and self.root_singleton_row is not None
):
return
pkce_idx = next(
(i for i, r in enumerate(self.r_rows)
if _is_oauth_pool_payload(r) and r.get("source") == "hermes_pkce"),
None,
)
if pkce_idx is None:
return
pkce_row = self.r_rows[pkce_idx]
if _oauth_freshness(pkce_row) > _oauth_freshness(self.root_singleton_row):
self.root_singleton_row = _adopt_oauth_material(self.root_singleton_row, pkce_row)
elif _oauth_freshness(self.root_singleton_row) > _oauth_freshness(pkce_row):
self.r_rows[pkce_idx] = _adopt_oauth_material(pkce_row, self.root_singleton_row)
self.root_changed = True
@property
def dirty(self) -> bool:
return bool(self.profile_changed or self.root_changed or self.summary["adopted"])
def _heal_forked_single_use_oauth_grants(provider_id: str) -> Optional[Dict[str, Any]]:
from hermes_cli.auth import _auth_file_path, _auth_store_lock, _global_auth_file_path, _load_auth_store, _oauth_heal_clean_marks, _oauth_heal_notices, _same_path, _save_auth_store
from hermes_cli.auth import (
_auth_file_path, _auth_store_lock, _global_auth_file_path, _load_auth_store,
_oauth_heal_clean_marks, _oauth_heal_notices, _same_path, _save_auth_store,
)
root_path = _global_auth_file_path()
if root_path is None:
return None # classic mode: nothing to consolidate into
if os.environ.get("PYTEST_CURRENT_TEST"):
# Same seat belt as the write-through paths: never touch the real
# user's ~/.hermes/auth.json from a test that forgot to isolate HOME.
# Same seat belt as the write-through paths: never touch the real user's
# ~/.hermes/auth.json from a test that forgot to isolate HOME.
real_home_env = os.environ.get("HOME", "")
if real_home_env and _same_path(root_path, Path(real_home_env) / ".hermes" / "auth.json"):
return None
profile_path = _auth_file_path()
profile_home = profile_path.parent
root_home = root_path.parent
profile_singleton = profile_home / ".anthropic_oauth.json" if provider_id == "anthropic" else None
is_anthropic = provider_id == "anthropic"
profile_singleton = profile_home / ".anthropic_oauth.json" if is_anthropic else None
root_singleton = root_path.parent / ".anthropic_oauth.json" if is_anthropic else None
# Hot-path short-circuit: load_pool() runs per model call. Once this
# profile's store was verified clean for *provider_id*, skip the locked
# read-modify-write until the profile's own files change (mtime key).
def _stamp(p: Optional[Path]) -> Optional[int]:
try:
return p.stat().st_mtime_ns if p is not None else None
except OSError:
return None
fingerprint = (str(profile_path), _stamp(profile_path), _stamp(profile_singleton))
# Hot-path short-circuit: load_pool() runs per model call. Once this profile's store was
# verified clean for *provider_id*, skip the locked read-modify-write until the profile's own
# files change (mtime key).
fingerprint = (str(profile_path), _mtime_ns(profile_path), _mtime_ns(profile_singleton))
if _oauth_heal_clean_marks.get(provider_id) == fingerprint:
return None
if fingerprint[1] is None and fingerprint[2] is None:
_oauth_heal_clean_marks[provider_id] = fingerprint
return None
summary: Dict[str, Any] = {"adopted": False, "stripped_ids": [], "files": [], "providers_block": False}
log_bits: List[str] = []
# Lock order: active (profile) store first, then the root source store —
# the same order ``_provider_state_transaction`` uses.
# Lock order: active (profile) store first, then the root source store — the same order
# ``_provider_state_transaction`` uses.
with _auth_store_lock():
profile_store = _load_auth_store(profile_path) if profile_path.exists() else {"providers": {}}
with _auth_store_lock(target_path=root_path):
root_store = _load_auth_store(root_path) if root_path.exists() else {"providers": {}}
profile_changed = False
root_changed = False
p_pool = profile_store.get("credential_pool")
p_rows = p_pool.get(provider_id) if isinstance(p_pool, dict) else None
p_rows = p_rows if isinstance(p_rows, list) else []
r_pool = root_store.get("credential_pool")
r_rows = r_pool.get(provider_id) if isinstance(r_pool, dict) else None
r_rows = r_rows if isinstance(r_rows, list) else []
r_oauth = [r for r in r_rows if _is_oauth_pool_payload(r)]
root_singleton = root_home / ".anthropic_oauth.json" if provider_id == "anthropic" else None
root_singleton_row = (
_singleton_as_row(root_singleton)
if root_singleton is not None and root_singleton.exists() else None
)
# ── credential_pool rows ────────────────────────────────────
kept_rows: List[Any] = []
for row in p_rows:
if not _is_oauth_pool_payload(row):
kept_rows.append(row) # API keys are safe to duplicate
continue
match_idx = _find_root_counterpart(row, r_rows)
if match_idx is not None:
root_row = r_rows[match_idx]
if _oauth_freshness(row) > _oauth_freshness(root_row):
r_rows[match_idx] = _adopt_oauth_material(root_row, row)
root_changed = True
summary["adopted"] = True
summary["stripped_ids"].append(row.get("id"))
profile_changed = True
continue
# No root pool counterpart. Root's grant may live only in its
# .anthropic_oauth.json (the ``hermes auth`` PKCE shape); a
# profile hermes_pkce-family row is that grant's copy.
is_pkce = str(row.get("source") or "").endswith("hermes_pkce")
if is_pkce and root_singleton_row is not None and not r_oauth:
if _oauth_freshness(row) > _oauth_freshness(root_singleton_row):
root_singleton_row = _adopt_oauth_material(root_singleton_row, row)
summary["adopted"] = True
summary["stripped_ids"].append(row.get("id"))
profile_changed = True
continue
# Root holds no copy of this lineage (independent account, or
# root never had the grant): the profile's row may be the
# only surviving copy — leave it alone.
kept_rows.append(row)
if profile_changed and isinstance(p_pool, dict):
if kept_rows:
p_pool[provider_id] = kept_rows
else:
p_pool.pop(provider_id, None)
# ── providers.<id> device-code blocks (Codex / xAI) ─────────
if provider_id in ("openai-codex", "xai-oauth"):
block_result = _heal_forked_provider_block(profile_store, root_store, provider_id)
if block_result is not None:
profile_changed = True
summary["providers_block"] = True
if block_result:
root_changed = True
summary["adopted"] = True
# ── profile-local .anthropic_oauth.json singleton ───────────
if profile_singleton is not None and profile_singleton.exists():
p_single = _singleton_as_row(profile_singleton)
root_has_grant = bool(r_oauth) or root_singleton_row is not None
if p_single is not None and root_has_grant:
if root_singleton_row is not None:
if _oauth_freshness(p_single) > _oauth_freshness(root_singleton_row):
root_singleton_row = _adopt_oauth_material(root_singleton_row, p_single)
summary["adopted"] = True
else:
# Root only has pool rows: fold the singleton's pair
# into the freshest-matching root pkce row, if any.
idx = next(
(i for i, r in enumerate(r_rows)
if _is_oauth_pool_payload(r)
and str(r.get("source") or "").endswith("hermes_pkce")),
None,
)
if idx is not None and _oauth_freshness(p_single) > _oauth_freshness(r_rows[idx]):
r_rows[idx] = _adopt_oauth_material(r_rows[idx], p_single)
root_changed = True
summary["adopted"] = True
try:
profile_singleton.unlink()
summary["files"].append(profile_singleton.name)
except OSError:
logger.debug("could not remove %s", profile_singleton, exc_info=True)
# Otherwise root has NO grant for this provider (or the file
# is not a grant): the profile's singleton may be the only
# surviving copy — never delete it.
if not (profile_changed or root_changed or summary["adopted"]):
run = _HealPass(profile_store, root_store, provider_id, root_singleton)
run.heal_pool_rows()
run.heal_provider_block()
run.heal_profile_singleton(profile_singleton)
if not run.dirty:
_oauth_heal_clean_marks[provider_id] = fingerprint
return None
if summary["adopted"] and root_singleton is not None and root_singleton_row is not None:
# Keep root's singleton and its ``hermes_pkce``-seeded pool row
# in step: root's next load_pool() re-seeds that row FROM the
# singleton file, so a stale file would resurrect the spent
# pair (and a stale row would be overwritten by a fresh file).
pkce_idx = next(
(i for i, r in enumerate(r_rows)
if _is_oauth_pool_payload(r) and r.get("source") == "hermes_pkce"),
None,
)
if pkce_idx is not None:
pkce_row = r_rows[pkce_idx]
if _oauth_freshness(pkce_row) > _oauth_freshness(root_singleton_row):
root_singleton_row = _adopt_oauth_material(root_singleton_row, pkce_row)
elif _oauth_freshness(root_singleton_row) > _oauth_freshness(pkce_row):
r_rows[pkce_idx] = _adopt_oauth_material(pkce_row, root_singleton_row)
root_changed = True
if root_changed:
if isinstance(r_pool, dict):
r_pool[provider_id] = r_rows
run.sync_root_singleton_with_pkce_row()
summary = run.summary
if run.root_changed:
if isinstance(run.r_pool, dict):
run.r_pool[provider_id] = run.r_rows
else:
root_store["credential_pool"] = {provider_id: r_rows}
root_store["credential_pool"] = {provider_id: run.r_rows}
_save_auth_store(root_store, target_path=root_path)
if summary["adopted"] and root_singleton is not None and root_singleton_row is not None:
singleton_row = run.root_singleton_row
if summary["adopted"] and root_singleton is not None and singleton_row is not None:
from agent.anthropic_credentials import _write_hermes_oauth_credentials
_write_hermes_oauth_credentials(
root_singleton_row.get("access_token") or "",
root_singleton_row.get("refresh_token"),
root_singleton_row.get("expires_at_ms"),
target=root_singleton,
singleton_row.get("access_token") or "", singleton_row.get("refresh_token"),
singleton_row.get("expires_at_ms"), target=root_singleton,
)
if profile_changed and profile_path.exists():
if run.profile_changed and profile_path.exists():
_save_auth_store(profile_store, target_path=profile_path)
log_bits: List[str] = []
if summary["stripped_ids"]:
log_bits.append(f"pool rows {summary['stripped_ids']}")
if summary["providers_block"]: