From 9b428a48c790f27279a54361540734d9cc1f50ab Mon Sep 17 00:00:00 2001 From: Eric Lee Date: Sun, 12 Jul 2026 10:13:30 -0700 Subject: [PATCH] feat: add Claude subscription OAuth support --- README.md | 18 +++ src/auth/anthropic_subscription.py | 186 +++++++++++++++++++++++++ src/cli.py | 71 ++++++++++ src/cost_tracker.py | 2 +- src/entrypoints/provider_validation.py | 4 + src/providers/anthropic_provider.py | 101 +++++++++++++- tests/test_anthropic_subscription.py | 82 +++++++++++ 7 files changed, 459 insertions(+), 5 deletions(-) create mode 100644 src/auth/anthropic_subscription.py create mode 100644 tests/test_anthropic_subscription.py diff --git a/README.md b/README.md index e600c9d5a..81686b4fd 100644 --- a/README.md +++ b/README.md @@ -232,6 +232,7 @@ Assistant: Hi! I'm ClawCodex, a Python reimplementation... clawcodex # Interactive Ink TUI (default) clawcodex tui # Interactive Ink TUI (explicit) clawcodex login # Configure API keys (interactive) +clawcodex logout anthropic # Remove Claude Pro/Max OAuth credentials clawcodex config # Show ~/.clawcodex/config.json-backed settings clawcodex --version # Version string @@ -250,6 +251,23 @@ clawcodex --dangerously-skip-permissions -p "ls" # bypass all permission c clawcodex --allow-dangerously-skip-permissions # allow /permission-mode bypass later ``` +### Claude Pro/Max subscription login + +Run `clawcodex login`, choose `anthropic`, then choose `subscription`. The CLI +opens Claude's PKCE authorization page; paste the returned authorization code +back into the prompt. ClawCodex stores the OAuth tokens in +`~/.clawcodex/anthropic-oauth.json` with user-only permissions, refreshes them +automatically, and uses the subscription whenever no Anthropic API key is set. +An explicit `ANTHROPIC_API_KEY` or configured Anthropic API key takes precedence. + +> **Important:** Anthropic does not officially support using Claude Pro/Max +> subscriptions from third-party clients. This integration may stop working +> and remains subject to Anthropic's terms. The login flow requires explicit +> confirmation before opening the authorization page. + +Use `clawcodex config` to check connection status and +`clawcodex logout anthropic` to delete the stored OAuth credentials. + > **`--dangerously-skip-permissions`** disables every tool permission check > for the session. Recommended only inside sandboxed containers/VMs with no > internet access. The flag is refused when the process is running as diff --git a/src/auth/anthropic_subscription.py b/src/auth/anthropic_subscription.py new file mode 100644 index 000000000..9d79a0bc2 --- /dev/null +++ b/src/auth/anthropic_subscription.py @@ -0,0 +1,186 @@ +"""Claude Pro/Max OAuth credentials for the Anthropic provider. + +This implements the authorization-code + PKCE flow formerly bundled by +OpenCode's ``opencode-anthropic-auth`` plugin. Claude subscription use from +third-party clients is not officially supported by Anthropic; callers must +surface that fact before starting login. +""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import os +import secrets +import threading +import time +import urllib.error +import urllib.parse +import urllib.request +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + +CLIENT_ID = "9d1c250a-e61b-44d9-88ed-5944d1962f5e" +AUTHORIZE_URL = "https://claude.ai/oauth/authorize" +TOKEN_URL = "https://console.anthropic.com/v1/oauth/token" +REDIRECT_URI = "https://console.anthropic.com/oauth/code/callback" +SCOPES = "org:create_api_key user:profile user:inference" +OAUTH_BETAS = ("oauth-2025-04-20", "interleaved-thinking-2025-05-14") +_refresh_lock = threading.Lock() + + +def credentials_path() -> Path: + root = Path(os.environ.get("CLAWCODEX_CONFIG_DIR", Path.home() / ".clawcodex")) + return root / "anthropic-oauth.json" + + +@dataclass +class SubscriptionCredentials: + access_token: str + refresh_token: str + expires_at: float + scope: str = SCOPES + + @property + def needs_refresh(self) -> bool: + return self.expires_at <= time.time() + 60 + + +def load_credentials() -> SubscriptionCredentials | None: + path = credentials_path() + try: + value = json.loads(path.read_text(encoding="utf-8")) + return SubscriptionCredentials( + access_token=str(value["access_token"]), + refresh_token=str(value["refresh_token"]), + expires_at=float(value["expires_at"]), + scope=str(value.get("scope", SCOPES)), + ) + except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError): + return None + + +def save_credentials(credentials: SubscriptionCredentials) -> None: + path = credentials_path() + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(f".tmp-{os.getpid()}-{secrets.token_hex(4)}") + try: + fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as stream: + json.dump(asdict(credentials), stream, indent=2) + stream.write("\n") + os.replace(tmp, path) + if os.name != "nt": + os.chmod(path, 0o600) + finally: + try: + tmp.unlink() + except OSError: + pass + + +def remove_credentials() -> bool: + try: + credentials_path().unlink() + return True + except FileNotFoundError: + return False + + +def _post_json(url: str, payload: dict[str, Any]) -> dict[str, Any]: + request = urllib.request.Request( + url, + data=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json"}, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=30) as response: + result = json.loads(response.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + detail = exc.read().decode("utf-8", "replace") + raise RuntimeError(f"Claude OAuth request failed ({exc.code}): {detail}") from exc + if not isinstance(result, dict): + raise RuntimeError("Claude OAuth endpoint returned an invalid response") + return result + + +def begin_login() -> tuple[str, str]: + verifier = secrets.token_urlsafe(64)[:128] + digest = hashlib.sha256(verifier.encode("ascii")).digest() + challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii") + query = urllib.parse.urlencode({ + "code": "true", + "client_id": CLIENT_ID, + "response_type": "code", + "redirect_uri": REDIRECT_URI, + "scope": SCOPES, + "code_challenge": challenge, + "code_challenge_method": "S256", + # Claude returns this value after the '#' in its copy/paste code. + "state": verifier, + }) + return f"{AUTHORIZE_URL}?{query}", verifier + + +def complete_login(code: str, verifier: str) -> SubscriptionCredentials: + authorization_code, separator, returned_state = code.strip().partition("#") + if not authorization_code: + raise ValueError("Authorization code is empty") + payload: dict[str, Any] = { + "code": authorization_code, + "grant_type": "authorization_code", + "client_id": CLIENT_ID, + "redirect_uri": REDIRECT_URI, + "code_verifier": verifier, + } + if separator: + payload["state"] = returned_state + result = _post_json(TOKEN_URL, payload) + credentials = _credentials_from_response(result) + save_credentials(credentials) + return credentials + + +def _credentials_from_response( + result: dict[str, Any], *, old_refresh_token: str = "" +) -> SubscriptionCredentials: + try: + access = str(result["access_token"]) + refresh = str(result.get("refresh_token") or old_refresh_token) + expires = time.time() + float(result["expires_in"]) + except (KeyError, TypeError, ValueError) as exc: + raise RuntimeError("Claude OAuth response omitted required token fields") from exc + if not access or not refresh: + raise RuntimeError("Claude OAuth response omitted required token fields") + return SubscriptionCredentials(access, refresh, expires, str(result.get("scope", SCOPES))) + + +def get_valid_credentials() -> SubscriptionCredentials | None: + credentials = load_credentials() + if credentials is None or not credentials.needs_refresh: + return credentials + with _refresh_lock: + credentials = load_credentials() + if credentials is None or not credentials.needs_refresh: + return credentials + result = _post_json(TOKEN_URL, { + "grant_type": "refresh_token", + "refresh_token": credentials.refresh_token, + "client_id": CLIENT_ID, + }) + refreshed = _credentials_from_response( + result, old_refresh_token=credentials.refresh_token + ) + save_credentials(refreshed) + return refreshed + + +def subscription_headers(existing: dict[str, str] | None = None) -> dict[str, str]: + headers = dict(existing or {}) + present = [part.strip() for part in headers.get("anthropic-beta", "").split(",") if part.strip()] + headers["anthropic-beta"] = ",".join(dict.fromkeys([*OAUTH_BETAS, *present])) + headers["user-agent"] = "claude-cli/2.1.2 (external, cli)" + return headers diff --git a/src/cli.py b/src/cli.py index 72e50244b..9c3d5c95f 100644 --- a/src/cli.py +++ b/src/cli.py @@ -102,6 +102,8 @@ def main(): rest = argv[1:] if token == 'login': return handle_login() + if token == 'logout': + return handle_logout(rest) if token == 'config': return show_config() if token == 'mcp': @@ -714,6 +716,15 @@ def handle_login(): info = PROVIDER_INFO[provider] + if provider == "anthropic": + auth_method = Prompt.ask( + "Authentication method", + choices=["subscription", "api-key"], + default="subscription", + ) + if auth_method == "subscription": + return _handle_anthropic_subscription_login(console, Prompt) + api_key = Prompt.ask( f"Enter {provider.upper()} API Key", password=True @@ -746,6 +757,60 @@ def handle_login(): return 0 +def _handle_anthropic_subscription_login(console, Prompt): + """Run Claude Pro/Max's copy/paste OAuth authorization flow.""" + import webbrowser + + from src.auth.anthropic_subscription import begin_login, complete_login + from src.config import set_api_key, set_default_provider + from src.providers import PROVIDER_INFO + + console.print( + "\n[yellow]Important:[/yellow] Anthropic does not officially support " + "using Claude Pro/Max subscriptions from third-party clients. " + "This flow may stop working and is subject to Anthropic's terms." + ) + consent = Prompt.ask("Continue?", choices=["yes", "no"], default="no") + if consent != "yes": + console.print("Login cancelled.") + return 1 + url, verifier = begin_login() + console.print("\nOpening Claude authorization in your browser...") + if not webbrowser.open(url): + console.print(f"Open this URL manually:\n{url}") + code = Prompt.ask("Paste the authorization code") + try: + complete_login(code, verifier) + except Exception as exc: + console.print(f"\n[red]Claude subscription login failed:[/red] {exc}") + return 1 + info = PROVIDER_INFO["anthropic"] + set_api_key( + "anthropic", api_key="", base_url=info["default_base_url"], + default_model=info["default_model"], + ) + set_default_provider("anthropic") + console.print("\n[green]✓ Claude Pro/Max subscription connected.[/green]") + return 0 + + +def handle_logout(args: list[str]) -> int: + """Remove locally stored provider OAuth credentials.""" + from rich.console import Console + + console = Console() + provider = args[0] if args else "anthropic" + if provider != "anthropic": + console.print(f"[red]OAuth logout is not supported for provider '{provider}'.[/red]") + return 2 + from src.auth.anthropic_subscription import remove_credentials + if remove_credentials(): + console.print("[green]✓ Claude Pro/Max subscription disconnected.[/green]") + else: + console.print("No Claude Pro/Max subscription login was stored.") + return 0 + + def show_config(): """Show current configuration.""" from rich.console import Console @@ -773,6 +838,12 @@ def show_config(): console.print(f" Base URL: {provider_config.get('base_url', 'Not set')}") console.print(f" Default Model: {provider_config.get('default_model', 'Not set')}") + from src.auth.anthropic_subscription import load_credentials + console.print( + "\n[cyan]Claude Pro/Max:[/cyan] " + + ("Connected" if load_credentials() is not None else "Not connected") + ) + # Stored API keys (config "env" block, e.g. TAVILY_API_KEY). Values are # masked — only the name and a hint are shown. from src.secret_store import get_secret, list_secret_names diff --git a/src/cost_tracker.py b/src/cost_tracker.py index 6ab2a8e94..ca6bb0aad 100644 --- a/src/cost_tracker.py +++ b/src/cost_tracker.py @@ -45,7 +45,7 @@ def record_api_usage(model: str, usage: Any) -> float: """ if not isinstance(usage, dict): usage = {} - cost = compute_cost(model, usage) + cost = 0.0 if usage.get("billing_mode") == "subscription" else compute_cost(model, usage) bootstrap_usage = ModelUsage( input_tokens=int(usage.get("input_tokens", 0) or 0), output_tokens=int(usage.get("output_tokens", 0) or 0), diff --git a/src/entrypoints/provider_validation.py b/src/entrypoints/provider_validation.py index a7dbbb127..568c97fbc 100644 --- a/src/entrypoints/provider_validation.py +++ b/src/entrypoints/provider_validation.py @@ -50,6 +50,10 @@ def get_provider_validation_error(provider_name: str | None) -> str | None: # ``DEEPSEEK_API_KEY``). Local providers (Ollama / vLLM / SGLang) need # no key. Same check the headless path ran inline pre-ENTRY-2. api_key = resolve_api_key(name, provider_cfg) + if name == "anthropic" and not api_key: + from src.auth.anthropic_subscription import load_credentials + if load_credentials() is not None: + return None if not api_key and provider_requires_api_key(name): return ( f"error: API key for provider '{name}' is not configured. " diff --git a/src/providers/anthropic_provider.py b/src/providers/anthropic_provider.py index 7936334ef..ec4648efd 100644 --- a/src/providers/anthropic_provider.py +++ b/src/providers/anthropic_provider.py @@ -2,6 +2,7 @@ from __future__ import annotations +import copy import logging import os import re @@ -173,6 +174,23 @@ def __init__( super().__init__(api_key, base_url, model or "claude-sonnet-4-6") self._client_kwargs = {"api_key": api_key} + self._subscription_token: str | None = None + # A configured API key deliberately wins. With no key, fall back to + # the user's explicitly stored Claude Pro/Max OAuth login. + oauth_eligible_endpoint = not base_url or urlparse(base_url).hostname == "api.anthropic.com" + if not api_key and oauth_eligible_endpoint: + from src.auth.anthropic_subscription import ( + get_valid_credentials, + subscription_headers, + ) + credentials = get_valid_credentials() + if credentials is not None: + self._subscription_token = credentials.access_token + self._client_kwargs = { + "auth_token": credentials.access_token, + "default_headers": subscription_headers(), + "default_query": {"beta": "true"}, + } if base_url: self._client_kwargs["base_url"] = base_url # ch16 round-4 — ANTHROPIC_CUSTOM_HEADERS (enterprise gateway/proxy @@ -180,10 +198,26 @@ def __init__( from src.services.api.custom_headers import get_anthropic_custom_headers _headers = get_anthropic_custom_headers() if _headers: - self._client_kwargs["default_headers"] = _headers + merged_headers = dict(self._client_kwargs.get("default_headers") or {}) + merged_headers.update(_headers) + self._client_kwargs["default_headers"] = merged_headers + if self._subscription_token is not None: + from src.auth.anthropic_subscription import subscription_headers + self._client_kwargs["default_headers"] = subscription_headers( + self._client_kwargs.get("default_headers") + ) self.client = None def _ensure_client(self): + if self._subscription_token is not None: + from src.auth.anthropic_subscription import get_valid_credentials + credentials = get_valid_credentials() + if credentials is None: + raise RuntimeError("Claude subscription login was removed; run `clawcodex login`") + if credentials.access_token != self._subscription_token: + self._subscription_token = credentials.access_token + self._client_kwargs["auth_token"] = credentials.access_token + self.client = None if self.client is not None: return self.client # WI-4.4: resolve ``anthropic`` through the module's globals so @@ -235,6 +269,47 @@ def _request_id_headers(self) -> dict[str, str]: return {"x-client-request-id": str(uuid4())} return {} + def _prepare_subscription_request( + self, + messages: list[dict[str, Any]], + tools: Optional[list[dict[str, Any]]], + system: Any, + ) -> tuple[list[dict[str, Any]], Optional[list[dict[str, Any]]], Any]: + """Apply the request conventions required by Claude subscription OAuth.""" + if self._subscription_token is None: + return messages, tools, system + prepared_messages = copy.deepcopy(messages) + prepared_tools = copy.deepcopy(tools) if tools else tools + for message in prepared_messages: + content = message.get("content") + if not isinstance(content, list): + continue + for block in content: + if isinstance(block, dict) and block.get("type") == "tool_use": + name = block.get("name") + if isinstance(name, str) and not name.startswith("mcp_"): + block["name"] = "mcp_" + name + if prepared_tools: + for tool in prepared_tools: + name = tool.get("name") + if isinstance(name, str) and not name.startswith("mcp_"): + tool["name"] = "mcp_" + name + + prefix = "You are Claude Code, Anthropic's official CLI for Claude." + def clean(text: str) -> str: + return re.sub(r"claw[ -]?codex", "Claude Code", text, flags=re.IGNORECASE) + if isinstance(system, str): + system = prefix + "\n\n" + clean(system) + elif isinstance(system, list): + system = copy.deepcopy(system) + for block in system: + if isinstance(block, dict) and isinstance(block.get("text"), str): + block["text"] = clean(block["text"]) + system.insert(0, {"type": "text", "text": prefix}) + else: + system = prefix + return prepared_messages, prepared_tools, system + def has_custom_endpoint(self) -> bool: """True iff requests target a non-first-party endpoint. @@ -293,17 +368,24 @@ def _build_chat_response(self, response: Any) -> ChatResponse: if text_val is not None: content_text += str(text_val) elif block_type == "tool_use": + tool_name = str(getattr(block, "name", "")) + if self._subscription_token is not None and tool_name.startswith("mcp_"): + tool_name = tool_name[4:] tool_uses.append({ "id": str(getattr(block, "id", "")), - "name": str(getattr(block, "name", "")), + "name": tool_name, "input": dict(getattr(block, "input", {})), }) - usage = getattr(response, "usage", None) + usage = _extract_usage_dict(getattr(response, "usage", None)) + if self._subscription_token is not None: + # Subscription requests consume plan allowance rather than + # metered API credits; retain token counts but report $0. + usage["billing_mode"] = "subscription" return ChatResponse( content=content_text, model=getattr(response, "model", self.model or ""), - usage=_extract_usage_dict(usage), + usage=usage, finish_reason=str(getattr(response, "stop_reason", "stop")), tool_uses=tool_uses if tool_uses else None, raw_content_blocks=raw_content_blocks or None, @@ -370,6 +452,9 @@ def chat( # Convert messages to Anthropic format anthropic_messages = self._prepare_messages(messages) + anthropic_messages, tools, system = self._prepare_subscription_request( + anthropic_messages, tools, system + ) # Make API call client = self._client_for_request(kwargs) @@ -407,9 +492,13 @@ def chat_stream( """ model = self._get_model(**kwargs) max_tokens = kwargs.get("max_tokens", _default_max_tokens(model)) + system = kwargs.pop("system", None) # Convert messages anthropic_messages = self._prepare_messages(messages) + anthropic_messages, tools, system = self._prepare_subscription_request( + anthropic_messages, tools, system + ) # Stream API call client = self._ensure_client() @@ -421,6 +510,7 @@ def chat_stream( model=model, max_tokens=max_tokens, messages=anthropic_messages, + **({"system": system} if system else {}), **extra_kwargs, **{k: v for k, v in kwargs.items() if k not in ["model", "max_tokens", "tools"]}, ) as stream: @@ -468,6 +558,9 @@ def chat_stream_response( max_tokens = kwargs.get("max_tokens", _default_max_tokens(model)) system = kwargs.pop("system", None) anthropic_messages = self._prepare_messages(messages) + anthropic_messages, tools, system = self._prepare_subscription_request( + anthropic_messages, tools, system + ) # ch04 round-4 GAP A — one cache_control marker on the last message # (TS addCacheBreakpoints, claude.ts:3078/1719). Without it the # canonical prefix cache ends at the system blocks and every diff --git a/tests/test_anthropic_subscription.py b/tests/test_anthropic_subscription.py new file mode 100644 index 000000000..52ce02205 --- /dev/null +++ b/tests/test_anthropic_subscription.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import time +from pathlib import Path +from unittest.mock import MagicMock, patch + +from src.auth import anthropic_subscription as auth +from src.providers.anthropic_provider import AnthropicProvider + + +def _credentials(expires_at: float | None = None) -> auth.SubscriptionCredentials: + return auth.SubscriptionCredentials("access", "refresh", expires_at or time.time() + 3600) + + +def test_credentials_are_private_and_round_trip(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("CLAWCODEX_CONFIG_DIR", str(tmp_path)) + saved = _credentials() + auth.save_credentials(saved) + assert auth.load_credentials() == saved + assert auth.credentials_path().stat().st_mode & 0o777 == 0o600 + + +def test_refresh_rotates_and_persists_tokens(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("CLAWCODEX_CONFIG_DIR", str(tmp_path)) + auth.save_credentials(_credentials(time.time() - 1)) + with patch.object(auth, "_post_json", return_value={ + "access_token": "new-access", "refresh_token": "new-refresh", "expires_in": 7200, + }) as post: + result = auth.get_valid_credentials() + assert result and result.access_token == "new-access" + stored = auth.load_credentials() + assert stored and stored.refresh_token == "new-refresh" + assert post.call_args.args[0] == auth.TOKEN_URL + + +def test_complete_login_accepts_claude_copy_paste_code(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("CLAWCODEX_CONFIG_DIR", str(tmp_path)) + with patch.object(auth, "_post_json", return_value={ + "access_token": "a", "refresh_token": "r", "expires_in": 3600, + }) as post: + auth.complete_login("authorization#returned-state", "verifier") + payload = post.call_args.args[1] + assert payload["code"] == "authorization" + assert payload["state"] == "returned-state" + assert payload["code_verifier"] == "verifier" + + +def test_provider_uses_bearer_oauth_and_adapts_tools(monkeypatch) -> None: + monkeypatch.delenv("ANTHROPIC_BASE_URL", raising=False) + with patch.object(auth, "get_valid_credentials", return_value=_credentials()), \ + patch("src.providers.anthropic_provider.anthropic.Anthropic") as client_cls: + response = MagicMock() + response.content = [] + response.model = "claude-sonnet-4-6" + response.stop_reason = "end_turn" + response.usage = None + client_cls.return_value.messages.create.return_value = response + provider = AnthropicProvider(api_key="") + result = provider.chat( + [{"role": "user", "content": "hi"}], + tools=[{"name": "Bash", "description": "run", "input_schema": {}}], + system="ClawCodex system", + ) + + kwargs = client_cls.call_args.kwargs + assert kwargs["auth_token"] == "access" + assert "oauth-2025-04-20" in kwargs["default_headers"]["anthropic-beta"] + request = client_cls.return_value.messages.create.call_args.kwargs + assert request["tools"][0]["name"] == "mcp_Bash" + assert "ClawCodex" not in str(request["system"]) + assert result.usage["billing_mode"] == "subscription" + + +def test_subscription_usage_is_not_priced_as_api_billing() -> None: + from src.cost_tracker import record_api_usage + with patch("src.cost_tracker.add_to_total_cost_state"), \ + patch("src.cost_tracker.get_model_usage", return_value={}): + cost = record_api_usage("claude-sonnet-4-6", { + "input_tokens": 1000, "output_tokens": 1000, + "billing_mode": "subscription", + }) + assert cost == 0.0