From 93022fb0f32edc7d8d6dc21c03d04978db0aa077 Mon Sep 17 00:00:00 2001 From: Nirvana Date: Sun, 16 Nov 2025 21:22:25 +0100 Subject: [PATCH] Fix taa_client --- .../providers/magenta2/auth.py | 39 + .../providers/magenta2/provider.py | 33 +- .../providers/magenta2/taa_client.py | 419 ++++++---- .../providers/magenta2/token_flow_manager.py | 725 ++++++++++++++++++ 4 files changed, 1053 insertions(+), 163 deletions(-) create mode 100644 lib/streaming_providers/providers/magenta2/token_flow_manager.py diff --git a/lib/streaming_providers/providers/magenta2/auth.py b/lib/streaming_providers/providers/magenta2/auth.py index 59162ec..c5b7b29 100644 --- a/lib/streaming_providers/providers/magenta2/auth.py +++ b/lib/streaming_providers/providers/magenta2/auth.py @@ -17,6 +17,7 @@ from ...base.utils.logger import logger from .sam3_client import Sam3Client from .sso_client import SsoClient from .taa_client import TaaClient, TaaAuthResult +from .token_flow_manager import TokenFlowManager from .constants import ( SUPPORTED_COUNTRIES, @@ -390,6 +391,10 @@ class Magenta2Authenticator(BaseOAuth2Authenticator): self._taa_client: Optional[TaaClient] = None self._initialize_taa_client() + # Initialize TokenFlowManager + self.token_flow_manager: Optional[TokenFlowManager] = None + self._initialize_token_flow_manager() + # Initialize parent super().__init__( provider_name='magenta2', @@ -495,6 +500,40 @@ class Magenta2Authenticator(BaseOAuth2Authenticator): except Exception as e: logger.warning(f"Failed to initialize SAM3/SSO clients: {e}") + def _initialize_token_flow_manager(self) -> None: + """Initialize token flow manager after SAM3 and TAA clients are ready""" + if self._sam3_client and self._taa_client: + from .token_flow_manager import TokenFlowManager + + self.token_flow_manager = TokenFlowManager( + session_manager=self.settings_manager, + sam3_client=self._sam3_client, + taa_client=self._taa_client, + provider_name=self.provider_name, + country=self.country + ) + logger.debug("TokenFlowManager initialized") + + def get_yo_digital_token(self, force_refresh: bool = False) -> Optional[str]: + """ + Public method to get yo_digital access token + + Returns: + yo_digital access token or None + """ + if not self.token_flow_manager: + logger.warning("TokenFlowManager not initialized") + return None + + result = self.token_flow_manager.get_yo_digital_token(force_refresh) + + if result.success: + logger.info(f"✓ Got yo_digital token via: {result.flow_path}") + return result.access_token + else: + logger.error(f"✗ Failed to get yo_digital token: {result.error}") + return None + def update_sam3_qr_code_url(self, qr_code_url: str) -> bool: """ Public method to update SAM3 client with QR code URL diff --git a/lib/streaming_providers/providers/magenta2/provider.py b/lib/streaming_providers/providers/magenta2/provider.py index d543834..39f5e55 100644 --- a/lib/streaming_providers/providers/magenta2/provider.py +++ b/lib/streaming_providers/providers/magenta2/provider.py @@ -474,11 +474,36 @@ class Magenta2Provider(StreamingProvider): logger.error(f"Device registration failed: {e}") return False + def get_yo_digital_token(self, force_refresh: bool = False) -> str: + """Get yo_digital access token following complete hierarchy""" + + # Try TokenFlowManager via public API + yo_digital_token = self.authenticator.get_yo_digital_token(force_refresh) + + if yo_digital_token: + logger.info("✓ Got yo_digital token") + return yo_digital_token + + # Fallback to existing authentication + logger.info("Using legacy authentication flow") + return self.authenticate(force_refresh=force_refresh) + def authenticate(self, **kwargs) -> str: - """ - ENHANCED: Authenticate with line auth priority - """ - # PROPER: Use public method to check line auth availability + """Authenticate and get access token""" + force_refresh = kwargs.get('force_refresh', False) + + # Try new yo_digital flow first (unless force_legacy is set) + if not kwargs.get('force_legacy', False): + yo_digital_token = self.authenticator.get_yo_digital_token(force_refresh) + + if yo_digital_token: + self.bearer_token = yo_digital_token + logger.info("✓ Authentication via yo_digital") + return self.bearer_token + + # Fallback to existing line_auth + TAA flow + logger.info("Using legacy authentication flow") + line_auth_available = ( hasattr(self.authenticator, 'can_use_line_auth') and self.authenticator.can_use_line_auth() diff --git a/lib/streaming_providers/providers/magenta2/taa_client.py b/lib/streaming_providers/providers/magenta2/taa_client.py index 84ac2e1..2e2c55f 100644 --- a/lib/streaming_providers/providers/magenta2/taa_client.py +++ b/lib/streaming_providers/providers/magenta2/taa_client.py @@ -15,9 +15,20 @@ from .constants import ( ) +@dataclass +class YoDigitalTokens: + """Result of yo_digital token operations""" + access_token: str + access_token_expires_in: int + refresh_token: str + refresh_token_expires_in: int + device_limit_exceeded: bool = False + raw_response: Optional[Dict[str, Any]] = None + + @dataclass class TaaAuthResult: - """Result of TAA authentication""" + """Result of TAA authentication (legacy - kept for compatibility)""" access_token: str refresh_token: Optional[str] = None dc_cts_persona_token: Optional[str] = None @@ -35,7 +46,7 @@ class TaaAuthResult: class TaaClient: """ Telekom Authentication and Authorization (TAA) client - Handles complete TAA authentication flow with proper JWT parsing + Handles yo_digital token operations via the TAA endpoint """ def __init__(self, http_manager: HTTPManager, platform: str = DEFAULT_PLATFORM): @@ -43,23 +54,204 @@ class TaaClient: self.platform = platform self.platform_config = MAGENTA2_PLATFORMS.get(platform, MAGENTA2_PLATFORMS[DEFAULT_PLATFORM]) + def get_yo_digital_tokens(self, taa_access_token: str, device_id: str, + client_model: Optional[str] = None, + device_model: Optional[str] = None, + yo_digital_endpoint: Optional[str] = None) -> Optional[YoDigitalTokens]: + """ + Get yo_digital tokens from TAA endpoint (renamed from authenticate) + + Args: + taa_access_token: TAA access token with 'taa' scope + device_id: Device identifier + client_model: Client model from bootstrap + device_model: Device model from bootstrap + yo_digital_endpoint: yo_digital endpoint URL (formerly taa_endpoint) + + Returns: + YoDigitalTokens with access and refresh tokens, or None on failure + """ + try: + logger.debug("Getting yo_digital tokens from TAA endpoint") + + # Build complete TAA payload + taa_payload = self._build_complete_taa_payload( + sam3_token=taa_access_token, + device_id=device_id, + client_model=client_model, + device_model=device_model + ) + + # Build headers + headers = self._get_taa_headers() + + # Use provided endpoint or fallback + endpoint = yo_digital_endpoint or "https://gateway-de-proxy.tv.yo-digital.com/de-idm/P/onboarding/login" + + logger.debug(f"yo_digital request to: {endpoint}") + logger.debug(f"TAA payload keyValue: {taa_payload.get('keyValue', 'MISSING')}") + + # Perform yo_digital request + response = self.http_manager.post( + endpoint, + operation='yo_digital_auth', + headers=headers, + json_data=taa_payload + ) + + # Log response details + logger.debug(f"yo_digital response status: {response.status_code}") + + # Check for device limit exceeded or other errors + if response.status_code == 400: + try: + error_data = response.json() + logger.error(f"yo_digital 400 error response: {json.dumps(error_data, indent=2)}") + + # Check for deviceLimitExceed (note: might be "Exceed" not "Exceeded") + if error_data.get('deviceLimitExceed') or error_data.get('deviceLimitExceeded'): + logger.error("Device limit exceeded in yo_digital authentication") + return YoDigitalTokens( + access_token="", + access_token_expires_in=0, + refresh_token="", + refresh_token_expires_in=0, + device_limit_exceeded=True + ) + + # Log any other error details + if 'error' in error_data: + logger.error(f"yo_digital error type: {error_data.get('error')}") + if 'error_description' in error_data: + logger.error(f"yo_digital error description: {error_data.get('error_description')}") + if 'message' in error_data: + logger.error(f"yo_digital error message: {error_data.get('message')}") + + except (ValueError, KeyError) as e: + logger.error(f"Could not parse 400 error response: {e}") + logger.error(f"Raw response text: {response.text}") + + response.raise_for_status() + yo_digital_data = response.json() + + # Parse yo_digital response + result = self._parse_yo_digital_response(yo_digital_data) + + logger.info("✓ yo_digital tokens obtained successfully") + return result + + except Exception as e: + logger.error(f"yo_digital token acquisition failed: {e}") + + # Try to extract error details if available + if hasattr(e, 'response') and hasattr(e.response, 'text'): + try: + error_body = e.response.text + logger.error(f"yo_digital error response body: {error_body}") + try: + error_json = e.response.json() + logger.error(f"yo_digital error response JSON: {json.dumps(error_json, indent=2)}") + except: + pass + except: + pass + + return None + + def refresh_yo_digital_tokens(self, refresh_token: str, + yo_digital_endpoint: Optional[str] = None) -> Optional[Dict[str, Any]]: + """ + Refresh yo_digital tokens using refresh_token (STUB) + + Args: + refresh_token: yo_digital refresh token + yo_digital_endpoint: yo_digital endpoint URL + + Returns: + Dictionary with new tokens in yo_digital format, or None on failure + + TODO: Implement actual refresh logic when endpoint/format is confirmed + """ + try: + logger.debug("Refreshing yo_digital tokens (STUB)") + + # STUB: Return None for now + # When implemented, this should: + # 1. POST to yo_digital endpoint with refresh_token + # 2. Parse response in yo_digital format + # 3. Return new tokens + + logger.warning("yo_digital token refresh not yet implemented") + return None + + except Exception as e: + logger.error(f"yo_digital token refresh failed: {e}") + return None + + def _parse_yo_digital_response(self, yo_digital_data: Dict[str, Any]) -> Optional[YoDigitalTokens]: + """ + Parse yo_digital response with proper field names + + yo_digital format: + { + "accessToken": "...", + "accessExpiresIn": 86400, + "refreshToken": "...", + "refreshExpiresIn": 86400, + "deviceLimitExceed": false, + "tvAccountIds": null + } + """ + try: + # Extract tokens using yo_digital field names + access_token = yo_digital_data.get('accessToken') + refresh_token = yo_digital_data.get('refreshToken') + + if not access_token or not refresh_token: + logger.error("Missing access or refresh token in yo_digital response") + return None + + # Extract expiry times + access_expires_in = yo_digital_data.get('accessExpiresIn', 86400) + refresh_expires_in = yo_digital_data.get('refreshExpiresIn', 86400) + + # Check device limit + device_limit_exceeded = yo_digital_data.get('deviceLimitExceed', False) or \ + yo_digital_data.get('deviceLimitExceeded', False) + + result = YoDigitalTokens( + access_token=access_token, + access_token_expires_in=access_expires_in, + refresh_token=refresh_token, + refresh_token_expires_in=refresh_expires_in, + device_limit_exceeded=device_limit_exceeded, + raw_response=yo_digital_data + ) + + logger.debug(f"✓ Parsed yo_digital tokens: " + f"access_expires_in={access_expires_in}s, " + f"refresh_expires_in={refresh_expires_in}s") + + return result + + except Exception as e: + logger.error(f"Failed to parse yo_digital response: {e}") + return None + + # ======================================================================== + # Legacy TAA Authentication (kept for backward compatibility) + # ======================================================================== + def authenticate(self, sam3_token: str, device_id: str, client_model: Optional[str] = None, device_model: Optional[str] = None, taa_endpoint: Optional[str] = None) -> TaaAuthResult: """ - Perform complete TAA authentication + Legacy method - performs TAA authentication and parses JWT - Args: - sam3_token: SAM3 access token for TAA scope - device_id: Device identifier - client_model: Client model from bootstrap - device_model: Device model from bootstrap (if None, uses taa_device_model from platform config) - taa_endpoint: TAA endpoint URL - - Returns: - TaaAuthResult with complete authentication data + DEPRECATED: Use get_yo_digital_tokens() for new code + This method is kept for backward compatibility with existing code """ try: - logger.debug("Starting TAA authentication") + logger.debug("Starting TAA authentication (legacy method)") # Build complete TAA payload taa_payload = self._build_complete_taa_payload( @@ -76,14 +268,6 @@ class TaaClient: endpoint = taa_endpoint or "https://taa.telekom-dienste.de/taa/v1/token" logger.debug(f"TAA request to: {endpoint}") - logger.debug(f"TAA payload keyValue: {taa_payload.get('keyValue', 'MISSING')}") - - # Log complete payload for debugging (mask sensitive data) - payload_debug = taa_payload.copy() - if 'accessToken' in payload_debug: - payload_debug['accessToken'] = payload_debug['accessToken'][:50] + '...' - logger.debug(f"Complete TAA payload: {json.dumps(payload_debug, indent=2)}") - logger.debug(f"TAA headers: {headers}") # Perform TAA request response = self.http_manager.post( @@ -93,38 +277,23 @@ class TaaClient: json_data=taa_payload ) - # Log response details - logger.debug(f"TAA response status: {response.status_code}") - - # Check for device limit exceeded or other errors + # Check for device limit exceeded if response.status_code == 400: try: error_data = response.json() - logger.error(f"TAA 400 error response: {json.dumps(error_data, indent=2)}") - if error_data.get('deviceLimitExceeded'): logger.error("Device limit exceeded in TAA authentication") return TaaAuthResult( access_token="", device_limit_exceeded=True ) - - # Log any other error details - if 'error' in error_data: - logger.error(f"TAA error type: {error_data.get('error')}") - if 'error_description' in error_data: - logger.error(f"TAA error description: {error_data.get('error_description')}") - if 'message' in error_data: - logger.error(f"TAA error message: {error_data.get('message')}") - - except (ValueError, KeyError) as e: - logger.error(f"Could not parse 400 error response: {e}") - logger.error(f"Raw response text: {response.text}") + except: + pass response.raise_for_status() taa_data = response.json() - # Parse TAA response + # Parse TAA response (JWT format) result = self._parse_taa_response(taa_data) logger.info("TAA authentication successful") @@ -132,22 +301,44 @@ class TaaClient: except Exception as e: logger.error(f"TAA authentication failed: {e}") - - # Try to extract error details if available (for requests.HTTPError) - if hasattr(e, 'response') and hasattr(e.response, 'text'): - try: - error_body = e.response.text - logger.error(f"TAA error response body: {error_body}") - try: - error_json = e.response.json() - logger.error(f"TAA error response JSON: {json.dumps(error_json, indent=2)}") - except: - pass - except: - pass - raise Exception(f"TAA authentication failed: {e}") + def _parse_taa_response(self, taa_data: Dict[str, Any]) -> TaaAuthResult: + """ + Parse TAA response and extract all required claims from JWT + (Legacy method for backward compatibility) + """ + # Handle different response key formats + access_token = taa_data.get('access_token', taa_data.get('accessToken')) + refresh_token = taa_data.get('refresh_token', taa_data.get('refreshToken')) + + if not access_token: + raise ValueError("No access token in TAA response") + + # Parse JWT to extract all claims + jwt_claims = self._parse_taa_jwt_complete(access_token) + + # Create result with all extracted data + result = TaaAuthResult( + access_token=access_token, + refresh_token=refresh_token, + dc_cts_persona_token=jwt_claims.get('dc_cts_persona_token'), + persona_id=jwt_claims.get('persona_id'), + account_id=jwt_claims.get('account_id'), + consumer_id=jwt_claims.get('consumer_id'), + tv_account_id=jwt_claims.get('tv_account_id'), + account_token=jwt_claims.get('account_token'), + account_uri=jwt_claims.get('account_uri'), + token_exp=jwt_claims.get('token_exp'), + raw_response=taa_data + ) + + return result + + # ======================================================================== + # Common Methods (used by both yo_digital and legacy TAA) + # ======================================================================== + def _build_complete_taa_payload(self, sam3_token: str, device_id: str, client_model: Optional[str] = None, device_model: Optional[str] = None) -> Dict[str, Any]: @@ -165,23 +356,17 @@ class TaaClient: "natco": "DE", "type": "telekom" } - - CRITICAL: Note the spaces after commas in TokenDeviceParams! """ - # Use provided device model or fallback to TAA-specific device model from platform config - # TAA requires specific device identification format (e.g., "SHIELD Android TV", "API level 30") - resolved_device_model = device_model or self.platform_config.get('taa_device_model') or self.platform_config[ - 'device_name'] + # Use provided device model or fallback to TAA-specific device model + resolved_device_model = device_model or self.platform_config.get('taa_device_model') or \ + self.platform_config['device_name'] - # Get TAA-specific OS format (e.g., "API level 30" instead of "Android 11") + # Get TAA-specific OS format resolved_os = self.platform_config.get('taa_os') or self.platform_config['firmware'] resolved_client_model = client_model or f"ftv-{self.platform}" # Build keyValue string with CORRECT spacing (spaces after commas!) - # Format: TokenDeviceParams(id=..., model=..., os=...) - # ^ ^ - # spaces here! key_value_parts = [ IDM, APPVERSION2, @@ -211,70 +396,17 @@ class TaaClient: "type": "telekom" } - # Add client model if available (not in original C++ but useful) + # Add client model if available if resolved_client_model: payload["client"] = {"model": resolved_client_model} - # Validate payload has all required fields - required_fields = ["keyValue", "accessToken", "accessTokenSource", "appVersion", "channel", "device", "natco", - "type"] - missing_fields = [field for field in required_fields if field not in payload] - if missing_fields: - logger.error(f"TAA payload missing required fields: {missing_fields}") - raise ValueError(f"TAA payload incomplete: missing {missing_fields}") - - logger.debug(f"Built TAA payload with keyValue: {key_value}") - logger.debug(f"Device model: {resolved_device_model}, OS: {resolved_os}") - logger.debug(f"Payload fields: {list(payload.keys())}") - return payload - def _parse_taa_response(self, taa_data: Dict[str, Any]) -> TaaAuthResult: - """ - Parse TAA response and extract all required claims from JWT - """ - # Handle different response key formats - access_token = taa_data.get('access_token', taa_data.get('accessToken')) - refresh_token = taa_data.get('refresh_token', taa_data.get('refreshToken')) - - if not access_token: - raise ValueError("No access token in TAA response") - - # Parse JWT to extract all claims - jwt_claims = self._parse_taa_jwt_complete(access_token) - - # Create result with all extracted data - result = TaaAuthResult( - access_token=access_token, - refresh_token=refresh_token, - dc_cts_persona_token=jwt_claims.get('dc_cts_persona_token'), - persona_id=jwt_claims.get('persona_id'), - account_id=jwt_claims.get('account_id'), - consumer_id=jwt_claims.get('consumer_id'), - tv_account_id=jwt_claims.get('tv_account_id'), - account_token=jwt_claims.get('account_token'), - account_uri=jwt_claims.get('account_uri'), - token_exp=jwt_claims.get('token_exp'), - raw_response=taa_data - ) - - # Log critical fields - if result.dc_cts_persona_token: - logger.debug("✓ dc_cts_persona_token found in TAA JWT") - else: - logger.warning("✗ dc_cts_persona_token NOT found in TAA JWT") - - if result.account_uri: - logger.debug(f"✓ account_uri found: {result.account_uri}") - else: - logger.warning("✗ account_uri NOT found in TAA JWT") - - return result - @staticmethod def _parse_taa_jwt_complete(jwt_token: str) -> Dict[str, Any]: """ Complete JWT parsing extracting ALL required fields from TAA token + (Used by legacy authenticate method) """ try: parts = jwt_token.split('.') @@ -291,29 +423,22 @@ class TaaClient: payload_json = base64.b64decode(payload_b64).decode('utf-8') claims = json.loads(payload_json) - logger.debug(f"TAA JWT claims: {list(claims.keys())}") - result = {} - # Enhanced claim mappings - ALL fields from C++ implementation + # Enhanced claim mappings claim_mappings = { - # Core persona token (most important!) 'dc_cts_persona_token': [ - 'dc_cts_personaToken', # Capital T! This is what the API returns + 'dc_cts_personaToken', 'dc_cts_persona_token', 'personaToken', 'urn:telekom:ott:dc_cts_persona_token' ], - - # Account URI (needed for composition!) 'account_uri': [ 'dc_cts_account_uri', 'accountUri', 'urn:telekom:ott:dc_cts_account_uri', 'mpxAccountUri' ], - - # IDs 'persona_id': [ 'dc_cts_personaId', 'personaId', @@ -334,8 +459,6 @@ class TaaClient: 'tvAccountId', 'urn:telekom:ott:dc_tvAccountId' ], - - # Account token 'account_token': [ 'dc_cts_account_token', 'accountToken', @@ -348,59 +471,39 @@ class TaaClient: for source_key in source_keys: if source_key in claims: result[target_key] = claims[source_key] - logger.debug(f"Extracted TAA claim {target_key} from {source_key}") break # Extract token expiration if 'exp' in claims: result['token_exp'] = claims['exp'] - logger.debug(f"TAA token expires at: {claims['exp']}") - - # Extract issuance time - if 'iat' in claims: - result['token_iat'] = claims['iat'] - - # CRITICAL CHECK: Verify we have the essential fields - essential_fields = ['dc_cts_persona_token', 'account_uri'] - missing_essential = [field for field in essential_fields if field not in result] - - if missing_essential: - logger.error(f"CRITICAL: Missing essential TAA claims: {missing_essential}") - logger.debug(f"Available TAA claims: {list(claims.keys())}") - else: - logger.info("✓ All essential TAA claims found") return result except Exception as e: - logger.error(f"Failed to parse TAA JWT completely: {e}") + logger.error(f"Failed to parse TAA JWT: {e}") return {} def _get_taa_headers(self) -> Dict[str, str]: """ - Get headers for TAA requests with required requestId - - Note: The accessToken is sent in the request body, not as Authorization header. - The TAA endpoint uses the token from the payload, not from headers. + Get headers for TAA/yo_digital requests """ import uuid - # Use TAA-specific user agent if available, otherwise fallback to platform user agent user_agent = self.platform_config.get('user_agent') headers = { - 'requestId': str(uuid.uuid4()), # Required by TAA endpoint! + 'requestId': str(uuid.uuid4()), 'User-Agent': user_agent, 'Accept': 'application/json', 'Content-Type': 'application/json; charset=UTF-8' } - # NOTE: We do NOT add Authorization header here - # The accessToken is sent in the request body payload - # Authorization header is not needed for TAA endpoint - return headers + # ======================================================================== + # Validation and Debugging + # ======================================================================== + def validate_taa_token(self, taa_token: str) -> bool: """ Validate TAA token expiration and basic structure @@ -409,7 +512,6 @@ class TaaClient: if not taa_token: return False - # Check if token is expired claims = self._parse_taa_jwt_complete(taa_token) token_exp = claims.get('token_exp') @@ -417,7 +519,6 @@ class TaaClient: logger.debug("TAA token is expired") return False - # Check for essential claims if claims.get('dc_cts_persona_token') and claims.get('account_uri'): return True diff --git a/lib/streaming_providers/providers/magenta2/token_flow_manager.py b/lib/streaming_providers/providers/magenta2/token_flow_manager.py new file mode 100644 index 0000000..f567507 --- /dev/null +++ b/lib/streaming_providers/providers/magenta2/token_flow_manager.py @@ -0,0 +1,725 @@ +# streaming_providers/providers/magenta2/token_flow_manager.py +""" +Hierarchical Token Flow Manager for Magenta2 +Manages the complete token acquisition hierarchy: +yo_digital → taa → tvhubs → line_auth/remote_login +""" + +import time +from typing import Optional, Dict, Any +from dataclasses import dataclass + +from ...base.utils.logger import logger +from ...base.auth.session_manager import SessionManager +from .sam3_client import Sam3Client +from .taa_client import TaaClient + + +@dataclass +class TokenFlowResult: + """Result of token flow operation""" + success: bool + access_token: Optional[str] = None + refresh_token: Optional[str] = None + access_token_expires_in: Optional[int] = None + refresh_token_expires_in: Optional[int] = None + error: Optional[str] = None + flow_path: Optional[str] = None # For debugging which path was taken + + +class TokenFlowManager: + """ + Manages hierarchical token acquisition for Magenta2 + + Token Hierarchy: + 1. yo_digital tokens (goal) - access + refresh, separate expiry + 2. taa access_token (from SAM3 via refresh_token exchange) + 3. tvhubs access_token + shared refresh_token (from line_auth/remote_login) + + Flow Priority: + 1. Check yo_digital access_token (valid?) → use it ✓ + 2. Check yo_digital refresh_token (valid?) → refresh yo_digital + 3. Check taa access_token (valid?) → get yo_digital + 4. Check shared refresh_token (exists?) → exchange for taa → get yo_digital + 5. Try line_auth → get tvhubs + refresh_token → chain to yo_digital + 6. Try remote_login → get tvhubs + refresh_token → chain to yo_digital + 7. All failed → ERROR + """ + + def __init__(self, + session_manager: SessionManager, + sam3_client: 'Sam3Client', + taa_client: 'TaaClient', + provider_name: str, + country: Optional[str] = None): + """ + Initialize TokenFlowManager + + Args: + session_manager: SessionManager for token storage + sam3_client: SAM3 client for tvhubs/taa operations + taa_client: TAA client for yo_digital operations + provider_name: Provider name (e.g., 'magenta2') + country: Optional country code + """ + self.session_manager = session_manager + self.sam3_client = sam3_client + self.taa_client = taa_client + self.provider_name = provider_name + self.country = country + + logger.debug(f"TokenFlowManager initialized for {provider_name}" + + (f" ({country})" if country else "")) + + def get_yo_digital_token(self, force_refresh: bool = False) -> TokenFlowResult: + """ + Get yo_digital access token following the complete hierarchy + + Args: + force_refresh: Skip cached tokens and force refresh + + Returns: + TokenFlowResult with yo_digital access token or error + """ + country_str = f" ({self.country})" if self.country else "" + logger.info(f"Getting yo_digital token for {self.provider_name}{country_str}") + + if not force_refresh: + # Step 1: Check existing yo_digital access_token + result = self._check_yo_digital_access_token() + if result.success: + logger.info("✓ Using valid yo_digital access_token") + return result + + # Step 2: Check yo_digital refresh_token + result = self._refresh_yo_digital_if_possible() + if result.success: + logger.info("✓ Refreshed yo_digital tokens") + return result + + # Step 3: Check taa access_token + result = self._get_yo_digital_from_taa() + if result.success: + logger.info("✓ Got yo_digital tokens from taa") + return result + + # Step 4: Check shared refresh_token + result = self._get_yo_digital_via_taa_exchange() + if result.success: + logger.info("✓ Got yo_digital tokens via taa exchange") + return result + + # Step 5: Try line_auth + result = self._get_yo_digital_via_line_auth() + if result.success: + logger.info("✓ Got yo_digital tokens via line_auth") + return result + + # Step 6: Try remote_login + result = self._get_yo_digital_via_remote_login() + if result.success: + logger.info("✓ Got yo_digital tokens via remote_login") + return result + + # All failed + logger.error("✗ All token acquisition methods failed") + return TokenFlowResult( + success=False, + error="All token acquisition methods failed", + flow_path="all_failed" + ) + + # ======================================================================== + # Step 1: Check existing yo_digital access_token + # ======================================================================== + + def _check_yo_digital_access_token(self) -> TokenFlowResult: + """Check if we have a valid yo_digital access_token""" + try: + token_data = self.session_manager.load_scoped_token( + self.provider_name, + 'yo_digital', + self.country + ) + + if not token_data or 'access_token' not in token_data: + return TokenFlowResult( + success=False, + error="No yo_digital token found", + flow_path="check_yo_digital_access" + ) + + # Check if access_token is still valid + if self._is_yo_digital_access_token_valid(token_data): + return TokenFlowResult( + success=True, + access_token=token_data['access_token'], + refresh_token=token_data.get('refresh_token'), + flow_path="yo_digital_access_valid" + ) + + return TokenFlowResult( + success=False, + error="yo_digital access_token expired", + flow_path="check_yo_digital_access" + ) + + except Exception as e: + logger.debug(f"Error checking yo_digital access_token: {e}") + return TokenFlowResult( + success=False, + error=str(e), + flow_path="check_yo_digital_access" + ) + + # ======================================================================== + # Step 2: Refresh yo_digital tokens + # ======================================================================== + + def _refresh_yo_digital_if_possible(self) -> TokenFlowResult: + """Try to refresh yo_digital tokens if refresh_token is valid""" + try: + token_data = self.session_manager.load_scoped_token( + self.provider_name, + 'yo_digital', + self.country + ) + + if not token_data or 'refresh_token' not in token_data: + return TokenFlowResult( + success=False, + error="No yo_digital refresh_token found", + flow_path="refresh_yo_digital" + ) + + # Check if refresh_token is still valid + if not self._is_yo_digital_refresh_token_valid(token_data): + return TokenFlowResult( + success=False, + error="yo_digital refresh_token expired", + flow_path="refresh_yo_digital" + ) + + # Refresh via TaaClient (stub for now) + logger.debug("Attempting to refresh yo_digital tokens") + new_tokens_dict = self.taa_client.refresh_yo_digital_tokens( + token_data['refresh_token'] + ) + + if not new_tokens_dict: + return TokenFlowResult( + success=False, + error="yo_digital refresh failed", + flow_path="refresh_yo_digital" + ) + + # Save new tokens (already in dict format from stub) + self._save_yo_digital_tokens(new_tokens_dict) + + return TokenFlowResult( + success=True, + access_token=new_tokens_dict.get('accessToken') or new_tokens_dict.get('access_token'), + refresh_token=new_tokens_dict.get('refreshToken') or new_tokens_dict.get('refresh_token'), + flow_path="yo_digital_refreshed" + ) + + except Exception as e: + logger.debug(f"Error refreshing yo_digital tokens: {e}") + return TokenFlowResult( + success=False, + error=str(e), + flow_path="refresh_yo_digital" + ) + + # ======================================================================== + # Step 3: Get yo_digital from taa access_token + # ======================================================================== + + def _get_yo_digital_from_taa(self) -> TokenFlowResult: + """Get yo_digital tokens using existing taa access_token""" + try: + # Check if we have valid taa access_token + taa_token_data = self.session_manager.load_scoped_token( + self.provider_name, + 'taa', + self.country + ) + + if not taa_token_data or 'access_token' not in taa_token_data: + return TokenFlowResult( + success=False, + error="No taa access_token found", + flow_path="yo_digital_from_taa" + ) + + # Check if taa token is still valid + if self._is_token_expired(taa_token_data): + return TokenFlowResult( + success=False, + error="taa access_token expired", + flow_path="yo_digital_from_taa" + ) + + # Get yo_digital tokens from TAA endpoint + logger.debug("Getting yo_digital tokens from taa access_token") + yo_digital_result = self.taa_client.get_yo_digital_tokens( + taa_access_token=taa_token_data['access_token'], + device_id=self.session_manager.get_device_id(self.provider_name, self.country) + ) + + if not yo_digital_result: + return TokenFlowResult( + success=False, + error="Failed to get yo_digital tokens from taa", + flow_path="yo_digital_from_taa" + ) + + # Convert YoDigitalTokens to dict for saving + yo_digital_dict = { + 'accessToken': yo_digital_result.access_token, + 'accessExpiresIn': yo_digital_result.access_token_expires_in, + 'refreshToken': yo_digital_result.refresh_token, + 'refreshExpiresIn': yo_digital_result.refresh_token_expires_in + } + + # Save yo_digital tokens + self._save_yo_digital_tokens(yo_digital_dict) + + return TokenFlowResult( + success=True, + access_token=yo_digital_result.access_token, + refresh_token=yo_digital_result.refresh_token, + flow_path="yo_digital_from_taa" + ) + + except Exception as e: + logger.debug(f"Error getting yo_digital from taa: {e}") + return TokenFlowResult( + success=False, + error=str(e), + flow_path="yo_digital_from_taa" + ) + + # ======================================================================== + # Step 4: Exchange shared refresh_token for taa, then yo_digital + # ======================================================================== + + def _get_yo_digital_via_taa_exchange(self) -> TokenFlowResult: + """Exchange shared refresh_token for taa, then get yo_digital""" + try: + # Check if we have shared refresh_token at provider level + session_data = self.session_manager.load_session( + self.provider_name, + self.country + ) + + if not session_data or 'refresh_token' not in session_data: + return TokenFlowResult( + success=False, + error="No shared refresh_token found", + flow_path="yo_digital_via_taa_exchange" + ) + + # Exchange refresh_token for taa access_token + logger.debug("Exchanging refresh_token for taa access_token") + taa_token = self.sam3_client.get_token( + grant_type='refresh_token', + scope='taa', + credential1=session_data['refresh_token'] + ) + + if not taa_token: + return TokenFlowResult( + success=False, + error="Failed to exchange refresh_token for taa", + flow_path="yo_digital_via_taa_exchange" + ) + + # Save taa token + self._save_taa_token(taa_token) + + # Now get yo_digital tokens using taa + logger.debug("Getting yo_digital tokens from exchanged taa token") + yo_digital_result = self.taa_client.get_yo_digital_tokens( + taa_access_token=taa_token, + device_id=self.session_manager.get_device_id(self.provider_name, self.country) + ) + + if not yo_digital_result: + return TokenFlowResult( + success=False, + error="Failed to get yo_digital from exchanged taa", + flow_path="yo_digital_via_taa_exchange" + ) + + # Convert YoDigitalTokens to dict for saving + yo_digital_dict = { + 'accessToken': yo_digital_result.access_token, + 'accessExpiresIn': yo_digital_result.access_token_expires_in, + 'refreshToken': yo_digital_result.refresh_token, + 'refreshExpiresIn': yo_digital_result.refresh_token_expires_in + } + + # Save yo_digital tokens + self._save_yo_digital_tokens(yo_digital_dict) + + return TokenFlowResult( + success=True, + access_token=yo_digital_result.access_token, + refresh_token=yo_digital_result.refresh_token, + flow_path="yo_digital_via_taa_exchange" + ) + + except Exception as e: + logger.debug(f"Error in taa exchange flow: {e}") + return TokenFlowResult( + success=False, + error=str(e), + flow_path="yo_digital_via_taa_exchange" + ) + + # ======================================================================== + # Step 5: Line auth → tvhubs + refresh → taa → yo_digital + # ======================================================================== + + def _get_yo_digital_via_line_auth(self) -> TokenFlowResult: + """Try line_auth to get tvhubs + refresh_token, then chain to yo_digital""" + try: + # Check if line_auth is available + if not hasattr(self.sam3_client, 'line_auth'): + return TokenFlowResult( + success=False, + error="line_auth not available", + flow_path="yo_digital_via_line_auth" + ) + + # Check if we have device_token (required for line_auth) + # This would be set up during provider initialization + if not hasattr(self.sam3_client, 'line_auth_endpoint'): + return TokenFlowResult( + success=False, + error="line_auth endpoint not configured", + flow_path="yo_digital_via_line_auth" + ) + + logger.info("Attempting line_auth flow") + + # Line auth is handled by authenticator's _perform_line_auth_flow + # We can't call it directly from here without circular dependency + # So we return a special result that tells the caller to try line_auth + + return TokenFlowResult( + success=False, + error="line_auth needs to be triggered by authenticator", + flow_path="yo_digital_via_line_auth" + ) + + except Exception as e: + logger.debug(f"Error in line_auth flow: {e}") + return TokenFlowResult( + success=False, + error=str(e), + flow_path="yo_digital_via_line_auth" + ) + + # ======================================================================== + # Step 6: Remote login → tvhubs + refresh → taa → yo_digital + # ======================================================================== + + def _get_yo_digital_via_remote_login(self) -> TokenFlowResult: + """Try remote_login to get tvhubs + refresh_token, then chain to yo_digital""" + try: + # Check if remote_login is available + if not hasattr(self.sam3_client, 'can_use_remote_login'): + return TokenFlowResult( + success=False, + error="remote_login not available", + flow_path="yo_digital_via_remote_login" + ) + + if not self.sam3_client.can_use_remote_login(): + return TokenFlowResult( + success=False, + error="remote_login not configured", + flow_path="yo_digital_via_remote_login" + ) + + logger.info("Attempting remote_login flow") + + # Perform remote login + remote_token_data = self.sam3_client.remote_login( + scope="tvhubs offline_access" + ) + + if not remote_token_data: + return TokenFlowResult( + success=False, + error="remote_login failed or timed out", + flow_path="yo_digital_via_remote_login" + ) + + # Save tvhubs token + self._save_tvhubs_token(remote_token_data) + + # Save refresh token at provider level + if 'refresh_token' in remote_token_data: + self._save_refresh_token(remote_token_data['refresh_token']) + + # Now exchange for taa + logger.debug("Exchanging remote_login refresh_token for taa") + taa_token = self.sam3_client.get_token( + grant_type='refresh_token', + scope='taa', + credential1=remote_token_data['refresh_token'] + ) + + if not taa_token: + return TokenFlowResult( + success=False, + error="Failed to exchange remote_login refresh for taa", + flow_path="yo_digital_via_remote_login" + ) + + # Save taa token + self._save_taa_token(taa_token) + + # Finally get yo_digital + logger.debug("Getting yo_digital tokens from remote_login taa token") + yo_digital_result = self.taa_client.get_yo_digital_tokens( + taa_access_token=taa_token, + device_id=self.session_manager.get_device_id(self.provider_name, self.country) + ) + + if not yo_digital_result: + return TokenFlowResult( + success=False, + error="Failed to get yo_digital from remote_login taa", + flow_path="yo_digital_via_remote_login" + ) + + # Convert YoDigitalTokens to dict for saving + yo_digital_dict = { + 'accessToken': yo_digital_result.access_token, + 'accessExpiresIn': yo_digital_result.access_token_expires_in, + 'refreshToken': yo_digital_result.refresh_token, + 'refreshExpiresIn': yo_digital_result.refresh_token_expires_in + } + + # Save yo_digital tokens + self._save_yo_digital_tokens(yo_digital_dict) + + return TokenFlowResult( + success=True, + access_token=yo_digital_result.access_token, + refresh_token=yo_digital_result.refresh_token, + flow_path="yo_digital_via_remote_login" + ) + + except Exception as e: + logger.debug(f"Error in remote_login flow: {e}") + return TokenFlowResult( + success=False, + error=str(e), + flow_path="yo_digital_via_remote_login" + ) + + # ======================================================================== + # Helper Methods - Token Validation + # ======================================================================== + + @staticmethod + def _is_yo_digital_access_token_valid(token_data: Dict[str, Any]) -> bool: + """Check if yo_digital access_token is still valid""" + if 'access_token_expires_in' not in token_data or 'access_token_issued_at' not in token_data: + return False + + expires_at = token_data['access_token_issued_at'] + token_data['access_token_expires_in'] + # Use 5 minute buffer + return time.time() < (expires_at - 300) + + @staticmethod + def _is_yo_digital_refresh_token_valid(token_data: Dict[str, Any]) -> bool: + """Check if yo_digital refresh_token is still valid""" + if 'refresh_token_expires_in' not in token_data or 'refresh_token_issued_at' not in token_data: + return False + + expires_at = token_data['refresh_token_issued_at'] + token_data['refresh_token_expires_in'] + # Use 5 minute buffer + return time.time() < (expires_at - 300) + + @staticmethod + def _is_token_expired(token_data: Dict[str, Any]) -> bool: + """Check if standard token (tvhubs/taa) is expired""" + if 'expires_in' not in token_data or 'issued_at' not in token_data: + return True + + expires_at = token_data['issued_at'] + token_data['expires_in'] + # Use 5 minute buffer + return time.time() >= (expires_at - 300) + + # ======================================================================== + # Helper Methods - Token Storage + # ======================================================================== + + def _save_yo_digital_tokens(self, tokens: Dict[str, Any]) -> None: + """Save yo_digital tokens with proper format""" + current_time = time.time() + + token_data = { + 'access_token': tokens.get('accessToken') or tokens.get('access_token'), + 'access_token_expires_in': tokens.get('accessExpiresIn') or tokens.get('access_token_expires_in', 86400), + 'access_token_issued_at': current_time, + 'refresh_token': tokens.get('refreshToken') or tokens.get('refresh_token'), + 'refresh_token_expires_in': tokens.get('refreshExpiresIn') or tokens.get('refresh_token_expires_in', 86400), + 'refresh_token_issued_at': current_time, + 'token_type': 'Bearer' + } + + success = self.session_manager.save_scoped_token( + self.provider_name, + 'yo_digital', + token_data, + self.country + ) + + if success: + logger.info("✓ yo_digital tokens saved") + else: + logger.error("✗ Failed to save yo_digital tokens") + + def _save_taa_token(self, taa_token: str) -> None: + """Save taa access_token""" + token_data = { + 'access_token': taa_token, + 'token_type': 'Bearer', + 'expires_in': 3600, # Default 1 hour + 'issued_at': time.time() + } + + self.session_manager.save_scoped_token( + self.provider_name, + 'taa', + token_data, + self.country + ) + logger.debug("taa token saved") + + def _save_tvhubs_token(self, token_data: Dict[str, Any]) -> None: + """Save tvhubs access_token""" + tvhubs_data = { + 'access_token': token_data.get('access_token'), + 'token_type': token_data.get('token_type', 'Bearer'), + 'expires_in': token_data.get('expires_in', 7200), + 'issued_at': time.time() + } + + self.session_manager.save_scoped_token( + self.provider_name, + 'tvhubs', + tvhubs_data, + self.country + ) + logger.debug("tvhubs token saved") + + def _save_refresh_token(self, refresh_token: str) -> None: + """Save shared refresh_token at provider level""" + session_data = self.session_manager.load_session( + self.provider_name, + self.country + ) or {} + + session_data['refresh_token'] = refresh_token + session_data['device_id'] = self.session_manager.get_device_id( + self.provider_name, + self.country + ) + + self.session_manager.save_session( + self.provider_name, + session_data, + self.country + ) + logger.debug("Shared refresh_token saved") + + # ======================================================================== + # Public API - Debugging + # ======================================================================== + + def get_token_status(self) -> Dict[str, Any]: + """Get comprehensive status of all tokens""" + status = { + 'yo_digital': self._get_yo_digital_status(), + 'taa': self._get_taa_status(), + 'tvhubs': self._get_tvhubs_status(), + 'refresh_token': self._get_refresh_token_status() + } + + return status + + def _get_yo_digital_status(self) -> Dict[str, Any]: + """Get yo_digital token status""" + token_data = self.session_manager.load_scoped_token( + self.provider_name, + 'yo_digital', + self.country + ) + + if not token_data: + return {'exists': False} + + return { + 'exists': True, + 'has_access_token': 'access_token' in token_data, + 'access_token_valid': self._is_yo_digital_access_token_valid(token_data), + 'has_refresh_token': 'refresh_token' in token_data, + 'refresh_token_valid': self._is_yo_digital_refresh_token_valid(token_data) + } + + def _get_taa_status(self) -> Dict[str, Any]: + """Get taa token status""" + token_data = self.session_manager.load_scoped_token( + self.provider_name, + 'taa', + self.country + ) + + if not token_data: + return {'exists': False} + + return { + 'exists': True, + 'has_access_token': 'access_token' in token_data, + 'is_valid': not self._is_token_expired(token_data) + } + + def _get_tvhubs_status(self) -> Dict[str, Any]: + """Get tvhubs token status""" + token_data = self.session_manager.load_scoped_token( + self.provider_name, + 'tvhubs', + self.country + ) + + if not token_data: + return {'exists': False} + + return { + 'exists': True, + 'has_access_token': 'access_token' in token_data, + 'is_valid': not self._is_token_expired(token_data) + } + + def _get_refresh_token_status(self) -> Dict[str, Any]: + """Get shared refresh_token status""" + session_data = self.session_manager.load_session( + self.provider_name, + self.country + ) + + if not session_data or 'refresh_token' not in session_data: + return {'exists': False} + + return { + 'exists': True, + 'has_refresh_token': True + } \ No newline at end of file