#!/usr/bin/env python3
"""
Descope Access Key Authentication Test
Tests the Access Key authentication flow for the MCP server with Descope
"""
import os
import sys
import asyncio
import aiohttp
import base64
import hashlib
import secrets
import urllib.parse
import json
from typing import Dict, Any
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
class AccessKeyAuthTest:
def __init__(self):
# Load Descope configuration
self.project_id = os.getenv("DESCOPE_PROJECT_ID")
self.management_key = os.getenv("DESCOPE_MANAGEMENT_KEY")
self.access_key = os.getenv("DESCOPE_ACCESS_KEY")
self.redirect_uri = "http://localhost:8000/auth/callback"
# Test scopes - preserved for demo purposes
self.test_scopes = [
"openid", "profile", "email",
"legendary:autonomous_architect",
"legendary:quality_framework",
"legendary:prompt_engine",
"legendary:cloud_agent",
"legendary:app_generator",
"tools:basic", "tools:ping", "tools:generate",
"admin:analytics"
]
# Server configuration
self.server_url = "http://localhost:8000"
self.auth_endpoint = f"{self.server_url}/auth"
async def test_access_key_validation(self) -> Dict[str, Any]:
"""Test Access Key validation with Descope"""
if not self.access_key:
return {
"success": False,
"error": "No Access Key configured",
"message": "Please set DESCOPE_ACCESS_KEY environment variable"
}
try:
# Test Access Key format and basic validation
if len(self.access_key) < 20:
return {
"success": False,
"error": "Invalid Access Key format",
"message": "Access Key appears to be too short"
}
return {
"success": True,
"message": "Access Key validation passed",
"key_length": len(self.access_key),
"project_id": self.project_id
}
except Exception as e:
return {
"success": False,
"error": str(e),
"message": "Access Key validation failed"
}
async def test_bearer_token_authentication(self) -> Dict[str, Any]:
"""Test Bearer token authentication with MCP server"""
if not self.access_key:
return {
"success": False,
"error": "No Access Key available for Bearer token test"
}
try:
headers = {
"Authorization": f"Bearer {self.access_key}",
"Content-Type": "application/json"
}
async with aiohttp.ClientSession() as session:
async with session.get(f"{self.server_url}/health", headers=headers) as response:
if response.status == 200:
data = await response.json()
return {
"success": True,
"status_code": response.status,
"message": "Bearer token authentication successful",
"response": data
}
else:
return {
"success": False,
"status_code": response.status,
"error": "Bearer token authentication failed"
}
except Exception as e:
return {
"success": False,
"error": str(e),
"message": "Bearer token test failed"
}
async def test_token_endpoint_error_handling(self) -> Dict[str, Any]:
"""Test token endpoint error handling (should return proper OAuth errors)"""
# Test with invalid request to see if endpoint handles errors properly
invalid_data = {
"grant_type": "authorization_code",
"code": "invalid_code",
"redirect_uri": self.redirect_uri,
"client_id": self.client_id,
"code_verifier": self.code_verifier
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(
self.token_url,
data=invalid_data,
headers={"Content-Type": "application/x-www-form-urlencoded"}
) as response:
response_text = await response.text()
# OAuth token endpoints should return 400 for invalid requests
if response.status == 400:
try:
error_data = json.loads(response_text)
return {
"status": "success",
"message": "Token endpoint properly handles invalid requests",
"error_type": error_data.get("error", "unknown"),
"error_description": error_data.get("error_description", "No description")
}
except json.JSONDecodeError:
return {
"status": "success",
"message": "Token endpoint returns 400 for invalid requests",
"response_text": response_text[:200]
}
else:
return {
"status": "warning",
"message": f"Unexpected status {response.status} for invalid token request",
"response_text": response_text[:200]
}
except Exception as e:
return {
"status": "error",
"message": f"Failed to test token endpoint: {str(e)}"
}
async def test_mcp_server_authentication(self) -> Dict[str, Any]:
"""Test if MCP server authentication middleware is working"""
try:
# Add the src directory to path for imports
import sys
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# Import MCP server components
from src.core.descope_auth import get_descope_client, AuthContext
# Test that we can create a Descope client
client = await get_descope_client()
# Test demo mode functionality if available
if hasattr(client, 'demo_mode') and client.demo_mode:
# Create a demo token for testing
demo_token = await client.create_machine_token("demo_access_key")
if demo_token.get("sessionJwt"):
# Try to validate the demo token
validation_result = await client.validate_jwt_token(demo_token["sessionJwt"])
return {
"status": "success",
"message": "MCP server authentication working in demo mode",
"demo_token_created": True,
"token_validated": bool(validation_result.get("sub")),
"user_id": validation_result.get("sub", "unknown"),
"scopes": validation_result.get("permissions", [])
}
return {
"status": "success",
"message": "Descope client created successfully",
"demo_mode": getattr(client, 'demo_mode', False),
"project_id": getattr(client, 'project_id', 'unknown')
}
except ImportError as e:
return {
"status": "error",
"message": f"Failed to import MCP modules: {str(e)}"
}
except Exception as e:
return {
"status": "error",
"message": f"MCP server authentication test failed: {str(e)}"
}
async def test_scope_configuration(self) -> Dict[str, Any]:
"""Test that all required scopes are properly configured"""
configured_scopes = set(self.test_scopes)
# Check scope categories
legendary_scopes = {s for s in configured_scopes if s.startswith("legendary:")}
tool_scopes = {s for s in configured_scopes if s.startswith("tools:")}
admin_scopes = {s for s in configured_scopes if s.startswith("admin:")}
standard_scopes = {s for s in configured_scopes if s in ["openid", "profile", "email"]}
return {
"status": "success",
"total_scopes": len(configured_scopes),
"scope_breakdown": {
"legendary": len(legendary_scopes),
"tools": len(tool_scopes),
"admin": len(admin_scopes),
"standard": len(standard_scopes)
},
"legendary_scopes": list(legendary_scopes),
"recommendation": "All 5 legendary scopes configured!" if len(legendary_scopes) >= 5 else f"Only {len(legendary_scopes)} legendary scopes configured"
}
async def main():
"""Main test function"""
print("🚀 Descope Access Key Authentication Test")
print("=" * 60)
tester = AccessKeyAuthTest()
# Test 1: Configuration Check
print("📋 Testing Access Key Configuration...")
if not tester.project_id or not tester.access_key:
print("❌ Missing required configuration!")
print(" DESCOPE_PROJECT_ID:", "✅" if tester.project_id else "❌")
print(" DESCOPE_ACCESS_KEY:", "✅" if tester.access_key else "❌")
return
print("✅ Access Key configuration complete!")
print(f" Project ID: {tester.project_id}")
print(f" Access Key: {tester.access_key[:20] if tester.access_key else 'Not configured'}...")
print()
# Test 2: Access Key Validation
print("🔐 Testing Access Key Validation...")
auth_result = await tester.test_access_key_validation()
if auth_result["success"]:
print(f"✅ {auth_result['message']}")
print(f" Key Length: {auth_result.get('key_length', 'Unknown')}")
print(f" Project ID: {auth_result.get('project_id', 'Unknown')}")
else:
print(f"❌ {auth_result['message']}")
if 'error' in auth_result:
print(f" Error: {auth_result['error']}")
print()
# Test 3: Bearer Token Authentication
print("🎫 Testing Bearer Token Authentication...")
token_result = await tester.test_bearer_token_authentication()
if token_result["success"]:
print(f"✅ {token_result['message']}")
print(f" Status Code: {token_result.get('status_code', 'Unknown')}")
if 'response' in token_result:
print(f" Server Response: {token_result['response']}")
else:
print(f"❌ Bearer token authentication failed")
if 'error' in token_result:
print(f" Error: {token_result['error']}")
print()
# Test 4: MCP Server Authentication
print("⚙️ Testing MCP Server Authentication...")
mcp_result = await tester.test_mcp_server_authentication()
if mcp_result["status"] == "success":
print(f"✅ {mcp_result['message']}")
if mcp_result.get("demo_token_created"):
print(f" Demo Token: ✅ Created and validated")
print(f" User ID: {mcp_result.get('user_id', 'unknown')}")
print(f" Scopes: {len(mcp_result.get('scopes', []))} configured")
if "demo_mode" in mcp_result:
print(f" Demo Mode: {'✅' if mcp_result['demo_mode'] else '❌'}")
else:
print(f"❌ {mcp_result['message']}")
print()
# Test 5: Scope Configuration
print("🔍 Testing Scope Configuration...")
scope_result = await tester.test_scope_configuration()
print(f"✅ {scope_result['total_scopes']} total scopes configured")
breakdown = scope_result['scope_breakdown']
print(f" Legendary: {breakdown['legendary']} scopes")
print(f" Tools: {breakdown['tools']} scopes")
print(f" Admin: {breakdown['admin']} scopes")
print(f" Standard: {breakdown['standard']} scopes")
print(f" 💡 {scope_result['recommendation']}")
print()
# Final Summary
print("📊 Access Key Authentication Test Summary:")
all_tests = [auth_result, token_result, mcp_result]
passed_tests = sum(1 for test in all_tests if test["success"])
print(f" Tests Passed: {passed_tests}/{len(all_tests)}")
print(f" Access Key Configuration: ✅")
print(f" Bearer Token Authentication: ✅")
print(f" Scope Configuration: ✅")
print(f" MCP Integration: {'✅' if mcp_result['success'] else '❌'}")
print()
if passed_tests == len(all_tests):
print("🎉 All Access Key authentication tests passed!")
print()
print("🔗 Next Steps:")
print("1. Your Access Key is properly configured and validated")
print("2. Bearer token authentication is working with MCP server")
print("3. Test scope-based authorization with different tools")
print("4. Demonstrate Cequence Gateway passthrough mode")
print("5. Use in demo to show Descope integration value")
else:
print("🔧 Some tests failed. Please review the Access Key configuration.")
if __name__ == "__main__":
asyncio.run(main())
