feat(provider): add Anthropic OAuth provider (Claude Pro/Max) #132
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | ||||||
|---|---|---|---|---|---|---|---|---|
| @@ -0,0 +1,71 @@ | ||||||||
| #!/usr/bin/env python3 | ||||||||
| """ | ||||||||
| Two-step Anthropic OAuth credential setup. | ||||||||
|
|
||||||||
| Step 1 (no args): Generate auth URL + save verifier | ||||||||
| python scripts/setup_anthropic_cred.py | ||||||||
|
|
||||||||
| Step 2 (with code): Exchange code for tokens | ||||||||
| python scripts/setup_anthropic_cred.py "CODE_FROM_BROWSER" | ||||||||
| """ | ||||||||
| import sys | ||||||||
| import os | ||||||||
| import json | ||||||||
| sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) | ||||||||
|
|
||||||||
| import asyncio | ||||||||
| from pathlib import Path | ||||||||
| from rotator_library.providers.anthropic_auth_base import ( | ||||||||
| _generate_pkce, _build_authorize_url, AnthropicAuthBase | ||||||||
| ) | ||||||||
|
|
||||||||
| STATE_FILE = Path(__file__).parent / ".anthropic_pkce_state.json" | ||||||||
| OAUTH_DIR = Path(__file__).parent / ".." / "oauth_creds" | ||||||||
|
|
||||||||
| async def exchange_code(auth_code: str): | ||||||||
| if not STATE_FILE.exists(): | ||||||||
| print("Error: PKCE state file not found. Please run Step 1 first.") | ||||||||
| sys.exit(1) | ||||||||
| state = json.loads(STATE_FILE.read_text()) | ||||||||
| verifier = state["verifier"] | ||||||||
|
|
||||||||
| auth = AnthropicAuthBase() | ||||||||
| tokens = await auth._exchange_code(auth_code.strip(), verifier) | ||||||||
|
|
||||||||
| import time | ||||||||
| creds = { | ||||||||
| **tokens, | ||||||||
| "email": "anthropic-oauth-user", | ||||||||
| "_proxy_metadata": { | ||||||||
| "email": "anthropic-oauth-user", | ||||||||
| "last_check_timestamp": time.time(), | ||||||||
| "credential_type": "oauth", | ||||||||
| }, | ||||||||
| } | ||||||||
|
|
||||||||
| oauth_dir = OAUTH_DIR.resolve() | ||||||||
| oauth_dir.mkdir(parents=True, exist_ok=True) | ||||||||
| existing = sorted(oauth_dir.glob("anthropic_oauth_*.json")) | ||||||||
| next_num = len(existing) + 1 | ||||||||
| file_path = oauth_dir / f"anthropic_oauth_{next_num}.json" | ||||||||
|
Comment on lines
+48
to
+50
There was a problem hiding this comment. Avoid If Suggested fix- existing = sorted(oauth_dir.glob("anthropic_oauth_*.json"))
- next_num = len(existing) + 1
+ existing_nums = [
+ int(path.stem.split("_")[-1])
+ for path in oauth_dir.glob("anthropic_oauth_*.json")
+ if path.stem.split("_")[-1].isdigit()
+ ]
+ next_num = max(existing_nums, default=0) + 1🤖 Prompt for AI Agents |
||||||||
|
|
||||||||
| file_path.write_text(json.dumps(creds, indent=2)) | ||||||||
| os.chmod(file_path, 0o600) | ||||||||
|
Comment on lines
+52
to
+53
There was a problem hiding this comment. OAuth token file briefly world-readable (TOCTOU race)
from rotator_library.utils.resilient_io import safe_write_json
import logging
_log = logging.getLogger(__name__)
# replace lines 52-53 with:
safe_write_json(str(file_path), creds, _log, secure_permissions=True)
Comment on lines
+52
to
+53
There was a problem hiding this comment. Create the credential file with secure permissions from creation time. Line 52 creates the JSON under the process umask, and Line 53 narrows it only afterward. For access/refresh tokens, this should use the same atomic 🤖 Prompt for AI Agents |
||||||||
| STATE_FILE.unlink(missing_ok=True) | ||||||||
|
|
||||||||
| print(f"Credential saved to: {file_path}") | ||||||||
| print(f"Access token prefix: {tokens['access_token'][:20]}...") | ||||||||
|
Comment on lines
+25
to
+57
There was a problem hiding this comment. if token exchange fails (line 33), STATE_FILE with PKCE verifier remains on disk; wrap in try-finally to ensure cleanup Prompt To Fix With AIThis is a comment left during a code review.
Path: scripts/setup_anthropic_cred.py
Line: 25-57
Comment:
if token exchange fails (line 33), STATE_FILE with PKCE verifier remains on disk; wrap in try-finally to ensure cleanup
How can I resolve this? If you propose a fix, please make it concise. |
||||||||
|
|
||||||||
| def step1(): | ||||||||
| verifier, challenge = _generate_pkce() | ||||||||
| url = _build_authorize_url(verifier, challenge) | ||||||||
| STATE_FILE.write_text(json.dumps({"verifier": verifier, "challenge": challenge})) | ||||||||
|
There was a problem hiding this comment. PKCE state file written world-readable
Suggested change
|
||||||||
| print("Open this URL in your browser, authorize, then copy the code:\n") | ||||||||
| print(url) | ||||||||
| print(f"\nThen run: python scripts/setup_anthropic_cred.py \"PASTE_CODE_HERE\"") | ||||||||
|
|
||||||||
| if __name__ == "__main__": | ||||||||
| if len(sys.argv) > 1: | ||||||||
| asyncio.run(exchange_code(sys.argv[1])) | ||||||||
| else: | ||||||||
| step1() | ||||||||
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.
This will raise a
FileNotFoundErrorif Step 2 is run before Step 1. Consider adding a check: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.
Good catch, fixed in 7aa20d4. Added the existence check before reading.