"""Anthropic credential sources, OAuth flows, and token resolution. Extracted from ``agent/anthropic_adapter.py``: the adapter is a message/HTTP translation layer, while everything below owns *where an Anthropic credential comes from* and *how a rotated one is committed*. Keeping the two apart means the refresh transaction has a single home instead of being interleaved with request building. Sources, in the order ``resolve_anthropic_token()`` consults them: 1. ``ANTHROPIC_TOKEN`` / ``CLAUDE_CODE_OAUTH_TOKEN`` (explicit OAuth env) 2. ``ANTHROPIC_API_KEY`` (explicit API key) 3. ``~/.hermes/.anthropic_oauth.json`` (Hermes PKCE login) 4. ``~/.claude/.credentials.json`` / macOS Keychain (Claude Code) 5. the credential pool in ``auth.json`` Sources 3 and 4 are *singletons*: ``credential_pool._seed_from_singletons()`` re-reads them on every ``load_pool()`` and writes what it finds over the pool row, which is why a failed write here is a failed refresh (see ``CredentialPersistError``) rather than a best-effort cache miss. ``agent.anthropic_adapter`` re-exports every public name below, so existing ``from agent.anthropic_adapter import resolve_anthropic_token`` imports keep working. """ import json import logging import os import platform import secrets import stat import subprocess import threading from collections import OrderedDict from pathlib import Path from typing import Any, Dict, Optional from hermes_constants import get_hermes_home from agent.secret_scope import get_secret as _get_secret logger = logging.getLogger(__name__) def _getenv(name: str, default: str = "") -> str: """Profile-scoped replacement for os.getenv on credential reads. Routes through the secret scope (Workstream A): identical to os.getenv when multiplexing is off, scope-aware (and fail-closed on an unscoped read) when on. Mirrors the same wrapper in hermes_cli/runtime_provider.py. """ val = _get_secret(name, default) return val if val is not None else default def _is_oauth_token(key: str) -> bool: """Check if the key is an Anthropic OAuth/setup token. Positively identifies Anthropic OAuth tokens by their key format: - ``sk-ant-`` prefix (but NOT ``sk-ant-api``) → setup tokens, managed keys - ``eyJ`` prefix → JWTs from the Anthropic OAuth flow - ``cc-`` prefix → Claude Code OAuth access tokens (from CLAUDE_CODE_OAUTH_TOKEN) Non-Anthropic keys (MiniMax, Alibaba, etc.) don't match any pattern and correctly return False. """ if not key: return False # Regular Anthropic Console API keys — x-api-key auth, never OAuth if key.startswith("sk-ant-api"): return False # Anthropic-issued tokens (setup-tokens sk-ant-oat-*, managed keys) if key.startswith("sk-ant-"): return True # JWTs from Anthropic OAuth flow if key.startswith("eyJ"): return True # Claude Code OAuth access tokens (opaque, from CLAUDE_CODE_OAUTH_TOKEN) if key.startswith("cc-"): return True return False class CredentialPersistError(RuntimeError): """A rotated single-use credential could not be durably committed. Anthropic OAuth refresh tokens are single-use: a successful refresh POST consumes the old refresh token server-side and returns a replacement. The replacement exists only in memory until it reaches its authoritative on-disk store (``~/.claude/.credentials.json`` for ``claude_code``, ``~/.hermes/.anthropic_oauth.json`` for ``hermes_pkce``). If that write fails and the caller reports success anyway, the on-disk (already consumed) pair survives and is re-seeded on the next ``load_pool()``, so the following refresh replays a spent token and fails with ``invalid_grant`` / ``refresh_token_reused``. Callers must therefore treat this as a failed refresh, not a successful one, and fail closed. """ def __init__(self, path: Any, cause: BaseException) -> None: super().__init__( f"failed to durably persist rotated Anthropic credentials to {path}: {cause}" ) self.path = path # Fingerprints of Anthropic secrets whose refresh POST succeeded (so the # server-side pair was rotated and the old refresh token is spent) but whose # replacement never reached its authoritative store. The pre-rotation pair # survives on disk and is re-seeded on the next ``load_pool()``, so without an # explicit verdict the resolver happily hands that already-consumed credential # back from a later source and the caller reads a silent success. # # Kept as non-reversible digests and bounded: a spent secret is spent forever, # so entries never need clearing (a re-auth mints new tokens with new # fingerprints). # # The registry has TWO scopes, because the credential it protects does: # * process-local (this OrderedDict) — fast path, always recorded; # * durable sidecar file next to the shared credential source — the # authority boundary of ``claude_code``/``hermes_pkce`` is the shared # singleton file, which other Hermes processes/profiles read with fresh # interpreters. A process-local verdict only stops the process that # lost the commit from lying to itself; the sidecar stops every OTHER # process from leasing the stale pair or re-POSTing the spent refresh # token. The sidecar stores only one-way fingerprints (never secrets) # and is written under the same path-keyed cross-process lock that # serializes refreshes of that source. _SPENT_ROTATION_LOCK = threading.Lock() _SPENT_ROTATION_FINGERPRINTS: "OrderedDict[str, None]" = OrderedDict() _SPENT_ROTATION_MAX_TRACKED = 64 _SPENT_ROTATION_SIDECAR_VERSION = 1 def _spent_rotation_sidecar_path(source_path: Path) -> Path: """Sidecar registry path for a shared credential source file.""" return source_path.with_name(source_path.name + ".hermes-spent-rotations.json") def spent_rotation_source_path(source: Any) -> Optional[Path]: """Map a pool-entry source to the shared singleton file it borrows from. Only singleton-backed sources have a cross-process authority boundary; profile-owned rows are already protected by the process-local registry plus the pool quarantine. """ if source == "claude_code": return claude_code_credentials_path() if source == "hermes_pkce": return _get_hermes_oauth_file() return None def _read_spent_rotation_sidecar(source_path: Optional[Path]) -> set: if source_path is None: return set() try: raw = json.loads( _spent_rotation_sidecar_path(source_path).read_text(encoding="utf-8") ) except (OSError, ValueError): return set() fingerprints = raw.get("fingerprints") if isinstance(raw, dict) else None if not isinstance(fingerprints, list): return set() return {fp for fp in fingerprints if isinstance(fp, str) and fp} def _append_spent_rotation_sidecar(source_path: Path, fingerprints: list) -> None: """Merge fingerprints into the sidecar registry (atomic replace). Callers on the refresh path already hold the path-keyed cross-process lock for ``source_path``, so concurrent merge-writes are serialized. Fail-soft: a sidecar write failure must never mask the fail-closed verdict already recorded in the process-local registry. """ sidecar = _spent_rotation_sidecar_path(source_path) try: merged = _read_spent_rotation_sidecar(source_path) merged.update(fingerprints) bounded = sorted(merged)[-_SPENT_ROTATION_MAX_TRACKED * 4 :] payload = json.dumps( { "version": _SPENT_ROTATION_SIDECAR_VERSION, "comment": ( "Non-secret one-way fingerprints of Anthropic OAuth " "credentials whose rotation was consumed server-side but " "never durably committed. Written by Hermes so sibling " "processes sharing this credential source fail closed " "instead of replaying a spent single-use refresh token." ), "fingerprints": bounded, }, indent=2, ) sidecar.parent.mkdir(parents=True, exist_ok=True) tmp = sidecar.with_name(sidecar.name + ".tmp") tmp.write_text(payload, encoding="utf-8") os.replace(tmp, sidecar) except Exception: logger.debug( "Failed to persist spent-rotation fingerprints to %s", sidecar, exc_info=True, ) def mark_rotation_consumed_uncommitted( *secrets: Any, source_path: Optional[Path] = None ) -> None: """Record secrets consumed by a refresh whose replacement never committed. Called from every commit-failure path (the direct resolver here and ``CredentialPool._fail_closed_unpersisted_rotation``). Recording the *pre-rotation* pair is what lets later resolution steps recognise the stale copy they read back off disk as unusable rather than as a working token. When ``source_path`` names the shared singleton file the credential was borrowed from, the verdict is additionally persisted to that source's sidecar registry so other processes/profiles sharing the file adopt it too. """ from agent.credential_persistence import fingerprint_secret_value recorded: list = [] with _SPENT_ROTATION_LOCK: for secret in secrets: value = str(secret or "").strip() if not value: continue fingerprint = fingerprint_secret_value(value) if not fingerprint: continue recorded.append(fingerprint) _SPENT_ROTATION_FINGERPRINTS.pop(fingerprint, None) _SPENT_ROTATION_FINGERPRINTS[fingerprint] = None while len(_SPENT_ROTATION_FINGERPRINTS) > _SPENT_ROTATION_MAX_TRACKED: _SPENT_ROTATION_FINGERPRINTS.popitem(last=False) if recorded and source_path is not None: _append_spent_rotation_sidecar(source_path, recorded) def is_rotation_consumed_uncommitted( secret: Any, *, source_path: Optional[Path] = None ) -> bool: """True when *secret* belongs to a rotation that was spent but not committed. Checks the process-local registry first, then (when ``source_path`` is given) the durable sidecar registry of the shared credential source, so a fresh interpreter in another process still sees the terminal verdict. """ from agent.credential_persistence import fingerprint_secret_value value = str(secret or "").strip() if not value: return False fingerprint = fingerprint_secret_value(value) if not fingerprint: return False with _SPENT_ROTATION_LOCK: if fingerprint in _SPENT_ROTATION_FINGERPRINTS: return True return fingerprint in _read_spent_rotation_sidecar(source_path) def _read_claude_code_credentials_from_keychain() -> Optional[Dict[str, Any]]: """Read Claude Code OAuth credentials from the macOS Keychain. Claude Code >=2.1.114 stores credentials in the macOS Keychain under the service name "Claude Code-credentials" rather than (or in addition to) the JSON file at ~/.claude/.credentials.json. The password field contains a JSON string with the same claudeAiOauth structure as the JSON file. Returns dict with {accessToken, refreshToken?, expiresAt?} or None. """ if platform.system() != "Darwin": return None try: # Read the "Claude Code-credentials" generic password entry result = subprocess.run( ["security", "find-generic-password", "-s", "Claude Code-credentials", "-w"], capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=5, stdin=subprocess.DEVNULL, ) except (OSError, subprocess.TimeoutExpired): logger.debug("Keychain: security command not available or timed out") return None if result.returncode != 0: logger.debug("Keychain: no entry found for 'Claude Code-credentials'") return None raw = result.stdout.strip() if not raw: return None try: data = json.loads(raw) except json.JSONDecodeError: logger.debug("Keychain: credentials payload is not valid JSON") return None oauth_data = data.get("claudeAiOauth") if oauth_data and isinstance(oauth_data, dict): access_token = oauth_data.get("accessToken", "") if access_token: return { "accessToken": access_token, "refreshToken": oauth_data.get("refreshToken", ""), "expiresAt": oauth_data.get("expiresAt", 0), "source": "macos_keychain", } return None def claude_code_credentials_path() -> Path: """Location Claude Code CLI writes its shared OAuth credentials file. This file is not profile-owned: every Hermes profile's credential pool reads and writes the *same* path, so cross-profile refresh races on a ``claude_code`` pool entry must be serialized against this exact path (see ``CredentialPool._claude_code_credentials_lock`` in ``agent/credential_pool.py``). """ return Path.home() / ".claude" / ".credentials.json" def _read_claude_code_credentials_from_file() -> Optional[Dict[str, Any]]: """Read Claude Code OAuth credentials from ~/.claude/.credentials.json. Returns dict with {accessToken, refreshToken?, expiresAt?, source} or None. """ cred_path = claude_code_credentials_path() if not cred_path.exists(): return None try: data = json.loads(cred_path.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError, IOError) as e: logger.debug("Failed to read ~/.claude/.credentials.json: %s", e) return None oauth_data = data.get("claudeAiOauth") if not (oauth_data and isinstance(oauth_data, dict)): return None access_token = oauth_data.get("accessToken", "") if not access_token: return None return { "accessToken": access_token, "refreshToken": oauth_data.get("refreshToken", ""), "expiresAt": oauth_data.get("expiresAt", 0), "source": "claude_code_credentials_file", } def read_claude_code_credentials() -> Optional[Dict[str, Any]]: """Read refreshable Claude Code OAuth credentials. Reads from two possible sources and reconciles them: 1. macOS Keychain (Darwin only) — "Claude Code-credentials" entry 2. ~/.claude/.credentials.json file Selection rules when both are present: - If exactly one is non-expired, prefer that one. (Handles the case where Claude Code refreshes one source but not the other — observed in the wild on Claude Code 2.1.x.) - Otherwise, prefer the source with the later ``expiresAt`` so that any subsequent refresh uses the most recent ``refreshToken``. This intentionally excludes ~/.claude.json primaryApiKey. Opencode's subscription flow is OAuth/setup-token based with refreshable credentials, and native direct Anthropic provider usage should follow that path rather than auto-detecting Claude's first-party managed key. Returns dict with {accessToken, refreshToken?, expiresAt?, source} or None. """ kc_creds = _read_claude_code_credentials_from_keychain() file_creds = _read_claude_code_credentials_from_file() if kc_creds and file_creds: kc_valid = is_claude_code_token_valid(kc_creds) file_valid = is_claude_code_token_valid(file_creds) if kc_valid and not file_valid: return kc_creds if file_valid and not kc_valid: return file_creds # Both valid or both expired: prefer the later expiresAt so the # downstream refresh path uses the freshest refresh_token. kc_exp = kc_creds.get("expiresAt", 0) or 0 file_exp = file_creds.get("expiresAt", 0) or 0 return kc_creds if kc_exp >= file_exp else file_creds return kc_creds or file_creds def is_claude_code_token_valid(creds: Dict[str, Any]) -> bool: """Check if Claude Code credentials have a non-expired access token.""" import time expires_at = creds.get("expiresAt", 0) if not expires_at: # No expiry set (managed keys) — valid if token is present return bool(creds.get("accessToken")) # expiresAt is in milliseconds since epoch now_ms = int(time.time() * 1000) # Allow 60 seconds of buffer return now_ms < (expires_at - 60_000) def refresh_anthropic_oauth_pure(refresh_token: str, *, use_json: bool = False) -> Dict[str, Any]: """Refresh an Anthropic OAuth token without mutating local credential files.""" import time import urllib.parse import urllib.request if not refresh_token: raise ValueError("refresh_token is required") client_id = "9d1c250a-e61b-44d9-88ed-5944d1962f5e" if use_json: data = json.dumps({ "grant_type": "refresh_token", "refresh_token": refresh_token, "client_id": client_id, }).encode() content_type = "application/json" else: data = urllib.parse.urlencode({ "grant_type": "refresh_token", "refresh_token": refresh_token, "client_id": client_id, }).encode() content_type = "application/x-www-form-urlencoded" token_endpoints = [ "https://platform.claude.com/v1/oauth/token", "https://console.anthropic.com/v1/oauth/token", ] last_error = None for endpoint in token_endpoints: req = urllib.request.Request( endpoint, data=data, headers={ "Content-Type": content_type, "User-Agent": _OAUTH_TOKEN_USER_AGENT, }, method="POST", ) try: with urllib.request.urlopen(req, timeout=10) as resp: result = json.loads(resp.read().decode()) except Exception as exc: last_error = exc logger.debug("Anthropic token refresh failed at %s: %s", endpoint, exc) continue access_token = result.get("access_token", "") if not access_token: raise ValueError("Anthropic refresh response was missing access_token") next_refresh = result.get("refresh_token", refresh_token) expires_in = result.get("expires_in", 3600) return { "access_token": access_token, "refresh_token": next_refresh, "expires_at_ms": int(time.time() * 1000) + (expires_in * 1000), } if last_error is not None: raise last_error raise ValueError("Anthropic token refresh failed") def _refresh_oauth_token(creds: Dict[str, Any]) -> Optional[str]: """Attempt to refresh an expired Claude Code OAuth token. Claude Code's OAuth refresh tokens are single-use: a successful refresh rotates the pair and invalidates the old refresh token. Claude Code itself also refreshes on its own schedule (IDE/CLI activity), so by the time Hermes notices an expired token, Claude Code may have already rotated it. POSTing our now-stale refresh token in that window races Claude Code and fails with ``invalid_grant``. So before refreshing, re-read the live credential sources. If Claude Code has already produced a valid token, adopt it and skip the POST entirely. Only fall back to refreshing ourselves when no fresh credential is found. """ # Claude Code may have already refreshed — adopt its token rather than # racing it with our (possibly already-rotated) refresh token. The read, # decision, POST, and write-back all belong to the shared credentials # source, so hold the same path-keyed cross-process lock used by the pool. # Without this direct resolver path, two profiles can still spend one # single-use refresh token even though CredentialPool is serialized. try: from hermes_cli.auth import AUTH_LOCK_TIMEOUT_SECONDS, _auth_store_lock, env_float refresh_timeout_seconds = env_float( "HERMES_ANTHROPIC_REFRESH_TIMEOUT_SECONDS", 20 ) lock_timeout_seconds = max( float(AUTH_LOCK_TIMEOUT_SECONDS), float(refresh_timeout_seconds) + 5.0, ) with _auth_store_lock( timeout_seconds=lock_timeout_seconds, target_path=claude_code_credentials_path(), ): # Only adopt when the live re-read produced a DIFFERENT token with # a real future expiry: re-adopting the same credential we were # just handed would be a no-op, and a 0/absent ``expiresAt`` means # "managed key / unknown expiry" (see is_claude_code_token_valid). current = read_claude_code_credentials() if current: current_token = current.get("accessToken", "") current_exp = current.get("expiresAt", 0) or 0 if ( current_token and current_token != creds.get("accessToken", "") and current_exp > 0 and is_claude_code_token_valid(current) ): logger.debug("Adopted Claude Code's already-refreshed OAuth token") return current_token refresh_token = ( (current or {}).get("refreshToken", "") or creds.get("refreshToken", "") ) if not refresh_token: logger.debug("No refresh token available — cannot refresh") return None # Another process may have spent this refresh token and lost the # commit; its durable sidecar verdict is authoritative for the # shared source. POSTing it again would just burn the family into # ``invalid_grant``. if is_rotation_consumed_uncommitted( refresh_token, source_path=claude_code_credentials_path() ): logger.debug( "Refresh token was already consumed by an uncommitted rotation " "- refusing to replay it; re-run 'claude setup-token'" ) return None try: refreshed = refresh_anthropic_oauth_pure(refresh_token, use_json=False) except Exception as e: logger.debug("Failed to refresh Claude Code token: %s", e) return None # The POST above already consumed ``refresh_token`` server-side. # Writing the replacement pair is the commit step of that # transaction, not a cache update: if it fails, the rotation is # unrecoverable and the pair still on disk is spent. Fail closed # rather than handing back an access token whose refresh half was # lost — reporting success here is what lets a later load replay # the consumed token and produce ``invalid_grant``. try: _write_claude_code_credentials( refreshed["access_token"], refreshed["refresh_token"], refreshed["expires_at_ms"], ) except Exception as e: logger.error( "Anthropic OAuth refresh rotated the single-use token but could not " "commit it to %s (%s) — treating the refresh as failed; " "re-run 'claude setup-token' to reauthenticate", claude_code_credentials_path(), e, ) # The POST already spent ``refresh_token`` server-side and the # replacement is gone. The pre-rotation pair is still on disk, # so mark it: without this, source 5 re-reads it through the # pool and returns the consumed credential as a success. mark_rotation_consumed_uncommitted( refresh_token, creds.get("accessToken", ""), (current or {}).get("accessToken", ""), (current or {}).get("refreshToken", ""), source_path=claude_code_credentials_path(), ) return None logger.debug("Successfully refreshed Claude Code OAuth token") return refreshed["access_token"] except Exception as e: # Lock acquisition/read failures should preserve the resolver's # existing fail-soft contract rather than taking down agent startup. logger.debug("Failed to acquire Claude Code refresh lock: %s", e) return None def _write_claude_code_credentials( access_token: str, refresh_token: str, expires_at_ms: int, *, scopes: Optional[list] = None, ) -> None: """Write refreshed credentials back to ~/.claude/.credentials.json. The optional *scopes* list (e.g. ``["user:inference", "user:profile", ...]``) is persisted so that Claude Code's own auth check recognises the credential as valid. Claude Code >=2.1.81 gates on the presence of ``"user:inference"`` in the stored scopes before it will use the token. Raises ``CredentialPersistError`` when the rotated pair does not reach the file. This write is the commit step of the refresh transaction, not a best-effort cache update: a swallowed failure leaves the consumed pre-rotation pair on disk to be re-seeded and replayed (see ``CredentialPersistError``). """ cred_path = claude_code_credentials_path() try: # Read existing file to preserve other fields existing = {} if cred_path.exists(): existing = json.loads(cred_path.read_text(encoding="utf-8")) oauth_data: Dict[str, Any] = { "accessToken": access_token, "refreshToken": refresh_token, "expiresAt": expires_at_ms, } if scopes is not None: oauth_data["scopes"] = scopes elif "claudeAiOauth" in existing and "scopes" in existing["claudeAiOauth"]: # Preserve previously-stored scopes when the refresh response # does not include a scope field. oauth_data["scopes"] = existing["claudeAiOauth"]["scopes"] existing["claudeAiOauth"] = oauth_data cred_path.parent.mkdir(parents=True, exist_ok=True) # Per-process random suffix avoids collisions between concurrent # writers and stale leftovers from a prior crashed write. _tmp_cred = cred_path.with_suffix(f".tmp.{os.getpid()}.{secrets.token_hex(4)}") try: # Create the temp file atomically at 0o600. The previous # write_text + post-replace chmod opened a TOCTOU window where # both the temp file and the destination briefly inherited the # process umask (commonly 0o644 = world-readable), exposing # Claude Code OAuth tokens to other local users between create # and chmod. Mirrors agent/google_oauth.py (#19673) and # tools/mcp_oauth.py (#21148). Parent dir (~/.claude/) is # owned by Claude Code itself, so we leave its mode alone. fd = os.open( str(_tmp_cred), os.O_WRONLY | os.O_CREAT | os.O_EXCL, stat.S_IRUSR | stat.S_IWUSR, ) with os.fdopen(fd, "w", encoding="utf-8") as fh: json.dump(existing, fh, indent=2) fh.flush() os.fsync(fh.fileno()) os.replace(_tmp_cred, cred_path) except OSError: try: _tmp_cred.unlink(missing_ok=True) except OSError: pass raise except (OSError, IOError, ValueError) as e: # ValueError covers a corrupt existing file (JSONDecodeError): the # merge-read is part of the commit, so failing it means the rotated # pair never landed either. logger.error("Failed to write refreshed credentials to %s: %s", cred_path, e) raise CredentialPersistError(cred_path, e) from e def _resolve_claude_code_token_from_credentials(creds: Optional[Dict[str, Any]] = None) -> Optional[str]: """Resolve a token from Claude Code credential files, refreshing if needed.""" creds = creds or read_claude_code_credentials() if creds and is_rotation_consumed_uncommitted( creds.get("accessToken", ""), source_path=claude_code_credentials_path() ): # This process already rotated this pair and failed to commit the # replacement. The file still holds the spent copy; treating it as # usable is exactly the silent success this transaction fails closed # to prevent. logger.debug( "Claude Code credentials hold a rotated-but-uncommitted token - refusing" ) return None if creds and is_claude_code_token_valid(creds): logger.debug("Using Claude Code credentials (auto-detected)") return creds["accessToken"] if creds: logger.debug("Claude Code credentials expired — attempting refresh") refreshed = _refresh_oauth_token(creds) if refreshed: return refreshed logger.debug("Token refresh failed — re-run 'claude setup-token' to reauthenticate") return None def _prefer_refreshable_claude_code_token(env_token: str, creds: Optional[Dict[str, Any]]) -> Optional[str]: """Prefer Claude Code creds when a persisted env OAuth token would shadow refresh. Hermes historically persisted setup tokens into ANTHROPIC_TOKEN. That makes later refresh impossible because the static env token wins before we ever inspect Claude Code's refreshable credential file. If we have a refreshable Claude Code credential record, prefer it over the static env OAuth token. """ if not env_token or not _is_oauth_token(env_token) or not isinstance(creds, dict): return None if not creds.get("refreshToken"): return None resolved = _resolve_claude_code_token_from_credentials(creds) if resolved and resolved != env_token: logger.debug( "Preferring Claude Code credential file over static env OAuth token so refresh can proceed" ) return resolved return None def _resolve_anthropic_pool_token() -> Optional[str]: """Return the first available Anthropic OAuth token from credential_pool. Read-only: enumerates with ``clear_expired=False, refresh=False`` so a bare token *resolve* (which runs from diagnostic/read-only call sites such as ``account_usage`` and ``hermes models``) never mutates ``~/.hermes/auth.json`` or makes a network refresh call. Refresh-on-expiry is owned by the API call path's pool recovery, not the resolver. """ try: from agent.credential_pool import AUTH_TYPE_OAUTH, load_pool except Exception: return None try: pool = load_pool("anthropic") # Enumerate read-only (clear_expired=False, refresh=False): never persist # to auth.json or trigger a network refresh from a bare resolve. select() # is deliberately NOT used — it runs clear_expired=True, refresh=True, # which would violate this read-only contract. entries, _pending = pool._available_entries(clear_expired=False, refresh=False) except Exception: logger.debug("Failed to read Anthropic credential_pool", exc_info=True) return None for entry in entries: if getattr(entry, "auth_type", None) != AUTH_TYPE_OAUTH: continue # access_token is a declared field but a persisted entry can carry an # explicit null (or a partially-written OAuth entry), so coerce before # strip — a bare None.strip() here would escape the try/excepts above # and crash the whole resolver, taking down the source #5 fallback too. # Matches the aux-client analog (auxiliary_client.py: str(key or "")). token = (getattr(entry, "access_token", None) or "").strip() if not token: continue # ``load_pool()`` re-seeds pool rows from the singleton files, so a # rotation that was consumed upstream but never committed comes back # here looking healthy. Enumeration is deliberately read-only # (refresh=False), which means nothing on this path would otherwise # notice that the credential is spent. Singleton-backed sources also # consult the durable sidecar registry: the failed commit may have # happened in a DIFFERENT process, whose process-local verdict this # interpreter never saw. entry_source_path = spent_rotation_source_path(getattr(entry, "source", None)) if is_rotation_consumed_uncommitted( token, source_path=entry_source_path ) or is_rotation_consumed_uncommitted( getattr(entry, "refresh_token", None), source_path=entry_source_path ): logger.debug( "Skipping Anthropic pool entry %s: rotated-but-uncommitted credential", getattr(entry, "id", "?"), ) continue return token return None def resolve_anthropic_token() -> Optional[str]: """Resolve an Anthropic token from all available sources. Priority: 1. ANTHROPIC_TOKEN env var (OAuth/setup token saved by Hermes) 2. CLAUDE_CODE_OAUTH_TOKEN env var 3. ANTHROPIC_API_KEY env var (explicit regular API key) 4. Claude Code credentials (~/.claude.json or ~/.claude/.credentials.json) — with automatic refresh if expired and a refresh token is available 5. Anthropic credential_pool OAuth entry (~/.hermes/auth.json) Returns the token string or None. """ creds: Optional[Dict[str, Any]] = None creds_loaded = False def _read_creds() -> Optional[Dict[str, Any]]: nonlocal creds, creds_loaded if not creds_loaded: creds = read_claude_code_credentials() creds_loaded = True return creds # 1. Hermes-managed OAuth/setup token env var token = _getenv("ANTHROPIC_TOKEN").strip() if token: preferred = _prefer_refreshable_claude_code_token(token, _read_creds()) if preferred: return preferred return token # 2. CLAUDE_CODE_OAUTH_TOKEN (used by Claude Code for setup-tokens) cc_token = _getenv("CLAUDE_CODE_OAUTH_TOKEN").strip() if cc_token: preferred = _prefer_refreshable_claude_code_token(cc_token, _read_creds()) if preferred: return preferred return cc_token # 3. Regular API key. An explicit user-configured key must not be shadowed # by auto-discovered Claude Code or credential-pool OAuth credentials. api_key = _getenv("ANTHROPIC_API_KEY").strip() if api_key: return api_key # 4. Claude Code credential file resolved_claude_token = _resolve_claude_code_token_from_credentials(_read_creds()) if resolved_claude_token: return resolved_claude_token # 5. Hermes credential_pool OAuth entry. resolved_pool_token = _resolve_anthropic_pool_token() if resolved_pool_token: return resolved_pool_token return None def run_oauth_setup_token() -> Optional[str]: """Run 'claude setup-token' interactively and return the resulting token. Checks multiple sources after the subprocess completes: 1. Claude Code credential files (may be written by the subprocess) 2. CLAUDE_CODE_OAUTH_TOKEN / ANTHROPIC_TOKEN env vars Returns the token string, or None if no credentials were obtained. Raises FileNotFoundError if the 'claude' CLI is not installed. """ import shutil import subprocess claude_path = shutil.which("claude") if not claude_path: raise FileNotFoundError( "The 'claude' CLI is not installed. " "Install it with: npm install -g @anthropic-ai/claude-code" ) # Run interactively — stdin/stdout/stderr inherited so the user can # complete the OAuth login prompt. Must keep inherited stdin; the TUI-EOF # concern does not apply to an interactive login the user explicitly # invokes. noqa: subprocess-stdin try: subprocess.run([claude_path, "setup-token"]) except (KeyboardInterrupt, EOFError): return None # Check if credentials were saved to Claude Code's config files creds = read_claude_code_credentials() if creds and is_claude_code_token_valid(creds): return creds["accessToken"] # Check env vars that may have been set for env_var in ("CLAUDE_CODE_OAUTH_TOKEN", "ANTHROPIC_TOKEN"): val = _getenv(env_var).strip() if val: return val return None # ── Hermes-native PKCE OAuth flow ──────────────────────────────────────── # Mirrors the flow used by Claude Code, pi-ai, and OpenCode. # Stores credentials in ~/.hermes/.anthropic_oauth.json (our own file). _OAUTH_CLIENT_ID = "9d1c250a-e61b-44d9-88ed-5944d1962f5e" # Anthropic migrated the OAuth token endpoint to platform.claude.com; # console.anthropic.com now 404s. Callers should iterate _OAUTH_TOKEN_URLS # (new host first, console fallback). _OAUTH_TOKEN_URL is kept as the primary # for backward compatibility with existing imports and now points at the live host. _OAUTH_TOKEN_URLS = [ "https://platform.claude.com/v1/oauth/token", "https://console.anthropic.com/v1/oauth/token", ] _OAUTH_TOKEN_URL = _OAUTH_TOKEN_URLS[0] # User-Agent sent on the OAuth *token endpoint* (login exchange + refresh). # Anthropic rate-limits (HTTP 429) any token-endpoint request whose UA starts # with ``claude-code/`` — verified empirically against platform.claude.com: # ``claude-code/2.1.200`` and ``Mozilla/5.0`` -> 429; ``axios/*``, ``node``, # and SDK-style UAs -> 400 (reached code validation). The real Claude Code CLI # exchanges the auth code with a bare axios client (``axios/``), NOT its # ``claude-code/`` inference UA. We mirror that here. NOTE: the *inference* path # (build_anthropic_kwargs) still uses the ``claude-code/`` UA + ``x-app: cli`` — # that fingerprint is required there and is NOT throttled on the messages API. _OAUTH_TOKEN_USER_AGENT = "axios/1.7.9" _OAUTH_REDIRECT_URI = "https://console.anthropic.com/oauth/code/callback" _OAUTH_SCOPES = "org:create_api_key user:profile user:inference" def _get_hermes_oauth_file() -> Path: return get_hermes_home() / ".anthropic_oauth.json" def _root_hermes_oauth_file() -> Optional[Path]: """Global-root ``.anthropic_oauth.json`` when running inside a named profile. ``None`` in classic mode (profile == root). Used to commit a rotation of a grant the profile borrowed through the credential-pool root fallback. """ try: from hermes_constants import get_default_hermes_root root = get_default_hermes_root() if root.resolve(strict=False) == get_hermes_home().resolve(strict=False): return None return root / ".anthropic_oauth.json" except Exception: return None def _generate_pkce() -> tuple: """Generate PKCE code_verifier and code_challenge (S256).""" import base64 import hashlib import secrets verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).rstrip(b"=").decode() challenge = base64.urlsafe_b64encode( hashlib.sha256(verifier.encode()).digest() ).rstrip(b"=").decode() return verifier, challenge def run_hermes_oauth_login_pure() -> Optional[Dict[str, Any]]: """Run Hermes-native OAuth PKCE flow and return credential state.""" import secrets import time import webbrowser verifier, challenge = _generate_pkce() oauth_state = secrets.token_urlsafe(32) params = { "code": "true", "client_id": _OAUTH_CLIENT_ID, "response_type": "code", "redirect_uri": _OAUTH_REDIRECT_URI, "scope": _OAUTH_SCOPES, "code_challenge": challenge, "code_challenge_method": "S256", "state": oauth_state, } from urllib.parse import urlencode auth_url = f"https://claude.ai/oauth/authorize?{urlencode(params)}" print() print("Authorize Hermes with your Claude Pro/Max subscription.") print() print("╭─ Claude Pro/Max Authorization ────────────────────╮") print("│ │") print("│ Open this link in your browser: │") print("╰───────────────────────────────────────────────────╯") print() print(f" {auth_url}") print() try: from hermes_cli.auth import _can_open_graphical_browser as _can_open_gui except Exception: _can_open_gui = lambda: True # noqa: E731 — degrade to prior behavior if _can_open_gui(): try: webbrowser.open(auth_url) print(" (Browser opened automatically)") except Exception: pass print() print("After authorizing, you'll see a code. Paste it below.") print() try: auth_code = input("Authorization code: ").strip() except (KeyboardInterrupt, EOFError): return None if not auth_code: print("No code entered.") return None splits = auth_code.split("#") code = splits[0] received_state = splits[1] if len(splits) > 1 else "" # Validate state to prevent CSRF (RFC 6749 §10.12) if received_state != oauth_state: logger.warning("OAuth state mismatch — possible CSRF, aborting") return None try: import urllib.request exchange_data = json.dumps({ "grant_type": "authorization_code", "client_id": _OAUTH_CLIENT_ID, "code": code, "state": received_state, "redirect_uri": _OAUTH_REDIRECT_URI, "code_verifier": verifier, }).encode() # Anthropic migrated the OAuth token endpoint to platform.claude.com; # console.anthropic.com now 404s. Try the new host first, then fall # back to console for older deployments (mirrors the refresh path). # UA is _OAUTH_TOKEN_USER_AGENT (a non-claude-code UA) — see the # constant's definition for why the token endpoint must not send # claude-code/ (429 UA-prefix block). result = None last_error = None for endpoint in _OAUTH_TOKEN_URLS: req = urllib.request.Request( endpoint, data=exchange_data, headers={ "Content-Type": "application/json", "User-Agent": _OAUTH_TOKEN_USER_AGENT, }, method="POST", ) try: with urllib.request.urlopen(req, timeout=15) as resp: result = json.loads(resp.read().decode()) break except Exception as exc: last_error = exc logger.debug("Anthropic token exchange failed at %s: %s", endpoint, exc) continue if result is None: raise last_error if last_error is not None else ValueError( "Anthropic token exchange failed" ) except Exception as e: print(f"Token exchange failed: {e}") return None access_token = result.get("access_token", "") refresh_token = result.get("refresh_token", "") expires_in = result.get("expires_in", 3600) if not access_token: print("No access token in response.") return None expires_at_ms = int(time.time() * 1000) + (expires_in * 1000) return { "access_token": access_token, "refresh_token": refresh_token, "expires_at_ms": expires_at_ms, } def read_hermes_oauth_credentials() -> Optional[Dict[str, Any]]: """Read Hermes-managed OAuth credentials from ~/.hermes/.anthropic_oauth.json.""" oauth_file = _get_hermes_oauth_file() if oauth_file.exists(): try: data = json.loads(oauth_file.read_text(encoding="utf-8")) if data.get("accessToken"): return data except (json.JSONDecodeError, OSError, IOError) as e: logger.debug("Failed to read Hermes OAuth credentials: %s", e) return None def _write_hermes_oauth_credentials( access_token: str, refresh_token: Optional[str], expires_at_ms: Optional[int], *, target: Optional[Path] = None, ) -> None: """Write refreshed hermes_pkce tokens back to ~/.hermes/.anthropic_oauth.json. ``target`` overrides the destination: a named profile that rotated a grant it BORROWED from the global root (credential-pool root fallback) must commit the new pair to the ROOT singleton, not create a forked copy under its own HERMES_HOME (#100339). Without this, a successful pool-level refresh of a ``hermes_pkce``-sourced entry is invisible to this singleton file. The next ``load_pool()`` call runs ``_seed_from_singletons()``, which reads the stale file and overwrites the freshly-rotated pool entry with the pre-refresh (and, for single-use Anthropic refresh tokens, already-consumed) token pair. Raises ``CredentialPersistError`` when the rotated pair does not reach the file, for the same reason ``_write_claude_code_credentials`` does: this is the commit step of the refresh transaction. """ oauth_file = target if target is not None else _get_hermes_oauth_file() try: oauth_data = { "accessToken": access_token, "refreshToken": refresh_token, "expiresAt": expires_at_ms, } oauth_file.parent.mkdir(parents=True, exist_ok=True) _tmp_oauth = oauth_file.with_suffix(f".tmp.{os.getpid()}.{secrets.token_hex(4)}") try: fd = os.open( str(_tmp_oauth), os.O_WRONLY | os.O_CREAT | os.O_EXCL, stat.S_IRUSR | stat.S_IWUSR, ) with os.fdopen(fd, "w", encoding="utf-8") as fh: json.dump(oauth_data, fh, indent=2) fh.flush() os.fsync(fh.fileno()) os.replace(_tmp_oauth, oauth_file) except OSError: try: _tmp_oauth.unlink(missing_ok=True) except OSError: pass raise except (OSError, IOError, ValueError) as e: logger.error( "Failed to write refreshed Hermes OAuth credentials to %s: %s", oauth_file, e ) raise CredentialPersistError(oauth_file, e) from e