You can close this window and return to the terminal.
""") elif 'error' in params: error = params['error'][0] error_desc = params.get('error_description', ['Unknown error'])[0] self.send_response(400) self.send_header('Content-type', 'text/html') self.end_headers() self.wfile.write(f"""Error: {error}
Description: {error_desc}
""".encode()) auth_code = None def log_message(self, format, *args): """Suppress logging""" pass def start_callback_server(): """Start local server to receive OAuth callback""" server = HTTPServer(('localhost', 8080), CallbackHandler) server_ready.set() server.handle_request() # Handle one request then stop def get_authorization_code(): """Step 1: Get authorization code via browser""" print("\n=== Step 1: Getting Authorization Code ===") # Start local server in background server_thread = threading.Thread(target=start_callback_server, daemon=True) server_thread.start() server_ready.wait() # Wait for server to be ready # Build authorization URL auth_params = { 'client_id': CLIENT_ID, 'response_type': 'code', 'redirect_uri': REDIRECT_URI, 'scope': SCOPE, 'state': 'test12345', 'prompt': 'select_account' } auth_request_url = f"{AUTH_URL}?" + "&".join([f"{k}={v}" for k, v in auth_params.items()]) print(f"Opening browser for authentication...") print(f"URL: {auth_request_url}\n") webbrowser.open(auth_request_url) print("Waiting for callback... (authenticate in your browser)") server_thread.join(timeout=120) # Wait up to 2 minutes if auth_code: print(f"✓ Authorization code received: {auth_code[:20]}...") return auth_code else: print("✗ Failed to receive authorization code") return None def exchange_code_for_token(code): """Step 2: Exchange authorization code for access token""" print("\n=== Step 2: Exchanging Code for Access Token ===") token_data = { 'client_id': CLIENT_ID, 'client_secret': CLIENT_SECRET, 'code': code, 'redirect_uri': REDIRECT_URI, 'grant_type': 'authorization_code', 'scope': SCOPE } try: response = requests.post(TOKEN_URL, data=token_data) response.raise_for_status() token_response = response.json() access_token = token_response.get('access_token') print(f"✓ Access token received: {access_token[:20]}...") print(f" Token type: {token_response.get('token_type')}") print(f" Expires in: {token_response.get('expires_in')} seconds") # Decode token to show claims (for debugging) import base64 token_parts = access_token.split('.') if len(token_parts) >= 2: # Decode payload (add padding if needed) payload = token_parts[1] payload += '=' * (4 - len(payload) % 4) decoded = base64.b64decode(payload) claims = json.loads(decoded) print(f"\n Token Claims:") print(f" - User: {claims.get('upn', claims.get('unique_name', 'N/A'))}") print(f" - Audience: {claims.get('aud', 'N/A')}") print(f" - Issuer: {claims.get('iss', 'N/A')}") return access_token except requests.exceptions.RequestException as e: print(f"✗ Token exchange failed: {e}") if hasattr(e.response, 'text'): print(f" Error details: {e.response.text}") return None def test_snowflake_connection(access_token): """Step 3: Test Snowflake connection with OAuth token""" print("\n=== Step 3: Testing Snowflake Connection ===") snowflake_url = f"https://{SNOWFLAKE_ACCOUNT}.snowflakecomputing.com/session/v1/login-request" login_data = { "data": { "AUTHENTICATOR": "OAUTH", "TOKEN": access_token, "ACCOUNT": SNOWFLAKE_ACCOUNT } } try: response = requests.post( snowflake_url, json=login_data, headers={'Content-Type': 'application/json'} ) if response.status_code == 200: result = response.json() print(f"✓ Snowflake authentication successful!") print(f" Session token: {result.get('data', {}).get('token', 'N/A')[:20]}...") print(f" Master token: {result.get('data', {}).get('masterToken', 'N/A')[:20]}...") return True else: print(f"✗ Snowflake authentication failed: {response.status_code}") print(f" Response: {response.text}") return False except requests.exceptions.RequestException as e: print(f"✗ Snowflake connection error: {e}") return False def main(): """Main test flow""" print("=" * 60) print("Azure AD OAuth + Snowflake Integration Test") print("=" * 60) # Check if client secret is set if CLIENT_SECRET == "YOUR_CLIENT_SECRET_HERE": print("\n⚠️ WARNING: Please set your CLIENT_SECRET in the script!") print(" Edit the CLIENT_SECRET variable at the top of this file.") return # Step 1: Get authorization code code = get_authorization_code() if not code: print("\n❌ Test failed: Could not get authorization code") return # Step 2: Exchange for token token = exchange_code_for_token(code) if not token: print("\n❌ Test failed: Could not get access token") return # Step 3: Test Snowflake success = test_snowflake_connection(token) if success: print("\n" + "=" * 60) print("✅ ALL TESTS PASSED!") print("=" * 60) print("\nYour OAuth integration is working correctly.") print("You can now configure Alteryx with confidence.") else: print("\n" + "=" * 60) print("❌ SNOWFLAKE TEST FAILED") print("=" * 60) print("\nPossible issues:") print("1. Check Snowflake external OAuth integration configuration") print("2. Verify audience list includes your client ID") print("3. Ensure user mapping is correct (UPN)") print("4. Check if OAuth integration is enabled in Snowflake") if __name__ == "__main__": try: main() except KeyboardInterrupt: print("\n\nTest interrupted by user") except Exception as e: print(f"\n❌ Unexpected error: {e}") import traceback traceback.print_exc()