Skip to content

Add automatic browser-based OAuth flow #6

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.

Sign up for GitHub

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

Merged
justinpecott merged 11 commits into from
Oct 15, 2025
Merged

Add automatic browser-based OAuth flow #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
46fa568
Add automatic browser-based OAuth flow with local HTTPS server
justinpecott Oct 15, 2025
60889e4
Fix URL construction bug in OAuth callback handling
justinpecott Oct 15, 2025
aa048d7
Fix server resource leak in callback handler
justinpecott Oct 15, 2025
39cf747
Fix race condition between server start and browser opening
justinpecott Oct 15, 2025
c5b10be
Remove unused imports
justinpecott Oct 15, 2025
ceba656
Fix thread safety issue in OAuth callback handler
justinpecott Oct 15, 2025
8a1643a
Fix race condition by using closure-bound handler class
justinpecott Oct 15, 2025
a4b37f5
Handle port binding failures gracefully
justinpecott Oct 15, 2025
1e17b06
Fix logic error in server readiness check
justinpecott Oct 15, 2025
6919066
Simplify OAuth flow with HTTP localhost callback
justinpecott Oct 15, 2025
47404bd
Update default redirect URL to include /cb path
justinpecott Oct 15, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,3 +163,6 @@ cython_debug/

# api-blaster OAuth credentials and tokens
.api-blaster/

# claude local settings
.claude/
222 changes: 208 additions & 14 deletions oauth_session.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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'
Copy link

Copilot AI Oct 15, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Setting OAUTHLIB_INSECURE_TRANSPORT globally affects all OAuth sessions in the process. Consider setting this only for localhost redirect URIs to prevent accidentally allowing insecure transport for production OAuth endpoints. You can check if the redirect_uri is localhost before setting this environment variable.

Copilot uses AI. Check for mistakes.




class OAuthSession:
"""
Expand Down Expand Up @@ -72,60 +84,242 @@ 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)")
Copy link

Copilot AI Oct 15, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Corrected capitalization of 'URL' in user-facing message.

Copilot uses AI. Check for mistakes.
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]:
"""Load existing token from file if available and valid."""
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.

Copy link

Copilot AI Oct 15, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The docstring doesn't document the host and port parameters. Add parameter descriptions to clarify their purpose and expected values.

Suggested change
Args:
host (str): Hostname or IP address for the local callback server to listen on (e.g., "localhost").
port (int): Port number for the local callback server (e.g., 8080).

Copilot uses AI. Check for mistakes.
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 = """
<!DOCTYPE html>
<html>
<head>
<title>Authentication Successful</title>
<style>
body {
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, Oxygen, Ubuntu, Cantarell, sans-serif;
display: flex;
align-items: center;
justify-content: center;
min-height: 100vh;
margin: 0;
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
}
.container {
text-align: center;
background: white;
padding: 60px 80px;
border-radius: 20px;
box-shadow: 0 20px 60px rgba(0,0,0,0.3);
}
.icon {
width: 80px;
height: 80px;
margin: 0 auto 30px;
animation: zap 0.6s ease-out;
}
h1 {
color: #2d3748;
margin: 0 0 15px 0;
font-size: 32px;
font-weight: 600;
}
p {
color: #718096;
margin: 0;
font-size: 16px;
}
@keyframes zap {
0% {
transform: scale(0) rotate(-180deg);
opacity: 0;
}
50% {
transform: scale(1.2) rotate(10deg);
}
100% {
transform: scale(1) rotate(0deg);
opacity: 1;
}
}
</style>
</head>
<body>
<div class="container">
<svg class="icon" viewBox="0 0 200 200" xmlns="http://www.w3.org/2000/svg">
<path d="M 110 20 L 70 100 L 95 100 L 80 180 L 140 90 L 110 90 Z"
fill="#000000"
stroke="#000000"
stroke-width="3"
stroke-linejoin="round"/>
</svg>
<h1>Authentication Successful!</h1>
<p>You can close this window and return to your terminal.</p>
</div>
</body>
</html>
"""
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
Comment on lines +240 to +243
Copy link

Copilot AI Oct 15, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If server_ready.wait() times out, the server thread may still be running. The thread is marked as daemon so it won't block process exit, but server.server_close() is called before the thread has confirmed startup. This could cause issues if the thread is still initializing. Consider adding a flag to signal the thread to stop or waiting briefly for the thread to complete.

Copilot uses AI. Check for mistakes.

# Give the server a moment to start listening
import time
time.sleep(0.1)
Comment on lines +246 to +247
Copy link

Copilot AI Oct 15, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The time module should be imported at the top of the file with other imports rather than inline within a function. This follows Python conventions and improves code readability.

Copilot uses AI. Check for mistakes.

# 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():
Copy link

Copilot AI Oct 15, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The condition 'not server_ready.is_set()' is redundant here. If the server is None (line 224), server_ready.set() is never called, so it will never be set. The second condition will always be True when server is None. Consider simplifying to just 'if server is None:' for clarity.

Suggested change
if server is None or not server_ready.is_set():
if server is None:

Copilot uses AI. Check for mistakes.
# 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)
Copy link

Copilot AI Oct 15, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The timeout value of 120 seconds is a magic number. Consider defining it as a named constant (e.g., OAUTH_CALLBACK_TIMEOUT_SECONDS = 120) at the class or module level to improve maintainability and make it easier to adjust.

Copilot uses AI. Check for mistakes.

# 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
))
Comment on lines +301 to +308
Copy link

Copilot AI Oct 15, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The logic for reconstructing the redirect URL could benefit from a comment explaining why we're combining components from both parsed_redirect and parsed_callback, particularly the fallback behavior for the path on line 304.

Copilot uses AI. Check for mistakes.
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
Expand Down