From 731d4d645e21792e35ec05f3c81fae71fd193cfe Mon Sep 17 00:00:00 2001 From: Adam Weeks Date: Wed, 24 Dec 2025 14:27:10 -0500 Subject: [PATCH] feat: Introduce service app token retrieval and enhance OAuth flow - Added a new script `get_service_app_token.py` for fetching fresh service app tokens. - Enhanced `setup_oauth.py` with improved OAuth credential handling and validation. - Refactored token management functions in `token_manager.py` for better clarity and functionality. - Implemented checks for existing tokens and streamlined the OAuth setup process for user convenience. --- get_service_app_token.py | 28 +++ setup_oauth.py | 375 ++++++++++++++++++++++++++++++++++----- token_manager.py | 225 ++++++++++++++++------- 3 files changed, 516 insertions(+), 112 deletions(-) create mode 100644 get_service_app_token.py diff --git a/get_service_app_token.py b/get_service_app_token.py new file mode 100644 index 0000000..2f90eb8 --- /dev/null +++ b/get_service_app_token.py @@ -0,0 +1,28 @@ +#!/usr/bin/env python3 +""" +Get a fresh service app token for external use + +Usage: + python3 get_service_app_token.py # Print token to stdout + python3 get_service_app_token.py | pbcopy # Copy to clipboard (macOS) + TOKEN=$(python3 get_service_app_token.py) # Save to variable +""" + +from token_manager import TokenManager +import sys + +def main(): + try: + tm = TokenManager() + token = tm.get_service_app_token() + + # Print only the token to stdout (clean output for piping) + print(token) + + except Exception as e: + print(f"Error: {e}", file=sys.stderr) + sys.exit(1) + +if __name__ == "__main__": + main() + diff --git a/setup_oauth.py b/setup_oauth.py index e7985a4..964b7de 100755 --- a/setup_oauth.py +++ b/setup_oauth.py @@ -10,6 +10,7 @@ import urllib.parse import webbrowser from typing import Dict +import requests def load_config() -> Dict: @@ -33,31 +34,224 @@ def save_config(config: Dict) -> None: json.dump(config, f, indent=4) -def setup_oauth_flow(): - """Set up OAuth flow for token manager integration.""" - print("OAuth Setup for Token Manager Integration") - print("=" * 50) - print() +def is_token_manager_token_valid(token: str, service_app_config: Dict) -> bool: + """ + Check if a Token Manager integration token is valid by using it + to fetch a service app token. - config = load_config() + Args: + token: The Token Manager integration token to validate + service_app_config: The serviceApp section from config - # Get OAuth credentials - print("First, you need the OAuth credentials from your Token Manager Integration:") - print("(This is DIFFERENT from your service app credentials)") - print() + Returns: + bool: True if valid, False otherwise + """ + try: + url = f"https://webexapis.com/v1/applications/{service_app_config['appId']}/token" + headers = { + "Authorization": f"Bearer {token}", + "Content-Type": "application/json", + } + payload = { + "clientId": service_app_config["clientId"], + "clientSecret": service_app_config["clientSecret"], + "targetOrgId": service_app_config["targetOrgId"], + } + + response = requests.post(url, headers=headers, json=payload) + + # 200 = valid token, 401 = invalid/expired token + return response.status_code == 200 + except Exception: + return False - client_id = input("Enter your Token Manager Integration Client ID: ").strip() - if not client_id: - print("Client ID is required") - return - client_secret = input( - "Enter your Token Manager Integration Client Secret: " - ).strip() - if not client_secret: - print("Client Secret is required") +def refresh_personal_token_oauth(token_config: Dict) -> str: + """ + Refresh the personal access token using OAuth. + + Args: + token_config: Configuration dictionary containing clientId, clientSecret, and refreshToken + + Returns: + str: New personal access token + + Raises: + Exception: If refresh fails + """ + url = "https://webexapis.com/v1/access_token" + headers = {"Content-Type": "application/x-www-form-urlencoded"} + data = { + "grant_type": "refresh_token", + "client_id": token_config["clientId"], + "client_secret": token_config["clientSecret"], + "refresh_token": token_config["refreshToken"], + } + + response = requests.post(url, headers=headers, data=data) + + if response.status_code == 401: + raise Exception( + "OAuth refresh token expired. Please re-authorize your integration." + ) + + response.raise_for_status() + + token_data = response.json() + new_access_token = token_data.get("access_token") + + if not new_access_token: + raise Exception("No access token in OAuth refresh response") + + return new_access_token + + +def can_refresh_oauth(config: Dict) -> bool: + """Check if OAuth refresh credentials exist.""" + return all(k in config for k in ["clientId", "clientSecret", "refreshToken"]) + + +def has_oauth_credentials(config: Dict) -> bool: + """Check if OAuth client credentials exist (without refresh token).""" + return all(k in config for k in ["clientId", "clientSecret"]) + + +def try_refresh_token(token_config: Dict, full_config: Dict, service_app_config: Dict) -> bool: + """ + Attempt to refresh using OAuth refresh token. + + Args: + token_config: The tokenManager section of the config + full_config: The complete config dictionary + service_app_config: The serviceApp section for validation + + Returns: + bool: True if refresh succeeded, False otherwise + """ + try: + print("Attempting to refresh token using OAuth...") + new_token = refresh_personal_token_oauth(token_config) + + # Validate the new token if we have service app config + if service_app_config.get("appId"): + print("Validating refreshed token...") + if not is_token_manager_token_valid(new_token, service_app_config): + print("✗ Refreshed token is not valid for Token Manager API") + return False + + full_config["tokenManager"]["personalAccessToken"] = new_token + save_config(full_config) + print("✓ Token refreshed successfully!") + print(f" New token: {new_token[:20]}...") + return True + except Exception as e: + print(f"✗ Refresh failed: {e}") + return False + + +def prompt_for_credential_method() -> str: + """ + Ask user to choose between PAT or OAuth flow. + + Returns: + str: "pat" or "oauth" + """ + print() + print("=" * 60) + print("No valid token manager credentials found.") + print("=" * 60) + print() + print("How would you like to proceed?") + print() + print(" 1. Paste a Personal Access Token from developer.webex.com") + print(" (Quick setup, but requires manual renewal)") + print() + print(" 2. Provide OAuth credentials for automatic refresh") + print(" (Recommended: enables automatic token renewal)") + print() + + while True: + choice = input("Enter your choice (1 or 2): ").strip() + if choice == "1": + return "pat" + elif choice == "2": + return "oauth" + else: + print("Invalid choice. Please enter 1 or 2.") + + +def handle_pat_input(config: Dict) -> None: + """ + Handle manual PAT input from user. + + Args: + config: The configuration dictionary + """ + print() + print("Manual Personal Access Token Setup") + print("-" * 40) + print() + print("To get a Personal Access Token:") + print("1. Go to https://developer.webex.com/docs/getting-started") + print("2. Log in with your Webex account") + print("3. Copy your Personal Access Token") + print() + + pat = input("Paste your Personal Access Token: ").strip() + if not pat: + print("✗ No token provided. Setup cancelled.") + return + + # Validate the token by trying to use it + print("Validating token...") + service_app_config = config.get("serviceApp", {}) + + if not service_app_config.get("appId"): + print("⚠️ Warning: Cannot validate token - serviceApp config missing.") + print(" Token will be saved but may not work correctly.") + elif not is_token_manager_token_valid(pat, service_app_config): + print("✗ Invalid token or token doesn't have required permissions.") + print(" Make sure this is a Token Manager Integration token with") + print(" 'spark:applications_token' scope.") return + + # Save to config + if "tokenManager" not in config: + config["tokenManager"] = {} + + config["tokenManager"]["personalAccessToken"] = pat + save_config(config) + + print() + print("✓ Personal Access Token saved successfully!") + print(f" Token: {pat[:20]}...") + print() + print("⚠️ Note: This token will need to be manually renewed when it expires.") + print(" Consider running this script again with OAuth credentials for automatic renewal.") + +def confirm_use_existing_credentials() -> bool: + """ + Ask user if they want to use existing OAuth credentials. + + Returns: + bool: True if user wants to use existing credentials + """ + print() + print("OAuth credentials found in configuration.") + response = input("Do you want to use these credentials? (Y/n): ").strip().lower() + return response in ["", "y", "yes"] + + +def do_oauth_flow_with_credentials(client_id: str, client_secret: str, config: Dict) -> None: + """ + Execute the OAuth authorization flow with provided credentials. + + Args: + client_id: OAuth client ID + client_secret: OAuth client secret + config: The configuration dictionary + """ # Use localhost as redirect URI redirect_uri = "http://localhost:3000/callback" @@ -76,7 +270,7 @@ def setup_oauth_flow(): print() print("OAuth Authorization Setup") - print("-" * 30) + print("-" * 40) print("1. Opening authorization URL in your browser...") print("2. Log in and authorize the application") print("3. You'll be redirected to localhost (which will fail)") @@ -97,14 +291,12 @@ def setup_oauth_flow(): auth_code = input("Enter the authorization code from the URL: ").strip() if not auth_code: - print("Authorization code is required") + print("✗ Authorization code is required. Setup cancelled.") return # Exchange code for tokens print("Exchanging authorization code for tokens...") - import requests - token_url = "https://webexapis.com/v1/access_token" token_data = { "grant_type": "authorization_code", @@ -123,51 +315,142 @@ def setup_oauth_flow(): refresh_token = tokens.get("refresh_token") if not access_token or not refresh_token: - print("Error: Did not receive both access and refresh tokens") + print("✗ Error: Did not receive both access and refresh tokens") return # Update configuration with flat structure - config["tokenManager"]["oauthClientId"] = client_id - config["tokenManager"]["oauthClientSecret"] = client_secret - config["tokenManager"]["oauthRefreshToken"] = refresh_token + if "tokenManager" not in config: + config["tokenManager"] = {} + + config["tokenManager"]["clientId"] = client_id + config["tokenManager"]["clientSecret"] = client_secret + config["tokenManager"]["refreshToken"] = refresh_token config["tokenManager"]["personalAccessToken"] = access_token save_config(config) print() - print("✅ OAuth setup completed successfully!") - print("Your token manager now supports automatic personal token refresh.") + print("✓ OAuth setup completed successfully!") + print(" Your token manager now supports automatic personal token refresh.") print() print("Updated configuration:") - print(f"- Personal Access Token: {access_token[:20]}...") - print(f"- Refresh Token: {refresh_token[:20]}...") - print("- OAuth credentials saved to token-config.json") + print(f" - Personal Access Token: {access_token[:20]}...") + print(f" - Refresh Token: {refresh_token[:20]}...") + print(" - OAuth credentials saved to token-config.json") except requests.exceptions.RequestException as e: - print(f"Error exchanging authorization code: {e}") + print(f"✗ Error exchanging authorization code: {e}") return except Exception as e: - print(f"Unexpected error: {e}") + print(f"✗ Unexpected error: {e}") return -def main(): - """Main function.""" - print("Token Manager OAuth Setup") - print("This will help you set up automatic refresh for your personal access token.") +def handle_oauth_credential_input(config: Dict) -> None: + """ + Handle OAuth credential input from user and start OAuth flow. + + Args: + config: The configuration dictionary + """ + print() + print("OAuth Credential Setup") + print("-" * 40) print() - print("Prerequisites:") - print("1. You have created a Token Manager Integration at developer.webex.com") - print("2. Your integration has the 'spark:applications_token' scope") - print("3. Your integration has 'http://localhost:3000/callback' as a redirect URI") + print("You need OAuth credentials from your Token Manager Integration:") + print("(This is DIFFERENT from your service app credentials)") print() + print("To get these credentials:") + print("1. Go to https://developer.webex.com/my-apps") + print("2. Create or select your Token Manager Integration") + print("3. Ensure it has 'spark:applications_token' scope") + print("4. Add 'http://localhost:3000/callback' as a redirect URI") + print() + + client_id = input("Enter your Token Manager Integration Client ID: ").strip() + if not client_id: + print("✗ Client ID is required. Setup cancelled.") + return - proceed = input("Do you want to continue? (y/N): ").strip().lower() - if proceed not in ["y", "yes"]: - print("Setup cancelled.") + client_secret = input( + "Enter your Token Manager Integration Client Secret: " + ).strip() + if not client_secret: + print("✗ Client Secret is required. Setup cancelled.") return - setup_oauth_flow() + do_oauth_flow_with_credentials(client_id, client_secret, config) + + +def smart_setup_oauth_flow(): + """ + Smart OAuth flow that checks existing tokens before prompting user. + + This follows a cascade: + 1. Check if PAT exists and is valid -> Done + 2. Try to refresh using OAuth if refresh token exists -> Done if successful + 3. Use existing OAuth credentials for new auth flow if they exist + 4. Prompt user to choose between PAT or OAuth setup + """ + print("Token Manager OAuth Setup") + print("=" * 60) + print() + + config = load_config() + token_manager_config = config.get("tokenManager", {}) + service_app_config = config.get("serviceApp", {}) + + # Step 1: Check if PAT exists and is valid + personal_token = token_manager_config.get("personalAccessToken") + if personal_token and service_app_config.get("appId"): + print("Checking existing personal access token...") + if is_token_manager_token_valid(personal_token, service_app_config): + print("✓ Existing personal access token is valid!") + print(f" Token: {personal_token[:20]}...") + print() + print("No further setup needed. Your token is ready to use.") + return + + print("✗ Existing token is expired or invalid.") + + # Step 2: Try OAuth refresh if available + if can_refresh_oauth(token_manager_config): + if try_refresh_token(token_manager_config, config, service_app_config): + print() + print("No further setup needed. Your token has been refreshed.") + return + + # Step 3: Try OAuth flow with existing credentials + if has_oauth_credentials(token_manager_config): + if confirm_use_existing_credentials(): + do_oauth_flow_with_credentials( + token_manager_config["clientId"], + token_manager_config["clientSecret"], + config + ) + return + else: + print("Proceeding with manual setup...") + + # Step 4: Prompt user for input method + choice = prompt_for_credential_method() + if choice == "pat": + handle_pat_input(config) + else: + handle_oauth_credential_input(config) + + +def main(): + """Main function.""" + print() + print("This script will help you set up token manager credentials.") + print("It supports both manual PAT entry and OAuth with automatic refresh.") + print() + + smart_setup_oauth_flow() + + print() + print("Setup complete!") if __name__ == "__main__": diff --git a/token_manager.py b/token_manager.py index 91f5960..74e456d 100644 --- a/token_manager.py +++ b/token_manager.py @@ -13,6 +13,67 @@ AWS_AVAILABLE = False +# Standalone utility functions for token validation and refresh +def is_personal_token_valid(token: str) -> bool: + """ + Check if a personal access token is valid. + + Args: + token: The personal access token to validate + + Returns: + bool: True if valid, False otherwise + """ + try: + headers = {"Authorization": f"Bearer {token}"} + response = requests.get( + "https://webexapis.com/v1/people/me", headers=headers + ) + return response.status_code == 200 + except Exception: + return False + + +def refresh_personal_token_oauth(config: Dict) -> str: + """ + Refresh the personal access token using OAuth. + + Args: + config: Configuration dictionary containing clientId, clientSecret, and refreshToken + + Returns: + str: New personal access token + + Raises: + Exception: If refresh fails + """ + url = "https://webexapis.com/v1/access_token" + headers = {"Content-Type": "application/x-www-form-urlencoded"} + data = { + "grant_type": "refresh_token", + "client_id": config["clientId"], + "client_secret": config["clientSecret"], + "refresh_token": config["refreshToken"], + } + + response = requests.post(url, headers=headers, data=data) + + if response.status_code == 401: + raise Exception( + "OAuth refresh token expired. Please re-authorize your integration." + ) + + response.raise_for_status() + + token_data = response.json() + new_access_token = token_data.get("access_token") + + if not new_access_token: + raise Exception("No access token in OAuth refresh response") + + return new_access_token + + class TokenManager: """Manages Webex service app authentication. @@ -196,7 +257,7 @@ def _load_config(self) -> Dict[str, str]: ] # OAuth fields are optional but must all be present together - oauth_fields = ["oauthClientId", "oauthClientSecret", "oauthRefreshToken"] + oauth_fields = ["clientId", "clientSecret", "refreshToken"] oauth_present = [field in config["tokenManager"] for field in oauth_fields] if any(oauth_present) and not all(oauth_present): @@ -224,7 +285,7 @@ def _load_config(self) -> Dict[str, str]: except Exception: raise - def _is_personal_token_valid(self, token: str) -> bool: + def is_personal_token_valid(self, token: str) -> bool: """ Check if a personal access token is valid. @@ -243,7 +304,7 @@ def _is_personal_token_valid(self, token: str) -> bool: except Exception: return False - def _refresh_personal_token_oauth(self, config: Dict) -> str: + def refresh_personal_token_oauth(self, config: Dict) -> str: """ Refresh the personal access token using OAuth. @@ -257,9 +318,9 @@ def _refresh_personal_token_oauth(self, config: Dict) -> str: headers = {"Content-Type": "application/x-www-form-urlencoded"} data = { "grant_type": "refresh_token", - "client_id": config["oauthClientId"], - "client_secret": config["oauthClientSecret"], - "refresh_token": config["oauthRefreshToken"], + "client_id": config["clientId"], + "client_secret": config["clientSecret"], + "refresh_token": config["refreshToken"], } response = requests.post(url, headers=headers, data=data) @@ -279,45 +340,51 @@ def _refresh_personal_token_oauth(self, config: Dict) -> str: return new_access_token - def _get_valid_personal_token(self, config: Dict) -> str: + def _try_refresh_personal_token(self, config: Dict) -> str: """ - Get a valid personal access token, refreshing via OAuth if needed. + Attempt to refresh the personal access token via OAuth. Args: config: The loaded configuration Returns: - str: A valid personal access token + str: A refreshed personal access token + + Raises: + Exception: If refresh is not configured or fails """ token_manager = config["tokenManager"] - personal_token = token_manager["personalAccessToken"] # Check if OAuth is configured (all three fields must be present) oauth_configured = all( key in token_manager - for key in ["oauthClientId", "oauthClientSecret", "oauthRefreshToken"] + for key in ["clientId", "clientSecret", "refreshToken"] ) - if oauth_configured: - # Test current personal token - if not self._is_personal_token_valid(personal_token): - print("Personal access token expired, refreshing via OAuth...") - try: - personal_token = self._refresh_personal_token_oauth(token_manager) - # Update the config file with new personal token - self._update_personal_token_in_config(personal_token) - print("Personal access token refreshed successfully") - except Exception as e: - print(f"Failed to refresh personal token via OAuth: {e}") - print("Using existing personal token (may be expired)") - + if not oauth_configured: + raise Exception( + "OAuth refresh is not configured. Cannot refresh token automatically. " + "Please run setup_oauth.py to configure automatic token refresh, " + "or manually update personalAccessToken in token-config.json" + ) + + print("Personal access token expired, refreshing via OAuth...") + personal_token = self.refresh_personal_token_oauth(token_manager) + + # Update the config file with new personal token + self._update_personal_token_in_config(personal_token) + print("Personal access token refreshed successfully") + return personal_token def get_service_app_token(self) -> str: """ Get a fresh service app access token. - This method always fetches a new token from the Webex API. + This method fetches a token from the Webex Token Manager API. + If the personal access token is expired, it will attempt to refresh it + automatically using OAuth (if configured) and retry. + The token is cached in memory for the duration of the script execution. Returns: @@ -330,54 +397,80 @@ def get_service_app_token(self) -> str: if self._service_app_token: return self._service_app_token + config = self._load_config() + + # Try to get token with current personal token try: - config = self._load_config() - - # Get a valid personal access token (refresh if needed) - personal_token = self._get_valid_personal_token(config) - - # Prepare the API request - service_app = config["serviceApp"] - - url = f"https://webexapis.com/v1/applications/{service_app['appId']}/token" - headers = { - "Authorization": f"Bearer {personal_token}", - "Content-Type": "application/json", - } - payload = { - "clientId": service_app["clientId"], - "clientSecret": service_app["clientSecret"], - "targetOrgId": service_app["targetOrgId"], - } - - # Make the API call - response = requests.post(url, headers=headers, json=payload) - - if response.status_code == 401: - raise Exception( - "Authentication failed. Your personal access token may have expired. " - "Get a new one from developer.webex.com and update token-config.json" - ) + return self._fetch_service_app_token(config) + except requests.exceptions.HTTPError as e: + # If we get a 401, the personal token might be expired + if e.response.status_code == 401: + print("Token Manager authentication failed (401), attempting to refresh personal token...") + try: + # Try to refresh the personal token + new_personal_token = self._try_refresh_personal_token(config) + + # Update config in memory with refreshed token + config["tokenManager"]["personalAccessToken"] = new_personal_token + + # Retry fetching service app token with refreshed personal token + print("Retrying with refreshed token...") + return self._fetch_service_app_token(config) + + except Exception as refresh_error: + raise Exception( + f"Failed to refresh personal token: {refresh_error}. " + "Please run setup_oauth.py to re-authorize." + ) + else: + # Some other HTTP error + raise Exception(f"Token request failed with status {e.response.status_code}: {e.response.text}") + except Exception as e: + raise Exception(f"Failed to get service app token: {e}") + + def _fetch_service_app_token(self, config: Dict) -> str: + """ + Internal method to fetch service app token from the API. + + Args: + config: The loaded configuration + + Returns: + str: Service app access token + + Raises: + requests.exceptions.HTTPError: If the API request fails + """ + personal_token = config["tokenManager"]["personalAccessToken"] + service_app = config["serviceApp"] - response.raise_for_status() + url = f"https://webexapis.com/v1/applications/{service_app['appId']}/token" + headers = { + "Authorization": f"Bearer {personal_token}", + "Content-Type": "application/json", + } + payload = { + "clientId": service_app["clientId"], + "clientSecret": service_app["clientSecret"], + "targetOrgId": service_app["targetOrgId"], + } - token_data = response.json() - access_token = token_data.get("access_token") - refresh_token = token_data.get("refresh_token") + # Make the API call + response = requests.post(url, headers=headers, json=payload) + response.raise_for_status() # Raises HTTPError for bad status codes - if not access_token: - raise Exception("No access token in API response") + token_data = response.json() + access_token = token_data.get("access_token") + refresh_token = token_data.get("refresh_token") - # Cache tokens in memory for this execution - self._service_app_token = access_token - self._service_app_refresh_token = refresh_token + if not access_token: + raise Exception("No access token in API response") - return access_token + # Cache tokens in memory for this execution + self._service_app_token = access_token + self._service_app_refresh_token = refresh_token - except requests.exceptions.RequestException as e: - raise Exception(f"Token request failed: {e}") - except Exception as e: - raise Exception(f"Failed to get service app token: {e}") + return access_token def extend_data_source_token( self, data_source_id: str, token_lifetime_minutes: int = 1440