diff --git a/.gitignore b/.gitignore
index 1f65ba1..fcf97a9 100644
--- a/.gitignore
+++ b/.gitignore
@@ -163,3 +163,6 @@ cython_debug/
# api-blaster OAuth credentials and tokens
.api-blaster/
+
+# claude local settings
+.claude/
diff --git a/oauth_session.py b/oauth_session.py
index 0ecec26..c149925 100644
--- a/oauth_session.py
+++ b/oauth_session.py
@@ -1,8 +1,20 @@
import json
import pathlib
+import webbrowser
+import threading
+import queue
+import os
+from http.server import HTTPServer, BaseHTTPRequestHandler
from typing import Dict, List, Optional
+from urllib.parse import urlparse, urlunparse
from requests_oauthlib import OAuth2Session
+# Allow OAuth2 over HTTP for localhost (development only)
+# This is safe because localhost traffic never leaves your machine
+os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'
+
+
+
class OAuthSession:
"""
@@ -72,16 +84,17 @@ def _load_config(self) -> Dict:
def _prompt_for_config(self) -> Dict:
"""Prompt user for OAuth2 configuration and save it."""
print("No config saved to file yet, lets get set up!")
+ print("\nFor the redirect URL, use http://localhost:8080/cb (recommended for OAuth)")
config = {
"client_id": input("Enter your Client ID: "),
"client_secret": input("Enter your Client Secret: "),
- "redirect_uri": input("Enter your Redirect URL (default: https://localhost:9090/cb): ") or "https://localhost:9090/cb"
+ "redirect_uri": input("Enter your Redirect URL (default: http://localhost:8080/cb): ") or "http://localhost:8080/cb"
}
-
+
# Save config
with open(self.conf_file_path, "w") as f:
json.dump(config, f)
-
+
return config
def _load_existing_token(self) -> Optional[Dict]:
@@ -89,43 +102,224 @@ def _load_existing_token(self) -> Optional[Dict]:
try:
with open(self.auth_file_path, "r") as f:
token = json.load(f)
-
+
# Validate required fields
required_fields = ["refresh_token", "expires_at"]
for field in required_fields:
if field not in token:
raise KeyError(f"Missing required field: {field}")
-
+
return token
-
+
except (FileNotFoundError, json.JSONDecodeError, KeyError):
return None
+
+ def _start_callback_server(self, host: str, port: int) -> tuple[Optional[HTTPServer], queue.Queue, threading.Event]:
+ """Start a local server to capture OAuth callback.
+
+ Returns:
+ Tuple of (server, response_queue, server_ready_event)
+ """
+ # Create queue for this server instance (thread-safe)
+ response_queue = queue.Queue()
+
+ # Create handler class with bound queue (via closure)
+ # This ensures each server instance has its own isolated queue
+ class BoundCallbackHandler(BaseHTTPRequestHandler):
+ def do_GET(self):
+ """Handle GET request with OAuth callback."""
+ # Store the response in the queue (captured from closure)
+ response_queue.put(self.path)
+
+ # Send success response
+ self.send_response(200)
+ self.send_header('Content-type', 'text/html')
+ self.end_headers()
+
+ # Show success message
+ success_html = """
+
+
+
+ Authentication Successful
+
+
+
+
+
+
Authentication Successful!
+
You can close this window and return to your terminal.
+
+
+
+ """
+ self.wfile.write(success_html.encode())
+
+ def log_message(self, format, *args):
+ """Suppress server log messages."""
+ pass
+
+ # Create server with the bound handler
+ # Event to signal when server is ready
+ server_ready = threading.Event()
+
+ try:
+ server = HTTPServer((host, port), BoundCallbackHandler)
+ except OSError as e:
+ # Port already in use or other binding error
+ print(f"Warning: Could not bind to {host}:{port} ({e}), falling back to manual flow")
+ return None, response_queue, server_ready
+
+ try:
+ # Start server in background thread
+ def run_server():
+ try:
+ server_ready.set() # Signal that server is ready
+ server.handle_request() # Handle one request then stop
+ finally:
+ server.server_close() # Clean up
+
+ server_thread = threading.Thread(target=run_server)
+ server_thread.daemon = True
+ server_thread.start()
+
+ # Wait for server to be ready (with short timeout)
+ if not server_ready.wait(timeout=5):
+ print("Warning: Server failed to start, falling back to manual flow")
+ server.server_close()
+ return None, response_queue, server_ready
+
+ # Give the server a moment to start listening
+ import time
+ time.sleep(0.1)
+
+ # Return server and queue - caller will wait for response
+ return server, response_queue, server_ready
+ except Exception as e:
+ # Always clean up on error
+ server.server_close()
+ raise
def _perform_oauth_flow(self) -> Dict:
"""Perform OAuth2 authorization flow to get initial token."""
print("No auth saved to file yet, lets get logged in!")
-
+
# Create session for authorization
oauth_session = OAuth2Session(
client_id=self.config["client_id"],
scope=self.scope,
redirect_uri=self.config["redirect_uri"]
)
-
+
# Get authorization URL
authorization_url, state = oauth_session.authorization_url(self.authorization_base_url)
- print("Please go here and authorize: ", authorization_url)
-
- # Get the authorization response
- redirect_response = input("Paste the full redirect URL here: ")
-
+
+ # Parse redirect URI to get host and port
+ redirect_uri_parsed = urlparse(self.config["redirect_uri"])
+ host = redirect_uri_parsed.hostname or "localhost"
+ port = redirect_uri_parsed.port or 8080
+
+ # Start local server to capture callback first
+ server, response_queue, server_ready = self._start_callback_server(host, port)
+
+ # Check if server started successfully
+ if server is None or not server_ready.is_set():
+ # Server failed to start, fall back to manual
+ print("\nAutomatic flow failed - server could not start.")
+ print(f"Opening browser for authorization...")
+ print(f"Please visit: {authorization_url}")
+ redirect_response = input("Paste the full redirect URL here: ")
+ else:
+ # Server started successfully, open browser for OAuth
+ print(f"\nOpening browser for authorization...")
+ print(f"If browser doesn't open automatically, visit: {authorization_url}")
+ webbrowser.open(authorization_url)
+
+ # Wait for callback from queue (with timeout)
+ try:
+ callback_path = response_queue.get(timeout=120)
+
+ # Success! Construct full redirect URL from path
+ # Parse the redirect URI to get base components
+ parsed_redirect = urlparse(self.config['redirect_uri'])
+ # Parse the callback path (includes query string)
+ parsed_callback = urlparse(callback_path)
+ # Combine: use base from redirect_uri, path and query from callback
+ redirect_response = urlunparse((
+ parsed_redirect.scheme,
+ parsed_redirect.netloc,
+ parsed_callback.path or parsed_redirect.path,
+ '', # params
+ parsed_callback.query,
+ '' # fragment
+ ))
+ print("✓ Authorization received!")
+ except queue.Empty:
+ # Timeout waiting for callback
+ print("\nAutomatic flow timed out waiting for authorization.")
+ print(f"Please visit: {authorization_url}")
+ redirect_response = input("Paste the full redirect URL here: ")
+
# Fetch the access token
token = oauth_session.fetch_token(
self.token_url,
client_secret=self.config["client_secret"],
authorization_response=redirect_response,
)
-
+
# Save the token
self._token_updater(token)
return token