test_oauth.py•6.72 kB
#!/usr/bin/env python3
import os
import sys
import asyncio
import httpx
sys.path.append('.')
from mcp_github_jira_server import setup_jira_oauth, load_config
async def test_oauth_comprehensive():
print("Comprehensive Jira OAuth Setup & Testing")
print("=" * 60)
print("\n1. Loading configuration...")
load_config()
jira_url = os.getenv("JIRA_URL")
jira_client_id = os.getenv("JIRA_CLIENT_ID")
jira_client_secret = os.getenv("JIRA_CLIENT_SECRET")
existing_access_token = os.getenv("JIRA_ACCESS_TOKEN")
existing_refresh_token = os.getenv("JIRA_REFRESH_TOKEN")
print(f" JIRA_URL: {jira_url}")
print(f" JIRA_CLIENT_ID: {jira_client_id[:8] if jira_client_id else 'Not set'}...")
print(f" JIRA_CLIENT_SECRET: {'Set' if jira_client_secret else 'Not set'}")
print(f" JIRA_ACCESS_TOKEN: {'Set' if existing_access_token else 'Not set'}")
print(f" JIRA_REFRESH_TOKEN: {'Set' if existing_refresh_token else 'Not set'}")
if not all([jira_url, jira_client_id, jira_client_secret]):
print("\nError: Missing required OAuth configuration!")
print("Please check your .env file has:")
print("- JIRA_URL (your Atlassian instance URL)")
print("- JIRA_CLIENT_ID (from Atlassian Developer Console)")
print("- JIRA_CLIENT_SECRET (from Atlassian Developer Console)")
return
if existing_access_token:
print("\n2. Testing existing access token...")
success = await test_access_token(existing_access_token)
if success:
print("Existing access token is valid!")
print("\nSkipping OAuth setup. Use setup_jira_oauth if you need to refresh.")
return
else:
print("Existing access token is invalid or expired.")
if existing_refresh_token:
print("\n3. Attempting to refresh token...")
success = await attempt_token_refresh(jira_client_id, jira_client_secret, existing_refresh_token)
if success:
print("Token refreshed successfully!")
return
else:
print("Token refresh failed. Proceeding with new OAuth flow.")
print("\n2. Starting OAuth authorization flow...")
print(" This will:")
print(" - Generate a secure state parameter for CSRF protection")
print(" - Start a local server on localhost:8080")
print(" - Open your browser for Atlassian authorization")
print(" - Exchange authorization code for access & refresh tokens")
print(" - Validate tokens with Atlassian API")
print("\n Important: Make sure port 8080 is available!")
print("\n Press Ctrl+C to cancel at any time.")
input("\nPress Enter to continue...")
try:
oauth_result = await setup_jira_oauth()
print(f"\nOAuth Result:")
print("=" * 40)
print(oauth_result)
access_token = os.getenv("JIRA_ACCESS_TOKEN")
refresh_token = os.getenv("JIRA_REFRESH_TOKEN")
print(f"\nToken Status:")
print(f" Access Token: {'Obtained' if access_token else 'Not obtained'}")
print(f" Refresh Token: {'Obtained' if refresh_token else 'Not obtained'}")
if access_token:
print("\nTesting new access token...")
success = await test_access_token(access_token)
if success:
print("Success! OAuth setup completed and validated.")
print("\nTo make tokens persistent, add them to your .env file:")
print(f"JIRA_ACCESS_TOKEN={access_token}")
if refresh_token:
print(f"JIRA_REFRESH_TOKEN={refresh_token}")
else:
print("Token validation failed. Please check your OAuth app configuration.")
else:
print("\nOAuth flow completed but no tokens were obtained.")
print("Please check the error messages above for troubleshooting.")
except KeyboardInterrupt:
print("\n\nOAuth setup cancelled by user.")
except Exception as e:
print(f"\n\nError during OAuth setup: {e}")
import traceback
traceback.print_exc()
async def test_access_token(access_token: str) -> bool:
try:
response = httpx.get(
"https://api.atlassian.com/oauth/token/accessible-resources",
headers={"Authorization": f"Bearer {access_token}"},
timeout=10.0
)
if response.status_code == 200:
resources = response.json()
print(f" Token valid! Found {len(resources)} accessible resource(s)")
for resource in resources:
name = resource.get('name', 'Unknown')
url = resource.get('url', 'No URL')
scopes = resource.get('scopes', [])
print(f" - {name}: {url}")
print(f" Scopes: {', '.join(scopes)}")
return True
else:
print(f" Token validation failed: {response.status_code}")
return False
except Exception as e:
print(f" Token test error: {e}")
return False
async def attempt_token_refresh(client_id: str, client_secret: str, refresh_token: str) -> bool:
try:
refresh_data = {
"grant_type": "refresh_token",
"client_id": client_id,
"client_secret": client_secret,
"refresh_token": refresh_token
}
response = httpx.post(
"https://auth.atlassian.com/oauth/token",
data=refresh_data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=30.0
)
if response.status_code == 200:
tokens = response.json()
new_access_token = tokens["access_token"]
new_refresh_token = tokens.get("refresh_token")
os.environ["JIRA_ACCESS_TOKEN"] = new_access_token
if new_refresh_token:
os.environ["JIRA_REFRESH_TOKEN"] = new_refresh_token
print(f" Access token refreshed successfully!")
if new_refresh_token:
print(f" New refresh token obtained (rotating refresh tokens)")
return True
else:
print(f" Token refresh failed: {response.status_code}")
return False
except Exception as e:
print(f" Token refresh error: {e}")
return False
if __name__ == "__main__":
asyncio.run(test_oauth_comprehensive())