nirvana-7777/script.service.ultimate
1
0
Fork 0
mirror of https://github.com/nirvana-7777/script.service.ultimate.git synced 2026-09-24 18:12:32 +02:00
Code Issues Packages Projects Releases Wiki Activity

Fix taa_client

Browse Source
This commit is contained in:
Nirvana
2025-11-16 21:22:25 +01:00
parent 727bcff258
commit 93022fb0f3
4 changed files with 1053 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/providers/magenta2/auth.py
+39
View File
@@ -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
lib/streaming_providers/providers/magenta2/provider.py
+29 -4
View File
@@ -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()
lib/streaming_providers/providers/magenta2/taa_client.py
+260 -159
View File
@@ -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
lib/streaming_providers/providers/magenta2/token_flow_manager.py
+725
View File
@@ -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
}