from __future__ import annotations import fcntl import hashlib import json import os import tempfile import time from dataclasses import dataclass from pathlib import Path from typing import Callable, Final, TextIO from claude_oauth import ( OAUTH_BETA_HEADER, USER_AGENT, ClaudeOAuthError, ClaudeOAuthRequest, ClaudeOAuthResponse, ClaudeOAuthTransport, default_transport, ) from runtime_paths import runtime_dir PROFILE_URL: Final[str] = "https://api.anthropic.com/api/oauth/profile" IDENTITY_CACHE_PATH: Final[Path] = runtime_dir() / "claude-token-identity.json" IDENTITY_RETRY_SECONDS: Final[float] = 5 * 60 IDENTITY_MIN_RETRY_SECONDS: Final[float] = 60 IDENTITY_REJECTED_RETRY_SECONDS: Final[float] = 30 * 60 IDENTITY_CACHE_CAPACITY: Final[int] = 64 @dataclass(frozen=True) class TokenIdentity: account_uuid: str email: str | None Records = dict[str, dict[str, object]] @dataclass class _Cache: tokens: Records failures: Records class ClaudeTokenIdentity: """Binds an OAuth access token to the account that owns it. The binding comes from the OAuth profile endpoint, the only source that speaks for the token itself; adjacent metadata files describe whichever account last wrote them. """ def __init__( self, cache_path: Path | None = None, transport: ClaudeOAuthTransport | None = None, timeout_secs: float = 10.0, now: Callable[[], float] = time.time, ) -> None: self._cache_path = IDENTITY_CACHE_PATH if cache_path is None else cache_path self._transport = default_transport if transport is None else transport self._timeout_secs = timeout_secs self._now = now def identify(self, credentials_path: Path) -> TokenIdentity | None: access_token = self._access_token(credentials_path) if access_token is None: return None return self.identify_token(access_token) def identify_token(self, access_token: str) -> TokenIdentity | None: key = hashlib.sha256(access_token.encode("utf-8")).hexdigest() with self._locked_cache() as cache: identity = self._cached_identity(cache.tokens, key) if identity is not None: return identity if self._now() < self._retry_not_before(cache.failures, key): return None identity, retry_after = self._fetch_identity(access_token) with self._locked_cache() as cache: if identity is None: cache.failures[key] = {"retry_not_before": self._now() + retry_after} else: cache.failures.pop(key, None) cache.tokens[key] = { "account_uuid": identity.account_uuid, "email": identity.email, "verified_at": self._now(), } self._prune(cache) self._write_cache(cache) return identity def account_uuid_for(self, credentials_path: Path) -> str | None: identity = self.identify(credentials_path) return None if identity is None else identity.account_uuid def _fetch_identity(self, access_token: str) -> tuple[TokenIdentity | None, float]: request = ClaudeOAuthRequest( method="GET", url=PROFILE_URL, headers={ "Accept": "application/json", "Authorization": f"Bearer {access_token}", "User-Agent": USER_AGENT, "anthropic-beta": OAUTH_BETA_HEADER, }, ) try: response = self._transport(request, self._timeout_secs) except (ClaudeOAuthError, OSError, TimeoutError): return None, IDENTITY_RETRY_SECONDS if not isinstance(response, ClaudeOAuthResponse): return None, IDENTITY_RETRY_SECONDS if response.status_code in {401, 403}: return None, IDENTITY_REJECTED_RETRY_SECONDS if response.status_code != 200: retry_after = self._retry_after_seconds(response) if retry_after is None: return None, IDENTITY_RETRY_SECONDS return None, max(retry_after, IDENTITY_MIN_RETRY_SECONDS) return self._identity_from_body(response.body), IDENTITY_RETRY_SECONDS @staticmethod def _identity_from_body(body: str) -> TokenIdentity | None: try: payload = json.loads(body) except (TypeError, ValueError): return None if not isinstance(payload, dict): return None account = payload.get("account") if not isinstance(account, dict): return None account_uuid = account.get("uuid") if not isinstance(account_uuid, str) or not account_uuid: return None email = account.get("email") return TokenIdentity(account_uuid, email if isinstance(email, str) and email else None) @staticmethod def _retry_after_seconds(response: ClaudeOAuthResponse) -> float | None: value = next( (header for key, header in response.headers.items() if key.lower() == "retry-after"), None, ) try: seconds = float(value) # type: ignore[arg-type] except (TypeError, ValueError): return None return seconds if seconds >= 0 else None @staticmethod def _access_token(credentials_path: Path) -> str | None: try: payload = json.loads(credentials_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return None if not isinstance(payload, dict): return None oauth = payload.get("claudeAiOauth") if not isinstance(oauth, dict): return None access_token = oauth.get("accessToken") return access_token if isinstance(access_token, str) and access_token else None @staticmethod def _cached_identity(tokens: Records, key: str) -> TokenIdentity | None: record = tokens.get(key) if record is None: return None account_uuid = record.get("account_uuid") if not isinstance(account_uuid, str) or not account_uuid: return None email = record.get("email") return TokenIdentity(account_uuid, email if isinstance(email, str) else None) @staticmethod def _retry_not_before(failures: Records, key: str) -> float: record = failures.get(key) if record is None: return 0.0 value = record.get("retry_not_before") if isinstance(value, bool) or not isinstance(value, (int, float)): return 0.0 return float(value) @staticmethod def _prune(cache: _Cache) -> None: tokens = cache.tokens if len(tokens) <= IDENTITY_CACHE_CAPACITY: return ordered = sorted(tokens.items(), key=lambda item: _verified_at(item[1])) for key, _ in ordered[: len(tokens) - IDENTITY_CACHE_CAPACITY]: tokens.pop(key, None) def _locked_cache(self) -> _LockedCache: return _LockedCache(self) def _read_cache(self) -> _Cache: try: payload = json.loads(self._cache_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): payload = None if not isinstance(payload, dict): payload = {} return _Cache(_records(payload.get("tokens")), _records(payload.get("failures"))) def _write_cache(self, cache: _Cache) -> None: self._cache_path.parent.mkdir(parents=True, exist_ok=True) temp_path: Path | None = None try: with tempfile.NamedTemporaryFile( "w", encoding="utf-8", dir=self._cache_path.parent, delete=False, ) as handle: handle.write( json.dumps({"tokens": cache.tokens, "failures": cache.failures}, indent=2) + "\n" ) handle.flush() os.fsync(handle.fileno()) temp_path = Path(handle.name) os.chmod(temp_path, 0o600) os.replace(temp_path, self._cache_path) except OSError: if temp_path is not None: temp_path.unlink(missing_ok=True) def _records(value: object) -> Records: if not isinstance(value, dict): return {} return {key: record for key, record in value.items() if isinstance(record, dict)} def _verified_at(record: dict[str, object]) -> float: value = record.get("verified_at") if isinstance(value, bool) or not isinstance(value, (int, float)): return 0.0 return float(value) class _LockedCache: def __init__(self, owner: ClaudeTokenIdentity) -> None: self._owner = owner self._handle: TextIO | None = None def __enter__(self) -> _Cache: lock_path = self._owner._cache_path.with_name(f"{self._owner._cache_path.name}.lock") lock_path.parent.mkdir(parents=True, exist_ok=True) self._handle = lock_path.open("w", encoding="utf-8") fcntl.flock(self._handle, fcntl.LOCK_EX) return self._owner._read_cache() def __exit__(self, *exc_info: object) -> None: if self._handle is not None: self._handle.close() self._handle = None __all__ = [ "IDENTITY_CACHE_PATH", "PROFILE_URL", "ClaudeTokenIdentity", "TokenIdentity", ]