-
Notifications
You must be signed in to change notification settings - Fork 2
✨ Add Claude OAuth token discovery chain (closes #102) #103
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 1 commit
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -5,19 +5,124 @@ | |
| import os | ||
| import subprocess | ||
| from abc import ABC, abstractmethod | ||
| from pathlib import Path | ||
| from typing import TYPE_CHECKING | ||
|
|
||
| from .settings import CLAUDE_ENV_DISABLE_ADAPTIVE_THINKING, CLAUDE_ENV_DISABLE_THINKING, CLAUDE_ENV_MAX_THINKING_TOKENS | ||
|
|
||
| if TYPE_CHECKING: | ||
| from pathlib import Path | ||
|
|
||
| from .definitions import AgentConfig | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
| DEBUG_OUTPUT_MAX_CHARS = 2000 | ||
|
|
||
| OAUTH_TOKEN_ENV = "CLAUDE_CODE_OAUTH_TOKEN" # noqa: S105 | ||
| OAUTH_TOKEN_FILE_ENV = "CLAUDE_OAUTH_TOKEN_FILE" # noqa: S105 | ||
| CONVENTIONAL_TOKEN_FILE: Path = Path.home() / ".tokens" / ".claude-oauth-token" | ||
| CREDENTIALS_JSON_FILE: Path = Path.home() / ".claude" / ".credentials.json" | ||
|
|
||
|
|
||
| class OAuthTokenNotFoundError(RuntimeError): | ||
| """Raised when no Claude OAuth token can be resolved from any source.""" | ||
|
|
||
|
|
||
| def _read_token_file(path: Path) -> str | None: | ||
| """Read and strip a token file. Returns None on FileNotFoundError or empty content. | ||
|
|
||
| Logs a WARNING and returns None on PermissionError. | ||
| """ | ||
| try: | ||
| content = path.read_text() | ||
| except FileNotFoundError: | ||
| return None | ||
| except PermissionError as exc: | ||
| logger.warning("auth: cannot read token file %s: %s", path, exc) | ||
| return None | ||
| token = content.strip() | ||
| return token or None | ||
|
|
||
|
|
||
| def _read_credentials_json(path: Path) -> str | None: | ||
|
Owner
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. move function to Runner method
Owner
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Done in 9c638ac — |
||
| """Read ``claudeAiOauth.accessToken`` from the Claude Code credentials file. | ||
|
|
||
| Schema observed: ``{"claudeAiOauth": {"accessToken": "..."}}``. | ||
| Returns None if missing/empty; logs a WARNING and returns None on parse or schema errors. | ||
| """ | ||
| try: | ||
| content = path.read_text() | ||
| except FileNotFoundError: | ||
| return None | ||
| except PermissionError as exc: | ||
| logger.warning("auth: cannot read credentials file %s: %s", path, exc) | ||
| return None | ||
| try: | ||
| data = json.loads(content) | ||
| except json.JSONDecodeError as exc: | ||
| logger.warning("auth: failed to parse %s as JSON: %s", path, exc) | ||
| return None | ||
| try: | ||
| token = data["claudeAiOauth"]["accessToken"] | ||
| except (KeyError, TypeError) as exc: | ||
| logger.warning("auth: %s missing claudeAiOauth.accessToken: %s", path, exc) | ||
| return None | ||
| if not isinstance(token, str): | ||
| logger.warning("auth: %s claudeAiOauth.accessToken is not a string", path) | ||
| return None | ||
| token = token.strip() | ||
| return token or None | ||
|
|
||
|
|
||
| def _xdg_token_path() -> Path: | ||
|
Owner
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. move function to Runner method
Owner
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Done in 9c638ac — |
||
| """Compute the XDG-compliant Claude OAuth token path at call time.""" | ||
| xdg_config_home = os.environ.get("XDG_CONFIG_HOME", "").strip() | ||
| base = Path(xdg_config_home) if xdg_config_home else Path.home() / ".config" | ||
| return base / "claude" / "oauth-token" | ||
|
|
||
|
|
||
| def _resolve_oauth_token() -> tuple[str, str]: | ||
|
Owner
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. move function to Runner method
Owner
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Done in 9c638ac — |
||
| """Resolve a Claude OAuth token from the discovery chain. | ||
|
|
||
| Returns ``(token, source_label)`` where ``source_label`` names the source that | ||
| produced the token. Raises :class:`OAuthTokenNotFoundError` when no source | ||
| yields a non-empty token. | ||
| """ | ||
| env_token = os.environ.get(OAUTH_TOKEN_ENV, "").strip() | ||
| if env_token: | ||
| return env_token, f"env {OAUTH_TOKEN_ENV}" | ||
|
|
||
| checked: list[str] = [f"env {OAUTH_TOKEN_ENV}"] | ||
|
|
||
| custom_path_str = os.environ.get(OAUTH_TOKEN_FILE_ENV, "").strip() | ||
| if custom_path_str: | ||
| custom_path = Path(custom_path_str).expanduser() | ||
| checked.append(f"env {OAUTH_TOKEN_FILE_ENV}={custom_path}") | ||
| token = _read_token_file(custom_path) | ||
| if token: | ||
| return token, f"file {custom_path} (via {OAUTH_TOKEN_FILE_ENV})" | ||
| else: | ||
| checked.append(f"env {OAUTH_TOKEN_FILE_ENV} (unset)") | ||
|
|
||
| checked.append(f"file {CONVENTIONAL_TOKEN_FILE}") | ||
| conventional = _read_token_file(CONVENTIONAL_TOKEN_FILE) | ||
| if conventional: | ||
| return conventional, f"file {CONVENTIONAL_TOKEN_FILE}" | ||
|
|
||
| xdg_path = _xdg_token_path() | ||
| checked.append(f"file {xdg_path}") | ||
| xdg_token = _read_token_file(xdg_path) | ||
| if xdg_token: | ||
| return xdg_token, f"file {xdg_path}" | ||
|
|
||
| checked.append(f"file {CREDENTIALS_JSON_FILE}") | ||
| credentials_token = _read_credentials_json(CREDENTIALS_JSON_FILE) | ||
| if credentials_token: | ||
| return credentials_token, f"file {CREDENTIALS_JSON_FILE}" | ||
|
|
||
| locations = ", ".join(checked) | ||
| msg = f"no Claude credentials found in any of: {locations}" | ||
| raise OAuthTokenNotFoundError(msg) | ||
|
|
||
|
|
||
| class Runner(ABC): | ||
| """Base class for task runners.""" | ||
|
|
@@ -88,6 +193,17 @@ def run( | |
| env = os.environ.copy() | ||
| env.pop("CLAUDECODE", None) | ||
|
|
||
| token, source = _resolve_oauth_token() | ||
| env[OAUTH_TOKEN_ENV] = token | ||
| if source != f"env {OAUTH_TOKEN_ENV}": | ||
| logger.info("[%s] auth: loaded %s from %s", issue_url, OAUTH_TOKEN_ENV, source) | ||
| if str(CREDENTIALS_JSON_FILE) in source: | ||
| logger.warning( | ||
| "[%s] auth: %s can be stale (Claude Code refreshes in RAM and may not write back)", | ||
| issue_url, | ||
| CREDENTIALS_JSON_FILE, | ||
| ) | ||
|
|
||
| if max_thinking_tokens is not None: | ||
| env[CLAUDE_ENV_MAX_THINKING_TOKENS] = str(max_thinking_tokens) | ||
| if disable_thinking: | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
move function to Runner method
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done in 9c638ac —
_read_token_fileis nowClaudeRunner._read_token_file(instance method).