diff --git a/.gitignore b/.gitignore index 154b600..957ce99 100644 --- a/.gitignore +++ b/.gitignore @@ -6,6 +6,9 @@ # Keep the sample file (it should be committed) !.env.sample +# Token configuration (sensitive) +token-config.json + # Data source operation files data_source_*.json diff --git a/README.md b/README.md index 9f133a3..1d13acb 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,39 @@ Watch the Vidcast of this app](https://app.vidcast.io/share/63e954e4-f0ae-4c20-8 - Updating existing data source configurations - Interactive menu-driven interface +## Token Management + +This project includes automated token refresh functionality to handle service app token expiration: + +**Files:** + +- **`token_manager.py`** - Core token management class with smart refresh logic +- **`refresh_token.py`** - Standalone script for manual token refresh +- **`TOKEN_MANAGEMENT.md`** - Complete setup and usage documentation + +**Key Features:** + +- **Smart refresh strategy**: Uses refresh tokens when available, falls back to personal access tokens +- **OAuth support**: Automatic refresh of personal access tokens via OAuth (optional) +- **Automatic validation**: Checks token validity before operations +- **Seamless integration**: Automatically refreshes tokens in the main data-sources.py script +- **Multiple setup options**: Portal tokens, integration tokens, or full OAuth automation + +**Quick Token Refresh:** + +```bash +# Activate virtual environment +source venv/bin/activate + +# Manually refresh tokens +python refresh_token.py + +# Set up OAuth for automatic personal token refresh (optional) +python setup_oauth.py +``` + +For complete setup instructions, see [TOKEN_MANAGEMENT.md](TOKEN_MANAGEMENT.md). + ## Requirements - Python 3.6 or higher @@ -40,8 +73,14 @@ Watch the Vidcast of this app](https://app.vidcast.io/share/63e954e4-f0ae-4c20-8 ``` 3. Configure your access token: + - Copy the sample environment file: `cp .env.sample .env` - - Edit the `.env` file and replace `your_service_app_token_here` with your actual Service App access token + - Edit the `.env` file and replace `your_service_app_token_here` with your actual Service App access token (stored as `WEBEX_SERVICE_APP_ACCESS_TOKEN`) + +4. **Optional**: Set up automated token refresh (recommended for production): + - Copy the token configuration template: `cp token-config.json.template token-config.json` + - Edit `token-config.json` with your service app and token manager credentials + - See [TOKEN_MANAGEMENT.md](TOKEN_MANAGEMENT.md) for detailed setup instructions ## Usage @@ -54,6 +93,7 @@ python data-sources.py **Features:** - Automatically loads and displays all your data sources on startup +- **Automated token refresh**: Automatically refreshes expired service app tokens - Interactive menu to view/update existing data sources or register new ones - **Schema-aware interface**: Automatically fetches and displays available schemas with friendly service type names - Real-time refresh capability @@ -144,6 +184,7 @@ The script automatically saves all operation details to JSON files: - **Successful updates**: `data_source_update_{ID}_{timestamp}.json` - **Failed operations**: `data_source_{operation}_failed_{timestamp}.json` - **Data source lists**: `data_sources_list_{timestamp}.json` (when using --save-list flag) +- **Token configuration**: `token-config.json` (if using automated token refresh) These files contain: @@ -163,6 +204,13 @@ The script requires a Service App access token with the following scopes: For full functionality, use a token that has both read and write scopes. +**Token Management:** + +- Service app tokens are automatically refreshed when they expire +- Refresh tokens are stored and used for efficient token renewal +- Manual token refresh available via `python refresh_token.py` +- See [TOKEN_MANAGEMENT.md](TOKEN_MANAGEMENT.md) for configuration details + ## Error Handling The script includes comprehensive error handling for: diff --git a/TOKEN_MANAGEMENT.md b/TOKEN_MANAGEMENT.md new file mode 100644 index 0000000..3a17642 --- /dev/null +++ b/TOKEN_MANAGEMENT.md @@ -0,0 +1,258 @@ +# Token Management + +This project now includes automated token refresh functionality for Webex service app tokens. + +## Setup + +1. **Create token configuration file:** + + ```bash + cp token-config.json.template token-config.json + ``` + +2. **Edit the configuration file** with your actual values: + + ```json + { + "serviceApp": { + "appId": "your_service_app_id", + "clientId": "your_service_app_client_id", + "clientSecret": "your_service_app_client_secret", + "targetOrgId": "your_target_org_id" + }, + "tokenManager": { + "personalAccessToken": "your_personal_access_token_or_integration_token", + "integration": { + "clientId": "your_token_manager_integration_client_id", + "clientSecret": "your_token_manager_integration_client_secret", + "refreshToken": "your_token_manager_integration_refresh_token" + } + } + } + ``` + + **Note**: The `integration` section is optional and only needed for OAuth-based personal token refresh (Option B2). + +## How to get the required values: + +The configuration requires two different sets of credentials: + +### 1. Service App Information (`serviceApp` section) + +These are the credentials for your **existing Webex service app** that manages data sources: + +- **appId**: Found in your Webex service app configuration (format: `Y2lzY29zcGFyazovL3VzL0FQUExJQ0FUSU9OL...`) +- **clientId**: Your **service app's** Client ID +- **clientSecret**: Your **service app's** Client Secret +- **targetOrgId**: The organization ID where the service app operates + +### 2. Token Manager Authentication (`tokenManager` section) + +This is the token used to **refresh** your service app tokens (choose one method): + +#### Option A: Quick Start - Developer Portal Token (Temporary) + +1. Go to [developer.webex.com](https://developer.webex.com) +2. Sign in and click on your profile in the top right +3. Copy your "Personal Access Token" +4. Use this token in the `personalAccessToken` field + +⚠️ **Note**: Portal tokens expire every 12 hours and are meant for testing only. + +#### Option B: Production - Create a Dedicated Integration (Recommended) + +1. Go to [developer.webex.com](https://developer.webex.com) +2. Click "Create a New App" → "Create an Integration" +3. Fill in the details: + - **Integration name**: "Token Refresh Helper" (or your preferred name) + - **Description**: "Integration for refreshing service app tokens" + - **Redirect URI**: `http://localhost:3000/callback` + - **Scopes**: Select `spark:applications_token` +4. Save the integration and copy the **Client ID** and **Client Secret** +5. **Option B1 - Manual Setup**: Use the test token provided in the `personalAccessToken` field +6. **Option B2 - OAuth Setup (Recommended)**: Run `python setup_oauth.py` for automated OAuth flow + +✅ **Recommended**: The OAuth setup (B2) provides automatic refresh of your personal access token and is suitable for production use. + +## OAuth Setup (Advanced) + +For production environments, you can set up OAuth to automatically refresh your personal access token: + +### Prerequisites + +- Token Manager Integration created with: + - Scope: `spark:applications_token` + - Redirect URI: `http://localhost:3000/callback` + +### Setup Process + +```bash +# Activate virtual environment +source venv/bin/activate + +# Run OAuth setup helper +python setup_oauth.py +``` + +This will: + +1. Open your browser for authorization +2. Exchange the authorization code for tokens +3. Update your `token-config.json` with OAuth credentials +4. Enable automatic personal token refresh + +## Development Approaches + +### Quick Start Approach (Option A) + +- **Time to setup**: ~2 minutes +- **Token lifespan**: 12 hours +- **Best for**: Testing, proof of concept, development +- **Maintenance**: Manual token refresh every 12 hours from developer portal + +### Production Approach (Option B) + +- **Time to setup**: ~10 minutes (includes OAuth setup) +- **Token lifespan**: Much longer (typically months) +- **Best for**: Production deployments, automated systems +- **Maintenance**: Minimal - tokens refresh automatically +- **Variants**: + - **B1 (Manual)**: Use integration test token (needs periodic manual refresh) + - **B2 (OAuth)**: Fully automated with personal token auto-refresh + +## Usage + +### Automatic Token Refresh + +The main `data-sources.py` script will automatically check token validity and refresh if needed when run. + +**Smart Refresh Strategy:** + +1. **First**: Tries to use the stored refresh token (faster, no personal token needed) +2. **Fallback**: Uses personal access token if refresh token fails or is unavailable +3. **Validation**: Uses the `/v1/dataSources` endpoint to check service app token validity + +### Manual Token Refresh + +You can manually refresh tokens anytime: + +```bash +# Activate virtual environment first +source venv/bin/activate + +# Then run the refresh script +python refresh_token.py +``` + +### Token Manager API + +You can also use the TokenManager class directly in your code: + +```python +# Activate virtual environment first +source venv/bin/activate + +from token_manager import TokenManager + +# Initialize the token manager +token_manager = TokenManager() + +# Check if current token is valid +if not token_manager.is_token_valid(): + # Refresh the token + new_token = token_manager.refresh_token() + print(f"New token: {new_token}") +``` + +## Security Notes + +- The `token-config.json` file is automatically excluded from git +- Keep your personal access token secure and never commit it to version control +- **Portal tokens** (Option A): Expire every 12 hours, good for development only +- **Integration tokens** (Option B): Last much longer, suitable for production +- The personal access token must have `spark:applications_token` scope +- Service app tokens are automatically updated in the `.env` file as: + - `WEBEX_SERVICE_APP_ACCESS_TOKEN`: The active service app token + - `WEBEX_SERVICE_APP_REFRESH_TOKEN`: Used for automatic token refresh +- **Refresh tokens** are automatically stored and used for efficient token refresh +- Consider using environment variables for the token config in production environments + +## How Refresh Tokens Work + +The system manages multiple types of tokens for maximum efficiency: + +### Service App Tokens + +- **Access Token**: Used for data source API calls → `WEBEX_SERVICE_APP_ACCESS_TOKEN` +- **Refresh Token**: Used to get new service app access tokens → `WEBEX_SERVICE_APP_REFRESH_TOKEN` + +### Token Manager Tokens (for refreshing service app tokens) + +- **Personal Access Token**: Used to authenticate service app token refresh requests +- **OAuth Refresh Token**: Used to automatically refresh the personal access token (optional) + +**Multi-Level Refresh Strategy:** + +1. **Service App Token Expired**: Use service app refresh token → New service app access token +2. **Service App Refresh Token Expired**: Use personal access token → New service app tokens +3. **Personal Access Token Expired**: Use OAuth refresh token → New personal access token +4. **OAuth Refresh Token Expired**: Manual re-authorization required + +**Benefits:** + +- Faster refresh (service app refresh tokens used first) +- Automatic personal token refresh (with OAuth setup) +- Minimal manual intervention required +- Production-ready automation + +## API Reference + +The token refresh uses the Webex Applications API: + +``` +POST https://webexapis.com/v1/applications/{serviceApp.appId}/token +Authorization: Bearer {tokenManager.personalAccessToken} +``` + +With the following payload (using **service app** credentials): + +```json +{ + "clientId": "serviceApp.clientId", + "clientSecret": "serviceApp.clientSecret", + "targetOrgId": "serviceApp.targetOrgId" +} +``` + +**Note**: The API call is authenticated with your `tokenManager.personalAccessToken` but requests a new token for your `serviceApp` credentials. + +## Troubleshooting + +### "Authentication failed" or 401 errors + +Your **token manager** personal access token has likely expired: + +**If using Portal Token (Option A):** + +1. Go to [developer.webex.com](https://developer.webex.com) +2. Sign in and get a fresh token from your profile +3. Update `tokenManager.personalAccessToken` in `token-config.json` + +**If using Integration Token (Option B):** + +1. Your integration's OAuth token has expired +2. Generate a new access token through your OAuth flow +3. Update `tokenManager.personalAccessToken` in `token-config.json` + +### "No access token in API response" + +- Verify your **service app** credentials in the `serviceApp` section are correct: + - `serviceApp.appId`, `serviceApp.clientId`, `serviceApp.clientSecret` +- Ensure the `serviceApp.targetOrgId` matches your service app's organization +- Check that your **token manager** `personalAccessToken` has `spark:applications_token` scope + +### Script fails to find config file + +- Ensure `token-config.json` exists in the same directory as the scripts +- Copy from `token-config.json.template` if needed +- Check file permissions diff --git a/data-sources.py b/data-sources.py index fa0877c..dcb3cf9 100755 --- a/data-sources.py +++ b/data-sources.py @@ -225,13 +225,35 @@ def load_env_token() -> str: load_dotenv(env_path) - token = os.getenv("WEBEX_ACCESS_TOKEN") + token = os.getenv("WEBEX_SERVICE_APP_ACCESS_TOKEN") if not token: - print("Error: WEBEX_ACCESS_TOKEN not found in .env file") + print("Error: WEBEX_SERVICE_APP_ACCESS_TOKEN not found in .env file") print(f"Please create a .env file in {script_dir} with:") - print("WEBEX_ACCESS_TOKEN=your_service_app_token_here") + print("WEBEX_SERVICE_APP_ACCESS_TOKEN=your_service_app_token_here") sys.exit(1) + # Try to import and use token manager for automatic refresh if config exists + try: + from token_manager import TokenManager + token_manager = TokenManager(env_path=env_path) + + # Check if token is valid, refresh if needed + if not token_manager.is_token_valid(): + print("Current token is invalid or expired. Attempting to refresh...") + try: + new_token = token_manager.refresh_token() + token = new_token + print("Token refreshed successfully!") + except Exception as refresh_error: + print(f"Warning: Could not refresh token automatically: {refresh_error}") + print("Please refresh manually using: python refresh_token.py") + except ImportError: + # Token manager not available, continue with existing token + pass + except Exception as e: + print(f"Warning: Token validation/refresh failed: {e}") + print("Continuing with existing token...") + return token diff --git a/refresh_token.py b/refresh_token.py new file mode 100755 index 0000000..dac6394 --- /dev/null +++ b/refresh_token.py @@ -0,0 +1,45 @@ +#!/usr/bin/env python3 +""" +Script to manually refresh Webex service app tokens. +""" + +import sys +from token_manager import TokenManager + + +def main(): + """Main function to refresh the token.""" + token_manager = TokenManager() + + try: + print("Checking current token validity...") + if token_manager.is_token_valid(): + print("Current token is still valid.") + response = input("Do you want to refresh it anyway? (y/N): ") + if response.lower() not in ['y', 'yes']: + print("Token refresh cancelled.") + return + + print("Refreshing token...") + new_token = token_manager.refresh_token() + print("Token refreshed successfully!") + print(f"New token starts with: {new_token[:20]}...") + + # Show which refresh method was used + if token_manager._get_current_refresh_token(): + print("Note: Future refreshes will use the stored refresh token for better efficiency.") + + except Exception as e: + print(f"Token refresh failed: {e}") + + # Provide guidance on how to fix token issues + if "Authentication failed" in str(e) or "401" in str(e): + print("\n" + "="*60) + print(token_manager.get_token_refresh_guidance()) + print("="*60) + + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/setup_oauth.py b/setup_oauth.py new file mode 100755 index 0000000..61b5692 --- /dev/null +++ b/setup_oauth.py @@ -0,0 +1,174 @@ +#!/usr/bin/env python3 +""" +OAuth Setup Helper for Token Manager Integration + +This script helps you set up OAuth authorization for your token manager integration. +Use this when you want to enable automatic refresh of your personal access token. +""" + +import json +import urllib.parse +import webbrowser +from typing import Dict + + +def load_config() -> Dict: + """Load the token configuration.""" + try: + with open('token-config.json', 'r') as f: + return json.load(f) + except FileNotFoundError: + print("Error: token-config.json not found. Please create it from the template first.") + exit(1) + except json.JSONDecodeError: + print("Error: Invalid JSON in token-config.json") + exit(1) + + +def save_config(config: Dict) -> None: + """Save the updated configuration.""" + with open('token-config.json', 'w') as f: + 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() + + config = load_config() + + # Check if OAuth integration is configured + if 'integration' not in config['tokenManager']: + config['tokenManager']['integration'] = {} + + integration = config['tokenManager']['integration'] + + # 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() + + 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") + return + + # Use localhost as redirect URI + redirect_uri = "http://localhost:3000/callback" + + # Build authorization URL + auth_params = { + 'client_id': client_id, + 'response_type': 'code', + 'redirect_uri': redirect_uri, + 'scope': 'spark:applications_token', + 'state': 'token_manager_setup' + } + + auth_url = "https://webexapis.com/v1/authorize?" + urllib.parse.urlencode(auth_params) + + print() + print("OAuth Authorization Setup") + print("-" * 30) + 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)") + print("4. Copy the 'code' parameter from the URL") + print() + print(f"Authorization URL: {auth_url}") + print() + + # Open browser + try: + webbrowser.open(auth_url) + except Exception: + print("Could not open browser automatically. Please copy the URL above.") + + print("After authorization, you'll see a URL like:") + print("http://localhost:3000/callback?code=ABC123...&state=token_manager_setup") + print() + + auth_code = input("Enter the authorization code from the URL: ").strip() + if not auth_code: + print("Authorization code is required") + 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', + 'client_id': client_id, + 'client_secret': client_secret, + 'code': auth_code, + 'redirect_uri': redirect_uri + } + + try: + response = requests.post(token_url, data=token_data) + response.raise_for_status() + + tokens = response.json() + access_token = tokens.get('access_token') + refresh_token = tokens.get('refresh_token') + + if not access_token or not refresh_token: + print("Error: Did not receive both access and refresh tokens") + return + + # Update configuration + integration['clientId'] = client_id + integration['clientSecret'] = client_secret + integration['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() + 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") + + except requests.exceptions.RequestException as e: + print(f"Error exchanging authorization code: {e}") + return + except Exception as 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.") + 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() + + proceed = input("Do you want to continue? (y/N): ").strip().lower() + if proceed not in ['y', 'yes']: + print("Setup cancelled.") + return + + setup_oauth_flow() + + +if __name__ == "__main__": + main() diff --git a/token-config.json.template b/token-config.json.template new file mode 100644 index 0000000..e2dd8ec --- /dev/null +++ b/token-config.json.template @@ -0,0 +1,16 @@ +{ + "serviceApp": { + "appId": "YOUR_SERVICE_APP_ID_HERE", + "clientId": "YOUR_SERVICE_APP_CLIENT_ID_HERE", + "clientSecret": "YOUR_SERVICE_APP_CLIENT_SECRET_HERE", + "targetOrgId": "YOUR_TARGET_ORG_ID_HERE" + }, + "tokenManager": { + "personalAccessToken": "YOUR_PERSONAL_ACCESS_TOKEN_HERE", + "integration": { + "clientId": "YOUR_TOKEN_MANAGER_INTEGRATION_CLIENT_ID_HERE", + "clientSecret": "YOUR_TOKEN_MANAGER_INTEGRATION_CLIENT_SECRET_HERE", + "refreshToken": "YOUR_TOKEN_MANAGER_INTEGRATION_REFRESH_TOKEN_HERE" + } + } +} diff --git a/token_manager.py b/token_manager.py new file mode 100644 index 0000000..1d42a3a --- /dev/null +++ b/token_manager.py @@ -0,0 +1,436 @@ +import json +import requests +from typing import Dict, Optional + + +class TokenManager: + """Manages Webex service app token refresh and updates.""" + + def __init__(self, env_path: str = ".env", config_path: str = "token-config.json"): + """ + Initialize TokenManager. + + Args: + env_path: Path to the .env file + config_path: Path to the token configuration file + """ + self.env_path = env_path + self.config_path = config_path + + def refresh_token(self) -> str: + """ + Refresh the Webex service app token using the API. + Tries to use stored refresh token first, falls back to full refresh if needed. + + Returns: + str: The new access token + + Raises: + Exception: If token refresh fails + """ + # First try to use the refresh token if available + current_refresh_token = self._get_current_refresh_token() + if current_refresh_token: + try: + return self._refresh_with_refresh_token(current_refresh_token) + except Exception as e: + print(f"Refresh token failed, falling back to full refresh: {e}") + + # Fall back to full refresh using personal access token + return self._refresh_with_personal_token() + + def _refresh_with_refresh_token(self, refresh_token: str) -> str: + """ + Refresh the token using the stored refresh token. + + Args: + refresh_token: The refresh token to use + + Returns: + str: The new access token + """ + try: + config = self._load_config() + service_app = config['serviceApp'] + + url = "https://webexapis.com/v1/access_token" + headers = {'Content-Type': 'application/x-www-form-urlencoded'} + data = { + 'grant_type': 'refresh_token', + 'client_id': service_app['clientId'], + 'client_secret': service_app['clientSecret'], + 'refresh_token': refresh_token + } + + response = requests.post(url, headers=headers, data=data) + + if response.status_code == 401: + raise Exception("Refresh token expired or invalid") + + response.raise_for_status() + + token_data = response.json() + new_access_token = token_data.get('access_token') + new_refresh_token = token_data.get('refresh_token', refresh_token) # Use new or keep old + + if not new_access_token: + raise Exception("No access token in refresh response") + + # Update the .env file with the new tokens + self._update_env_file(new_access_token, new_refresh_token) + + print("Token refreshed successfully using refresh token") + return new_access_token + + except requests.exceptions.RequestException as e: + raise Exception(f"Refresh token request failed: {e}") + + def _get_valid_personal_token(self, config: Dict) -> str: + """ + Get a valid personal access token, refreshing via OAuth if needed. + + Args: + config: The loaded configuration + + Returns: + str: A valid personal access token + """ + token_manager = config['tokenManager'] + personal_token = token_manager['personalAccessToken'] + + # Check if OAuth integration is configured + if 'integration' in token_manager and all( + key in token_manager['integration'] + for key in ['clientId', 'clientSecret', 'refreshToken'] + ): + # 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['integration']) + # 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)") + + return personal_token + + def _is_personal_token_valid(self, 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(self, integration_config: Dict) -> str: + """ + Refresh the personal access token using OAuth. + + Args: + integration_config: OAuth integration configuration + + Returns: + str: New personal access token + """ + url = "https://webexapis.com/v1/access_token" + headers = {'Content-Type': 'application/x-www-form-urlencoded'} + data = { + 'grant_type': 'refresh_token', + 'client_id': integration_config['clientId'], + 'client_secret': integration_config['clientSecret'], + 'refresh_token': integration_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') + new_refresh_token = token_data.get('refresh_token') + + if not new_access_token: + raise Exception("No access token in OAuth refresh response") + + # Update refresh token if provided + if new_refresh_token: + integration_config['refreshToken'] = new_refresh_token + # Note: This updates the in-memory config, _update_personal_token_in_config will save it + + return new_access_token + + def _update_personal_token_in_config(self, new_personal_token: str) -> None: + """ + Update the personal access token in the config file. + + Args: + new_personal_token: The new personal access token + """ + try: + with open(self.config_path, 'r') as f: + config = json.load(f) + + config['tokenManager']['personalAccessToken'] = new_personal_token + + with open(self.config_path, 'w') as f: + json.dump(config, f, indent=4) + + except Exception as e: + raise Exception(f"Failed to update personal token in config: {e}") + + def _refresh_with_personal_token(self) -> str: + """ + Refresh the token using the personal access token (full refresh). + + Returns: + str: The new access token + """ + def _refresh_with_personal_token(self) -> str: + """ + Refresh the token using the personal access token (full refresh). + + Returns: + str: The new access token + """ + try: + config = self._load_config() + + # Try to refresh the personal access token first if OAuth is configured + 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. " + "If using a portal token, get a new one from developer.webex.com. " + "If using an integration token, refresh your OAuth token.") + + response.raise_for_status() + + token_data = response.json() + new_access_token = token_data.get('access_token') + new_refresh_token = token_data.get('refresh_token') + + if not new_access_token: + raise Exception("No access token in API response") + + # Update the .env file with the new tokens + self._update_env_file(new_access_token, new_refresh_token) + + print("Token refreshed successfully using personal access token") + return new_access_token + + except requests.exceptions.RequestException as e: + print(f"API request failed: {e}") + raise Exception(f"Token refresh failed: {e}") + except Exception as e: + print(f"Failed to refresh token: {e}") + raise + + def _load_config(self) -> Dict[str, str]: + """ + Load configuration from the token config file. + + Returns: + Dict containing the configuration + + Raises: + Exception: If config file is not found or invalid + """ + try: + with open(self.config_path, 'r') as f: + config = json.load(f) + + # Validate structure + if 'serviceApp' not in config: + raise Exception("Missing 'serviceApp' section in config") + if 'tokenManager' not in config: + raise Exception("Missing 'tokenManager' section in config") + + # Validate service app fields + service_app_fields = ['appId', 'clientId', 'clientSecret', 'targetOrgId'] + missing_service_fields = [field for field in service_app_fields if field not in config['serviceApp']] + + # Validate token manager fields + token_manager_fields = ['personalAccessToken'] + missing_token_fields = [field for field in token_manager_fields if field not in config['tokenManager']] + + # OAuth integration fields are optional + if 'integration' in config['tokenManager']: + integration_fields = ['clientId', 'clientSecret', 'refreshToken'] + missing_integration_fields = [ + field for field in integration_fields + if field not in config['tokenManager']['integration'] + ] + if missing_integration_fields: + print(f"Warning: OAuth integration partially configured. Missing: {missing_integration_fields}") + print("OAuth token refresh will not be available.") + + all_missing = [] + if missing_service_fields: + all_missing.extend([f"serviceApp.{field}" for field in missing_service_fields]) + if missing_token_fields: + all_missing.extend([f"tokenManager.{field}" for field in missing_token_fields]) + + if all_missing: + raise Exception(f"Missing required fields in config: {all_missing}") + + return config + + except FileNotFoundError: + raise Exception(f"Token config file not found: {self.config_path}") + except json.JSONDecodeError: + raise Exception("Invalid JSON in token config file") + + def _update_env_file(self, new_access_token: str, new_refresh_token: Optional[str] = None) -> None: + """ + Update the .env file with the new tokens. + + Args: + new_access_token: The new access token to write to the file + new_refresh_token: The new refresh token to write to the file (optional) + """ + try: + # Read the current .env file + with open(self.env_path, 'r') as f: + content = f.read() + + # Replace the existing token lines + lines = content.split('\n') + updated_lines = [] + refresh_token_updated = False + + for line in lines: + if line.startswith('WEBEX_SERVICE_APP_ACCESS_TOKEN='): + updated_lines.append(f'WEBEX_SERVICE_APP_ACCESS_TOKEN={new_access_token}') + elif line.startswith('WEBEX_SERVICE_APP_REFRESH_TOKEN='): + if new_refresh_token: + updated_lines.append(f'WEBEX_SERVICE_APP_REFRESH_TOKEN={new_refresh_token}') + refresh_token_updated = True + # If no new refresh token, keep the existing line + else: + updated_lines.append(line) + else: + updated_lines.append(line) + + # Add refresh token if we have one but it wasn't in the file + if new_refresh_token and not refresh_token_updated: + updated_lines.append(f'WEBEX_SERVICE_APP_REFRESH_TOKEN={new_refresh_token}') + + # Write the updated content back to the file + with open(self.env_path, 'w') as f: + f.write('\n'.join(updated_lines)) + + except Exception as e: + raise Exception(f"Failed to update .env file: {e}") + + def is_token_valid(self) -> bool: + """ + Check if the current token is valid by making a test API call. + + Returns: + bool: True if token is valid, False otherwise + """ + try: + # Get current token from .env + current_token = self._get_current_token() + if not current_token: + return False + + # Test the token with the data sources endpoint (service app compatible) + headers = {'Authorization': f'Bearer {current_token}'} + response = requests.get('https://webexapis.com/v1/dataSources', headers=headers) + + return response.status_code == 200 + + except Exception: + return False + + def _get_current_token(self) -> Optional[str]: + """ + Get the current token from the .env file. + + Returns: + The current token or None if not found + """ + try: + with open(self.env_path, 'r') as f: + for line in f: + if line.startswith('WEBEX_SERVICE_APP_ACCESS_TOKEN='): + return line.split('=', 1)[1].strip() + return None + except Exception: + return None + + def _get_current_refresh_token(self) -> Optional[str]: + """ + Get the current refresh token from the .env file. + + Returns: + The current refresh token or None if not found + """ + try: + with open(self.env_path, 'r') as f: + for line in f: + if line.startswith('WEBEX_SERVICE_APP_REFRESH_TOKEN='): + return line.split('=', 1)[1].strip() + return None + except Exception: + return None + + def get_token_refresh_guidance(self) -> str: + """ + Provide guidance on how to refresh tokens based on common failure scenarios. + + Returns: + str: Guidance message for token refresh + """ + return """ +Token Refresh Guidance: + +If you're using a PORTAL TOKEN (Quick Start - Option A): +1. Go to developer.webex.com +2. Sign in and click your profile (top right) +3. Copy the new "Personal Access Token" +4. Update the 'personalAccessToken' in your token-config.json +5. Run: python refresh_token.py + +If you're using an INTEGRATION TOKEN (Production - Option B): +1. Your integration token may have expired +2. Check your integration's OAuth setup +3. Generate a new access token through the OAuth flow +4. Update the 'personalAccessToken' in your token-config.json +5. Run: python refresh_token.py + +For production use, consider Option B (Integration) for longer-lasting tokens. + """.strip()