From 40a09152f1335ef5ad3ec5cba35d087016cf90ce Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:09:37 -0700 Subject: [PATCH] refactor(hermes_cli): compact auth_device_flow + split forked-OAuth heal into _HealPass phases --- hermes_cli/auth_device_flow.py | 214 +++++---------- hermes_cli/auth_oauth_grants.py | 464 ++++++++++++++++---------------- 2 files changed, 301 insertions(+), 377 deletions(-) diff --git a/hermes_cli/auth_device_flow.py b/hermes_cli/auth_device_flow.py index 95bc69c12b..d0dd4e7ad4 100644 --- a/hermes_cli/auth_device_flow.py +++ b/hermes_cli/auth_device_flow.py @@ -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.`` keeps resolving (and monkeypatching) as before. Origin-internal -helpers are imported lazily inside each function (no import cycle; patches on -``hermes_cli.auth.`` still intercept). +Split out of ``hermes_cli/auth.py``; every name is re-exported there so ``hermes_cli.auth.`` +keeps resolving (and monkeypatching). Origin-internal helpers are imported lazily inside each +function (no import cycle; patches on ``hermes_cli.auth.`` 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 "" @@ -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:/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:/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. diff --git a/hermes_cli/auth_oauth_grants.py b/hermes_cli/auth_oauth_grants.py index 7005a122a2..6ecec5ced7 100644 --- a/hermes_cli/auth_oauth_grants.py +++ b/hermes_cli/auth_oauth_grants.py @@ -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.`` keeps resolving (and monkeypatching) as before. Origin-internal -helpers are imported lazily inside each function (no import cycle; patches on -``hermes_cli.auth.`` still intercept). +Split out of ``hermes_cli/auth.py``; every name is re-exported there so ``hermes_cli.auth.`` +keeps resolving (and monkeypatching). Origin-internal helpers are imported lazily inside each +function (no import cycle; patches on ``hermes_cli.auth.`` 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.`` 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.`` (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.; - # _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

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

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. 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"]: