827 lines
41 KiB
Python
827 lines
41 KiB
Python
"""Anthropic credential sources, OAuth flows, and token resolution.
|
|
|
|
``resolve_anthropic_token()`` order: ``ANTHROPIC_TOKEN`` / ``CLAUDE_CODE_OAUTH_TOKEN``,
|
|
``ANTHROPIC_API_KEY``, Hermes-owned OAuth grants in the ``auth.json`` credential
|
|
pool, then ``~/.claude/.credentials.json`` / macOS Keychain as a borrowed fallback.
|
|
``~/.hermes/.anthropic_oauth.json`` (Hermes PKCE) and
|
|
the Claude Code file are *singletons*: ``credential_pool._seed_from_singletons()``
|
|
re-reads them on every ``load_pool()``, so a failed write here is a failed refresh
|
|
(``CredentialPersistError``), not a cache miss.
|
|
"""
|
|
|
|
import base64
|
|
import contextlib
|
|
import functools
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import platform
|
|
import re
|
|
import secrets
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
from collections import OrderedDict
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
from urllib.parse import urlparse
|
|
|
|
from hermes_constants import get_hermes_home
|
|
from utils import atomic_json_write
|
|
from agent.secret_scope import get_secret as _get_secret
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_OAUTH_CLIENT_ID = "9d1c250a-e61b-44d9-88ed-5944d1962f5e"
|
|
# platform.claude.com is the live token host; console.anthropic.com 404s but is kept as a fallback.
|
|
_OAUTH_TOKEN_URLS = [
|
|
"https://platform.claude.com/v1/oauth/token", "https://console.anthropic.com/v1/oauth/token"
|
|
]
|
|
# Anthropic 429s token-endpoint requests whose UA starts with ``claude-code/`` (or Mozilla); the real CLI uses
|
|
# bare axios there. Inference (build_anthropic_kwargs) still needs claude-code/.
|
|
_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"
|
|
# Claude Code's macOS Keychain entry (generic password). Hermes reads it
|
|
# (_read_claude_code_credentials_from_keychain) and, since #98334, mirrors the
|
|
# refresh write into it so the two stores stop diverging on a single-use rotation.
|
|
_CLAUDE_CODE_KEYCHAIN_SERVICE = "Claude Code-credentials"
|
|
|
|
|
|
def _getenv(name: str, default: str = "") -> str:
|
|
"""Profile-scoped os.getenv for credential reads (fail-closed on unscoped reads when multiplexing)."""
|
|
val = _get_secret(name, default)
|
|
return val if val is not None else default
|
|
|
|
|
|
def _first_env(*names: str) -> str:
|
|
"""First non-blank (stripped) value among *names*, else ''."""
|
|
return next((v for v in (_getenv(n).strip() for n in names) if v), "")
|
|
|
|
|
|
def _is_oauth_token(key: str) -> bool:
|
|
"""True for Anthropic OAuth/setup tokens (sk-ant-*, eyJ JWTs, cc-); False for sk-ant-api* Console keys."""
|
|
if not key or key.startswith("sk-ant-api"):
|
|
return False
|
|
return key.startswith(("sk-ant-", "eyJ", "cc-"))
|
|
|
|
|
|
def anthropic_route_is_oauth(base_url: Any, credential: Any, *, provider: Optional[str] = None) -> bool:
|
|
"""Claude Code OAuth identity for one Anthropic Messages route (#114967).
|
|
|
|
The route qualifies when it is the ``anthropic`` provider itself or its host is exactly
|
|
``api.anthropic.com`` (an empty base_url is the native default) — a named custom provider
|
|
pointed at the native host carries the same identity, while third-party Anthropic-protocol
|
|
endpoints never do (Claude Code headers and tool-name transforms 401/403 there). ``credential``
|
|
is a static string or a ``key_cmd``/per-request callable token source; a callable is
|
|
materialized once for the shape test (``CommandTokenSource`` caches, so this never double-mints)
|
|
and a mint failure classifies as non-OAuth — the wire client surfaces the real error.
|
|
"""
|
|
text = str(base_url or "").strip()
|
|
native_host = not text or (urlparse(text).hostname or "").lower().rstrip(".") == "api.anthropic.com"
|
|
if not (native_host or (provider or "").strip().lower() == "anthropic"):
|
|
return False
|
|
if callable(credential) and not isinstance(credential, str):
|
|
try:
|
|
credential = credential()
|
|
except Exception: # noqa: BLE001 — classification must never raise
|
|
return False
|
|
return isinstance(credential, str) and _is_oauth_token(credential)
|
|
|
|
|
|
class CredentialPersistError(RuntimeError):
|
|
"""A rotated single-use credential could not be durably committed. The refresh POST already spent the old
|
|
refresh token, so a swallowed write failure leaves a consumed pair on disk that later replays as invalid_grant."""
|
|
|
|
def __init__(self, path: Any, cause: BaseException) -> None:
|
|
super().__init__(f"failed to durably persist rotated Anthropic credentials to {path}: {cause}")
|
|
self.path = path
|
|
|
|
|
|
def _load_json_if_exists(path: Path, what: str) -> Optional[Any]:
|
|
"""Parsed JSON from *path*, or None when missing/unreadable/corrupt (debug-logged)."""
|
|
if not path.exists():
|
|
return None
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8-sig"))
|
|
except (json.JSONDecodeError, OSError) as e:
|
|
logger.debug("Failed to read %s: %s", what, e)
|
|
return None
|
|
|
|
|
|
def _atomic_write_private_json(path: Path, payload: Any) -> None:
|
|
"""0600-from-creation temp file + fsync + atomic replace (the token is never briefly umask-readable).
|
|
The parent dir's mode is left alone (~/.claude/ is owned by Claude Code)."""
|
|
atomic_json_write(path, payload, mode=0o600)
|
|
|
|
|
|
def _commit_private_json(path: Path, payload: Any, what: str) -> None:
|
|
"""Atomic private write; any failure becomes ``CredentialPersistError`` (the commit step of a rotation)."""
|
|
try:
|
|
_atomic_write_private_json(path, payload)
|
|
except (OSError, ValueError) as e:
|
|
logger.error("Failed to write refreshed %s to %s: %s", what, path, e)
|
|
raise CredentialPersistError(path, e) from e
|
|
|
|
|
|
# ── Spent-rotation registry: fingerprints of secrets whose refresh POST succeeded but whose replacement never
|
|
# reached its store. Two scopes: process-local (OrderedDict) and a durable sidecar next to the shared singleton
|
|
# file so OTHER processes fail closed too. Non-reversible digests; never cleared.
|
|
_SPENT_ROTATION_LOCK = threading.Lock()
|
|
# Fingerprints of Claude Code refresh tokens the endpoint rejected terminally: the WARNING fires once per token
|
|
# per process and later attempts skip the POST (a re-login rotates the token, so a new one is tried normally).
|
|
_DEAD_REFRESH_TOKEN_FINGERPRINTS: set = set()
|
|
_SPENT_ROTATION_FINGERPRINTS: "OrderedDict[str, None]" = OrderedDict()
|
|
_SPENT_ROTATION_MAX_TRACKED = 64
|
|
_SPENT_ROTATION_SIDECAR_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."
|
|
)
|
|
|
|
|
|
def _spent_rotation_sidecar_path(source_path: Path) -> Path:
|
|
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 (or None)."""
|
|
getter = _SINGLETON_SOURCE_PATHS.get(source) if isinstance(source, str) else None
|
|
return getter() if getter else 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-sig")
|
|
)
|
|
except (OSError, ValueError):
|
|
return set()
|
|
fingerprints = raw.get("fingerprints") if isinstance(raw, dict) else None
|
|
return {fp for fp in fingerprints if isinstance(fp, str) and fp} if isinstance(fingerprints, list) else set()
|
|
|
|
|
|
def _append_spent_rotation_sidecar(source_path: Path, fingerprints: list) -> None:
|
|
"""Merge fingerprints into the sidecar (atomic replace; caller holds the path lock). Fail-soft: a sidecar
|
|
write failure must never mask the process-local verdict."""
|
|
sidecar = _spent_rotation_sidecar_path(source_path)
|
|
try:
|
|
merged = _read_spent_rotation_sidecar(source_path)
|
|
merged.update(fingerprints)
|
|
payload = json.dumps({
|
|
"version": 1,
|
|
"comment": _SPENT_ROTATION_SIDECAR_COMMENT,
|
|
"fingerprints": sorted(merged)[-_SPENT_ROTATION_MAX_TRACKED * 4 :],
|
|
}, 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 _fingerprint(secret: Any) -> Optional[str]:
|
|
from agent.credential_persistence import fingerprint_secret_value
|
|
value = str(secret or "").strip()
|
|
return fingerprint_secret_value(value) if value else None
|
|
|
|
|
|
def mark_rotation_consumed_uncommitted(*secrets: Any, source_path: Optional[Path] = None) -> None:
|
|
"""Record the pre-rotation pair of a refresh whose replacement never committed; with ``source_path`` the
|
|
verdict is also persisted to that singleton's sidecar."""
|
|
recorded = [fp for fp in map(_fingerprint, secrets) if fp]
|
|
with _SPENT_ROTATION_LOCK:
|
|
for fingerprint in recorded:
|
|
_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."""
|
|
fingerprint = _fingerprint(secret)
|
|
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)
|
|
|
|
|
|
# ── Claude Code credentials (Keychain / ~/.claude/.credentials.json) ──
|
|
# Only singleton-backed pool sources have a cross-process authority boundary.
|
|
_SINGLETON_SOURCE_PATHS = {
|
|
"claude_code": lambda: claude_code_credentials_path(), "hermes_pkce": lambda: _get_hermes_oauth_file()
|
|
}
|
|
|
|
|
|
def _claude_oauth_record(data: Any, source: str) -> Optional[Dict[str, Any]]:
|
|
"""Normalise a ``{"claudeAiOauth": {...}}`` payload into our credential dict."""
|
|
oauth_data = data.get("claudeAiOauth")
|
|
access_token = oauth_data.get("accessToken", "") if isinstance(oauth_data, dict) else ""
|
|
if not access_token:
|
|
return None
|
|
return {
|
|
"accessToken": access_token, "refreshToken": oauth_data.get("refreshToken", ""),
|
|
"expiresAt": oauth_data.get("expiresAt", 0), "source": source,
|
|
}
|
|
|
|
|
|
_KEYCHAIN_ATTR = r'(?:0x(?P<hex>[0-9A-Fa-f]+)\b.*|"(?P<text>.*)")'
|
|
|
|
|
|
def _decode_keychain_attr(match: Optional["re.Match[str]"]) -> str:
|
|
"""``security`` prints an attribute as ``"text"`` when it is plain printable ASCII and as
|
|
``0x<HEX> "<octal-escaped echo>"`` otherwise; the quoted form is NOT escaped (an embedded
|
|
``"`` appears raw), so the text group must run to the last quote on the line."""
|
|
if match is None:
|
|
return ""
|
|
if match.group("hex"):
|
|
try:
|
|
return bytes.fromhex(match.group("hex")).decode("utf-8")
|
|
except ValueError:
|
|
return ""
|
|
return match.group("text") or ""
|
|
|
|
|
|
def _find_claude_code_keychain_item() -> Optional[tuple[str, Dict[str, Any]]]:
|
|
"""``(account, payload)`` of the ``Claude Code-credentials`` login Keychain item, or None.
|
|
|
|
One ``find-generic-password -g`` call: attributes on stdout, ``password: …`` on stderr. The
|
|
account matters because ``add-generic-password -U`` matches on account AND service — writing
|
|
under another account would create a second item instead of updating the one Claude Code reads.
|
|
"""
|
|
if platform.system() != "Darwin":
|
|
return None
|
|
try:
|
|
result = subprocess.run(
|
|
["security", "find-generic-password", "-s", _CLAUDE_CODE_KEYCHAIN_SERVICE, "-g"],
|
|
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5, stdin=subprocess.DEVNULL,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
return None
|
|
if result.returncode != 0:
|
|
return None
|
|
account = _decode_keychain_attr(re.search(r'^\s*"acct"<blob>=' + _KEYCHAIN_ATTR + r"\s*$", result.stdout, re.M))
|
|
raw = _decode_keychain_attr(re.search(r"^password: " + _KEYCHAIN_ATTR + r"\s*$", result.stderr, re.M))
|
|
if not account or not raw:
|
|
return None
|
|
try:
|
|
payload = json.loads(raw)
|
|
except ValueError:
|
|
return None
|
|
return (account, payload) if isinstance(payload, dict) else None
|
|
|
|
|
|
def _read_claude_code_keychain_payload() -> Optional[Dict[str, Any]]:
|
|
"""Raw ``{"claudeAiOauth": {...}, ...}`` payload from the macOS Keychain, or None.
|
|
|
|
Returns the full entry (not the normalised credential record) so a refresh
|
|
write can merge the rotated token triple over the existing metadata
|
|
(``subscriptionType`` / ``rateLimitTier`` / ``scopes``) instead of clobbering it.
|
|
"""
|
|
if platform.system() != "Darwin":
|
|
return None
|
|
try:
|
|
result = subprocess.run(
|
|
["security", "find-generic-password", "-s", _CLAUDE_CODE_KEYCHAIN_SERVICE, "-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 %r", _CLAUDE_CODE_KEYCHAIN_SERVICE)
|
|
return None
|
|
raw = result.stdout.strip()
|
|
if not raw:
|
|
return None
|
|
try:
|
|
payload = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
logger.debug("Keychain: credentials payload is not valid JSON")
|
|
return None
|
|
return payload if isinstance(payload, dict) else None
|
|
|
|
|
|
def _keychain_mirror_command(account: str, payload: Dict[str, Any]) -> tuple[list[str], str]:
|
|
"""``(argv, stdin)`` that updates the Claude Code Keychain item with ``payload``.
|
|
|
|
The command line goes to ``security -i`` on stdin, with the secret hex-encoded (``-X``):
|
|
a bare ``-w`` prompts twice on /dev/tty when a terminal exists (hangs the CLI) and, with
|
|
no terminal, reads only the first line and stores an EMPTY password when the confirmation
|
|
read hits EOF — either way the live token must never sit on argv.
|
|
"""
|
|
def quoted(value: str) -> str: # the ``security -i`` tokenizer: double quotes, backslash escapes
|
|
return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'
|
|
|
|
encoded = json.dumps(payload, separators=(",", ":"), ensure_ascii=True).encode("utf-8").hex()
|
|
line = f"add-generic-password -U -a {quoted(account)} -s {quoted(_CLAUDE_CODE_KEYCHAIN_SERVICE)} -X {encoded}\n"
|
|
return ["security", "-i"], line
|
|
|
|
|
|
def _read_claude_code_credentials_from_keychain() -> Optional[Dict[str, Any]]:
|
|
"""Read the "Claude Code-credentials" macOS Keychain entry (Claude Code >=2.1.114)."""
|
|
payload = _read_claude_code_keychain_payload()
|
|
return _claude_oauth_record(payload, "macos_keychain") if payload else None
|
|
|
|
|
|
def claude_code_credentials_path() -> Path:
|
|
"""Claude Code's shared OAuth file; every profile reads/writes this same path. Honours ``CLAUDE_CONFIG_DIR``
|
|
like the Claude CLI itself (blank = unset, as in ``hermes_cli.foreign_sessions``). The supported opt-out of
|
|
borrowing the login is ``auth.adopt_external_logins: false`` in config.yaml."""
|
|
override = os.environ.get("CLAUDE_CONFIG_DIR", "").strip()
|
|
root = Path(override).expanduser() if override else Path.home() / ".claude"
|
|
return root / ".credentials.json"
|
|
|
|
|
|
def _read_claude_code_credentials_from_file() -> Optional[Dict[str, Any]]:
|
|
data = _load_json_if_exists(claude_code_credentials_path(), "~/.claude/.credentials.json")
|
|
return _claude_oauth_record(data, "claude_code_credentials_file") if data is not None else None
|
|
|
|
|
|
def read_claude_code_credentials() -> Optional[Dict[str, Any]]:
|
|
"""Read refreshable Claude Code OAuth credentials (Keychain and/or file). When both exist: prefer the only
|
|
non-expired one (Claude Code 2.1.x refreshes one source but not the other), else the later ``expiresAt`` so a
|
|
refresh uses the freshest refreshToken. ~/.claude.json primaryApiKey is deliberately excluded.
|
|
|
|
This is the only reader of the borrowed login, so ``auth.adopt_external_logins: false`` is enforced here:
|
|
every resolver, pool seed/sync and 401 refresher then sees "no Claude Code login" and never touches the file."""
|
|
from agent.credential_sources import adopt_external_logins_enabled
|
|
if not adopt_external_logins_enabled():
|
|
return None
|
|
kc_creds = _read_claude_code_credentials_from_keychain()
|
|
file_creds = _read_claude_code_credentials_from_file()
|
|
if not (kc_creds and file_creds):
|
|
return kc_creds or file_creds
|
|
kc_valid, file_valid = is_claude_code_token_valid(kc_creds), is_claude_code_token_valid(file_creds)
|
|
if kc_valid != file_valid:
|
|
return kc_creds if kc_valid else file_creds
|
|
return kc_creds if (kc_creds.get("expiresAt", 0) or 0) >= (file_creds.get("expiresAt", 0) or 0) else file_creds
|
|
|
|
|
|
def is_claude_code_token_valid(creds: Dict[str, Any]) -> bool:
|
|
"""Non-expired access token (60s buffer); no expiresAt means managed key → valid if present."""
|
|
expires_at = creds.get("expiresAt", 0)
|
|
return int(time.time() * 1000) < (expires_at - 60_000) if expires_at else bool(creds.get("accessToken"))
|
|
|
|
|
|
# ── OAuth token endpoint ──
|
|
|
|
|
|
# OAuth ``error`` codes (RFC 6749 §5.2 + the provider's reuse detection) after which replaying the
|
|
# same refresh token can never succeed; only a fresh login recovers.
|
|
_OAUTH_GRANT_DEAD_CODES = frozenset({"invalid_grant", "invalid_token", "refresh_token_reused"})
|
|
|
|
|
|
class AnthropicOAuthError(ValueError):
|
|
"""Token endpoint rejected the request. ``code`` is the OAuth ``error`` field of the response body."""
|
|
|
|
def __init__(self, status: int, code: str, description: str, *, what: str) -> None:
|
|
self.status = status
|
|
self.code = code
|
|
detail = f" ({description})" if description else ""
|
|
super().__init__(f"Anthropic token {what} failed: HTTP {status} {code or 'error'}{detail}")
|
|
|
|
@property
|
|
def relogin_required(self) -> bool:
|
|
return self.status in (400, 401) and self.code in _OAUTH_GRANT_DEAD_CODES
|
|
|
|
|
|
def is_terminal_anthropic_refresh_error(exc: BaseException) -> bool:
|
|
"""True when retrying the same Anthropic refresh token cannot succeed (dead grant)."""
|
|
return isinstance(exc, AnthropicOAuthError) and exc.relogin_required
|
|
|
|
|
|
def _oauth_http_error(exc: Any, *, what: str) -> AnthropicOAuthError:
|
|
"""``urllib.error.HTTPError`` -> structured error carrying the body's OAuth ``error`` code."""
|
|
code, description = "", ""
|
|
try:
|
|
payload = json.loads(exc.read().decode() or "{}")
|
|
code = str(payload.get("error") or "")
|
|
description = str(payload.get("error_description") or "")
|
|
except Exception:
|
|
pass
|
|
return AnthropicOAuthError(int(exc.code), code, description, what=what)
|
|
|
|
|
|
def _post_oauth_token(
|
|
data: bytes, *, content_type: str, timeout: int, what: str, user_agent: str = _OAUTH_TOKEN_USER_AGENT
|
|
) -> Dict[str, Any]:
|
|
"""POST to the token endpoints in order; raise the last error if all fail."""
|
|
import urllib.error
|
|
import urllib.request
|
|
last_error = None
|
|
for endpoint in _OAUTH_TOKEN_URLS:
|
|
req = urllib.request.Request(
|
|
endpoint, data=data, method="POST", headers={"Content-Type": content_type, "User-Agent": user_agent}
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
return json.loads(resp.read().decode())
|
|
except urllib.error.HTTPError as exc:
|
|
last_error = _oauth_http_error(exc, what=what)
|
|
logger.debug("Anthropic token %s failed at %s: %s", what, endpoint, last_error)
|
|
if last_error.relogin_required:
|
|
break # a dead grant is dead at every endpoint; do not replay it
|
|
except Exception as exc:
|
|
last_error = exc
|
|
logger.debug("Anthropic token %s failed at %s: %s", what, endpoint, exc)
|
|
raise last_error or ValueError(f"Anthropic token {what} failed")
|
|
|
|
|
|
def _oauth_token_state(result: Dict[str, Any], *, fallback_refresh_token: str = "") -> Dict[str, Any]:
|
|
"""Token-endpoint JSON -> ``{access_token, refresh_token, expires_at_ms}`` (expires_in defaults to 3600s)."""
|
|
return {
|
|
"access_token": result.get("access_token", ""),
|
|
"refresh_token": result.get("refresh_token", fallback_refresh_token),
|
|
"expires_at_ms": int(time.time() * 1000) + (result.get("expires_in", 3600) * 1000),
|
|
}
|
|
|
|
|
|
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 urllib.parse
|
|
if not refresh_token:
|
|
raise ValueError("refresh_token is required")
|
|
payload = {"grant_type": "refresh_token", "refresh_token": refresh_token, "client_id": _OAUTH_CLIENT_ID}
|
|
encode, content_type = ((json.dumps, "application/json") if use_json
|
|
else (urllib.parse.urlencode, "application/x-www-form-urlencoded"))
|
|
result = _post_oauth_token(encode(payload).encode(), content_type=content_type, timeout=10, what="refresh",
|
|
user_agent=_OAUTH_TOKEN_USER_AGENT)
|
|
if not result.get("access_token"):
|
|
raise ValueError("Anthropic refresh response was missing access_token")
|
|
return _oauth_token_state(result, fallback_refresh_token=refresh_token)
|
|
|
|
|
|
def _refresh_oauth_token(creds: Dict[str, Any]) -> Optional[str]:
|
|
"""Refresh an expired Claude Code OAuth token, returning the new access token. Refresh tokens are single-use and
|
|
Claude Code refreshes on its own schedule, so we first re-read the live sources and adopt an already-rotated
|
|
token instead of racing it into ``invalid_grant``. Read, decision, POST and write-back share the pool's
|
|
path-keyed cross-process lock (else two profiles can spend one refresh token)."""
|
|
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)
|
|
cred_path = claude_code_credentials_path()
|
|
with _auth_store_lock(timeout_seconds=lock_timeout_seconds, target_path=cred_path):
|
|
# Adopt only a DIFFERENT token with a real future expiry (0/absent expiresAt = managed key/unknown).
|
|
current = read_claude_code_credentials() or {}
|
|
current_token = current.get("accessToken", "")
|
|
if (current_token and current_token != creds.get("accessToken", "")
|
|
and (current.get("expiresAt", 0) or 0) > 0 and is_claude_code_token_valid(current)):
|
|
logger.debug("Adopted Claude Code's already-refreshed OAuth token")
|
|
return current_token
|
|
|
|
refresh_token = current.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 token and lost the commit; its sidecar verdict is authoritative.
|
|
if is_rotation_consumed_uncommitted(refresh_token, source_path=cred_path):
|
|
logger.debug("Refresh token was already consumed by an uncommitted rotation "
|
|
"- refusing to replay it; run 'hermes auth add anthropic'")
|
|
return None
|
|
fingerprint = hashlib.sha256(refresh_token.encode("utf-8")).hexdigest()[:32]
|
|
if fingerprint in _DEAD_REFRESH_TOKEN_FINGERPRINTS:
|
|
logger.debug("Claude Code refresh token was already rejected as terminally invalid - not replaying it")
|
|
return None
|
|
try:
|
|
refreshed = refresh_anthropic_oauth_pure(refresh_token, use_json=False)
|
|
except Exception as e:
|
|
if is_terminal_anthropic_refresh_error(e):
|
|
_DEAD_REFRESH_TOKEN_FINGERPRINTS.add(fingerprint)
|
|
logger.warning(
|
|
"Claude Code OAuth refresh token is terminally invalid (%s); Hermes cannot use this "
|
|
"login. Run 'hermes auth add anthropic' to give Hermes its own login.", e)
|
|
else:
|
|
logger.debug("Failed to refresh Claude Code token: %s", e)
|
|
return None
|
|
# The POST spent ``refresh_token``; this write is the commit step. On failure, fail closed and
|
|
# mark the pre-rotation pair as spent.
|
|
try:
|
|
_write_claude_code_credentials(
|
|
refreshed["access_token"], refreshed["refresh_token"], refreshed["expires_at_ms"],
|
|
spent_refresh_token=refresh_token,
|
|
)
|
|
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; "
|
|
"run 'hermes auth add anthropic' to give Hermes its own login",
|
|
cred_path, e,
|
|
)
|
|
mark_rotation_consumed_uncommitted(
|
|
refresh_token, creds.get("accessToken", ""), current.get("accessToken", ""),
|
|
current.get("refreshToken", ""), source_path=cred_path,
|
|
)
|
|
return None
|
|
logger.debug("Successfully refreshed Claude Code OAuth token")
|
|
return refreshed["access_token"]
|
|
except Exception as e:
|
|
# Lock/read failures keep the resolver's fail-soft contract.
|
|
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,
|
|
spent_refresh_token: str = "",
|
|
) -> None:
|
|
"""Commit refreshed credentials to ~/.claude/.credentials.json; ``CredentialPersistError`` on any failure (a
|
|
corrupt existing file included). *scopes* (or the previously stored scopes) are persisted because Claude Code
|
|
>=2.1.81 gates on ``"user:inference"`` being present."""
|
|
cred_path = claude_code_credentials_path()
|
|
try:
|
|
existing = json.loads(cred_path.read_text(encoding="utf-8-sig")) if cred_path.exists() else {}
|
|
except (OSError, ValueError) as e:
|
|
logger.error("Failed to write refreshed credentials to %s: %s", cred_path, e)
|
|
raise CredentialPersistError(cred_path, e) from e
|
|
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"]:
|
|
oauth_data["scopes"] = existing["claudeAiOauth"]["scopes"]
|
|
existing["claudeAiOauth"] = oauth_data
|
|
_commit_private_json(cred_path, existing, "credentials")
|
|
_mirror_claude_code_credentials_to_keychain(
|
|
access_token, refresh_token, expires_at_ms, spent_refresh_token=spent_refresh_token)
|
|
|
|
|
|
def _merge_keychain_credential_payload(
|
|
existing_payload: Dict[str, Any], access_token: str, refresh_token: str, expires_at_ms: int
|
|
) -> Dict[str, Any]:
|
|
"""Rotate the ``claudeAiOauth`` token triple over the existing Keychain payload,
|
|
preserving its metadata (``subscriptionType`` / ``rateLimitTier`` / ``scopes``).
|
|
|
|
Pure and host-agnostic so the merge semantics are unit-testable without a Keychain.
|
|
"""
|
|
merged = dict(existing_payload)
|
|
oauth = dict(existing_payload.get("claudeAiOauth") or {})
|
|
oauth.update({"accessToken": access_token, "refreshToken": refresh_token, "expiresAt": expires_at_ms})
|
|
merged["claudeAiOauth"] = oauth
|
|
return merged
|
|
|
|
|
|
def _mirror_claude_code_credentials_to_keychain(
|
|
access_token: str, refresh_token: str, expires_at_ms: int, *, spent_refresh_token: str
|
|
) -> None:
|
|
"""After a Hermes refresh, write the rotated pair into the Claude Code Keychain item too (#98334).
|
|
|
|
Claude Code on macOS reads the login Keychain first. Refresh tokens are single-use, so a refresh
|
|
that only updates the file leaves the Keychain holding a spent token and Claude Code logs itself
|
|
out. Only the item that held the pair we just spent is updated — a different pair there means a
|
|
different login (``CLAUDE_CONFIG_DIR``) or a rotation Claude Code already made, and clobbering it
|
|
would be the bug in the other direction. Best-effort: never raises, never creates an item.
|
|
"""
|
|
if platform.system() != "Darwin":
|
|
return
|
|
try:
|
|
item = _find_claude_code_keychain_item()
|
|
if item is None:
|
|
return
|
|
account, existing = item
|
|
oauth = existing.get("claudeAiOauth")
|
|
if not isinstance(oauth, dict) or oauth.get("refreshToken") != spent_refresh_token:
|
|
logger.debug("Keychain mirror skipped: item does not hold the pair that was just rotated")
|
|
return
|
|
argv, line = _keychain_mirror_command(
|
|
account, _merge_keychain_credential_payload(existing, access_token, refresh_token, expires_at_ms))
|
|
result = subprocess.run(
|
|
argv, input=line, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10,
|
|
)
|
|
except Exception as e: # the file commit already succeeded; a Keychain hiccup must not fail the rotation
|
|
logger.debug("Keychain mirror skipped (%s)", e)
|
|
return
|
|
if result.returncode != 0:
|
|
logger.debug("Keychain mirror failed (rc=%s): %s", result.returncode, (result.stderr or "").strip()[:200])
|
|
|
|
|
|
# ── Resolution ──
|
|
|
|
|
|
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 not creds:
|
|
return None
|
|
if is_rotation_consumed_uncommitted(creds.get("accessToken", ""), source_path=claude_code_credentials_path()):
|
|
# The file still holds the spent pre-rotation copy of a failed commit.
|
|
logger.debug("Claude Code credentials hold a rotated-but-uncommitted token - refusing")
|
|
return None
|
|
if is_claude_code_token_valid(creds):
|
|
logger.debug("Using Claude Code credentials (auto-detected)")
|
|
return creds["accessToken"]
|
|
logger.debug("Claude Code credentials expired — attempting refresh")
|
|
refreshed = _refresh_oauth_token(creds)
|
|
if not refreshed:
|
|
logger.debug("Token refresh failed — run 'hermes auth add anthropic' to give Hermes its own login")
|
|
return refreshed or None
|
|
|
|
|
|
def _prefer_refreshable_claude_code_token(env_token: str, creds: Optional[Dict[str, Any]]) -> Optional[str]:
|
|
"""Prefer refreshable Claude Code creds over a static env OAuth token: Hermes historically persisted setup tokens
|
|
into ANTHROPIC_TOKEN, and that static token would otherwise win before the refreshable file is inspected."""
|
|
if not (env_token and _is_oauth_token(env_token) and isinstance(creds, dict) and 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(*, skip_borrowed: bool = False) -> Optional[str]:
|
|
"""First available Anthropic OAuth token from credential_pool, read-only: enumerates with ``clear_expired=False,
|
|
refresh=False`` (never ``select()``) so diagnostic call sites (account_usage, ``hermes models``) never mutate
|
|
auth.json or hit the network; refresh-on-expiry belongs to the API call path's pool recovery."""
|
|
try:
|
|
from agent.credential_pool import AUTH_TYPE_OAUTH, load_pool
|
|
entries, _pending = load_pool("anthropic")._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 skip_borrowed and entry.source == "claude_code":
|
|
continue
|
|
# access_token may be an explicit null on a persisted entry; None.strip() would crash the resolver.
|
|
token = (getattr(entry, "access_token", None) or "").strip()
|
|
if getattr(entry, "auth_type", None) != AUTH_TYPE_OAUTH or not token:
|
|
continue
|
|
# load_pool() re-seeds rows from the singleton files, so a spent-but-uncommitted rotation
|
|
# (possibly from another process) looks healthy here.
|
|
entry_source_path = spent_rotation_source_path(getattr(entry, "source", None))
|
|
if any(
|
|
is_rotation_consumed_uncommitted(secret, source_path=entry_source_path)
|
|
for secret in (token, getattr(entry, "refresh_token", None))
|
|
):
|
|
logger.debug("Skipping Anthropic pool entry %s: rotated-but-uncommitted credential", getattr(entry, "id", "?"))
|
|
continue
|
|
return token
|
|
return None
|
|
|
|
|
|
def _available_anthropic_token(token: Optional[str], model: Optional[str]) -> Optional[str]:
|
|
"""Return *token* unless the pool holds an active cooldown for it on *model*.
|
|
|
|
Only model-aware callers (the API-call paths) are gated: diagnostics that
|
|
resolve a token without a model (usage display, model discovery) keep it.
|
|
"""
|
|
if not token or not model:
|
|
return token or None
|
|
try:
|
|
from agent.credential_pool import load_pool
|
|
if load_pool("anthropic").token_is_blocked(token, model=model):
|
|
return None
|
|
except Exception:
|
|
# Credential discovery must remain available when the pool store is
|
|
# unavailable or malformed.
|
|
logger.debug("Failed to check Anthropic model cooldown", exc_info=True)
|
|
return token
|
|
|
|
|
|
def resolve_anthropic_token(*, model: Optional[str] = None) -> Optional[str]:
|
|
"""Resolve an Anthropic token from all sources in priority order (see module docstring).
|
|
|
|
With *model*, a token the credential pool has benched for that model resolves to ``None``
|
|
instead of being handed straight back to the caller that just saw it rate-limited."""
|
|
_read_creds = functools.cache(read_claude_code_credentials) # read the file at most once per resolve
|
|
token = _first_env("ANTHROPIC_TOKEN", "CLAUDE_CODE_OAUTH_TOKEN")
|
|
if token:
|
|
return _available_anthropic_token(
|
|
_prefer_refreshable_claude_code_token(token, _read_creds()) or token, model,
|
|
)
|
|
api_key = _first_env("ANTHROPIC_API_KEY") # an explicit API key must not be shadowed by discovered OAuth creds
|
|
if api_key:
|
|
return _available_anthropic_token(api_key, model)
|
|
# The pool's claude_code row mirrors the same externally owned refresh grant.
|
|
return _available_anthropic_token(
|
|
_resolve_anthropic_pool_token(skip_borrowed=True) or _resolve_claude_code_token_from_credentials(_read_creds()),
|
|
model,
|
|
)
|
|
|
|
|
|
def run_oauth_setup_token() -> Optional[str]:
|
|
"""Run 'claude setup-token' interactively; the resulting token or None. FileNotFoundError if no 'claude' CLI."""
|
|
import shutil
|
|
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")
|
|
# Interactive: stdio inherited so the user can complete the OAuth prompt. noqa: subprocess-stdin
|
|
try:
|
|
subprocess.run([claude_path, "setup-token"])
|
|
except (KeyboardInterrupt, EOFError):
|
|
return None
|
|
creds = read_claude_code_credentials()
|
|
if creds and is_claude_code_token_valid(creds):
|
|
return creds["accessToken"]
|
|
return _first_env("CLAUDE_CODE_OAUTH_TOKEN", "ANTHROPIC_TOKEN") or None
|
|
|
|
|
|
# ── Hermes-native PKCE OAuth flow (~/.hermes/.anthropic_oauth.json); mirrors Claude Code / pi-ai / OpenCode ──
|
|
|
|
|
|
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`` inside a named profile (None in classic mode); used to commit a
|
|
rotation of a grant the profile borrowed via the pool's root fallback."""
|
|
try:
|
|
from hermes_constants import get_default_hermes_root
|
|
root = get_default_hermes_root()
|
|
return None if root.resolve(strict=False) == get_hermes_home().resolve(strict=False) else root / ".anthropic_oauth.json"
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _generate_pkce() -> tuple:
|
|
"""Generate PKCE code_verifier and code_challenge (S256)."""
|
|
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 webbrowser
|
|
from urllib.parse import urlencode
|
|
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,
|
|
}
|
|
auth_url = f"https://claude.ai/oauth/authorize?{urlencode(params)}"
|
|
print("\n".join([
|
|
"", "Authorize Hermes with your Claude Pro/Max subscription.", "",
|
|
"╭─ Claude Pro/Max Authorization ────────────────────╮",
|
|
"│ │",
|
|
"│ Open this link in your browser: │",
|
|
"╰───────────────────────────────────────────────────╯",
|
|
"", f" {auth_url}", "",
|
|
]))
|
|
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():
|
|
with contextlib.suppress(Exception):
|
|
webbrowser.open(auth_url)
|
|
print(" (Browser opened automatically)")
|
|
print("\nAfter authorizing, you'll see a code. Paste it below.\n")
|
|
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, received_state = splits[0], (splits[1] if len(splits) > 1 else "")
|
|
if received_state != oauth_state: # CSRF guard (RFC 6749 §10.12)
|
|
logger.warning("OAuth state mismatch — possible CSRF, aborting")
|
|
return None
|
|
try:
|
|
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()
|
|
result = _post_oauth_token(exchange_data, content_type="application/json", timeout=15, what="exchange")
|
|
except Exception as e:
|
|
print(f"Token exchange failed: {e}")
|
|
return None
|
|
if not result.get("access_token"):
|
|
print("No access token in response.")
|
|
return None
|
|
return _oauth_token_state(result)
|
|
|
|
|
|
def read_hermes_oauth_credentials() -> Optional[Dict[str, Any]]:
|
|
"""Read Hermes-managed OAuth credentials from ~/.hermes/.anthropic_oauth.json."""
|
|
data = _load_json_if_exists(_get_hermes_oauth_file(), "Hermes OAuth credentials")
|
|
return data if data is not None and data.get("accessToken") else None
|
|
|
|
|
|
def _write_hermes_oauth_credentials(
|
|
access_token: str, refresh_token: Optional[str], expires_at_ms: Optional[int], *, target: Optional[Path] = None
|
|
) -> None:
|
|
"""Commit refreshed hermes_pkce tokens to ~/.hermes/.anthropic_oauth.json (``CredentialPersistError`` on failure).
|
|
``target`` lets a named profile commit a grant it BORROWED from the global root back to the ROOT singleton
|
|
instead of forking a copy under its own HERMES_HOME; without this write-through the next ``load_pool()``
|
|
re-seeds the stale (consumed) pair from the file over the rotated pool entry."""
|
|
_commit_private_json(
|
|
target if target is not None else _get_hermes_oauth_file(),
|
|
{"accessToken": access_token, "refreshToken": refresh_token, "expiresAt": expires_at_ms},
|
|
"Hermes OAuth credentials",
|
|
)
|