From a7a44f4c778ba311f9e0bd2ba224788f109aaaca Mon Sep 17 00:00:00 2001 From: Adam Weeks Date: Fri, 10 Jul 2026 10:23:17 -0400 Subject: [PATCH 1/2] refactor: consume provider-based BYODS SDK --- .gitignore | 2 +- README.md | 14 +- TOKEN_MANAGEMENT.md | 12 +- data-sources.py | 264 +------------ deploy/AWS_SETUP.md | 2 +- deploy/package_lambda.sh | 5 +- get_service_app_token.py | 1 - lambda_function.py | 1 - requirements.txt | 1 + tests/test_token_manager.py | 54 +++ token_manager.py | 730 ++++++------------------------------ 11 files changed, 205 insertions(+), 881 deletions(-) create mode 100644 tests/test_token_manager.py diff --git a/.gitignore b/.gitignore index 60d3f04..5ad1904 100644 --- a/.gitignore +++ b/.gitignore @@ -32,4 +32,4 @@ node_modules/ # AWS Lambda deployment artifacts deploy/lambda_deployment.zip -deploy/lambda_package/ \ No newline at end of file +deploy/lambda_package/ diff --git a/README.md b/README.md index 2b00f45..2b143a8 100644 --- a/README.md +++ b/README.md @@ -27,13 +27,21 @@ Watch the Vidcast of this app](https://app.vidcast.io/share/63e954e4-f0ae-4c20-8 - Configurable token lifetime - Command-line automation friendly +## SDK Dependency + +This CLI and Lambda consume +[`webex-python-byods-sdk`](https://github.com/WebexCommunity/webex-python-byods-sdk) +at the commit pinned in `requirements.txt`. The SDK owns BYODS API operations +and token-provider behavior; this repository owns the interactive prompts, +JSON configuration, AWS Secrets Manager adapter, and operation records. + ## 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 +- **`token_manager.py`** - CLI/Lambda adapter that creates SDK token providers - **`refresh_token.py`** - Standalone script for manual token refresh - **`TOKEN_MANAGEMENT.md`** - Complete setup and usage documentation @@ -64,7 +72,7 @@ For complete setup instructions, see [TOKEN_MANAGEMENT.md](TOKEN_MANAGEMENT.md). ## Requirements -- Python 3.6 or higher +- Python 3.8 or higher - **A Webex Service App** with appropriate scopes: - `spark-admin:datasource_write` (for registration and updates) - `spark-admin:datasource_read` (for listing/viewing) @@ -259,7 +267,7 @@ The `TokenManager` class automatically detects its environment: - **Local execution**: Uses `token-config.json` for all credentials and tokens - **Lambda execution**: Uses AWS Secrets Manager for all credentials and tokens -This means the same code works in both environments without modification. +The adapter supplies either configuration source to the same SDK provider API. ### Monitoring and Troubleshooting diff --git a/TOKEN_MANAGEMENT.md b/TOKEN_MANAGEMENT.md index 9bc5a60..5f91f4f 100644 --- a/TOKEN_MANAGEMENT.md +++ b/TOKEN_MANAGEMENT.md @@ -144,12 +144,10 @@ python refresh_token.py ### Token Manager API -You can also use the TokenManager class directly in your code: +CLI and Lambda code can use this repository's adapter directly. It reads local +JSON or AWS Secrets Manager, then supplies explicit token providers to the SDK: ```python -# Activate virtual environment first -source venv/bin/activate - from token_manager import TokenManager # Initialize the token manager @@ -169,10 +167,8 @@ if not token_manager.is_token_valid(): - **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 `token-config.json` under the `env` section: - - `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 +- Service app tokens are cached in memory for the running process +- The SDK does not write credentials or token values to `token-config.json` - Consider using environment variables for the token config in production environments ## How Refresh Tokens Work diff --git a/data-sources.py b/data-sources.py index 3286484..e958588 100755 --- a/data-sources.py +++ b/data-sources.py @@ -16,280 +16,38 @@ import os import sys import json -import requests import argparse import uuid -import jwt from typing import Dict, Any, List from datetime import datetime from token_manager import TokenManager +from webex_byods import ( + WebexDataSourceClient as WebexDataSourceManager, + enhance_data_source_with_jwt, + get_token_expiration_display, +) -class WebexDataSourceManager: - """Handle Webex Data Source operations via API""" - - BASE_URL = "https://webexapis.com/v1" - - def __init__(self, access_token: str): - """Initialize with access token""" - self.access_token = access_token - self.headers = { - "Authorization": f"Bearer {access_token}", - "Accept": "application/json", - "Content-Type": "application/json", - } - self.schemas_cache = None # Cache for schemas - - # Initialize token manager for automatic refresh - try: - self.token_manager = TokenManager() - except Exception: - self.token_manager = None - - def _refresh_token_if_needed(self, response: requests.Response) -> bool: - """ - Check if a 401 response indicates token expiration and refresh if possible. - - Args: - response: The response object to check - - Returns: - bool: True if token was refreshed, False otherwise - """ - if response.status_code == 401 and self.token_manager: - try: - print("🔄 Access token appears to be expired, attempting refresh...") - new_token = self.token_manager.refresh_token() - - # Update the instance with new token - self.access_token = new_token - self.headers["Authorization"] = f"Bearer {new_token}" - - print("✅ Token refreshed successfully!") - return True - - except Exception as e: - print(f"❌ Failed to refresh token: {e}") - return False - return False - - def _make_request(self, method: str, url: str, **kwargs) -> Dict[str, Any]: - """ - Make an HTTP request with automatic token refresh on 401 errors. - - Args: - method: HTTP method (GET, POST, PUT, DELETE) - url: Request URL - **kwargs: Additional arguments for requests - - Returns: - Dict containing success status, data/error, and status code - """ - try: - # Make the initial request - response = requests.request( - method, url, headers=self.headers, timeout=30, **kwargs - ) - - # If we get a 401, try to refresh the token and retry once - if response.status_code == 401 and self._refresh_token_if_needed(response): - print("🔄 Retrying request with refreshed token...") - response = requests.request( - method, url, headers=self.headers, timeout=30, **kwargs - ) - - if response.status_code in [200, 201]: - return { - "success": True, - "data": response.json(), - "status_code": response.status_code, - } - else: - return { - "success": False, - "error": response.text, - "status_code": response.status_code, - } - - except requests.exceptions.RequestException as e: - return { - "success": False, - "error": f"Request failed: {str(e)}", - "status_code": None, - } - - def list_all_data_sources(self) -> Dict[str, Any]: - """Retrieve all data sources""" - url = f"{self.BASE_URL}/dataSources" - return self._make_request("GET", url) - - def get_data_source_details(self, data_source_id: str) -> Dict[str, Any]: - """Retrieve details for a specific data source""" - url = f"{self.BASE_URL}/dataSources/{data_source_id}" - return self._make_request("GET", url) - - def register_data_source( - self, data_source_config: Dict[str, Any] - ) -> Dict[str, Any]: - """Register a new data source""" - url = f"{self.BASE_URL}/dataSources" - return self._make_request("POST", url, json=data_source_config) - - def update_data_source( - self, data_source_id: str, update_config: Dict[str, Any] - ) -> Dict[str, Any]: - """Update a data source""" - url = f"{self.BASE_URL}/dataSources/{data_source_id}" - return self._make_request("PUT", url, json=update_config) - - def get_data_source_schemas(self) -> Dict[str, Any]: - """Retrieve all available data source schemas""" - url = f"{self.BASE_URL}/dataSources/schemas" - return self._make_request("GET", url) - - def load_schemas_cache(self) -> bool: - """Load schemas into cache for friendly display""" - if self.schemas_cache is not None: - return True # Already loaded - - result = self.get_data_source_schemas() - if result["success"]: - self.schemas_cache = result["data"].get("items", []) - return True - else: - print(f"Warning: Could not load schemas: {result['error']}") - self.schemas_cache = [] - return False - - def get_schema_display_name(self, schema_id: str) -> str: - """Get friendly display name for a schema ID""" - if self.schemas_cache is None: - self.load_schemas_cache() - - for schema in self.schemas_cache: - if schema.get("id") == schema_id: - service_type = schema.get("serviceType", "Unknown Service") - return f"{service_type} ({schema_id[:8]}...)" - - return f"Unknown Schema ({schema_id[:8]}...)" - - def get_available_schemas(self) -> List[Dict[str, Any]]: - """Get list of available schemas for selection""" - if self.schemas_cache is None: - self.load_schemas_cache() - - return self.schemas_cache - - -def load_env_token() -> str: - """Get a fresh service app access token from token-config.json""" +def load_env_token() -> tuple: + """Build the CLI adapter and retrieve an initial service-app token.""" # Load token-config.json from the same directory as the script script_dir = os.path.dirname(os.path.abspath(__file__)) config_path = os.path.join(script_dir, "token-config.json") try: - from token_manager import TokenManager - token_manager = TokenManager(config_path=config_path) # Get fresh service app token print("Fetching fresh service app token...") token = token_manager.get_service_app_token() print("Service app token retrieved successfully!") - return token - - except ImportError: - print("Error: TokenManager not available") - print("Please ensure token_manager.py is in the same directory") - sys.exit(1) + return token, token_manager except Exception as e: print(f"Error: Could not get service app token: {e}") print("Please ensure your token-config.json is properly configured") sys.exit(1) -def decode_jwt_token(token: str) -> Dict[str, Any]: - """Decode JWT token to extract audience and subject""" - try: - # Decode without verification since we just need to read the payload - decoded = jwt.decode(token, options={"verify_signature": False}) - return decoded - except Exception as e: - print(f"Warning: Could not decode JWT token: {str(e)}") - return {} - - -def enhance_data_source_with_jwt(data_source: Dict[str, Any]) -> Dict[str, Any]: - """Enhance data source data by decoding JWT token if present""" - enhanced = data_source.copy() - - # Check if there's a JWT token field - jwt_token = data_source.get("jwtToken") or data_source.get("jwsToken") - - if jwt_token: - jwt_claims = decode_jwt_token(jwt_token) - - # Extract audience and subject from JWT if not present in main data - if not enhanced.get("audience") and jwt_claims.get("aud"): - enhanced["audience"] = jwt_claims["aud"] - - if not enhanced.get("subject") and jwt_claims.get("sub"): - enhanced["subject"] = jwt_claims["sub"] - - # Store JWT claims for reference - enhanced["jwt_claims"] = jwt_claims - - return enhanced - - -def get_token_expiration_display(data_source: Dict[str, Any]) -> str: - """ - Get a human-readable token expiration display string by parsing the JWT token. - - Args: - data_source: Data source dictionary from API response - - Returns: - String showing token expiration status - """ - try: - # Get the JWT token from the data source - jws_token = data_source.get("jwsToken") or data_source.get("jwtToken") - - if not jws_token: - return "No token found" - - # Decode the JWT token to get expiration - decoded = jwt.decode(jws_token, options={"verify_signature": False}) - exp_timestamp = decoded.get("exp") - - if not exp_timestamp: - return "No expiration in token" - - # Convert timestamp to datetime and calculate remaining time - exp_date = datetime.fromtimestamp(exp_timestamp) - now = datetime.now() - - if exp_date > now: - time_diff = exp_date - now - total_seconds = int(time_diff.total_seconds()) - - if total_seconds > 3600: # More than 1 hour - hours = total_seconds // 3600 - minutes = (total_seconds % 3600) // 60 - return f"{hours}h {minutes}m" - elif total_seconds > 60: # More than 1 minute - minutes = total_seconds // 60 - return f"{minutes}m" - else: # Less than 1 minute - return f"{total_seconds}s" - else: - return "EXPIRED" - - except Exception as e: - return f"Parse error: {str(e)[:15]}" - - def display_data_sources_list( data_sources: List[Dict[str, Any]], manager: WebexDataSourceManager = None ) -> None: @@ -690,10 +448,12 @@ def main(): try: # Load access token - access_token = load_env_token() + access_token, token_manager = load_env_token() # Create manager instance - manager = WebexDataSourceManager(access_token) + manager = WebexDataSourceManager( + token_provider=token_manager.get_token_provider() + ) print("Webex Data Source Manager") print("=" * 30) diff --git a/deploy/AWS_SETUP.md b/deploy/AWS_SETUP.md index 90731db..8bce7d7 100644 --- a/deploy/AWS_SETUP.md +++ b/deploy/AWS_SETUP.md @@ -716,7 +716,7 @@ Should show the complete JSON structure with all required fields. To update the code: -1. Make changes locally to `lambda_function.py` or `token_manager.py` +1. Make changes locally to `lambda_function.py`, `token_manager.py`, or `requirements.txt` 2. Run the packaging script: `./deploy/package_lambda.sh` 3. Upload new `lambda_deployment.zip` to Lambda 4. Test the function diff --git a/deploy/package_lambda.sh b/deploy/package_lambda.sh index d7972be..c9635c5 100755 --- a/deploy/package_lambda.sh +++ b/deploy/package_lambda.sh @@ -38,9 +38,7 @@ fi # Install dependencies to package directory pip install --target "$PACKAGE_DIR" \ - requests \ - pyjwt \ - boto3 \ + -r ../requirements.txt \ --upgrade \ --quiet @@ -92,4 +90,3 @@ echo "4. Set up EventBridge trigger for hourly execution" echo "" echo "For detailed instructions, see deploy/AWS_SETUP.md" echo "" - diff --git a/get_service_app_token.py b/get_service_app_token.py index 2f90eb8..fe29801 100644 --- a/get_service_app_token.py +++ b/get_service_app_token.py @@ -25,4 +25,3 @@ def main(): if __name__ == "__main__": main() - diff --git a/lambda_function.py b/lambda_function.py index 4d9cafb..d9a7e1f 100644 --- a/lambda_function.py +++ b/lambda_function.py @@ -152,4 +152,3 @@ def __init__(self): print("-" * 60) print("Response:") print(json.dumps(response, indent=2)) - diff --git a/requirements.txt b/requirements.txt index ab38fe3..816f55e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,3 +2,4 @@ requests>=2.31.0 python-dotenv>=1.0.0 pyjwt>=2.8.0 boto3>=1.28.0 +webex-byods-sdk @ git+https://github.com/WebexCommunity/webex-python-byods-sdk.git@dd89e8fed4bd52db0e1d21bd91232e3c92999029 diff --git a/tests/test_token_manager.py b/tests/test_token_manager.py new file mode 100644 index 0000000..2f18d47 --- /dev/null +++ b/tests/test_token_manager.py @@ -0,0 +1,54 @@ +import json + +from webex_byods import OAuthRefreshTokenProvider, StaticAccessTokenProvider + +from token_manager import TokenManager + + +def write_config(path, token_manager): + path.write_text( + json.dumps( + { + "serviceApp": { + "appId": "app-id", + "clientId": "service-client-id", + "clientSecret": "service-client-secret", + "targetOrgId": "org-id", + }, + "tokenManager": token_manager, + } + ) + ) + + +def test_token_manager_builds_service_provider_from_json_config(tmp_path): + config_path = tmp_path / "token-config.json" + write_config(config_path, {"personalAccessToken": "personal-token"}) + + provider = TokenManager(config_path=str(config_path)).get_token_provider() + + assert provider.credentials.app_id == "app-id" + assert provider.credentials.client_id == "service-client-id" + assert isinstance(provider.personal_token_provider, StaticAccessTokenProvider) + assert provider.personal_token_provider.get_access_token() == "personal-token" + assert TokenManager(config_path=str(config_path)).get_client().token_provider.credentials.app_id == "app-id" + + +def test_token_manager_uses_oauth_refresh_credentials_from_config(tmp_path): + config_path = tmp_path / "token-config.json" + write_config( + config_path, + { + "personalAccessToken": "personal-token", + "oauthClientId": "oauth-client-id", + "oauthClientSecret": "oauth-client-secret", + "oauthRefreshToken": "oauth-refresh-token", + }, + ) + + provider = TokenManager(config_path=str(config_path)).get_token_provider() + + assert isinstance(provider.personal_token_provider, OAuthRefreshTokenProvider) + assert provider.personal_token_provider.client_id == "oauth-client-id" + assert provider.personal_token_provider.client_secret == "oauth-client-secret" + assert provider.personal_token_provider.refresh_token == "oauth-refresh-token" diff --git a/token_manager.py b/token_manager.py index 603d992..a20dcdc 100644 --- a/token_manager.py +++ b/token_manager.py @@ -1,660 +1,170 @@ +"""CLI and Lambda adapter for configuring the Webex BYODS SDK.""" + import json import os -import requests -import uuid from typing import Any, Dict, Optional -# AWS SDK import - only used in Lambda environment +from webex_byods import ( + OAuthRefreshTokenProvider, + ServiceAppCredentials, + StaticAccessTokenProvider, + WebexDataSourceClient, + WebexServiceAppTokenProvider, +) + try: import boto3 from botocore.exceptions import ClientError + AWS_AVAILABLE = True except ImportError: 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. - - Simplified approach: Always fetches fresh service app tokens on demand. - Tokens are kept in memory only and never persisted. - - Key Features: - - Fetches fresh service app tokens using personal access token - - OAuth support for automatic personal token refresh - - Data source token extension - - AWS Secrets Manager support for Lambda deployments - """ + """Build SDK token providers from CLI files or Lambda Secrets Manager data.""" - def __init__(self, config_path: str = "token-config.json", secret_name: Optional[str] = None): - """ - Initialize TokenManager. - - Args: - config_path: Path to the token configuration file (used for local execution) - secret_name: AWS Secrets Manager secret name (used for Lambda execution) - """ + def __init__( + self, config_path: str = "token-config.json", secret_name: Optional[str] = None + ) -> None: self.config_path = config_path self.secret_name = secret_name - self.use_aws = self._should_use_aws() - - # In-memory token cache (per script execution only) - self._service_app_token = None - self._service_app_refresh_token = None - - # Initialize AWS Secrets Manager client if needed - if self.use_aws: - if not AWS_AVAILABLE: - raise Exception("boto3 is not installed. Install it with: pip install boto3") - self.secrets_client = boto3.client('secretsmanager', region_name=os.environ.get('AWS_REGION', 'us-east-1')) - self._secret_cache = None # Cache the secret to reduce API calls - - def _should_use_aws(self) -> bool: - """ - Determine if we should use AWS Secrets Manager. - - Returns: - bool: True if running in AWS Lambda environment, False otherwise - """ - # Check if we're in Lambda environment - if 'AWS_LAMBDA_FUNCTION_NAME' in os.environ: - return True - # Check if secret_name is provided and AWS is available - if self.secret_name and AWS_AVAILABLE: - return True - return False - - def _get_secret_from_aws(self) -> Dict: - """ - Retrieve secret from AWS Secrets Manager. - - Returns: - Dict: The secret data containing credentials - """ - if self._secret_cache: - return self._secret_cache - - try: - response = self.secrets_client.get_secret_value(SecretId=self.secret_name) - secret_data = json.loads(response['SecretString']) - self._secret_cache = secret_data - return secret_data - except ClientError as e: - error_code = e.response['Error']['Code'] - if error_code == 'ResourceNotFoundException': - raise Exception(f"Secret '{self.secret_name}' not found in AWS Secrets Manager") - elif error_code == 'AccessDeniedException': - raise Exception(f"Access denied to secret '{self.secret_name}'. Check IAM permissions.") - else: - raise Exception(f"Failed to retrieve secret from AWS: {e}") - - def _update_personal_token_in_config(self, new_personal_token: str) -> None: - """ - Update the personal access token in the config file or AWS Secrets Manager. + self.use_aws = "AWS_LAMBDA_FUNCTION_NAME" in os.environ or bool( + secret_name and AWS_AVAILABLE + ) + self._config_cache: Optional[Dict[str, Any]] = None + self._token_provider: Optional[WebexServiceAppTokenProvider] = None - Args: - new_personal_token: The new personal access token - """ - # Use AWS Secrets Manager if in Lambda environment if self.use_aws: - try: - secret_data = self._get_secret_from_aws() - if 'tokenManager' not in secret_data: - secret_data['tokenManager'] = {} - secret_data['tokenManager']['personalAccessToken'] = new_personal_token - - # Update the secret - self.secrets_client.update_secret( - SecretId=self.secret_name, - SecretString=json.dumps(secret_data) - ) - # Update cache - self._secret_cache = secret_data - return - except Exception as e: - raise Exception(f"Failed to update personal token in AWS Secrets Manager: {e}") - - # Otherwise use local config file - try: - import tempfile - - with open(self.config_path, "r") as f: - config = json.load(f) - - config["tokenManager"]["personalAccessToken"] = new_personal_token - - # Write back to file atomically - temp_file = tempfile.NamedTemporaryFile( - mode='w', - delete=False, - dir=os.path.dirname(self.config_path) or '.', - prefix='.token-config-', - suffix='.tmp' + if not AWS_AVAILABLE: + raise RuntimeError("boto3 is required for AWS Secrets Manager support") + self.secrets_client = boto3.client( + "secretsmanager", region_name=os.environ.get("AWS_REGION", "us-east-1") ) - try: - json.dump(config, temp_file, indent=4) - temp_file.close() - os.replace(temp_file.name, self.config_path) - except Exception: - try: - os.unlink(temp_file.name) - except Exception: - pass - raise - - except Exception as e: - raise Exception(f"Failed to update personal token in config: {e}") def _load_config(self) -> Dict[str, Any]: - """ - Load configuration from AWS Secrets Manager or local config file. + if self._config_cache is not None: + return self._config_cache - Returns: - Dict containing the configuration - - Raises: - Exception: If config is not found or invalid - """ - # Use AWS Secrets Manager if in Lambda environment if self.use_aws: - secret_data = self._get_secret_from_aws() - config = { - 'serviceApp': secret_data.get('serviceApp', {}), - 'tokenManager': secret_data.get('tokenManager', {}) - } + try: + response = self.secrets_client.get_secret_value(SecretId=self.secret_name) + config = json.loads(response["SecretString"]) + except ClientError as error: + raise RuntimeError(f"Failed to read AWS secret: {error}") from error else: - # Load from local file try: - with open(self.config_path, "r") as f: - config = json.load(f) - except FileNotFoundError: - raise Exception(f"Token config file not found: {self.config_path}") - except json.JSONDecodeError: - raise Exception("Invalid JSON in token config file") - - try: - # 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 fields are optional but must all be present together - 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): - missing_oauth = [field for field in oauth_fields if field not in config["tokenManager"]] - print( - f"Warning: OAuth partially configured. Missing: {missing_oauth}" - ) - 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] - ) + with open(self.config_path, "r", encoding="utf-8") as config_file: + config = json.load(config_file) + except FileNotFoundError as error: + raise RuntimeError( + f"Token config file not found: {self.config_path}" + ) from error + except json.JSONDecodeError as error: + raise RuntimeError("Invalid JSON in token config file") from error + + self._validate_config(config) + self._config_cache = config + return config + + @staticmethod + def _validate_config(config: Dict[str, Any]) -> None: + service_app = config.get("serviceApp", {}) + token_manager = config.get("tokenManager", {}) + missing_fields = [ + f"serviceApp.{field}" + for field in ("appId", "clientId", "clientSecret", "targetOrgId") + if not service_app.get(field) + ] + if not token_manager.get("personalAccessToken"): + missing_fields.append("tokenManager.personalAccessToken") + if missing_fields: + raise RuntimeError(f"Missing required fields in config: {missing_fields}") + + @staticmethod + def _oauth_value(config: Dict[str, Any], primary: str, legacy: str) -> Optional[str]: + return config.get(primary) or config.get(legacy) + + def get_token_provider(self) -> WebexServiceAppTokenProvider: + """Build and cache the SDK service-app token provider.""" + if self._token_provider is not None: + return self._token_provider - if all_missing: - raise Exception(f"Missing required fields in config: {all_missing}") - - return config - - except Exception: - raise - - 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 + config = self._load_config() + service_app = config["serviceApp"] + token_config = config["tokenManager"] + oauth_client_id = self._oauth_value(token_config, "oauthClientId", "clientId") + oauth_client_secret = self._oauth_value( + token_config, "oauthClientSecret", "clientSecret" + ) + oauth_refresh_token = self._oauth_value( + token_config, "oauthRefreshToken", "refreshToken" + ) - Returns: - bool: True if valid, False otherwise - """ - try: - headers = {"Authorization": f"Bearer {token}"} - response = requests.get( - "https://webexapis.com/v1/people/me", headers=headers + if all((oauth_client_id, oauth_client_secret, oauth_refresh_token)): + personal_token_provider = OAuthRefreshTokenProvider( + client_id=oauth_client_id, + client_secret=oauth_client_secret, + refresh_token=oauth_refresh_token, ) - return response.status_code == 200 - except Exception: - return False - - def refresh_personal_token_oauth(self, config: Dict) -> str: - """ - Refresh the personal access token using OAuth. - - Args: - config: Configuration containing OAuth fields - - 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": 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." + else: + personal_token_provider = StaticAccessTokenProvider( + token_config["personalAccessToken"] ) - response.raise_for_status() - - token_data = response.json() - new_access_token = token_data.get("access_token") + self._token_provider = WebexServiceAppTokenProvider( + ServiceAppCredentials( + app_id=service_app["appId"], + client_id=service_app["clientId"], + client_secret=service_app["clientSecret"], + target_org_id=service_app["targetOrgId"], + ), + personal_token_provider, + ) + return self._token_provider - if not new_access_token: - raise Exception("No access token in OAuth refresh response") + def get_service_app_token(self, force_refresh: bool = False) -> str: + """Return a service-app token through the configured SDK provider.""" + return self.get_token_provider().get_access_token(force_refresh=force_refresh) - return new_access_token + def get_client(self) -> WebexDataSourceClient: + """Return a data-source client that can refresh through this adapter.""" + return WebexDataSourceClient(token_provider=self.get_token_provider()) def is_token_valid(self) -> bool: - """ - Check if the current personal access token (from config) is valid. - - Returns: - bool: True if valid, False otherwise - """ + """Check that the configured credentials can obtain a service-app token.""" try: - config = self._load_config() - token = config["tokenManager"]["personalAccessToken"] - return self.is_personal_token_valid(token) + self.get_service_app_token() + return True except Exception: return False def refresh_token(self) -> str: - """ - Refresh the personal access token via OAuth and return the new token. - - Returns: - str: The new personal access token - - Raises: - Exception: If OAuth is not configured or refresh fails - """ - config = self._load_config() - return self._try_refresh_personal_token(config) + """Refresh the personal token when OAuth credentials are configured.""" + provider = self.get_token_provider().personal_token_provider + if not isinstance(provider, OAuthRefreshTokenProvider): + raise RuntimeError( + "OAuth refresh is not configured. Update tokenManager credentials " + "or run setup_oauth.py." + ) + return provider.get_access_token(force_refresh=True) def _get_current_refresh_token(self) -> Optional[str]: - """ - Get the current OAuth refresh token if configured. - - Returns: - Optional[str]: The refresh token string if OAuth is configured, - None otherwise - """ - try: - config = self._load_config() - token_manager = config["tokenManager"] - return token_manager.get("refreshToken") - except Exception: - return None + token_config = self._load_config()["tokenManager"] + return self._oauth_value(token_config, "oauthRefreshToken", "refreshToken") def get_token_refresh_guidance(self) -> str: - """ - Get guidance for resolving token refresh issues. - - Returns: - str: Human-readable guidance for fixing token problems - """ return ( "Token Refresh Guidance:\n" - "1. If OAuth is not configured: Run setup_oauth.py to configure " - "automatic token refresh.\n" - "2. If OAuth refresh token expired: Re-run setup_oauth.py to " - "re-authorize your integration.\n" - "3. Manual update: Edit token-config.json and set " - "tokenManager.personalAccessToken to a valid token from " - "https://developer.webex.com." - ) - - def _try_refresh_personal_token(self, config: Dict) -> str: - """ - Attempt to refresh the personal access token via OAuth. - - Args: - config: The loaded configuration - - Returns: - str: A refreshed personal access token - - Raises: - Exception: If refresh is not configured or fails - """ - token_manager = config["tokenManager"] - - # Check if OAuth is configured (all three fields must be present) - oauth_configured = all( - key in token_manager - for key in ["clientId", "clientSecret", "refreshToken"] + "1. Configure OAuth credentials with setup_oauth.py for automatic refresh.\n" + "2. Re-authorize the integration if its OAuth refresh token expires.\n" + "3. Otherwise update tokenManager.personalAccessToken in token-config.json." ) - - 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 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: - str: Service app access token - - Raises: - Exception: If token request fails - """ - # Return cached token if we already have one for this execution - if self._service_app_token: - return self._service_app_token - - config = self._load_config() - - # Try to get token with current personal token - try: - 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"] - - 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) - response.raise_for_status() # Raises HTTPError for bad status codes - - token_data = response.json() - access_token = token_data.get("access_token") - refresh_token = token_data.get("refresh_token") - - if not access_token: - raise Exception("No access token in API response") - - # Cache tokens in memory for this execution - self._service_app_token = access_token - self._service_app_refresh_token = refresh_token - - return access_token def extend_data_source_token( self, data_source_id: str, token_lifetime_minutes: int = 1440 ) -> Dict[str, Any]: - """ - Extend a data source token by updating only the nonce. - - Args: - data_source_id: The ID of the data source to update - token_lifetime_minutes: Token lifetime in minutes (default: 1440 = 24 hours, max: 1440) - - Returns: - Dict containing the result of the operation - """ - try: - # Validate token lifetime - if token_lifetime_minutes > 1440: - return { - "success": False, - "error": f"Token lifetime cannot exceed 1440 minutes (24 hours). Requested: {token_lifetime_minutes} minutes", - } - - if token_lifetime_minutes <= 0: - return { - "success": False, - "error": f"Token lifetime must be positive. Requested: {token_lifetime_minutes} minutes", - } - - # Get fresh service app token - access_token = self.get_service_app_token() - - # First, get the current data source configuration - headers = {"Authorization": f"Bearer {access_token}"} - get_url = f"https://webexapis.com/v1/dataSources/{data_source_id}" - - response = requests.get(get_url, headers=headers) - - if response.status_code != 200: - return { - "success": False, - "error": f"Failed to retrieve data source: {response.text}", - "status_code": response.status_code, - } - - current_config = response.json() - - # Parse the JWT token to extract audience, subject, and schema info - jws_token = current_config.get("jwsToken", "") - audience = None - subject = None - schema_uuid = None - - if jws_token: - try: - # Decode JWT without verification to extract claims - import jwt - - decoded = jwt.decode(jws_token, options={"verify_signature": False}) - audience = decoded.get("aud") - subject = decoded.get("sub") - schema_uuid = decoded.get("com.cisco.datasource.schema.uuid") - except Exception as e: - print(f"Warning: Could not decode JWT token: {e}") - - # Fallback to current config values if JWT parsing failed - if not audience: - audience = current_config.get("audience", "") - if not subject: - subject = current_config.get("subject", "subject") - if not schema_uuid: - schema_uuid = current_config.get("schemaId", "") - - # Generate a new nonce (this is what triggers the token refresh) - new_nonce = str(uuid.uuid4()) - - # Create update configuration with all required fields - update_config = { - "audience": audience, - "nonce": new_nonce, - "schemaId": schema_uuid, - "subject": subject, - "url": current_config.get("url"), - "tokenLifetimeMinutes": token_lifetime_minutes, - "status": current_config.get("status", "active"), - } - - # Validate that we have all required fields - required_fields = ["audience", "schemaId", "url"] - missing_fields = [ - field for field in required_fields if not update_config.get(field) - ] - - if missing_fields: - return { - "success": False, - "error": f"Missing required fields: {missing_fields}. Could not extract from current data source.", - } - - # Update the data source - update_url = f"https://webexapis.com/v1/dataSources/{data_source_id}" - headers["Content-Type"] = "application/json" - - update_response = requests.put( - update_url, headers=headers, json=update_config - ) - - if update_response.status_code == 200: - result_data = update_response.json() - return { - "success": True, - "data": result_data, - "nonce_updated": new_nonce, - "token_lifetime_minutes": token_lifetime_minutes, - "token_expiry": result_data.get("tokenExpiryTime"), - "message": f"Data source token extended successfully. New expiry: {result_data.get('tokenExpiryTime')}", - } - else: - return { - "success": False, - "error": f"Failed to update data source: {update_response.text}", - "status_code": update_response.status_code, - } - - except requests.exceptions.RequestException as e: - return {"success": False, "error": f"Request failed: {str(e)}"} - except Exception as e: - return {"success": False, "error": f"Unexpected error: {str(e)}"} + """Delegate data-source token extension to the SDK client.""" + return self.get_client().extend_data_source_token( + data_source_id, token_lifetime_minutes + ) From b21a3370c7e51d2dcd81cdc673d06e5939a55dd1 Mon Sep 17 00:00:00 2001 From: Adam Weeks Date: Fri, 10 Jul 2026 15:36:14 -0400 Subject: [PATCH 2/2] chore: use published BYODS SDK --- README.md | 8 ++++---- requirements.txt | 2 +- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 2b143a8..3a71ff9 100644 --- a/README.md +++ b/README.md @@ -30,10 +30,10 @@ Watch the Vidcast of this app](https://app.vidcast.io/share/63e954e4-f0ae-4c20-8 ## SDK Dependency This CLI and Lambda consume -[`webex-python-byods-sdk`](https://github.com/WebexCommunity/webex-python-byods-sdk) -at the commit pinned in `requirements.txt`. The SDK owns BYODS API operations -and token-provider behavior; this repository owns the interactive prompts, -JSON configuration, AWS Secrets Manager adapter, and operation records. +[`webex-byods-sdk`](https://pypi.org/project/webex-byods-sdk/) at the version +pinned in `requirements.txt`. The SDK owns BYODS API operations and +token-provider behavior; this repository owns the interactive prompts, JSON +configuration, AWS Secrets Manager adapter, and operation records. ## Token Management diff --git a/requirements.txt b/requirements.txt index 816f55e..035e1e8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,4 +2,4 @@ requests>=2.31.0 python-dotenv>=1.0.0 pyjwt>=2.8.0 boto3>=1.28.0 -webex-byods-sdk @ git+https://github.com/WebexCommunity/webex-python-byods-sdk.git@dd89e8fed4bd52db0e1d21bd91232e3c92999029 +webex-byods-sdk==0.2.0