#!/usr/bin/env python3 """ ADR-004 Manual OAuth Flow Test This is a simplified version that doesn't use Playwright automation. Instead, it prints URLs and waits for manual browser interaction. Usage: uv run python tests/manual/test_adr004_manual.py --provider nextcloud """ import argparse import asyncio import hashlib import logging import secrets from base64 import urlsafe_b64encode from http.server import BaseHTTPRequestHandler, HTTPServer from threading import Thread from urllib.parse import parse_qs, urlencode, urlparse import httpx logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) class CallbackHandler(BaseHTTPRequestHandler): """Handles OAuth callback redirect to localhost""" authorization_code = None state = None def do_GET(self): """Handle GET request with authorization code""" parsed = urlparse(self.path) params = parse_qs(parsed.query) # Ignore favicon requests if parsed.path == "/favicon.ico": self.send_response(200) self.send_header("Content-type", "image/x-icon") self.end_headers() return CallbackHandler.authorization_code = params.get("code", [None])[0] CallbackHandler.state = params.get("state", [None])[0] # Send success page self.send_response(200) self.send_header("Content-type", "text/html") self.end_headers() code_display = ( CallbackHandler.authorization_code[:50] + "..." if CallbackHandler.authorization_code else "No code received" ) html = """ Authorization Success

āœ“ Authorization Successful

Authorization code received. You can close this window and return to the terminal.

{} """.format(code_display) self.wfile.write(html.encode()) def log_message(self, format, *args): """Log HTTP requests""" logger.info(f"Callback server: {format % args}") def generate_pkce_challenge(): """Generate PKCE code verifier and challenge""" code_verifier = secrets.token_urlsafe(32) digest = hashlib.sha256(code_verifier.encode()).digest() code_challenge = urlsafe_b64encode(digest).decode().rstrip("=") return code_verifier, code_challenge async def test_oauth_manual( provider: str, mcp_server_url: str, nextcloud_host: str, ): """ Manual OAuth flow test - prints URLs for manual browser interaction. """ print("\n" + "=" * 70) print("ADR-004 MANUAL OAUTH FLOW TEST") print("=" * 70) print(f"Provider: {provider}") print(f"MCP Server: {mcp_server_url}") print(f"Nextcloud: {nextcloud_host}") print("=" * 70 + "\n") # Generate PKCE challenge code_verifier, code_challenge = generate_pkce_challenge() logger.info(f"āœ“ Generated PKCE challenge: {code_challenge[:16]}...") # Generate state for CSRF protection state = secrets.token_urlsafe(32) # Start local HTTP server for OAuth callback callback_port = 8765 redirect_uri = f"http://localhost:{callback_port}/callback" server = HTTPServer(("localhost", callback_port), CallbackHandler) server_thread = Thread(target=server.serve_forever, daemon=True) server_thread.start() logger.info(f"āœ“ Started callback server at {redirect_uri}") try: # Build authorization URL auth_params = { "response_type": "code", "client_id": "test-mcp-client", "redirect_uri": redirect_uri, "scope": "openid profile email offline_access notes:read notes:write", "state": state, "code_challenge": code_challenge, "code_challenge_method": "S256", } auth_url = f"{mcp_server_url}/oauth/authorize?{urlencode(auth_params)}" print("\n" + "=" * 70) print("STEP 1: AUTHORIZE THE MCP SERVER") print("=" * 70) print("\nšŸ“‹ Open this URL in your browser:\n") print(f" {auth_url}") print("\nšŸ“Œ What will happen:") print(" 1. You'll be redirected to Nextcloud/Keycloak login") print(" 2. Login with username: admin, password: admin") print(" 3. You'll see a consent screen asking to authorize the MCP server") print(" 4. Click 'Authorize' or 'Allow'") print(" 5. You'll be redirected to localhost:8765/callback") print(" 6. The authorization code will appear in the terminal\n") print("=" * 70) print("\nā³ Waiting for authorization... (timeout: 5 minutes)\n") # Wait for authorization code (with timeout) timeout = 300 # 5 minutes elapsed = 0 while not CallbackHandler.authorization_code and elapsed < timeout: await asyncio.sleep(1) elapsed += 1 if not CallbackHandler.authorization_code: raise RuntimeError("Timeout waiting for authorization code") authorization_code = CallbackHandler.authorization_code returned_state = CallbackHandler.state print("\nāœ“ Received authorization code!") logger.info(f"Code: {authorization_code[:16]}...") # Verify state if returned_state != state: raise RuntimeError( f"State mismatch! Expected {state}, got {returned_state}" ) logger.info("āœ“ State parameter verified (CSRF protection)") # Exchange authorization code for access token print("\n" + "=" * 70) print("STEP 2: EXCHANGE CODE FOR ACCESS TOKEN") print("=" * 70) async with httpx.AsyncClient() as client: token_response = await client.post( f"{mcp_server_url}/oauth/token", data={ "grant_type": "authorization_code", "code": authorization_code, "code_verifier": code_verifier, "redirect_uri": redirect_uri, "client_id": "test-mcp-client", }, timeout=30.0, ) if token_response.status_code != 200: print(f"\nāŒ Token exchange failed: {token_response.status_code}") print(f"Response: {token_response.text}") raise RuntimeError("Token exchange failed") token_data = token_response.json() access_token = token_data["access_token"] print("\nāœ“ Successfully received access token") print(f" Token: {access_token[:30]}...") print(f" Type: {token_data.get('token_type', 'Bearer')}") print(f" Expires: {token_data.get('expires_in', 'unknown')}s") # Test MCP tool call print("\n" + "=" * 70) print("STEP 3: CALL MCP TOOL WITH ACCESS TOKEN") print("=" * 70) async with httpx.AsyncClient() as client: mcp_request = { "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "nc_notes_search_notes", "arguments": {"query": "test"}, }, } mcp_response = await client.post( f"{mcp_server_url}/mcp", json=mcp_request, headers={ "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", "Accept": "application/json, text/event-stream", }, timeout=30.0, ) if mcp_response.status_code != 200: print(f"\nāŒ MCP tool call failed: {mcp_response.status_code}") print(f"Response: {mcp_response.text}") raise RuntimeError("MCP tool call failed") mcp_result = mcp_response.json() if "error" in mcp_result: print(f"\nāŒ MCP tool returned error: {mcp_result['error']}") raise RuntimeError(f"MCP tool error: {mcp_result['error']}") print("\nāœ“ MCP tool call succeeded!") print(f" Result: {mcp_result.get('result', {})}") # Summary print("\n" + "=" * 70) print("šŸŽ‰ ADR-004 OAUTH FLOW TEST - SUCCESS") print("=" * 70) print(f"Provider: {provider}") print(f"MCP Server: {mcp_server_url}") print(f"Nextcloud: {nextcloud_host}") print("") print("āœ“ User consented to MCP server access") print("āœ“ User consented to offline_access (refresh tokens)") print("āœ“ MCP server stored master refresh token") print("āœ“ Client received MCP access token via PKCE") print("āœ“ MCP tool call succeeded") print("āœ“ MCP server exchanged tokens in background") print("āœ“ Nextcloud data fetched successfully") print("=" * 70 + "\n") return {"success": True} finally: server.shutdown() logger.info("Stopped callback server") async def main(): parser = argparse.ArgumentParser( description="Manual test for ADR-004 OAuth Hybrid Flow" ) parser.add_argument( "--provider", choices=["nextcloud", "keycloak"], required=True, help="OAuth provider to test", ) parser.add_argument( "--mcp-server-url", default="http://localhost:8001", help="MCP server URL (default: http://localhost:8001)", ) parser.add_argument( "--nextcloud-host", default="http://localhost:8080", help="Nextcloud host URL (default: http://localhost:8080)", ) args = parser.parse_args() try: result = await test_oauth_manual( provider=args.provider, mcp_server_url=args.mcp_server_url, nextcloud_host=args.nextcloud_host, ) return 0 if result["success"] else 1 except KeyboardInterrupt: print("\n\nāš ļø Test interrupted by user") return 1 except Exception as e: logger.error(f"OAuth flow test failed: {e}", exc_info=True) print("\n" + "=" * 70) print("āŒ ADR-004 OAUTH FLOW TEST - FAILED") print("=" * 70) print(f"Error: {e}") print("=" * 70) return 1 if __name__ == "__main__": exit_code = asyncio.run(main()) exit(exit_code)