Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 63 additions & 3 deletions token_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
import os
import requests
import uuid
from typing import Dict, Optional
from typing import Any, Dict, Optional

# AWS SDK import - only used in Lambda environment
try:
Expand Down Expand Up @@ -206,7 +206,7 @@ def _update_personal_token_in_config(self, new_personal_token: str) -> None:
except Exception as e:
raise Exception(f"Failed to update personal token in config: {e}")

def _load_config(self) -> Dict[str, str]:
def _load_config(self) -> Dict[str, Any]:
"""
Load configuration from AWS Secrets Manager or local config file.

Expand Down Expand Up @@ -340,6 +340,66 @@ def refresh_personal_token_oauth(self, config: Dict) -> str:

return new_access_token

def is_token_valid(self) -> bool:
"""
Check if the current personal access token (from config) is valid.

Returns:
bool: True if valid, False otherwise
"""
try:
config = self._load_config()
token = config["tokenManager"]["personalAccessToken"]
return self.is_personal_token_valid(token)
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)

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

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.
Expand Down Expand Up @@ -474,7 +534,7 @@ def _fetch_service_app_token(self, config: Dict) -> str:

def extend_data_source_token(
self, data_source_id: str, token_lifetime_minutes: int = 1440
) -> Dict[str, any]:
) -> Dict[str, Any]:
"""
Extend a data source token by updating only the nonce.

Expand Down
Loading