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

Add Magenta 2.0 (DE)

Browse Source
This commit is contained in:
Nirvana
2025-11-13 16:36:39 +01:00
parent faef6ae781
commit e9972d650f
21 changed files with 6844 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1
View File
@@ -27,6 +27,7 @@ Currently supported:
- 🇦🇹 **Joyn (AT)**
- 🇨🇭 **Joyn (CH)**
- 🇩🇪 **RTL+**
- 🇩🇪 **Magenta TV 2.0**
- 🇦🇹 **Magenta TV (AT)**
- 🇭🇷 **Max TV (HR)**
- 🇵🇱 **Magenta TV (PL)**
lib/streaming_providers/base/auth/credentials.py
+1 -1
View File
@@ -63,7 +63,7 @@ class ClientCredentials(BaseCredentials):
Client credentials (client_id/client_secret) based authentication
"""
client_id: str
client_secret: str
client_secret: Optional[str] = ""
grant_type: str = 'client_credentials'
def validate(self) -> bool:
lib/streaming_providers/base/auth/session_manager.py
+170 -115
View File
@@ -13,7 +13,7 @@ class SessionManager:
"""
Manages persistent session data including tokens and device IDs
Compatible with both Kodi and standalone environments via VFS abstraction
Now supports country-specific sessions
Now supports country-specific sessions and scope-based token storage
"""
def __init__(self, config_dir: Optional[str] = None):
@@ -25,12 +25,9 @@ class SessionManager:
"""
# Initialize VFS with optional config directory override
if config_dir:
# For custom config directories, we'll use a VFS instance that treats
# the config_dir as the base path directly
self.vfs = VFS()
self.vfs._base_path = config_dir
else:
# Use default VFS (handles Kodi vs standard filesystem automatically)
self.vfs = VFS()
# Session file is always in the root of the VFS base path
@@ -92,12 +89,7 @@ class SessionManager:
if session_data:
# Log what we found (without sensitive data)
safe_data = {}
for key, value in session_data.items():
if key in ['access_token', 'refresh_token', 'client_secret', 'password']:
safe_data[key] = f"<{key}_present>" if value else f"<{key}_missing>"
else:
safe_data[key] = value
safe_data = self._get_safe_representation(session_data)
logger.info(f"Loaded session data for {provider}{country_str}: {safe_data}")
else:
logger.info(f"No session data found for {provider}{country_str} in session file")
@@ -132,11 +124,9 @@ class SessionManager:
# Extract and preserve token classification data if it's a token object
if hasattr(session_data, 'auth_level'):
# If session_data is a token-like object, convert to dict first
if hasattr(session_data, 'to_dict'):
session_data = session_data.to_dict()
else:
# Extract classification fields from object
token_dict = {}
for key in ['access_token', 'refresh_token', 'token_type', 'expires_in', 'issued_at',
'refresh_expires_in']:
@@ -145,12 +135,7 @@ class SessionManager:
session_data = token_dict
# Log what we're about to save (without sensitive data)
safe_data = {}
for key, value in session_data.items():
if key in ['access_token', 'refresh_token', 'client_secret', 'password']:
safe_data[key] = f"<{key}_present>" if value else f"<{key}_missing>"
else:
safe_data[key] = value
safe_data = self._get_safe_representation(session_data)
logger.info(f"Saving session data for {provider}{country_str}: {safe_data}")
# Navigate and create nested structure if needed
@@ -178,7 +163,6 @@ class SessionManager:
# Verify by reading it back
verify_data = self.vfs.read_json(self.session_file)
if verify_data:
# Navigate to verify
verify_current = verify_data
found = True
for key in keys_path:
@@ -207,10 +191,115 @@ class SessionManager:
logger.error(f"Full traceback: {traceback.format_exc()}")
return False
def save_scoped_token(self, provider: str, scope: str, token_data: Dict[str, Any],
country: Optional[str] = None) -> bool:
"""
Save authentication token for a specific scope
Args:
provider: Provider name
scope: Token scope (e.g., 'line_auth', 'taa', 'yo_digital')
token_data: Token data to save
country: Optional country code
Returns:
True if successful, False otherwise
"""
country_str = f" (country: {country})" if country else ""
try:
logger.debug(f"Saving scoped token for {provider}{country_str}, scope: {scope}")
# Load existing session data
session_data = self.load_session(provider, country) or {}
# Update token data for this scope
session_data[scope] = token_data
# Log what we're saving
safe_token_data = self._get_safe_representation(token_data)
logger.info(f"Scoped token data for {provider}{country_str}/{scope}: {safe_token_data}")
success = self.save_session(provider, session_data, country)
if success:
logger.info(f"Successfully saved scoped token for {provider}{country_str}/{scope}")
else:
logger.error(f"Failed to save scoped token for {provider}{country_str}/{scope}")
return success
except Exception as e:
logger.error(f"Error saving scoped token for {provider}{country_str}/{scope}: {e}")
return False
def load_scoped_token(self, provider: str, scope: str,
country: Optional[str] = None) -> Optional[Dict[str, Any]]:
"""
Load token data for a specific scope
Args:
provider: Provider name
scope: Token scope (e.g., 'line_auth', 'taa', 'yo_digital')
country: Optional country code
Returns:
Token data dictionary or None
"""
country_str = f" (country: {country})" if country else ""
logger.debug(f"Loading scoped token for {provider}{country_str}, scope: {scope}")
session_data = self.load_session(provider, country)
if not session_data:
logger.info(f"No session data available for {provider}{country_str}")
return None
# Check if scope exists
if scope not in session_data:
logger.info(f"No token found for scope '{scope}' in {provider}{country_str}")
logger.debug(f"Available scopes: {list(session_data.keys())}")
return None
token_data = session_data[scope]
# Validate it's actually token data
if not isinstance(token_data, dict) or 'access_token' not in token_data:
logger.warning(f"Scope '{scope}' exists but doesn't contain valid token data")
return None
# Check if token is expired
if self._is_token_expired(token_data):
logger.info(f"Token for scope '{scope}' is expired")
return token_data # Return anyway so refresh can be attempted
logger.info(f"Loaded valid token for {provider}{country_str}/{scope}")
return token_data
@staticmethod
def _is_token_expired(token_data: Dict[str, Any], buffer_seconds: int = 300) -> bool:
"""
Check if token is expired with buffer
Args:
token_data: Token data dictionary
buffer_seconds: Seconds buffer before expiry (default 5 minutes)
Returns:
True if expired, False otherwise
"""
if 'expires_in' not in token_data or 'issued_at' not in token_data:
return False # Can't determine, assume valid
expires_in = token_data.get('expires_in', 0)
issued_at = token_data.get('issued_at', 0)
current_time = time.time()
expires_at = issued_at + expires_in
return current_time >= (expires_at - buffer_seconds)
def save_token(self, provider: str, token: BaseAuthToken,
country: Optional[str] = None) -> bool:
"""
Save authentication token for a provider
Save authentication token for a provider (legacy compatibility)
Args:
provider: Provider name
@@ -227,24 +316,15 @@ class SessionManager:
# Load existing session data
session_data = self.load_session(provider, country) or {}
logger.debug(f"Current session data keys before token save: {list(session_data.keys())}")
# Update token data - include classification fields
# Update token data
token_data = token.to_dict()
# Log token info (without sensitive data)
safe_token_data = {}
for key, value in token_data.items():
if key in ['access_token', 'refresh_token']:
safe_token_data[key] = f"<{key}_present>" if value else f"<{key}_missing>"
elif key in ['auth_level', 'credential_type']:
safe_token_data[key] = value # Include classification info
else:
safe_token_data[key] = value
safe_token_data = self._get_safe_representation(token_data)
logger.info(f"Token data to save for {provider}{country_str}: {safe_token_data}")
session_data.update(token_data)
logger.debug(f"Updated session data keys after token merge: {list(session_data.keys())}")
success = self.save_session(provider, session_data, country)
if success:
@@ -258,7 +338,7 @@ class SessionManager:
return False
def load_token_data(self, provider: str, country: Optional[str] = None) -> Optional[Dict[str, Any]]:
"""Load token data for a provider"""
"""Load token data for a provider (legacy compatibility)"""
country_str = f" (country: {country})" if country else ""
logger.debug(f"Loading token data for {provider}{country_str}")
@@ -274,55 +354,27 @@ class SessionManager:
logger.debug(f"Available session keys: {list(session_data.keys())}")
return None
# Check if token is expired (with 5 minute buffer)
expires_in = session_data.get('expires_in', 0)
issued_at = session_data.get('issued_at', 0)
current_time = time.time()
expires_at = issued_at + expires_in
time_until_expiry = expires_at - current_time
logger.debug(f"Token expiry check for {provider}{country_str}: issued_at={issued_at}, "
f"expires_in={expires_in}, current_time={current_time}, "
f"time_until_expiry={time_until_expiry:.0f}")
is_expired = current_time >= (expires_at - 300) # 5 minute buffer
has_refresh_token = bool(session_data.get('refresh_token'))
logger.debug(f"Token status - is_expired: {is_expired}, has_refresh_token: {has_refresh_token}")
if is_expired:
# Check if token is expired
if self._is_token_expired(session_data):
has_refresh_token = bool(session_data.get('refresh_token'))
if has_refresh_token:
# Token is expired BUT we have a refresh token - return the data so refresh can be attempted
logger.info(
f"Token expired for {provider}{country_str} but refresh token available - returning data for refresh")
logger.info(f"Token expired for {provider}{country_str} but refresh token available")
return session_data
else:
logger.info(f"Token expired for {provider}{country_str} and no refresh token available")
return None
logger.info(f"Loaded valid token data for {provider}{country_str} "
f"(expires in {time_until_expiry:.0f}s, "
f"auth_level={session_data.get('auth_level')})")
logger.info(f"Loaded valid token data for {provider}{country_str}")
return session_data
def get_device_id(self, provider: str, country: Optional[str] = None) -> str:
"""
Get or generate device ID for a provider
Args:
provider: Provider name
country: Optional country code
Returns:
Device ID (UUID string)
"""
"""Get or generate device ID for a provider"""
country_str = f" (country: {country})" if country else ""
session_data = self.load_session(provider, country) or {}
device_id = session_data.get('device_id')
if not device_id:
# Generate new device ID
device_id = str(uuid.uuid4())
session_data['device_id'] = device_id
self.save_session(provider, session_data, country)
@@ -333,16 +385,7 @@ class SessionManager:
return device_id
def clear_session(self, provider: str, country: Optional[str] = None) -> bool:
"""
Clear session data for a provider and optional country
Args:
provider: Provider name
country: Optional country code (if None, clears entire provider)
Returns:
True if successful, False otherwise
"""
"""Clear session data for a provider and optional country"""
country_str = f" (country: {country})" if country else ""
try:
@@ -352,12 +395,10 @@ class SessionManager:
return True
if country:
# Clear specific country data
if provider in data and isinstance(data[provider], dict) and country in data[provider]:
del data[provider][country]
logger.info(f"Cleared session data for {provider}{country_str}")
# If provider dict is now empty, remove it entirely
if not data[provider]:
del data[provider]
logger.debug(f"Provider {provider} had no more countries, removed entirely")
@@ -366,7 +407,6 @@ class SessionManager:
else:
logger.debug(f"No session data found to clear for {provider}{country_str}")
else:
# Clear entire provider data (all countries)
if provider in data:
del data[provider]
logger.info(f"Cleared all session data for {provider}")
@@ -380,12 +420,14 @@ class SessionManager:
logger.error(f"Error clearing session for {provider}{country_str}: {e}")
return False
def clear_token(self, provider: str, country: Optional[str] = None) -> bool:
def clear_scoped_token(self, provider: str, scope: str,
country: Optional[str] = None) -> bool:
"""
Clear only token data but keep other session data (like device_id)
Clear token for a specific scope
Args:
provider: Provider name
scope: Token scope to clear
country: Optional country code
Returns:
@@ -393,6 +435,28 @@ class SessionManager:
"""
country_str = f" (country: {country})" if country else ""
try:
session_data = self.load_session(provider, country)
if not session_data:
logger.debug(f"No session data found for {provider}{country_str}")
return True
if scope in session_data:
del session_data[scope]
logger.info(f"Cleared scoped token for {provider}{country_str}/{scope}")
return self.save_session(provider, session_data, country)
else:
logger.debug(f"No token found for scope '{scope}' in {provider}{country_str}")
return True
except Exception as e:
logger.error(f"Error clearing scoped token for {provider}{country_str}/{scope}: {e}")
return False
def clear_token(self, provider: str, country: Optional[str] = None) -> bool:
"""Clear only token data but keep other session data (legacy compatibility)"""
country_str = f" (country: {country})" if country else ""
try:
session_data = self.load_session(provider, country)
if not session_data:
@@ -417,15 +481,7 @@ class SessionManager:
return False
def get_all_countries(self, provider: str) -> list:
"""
Get all countries that have session data for a provider
Args:
provider: Provider name
Returns:
List of country codes
"""
"""Get all countries that have session data for a provider"""
try:
data = self.vfs.read_json(self.session_file)
if not data or provider not in data:
@@ -433,10 +489,7 @@ class SessionManager:
provider_data = data[provider]
# Check if this is a nested (country-aware) structure
if isinstance(provider_data, dict):
# Check if any keys look like country codes (2-3 char strings)
# and their values are dicts (session data)
countries = []
for key, value in provider_data.items():
if isinstance(value, dict) and len(key) <= 3:
@@ -449,14 +502,29 @@ class SessionManager:
logger.error(f"Error getting countries for {provider}: {e}")
return []
@staticmethod
def _get_safe_representation(data: Any) -> Dict[str, Any]:
"""Get safe representation of data hiding sensitive fields"""
if not isinstance(data, dict):
return {}
safe_data = {}
for key, value in data.items():
if key in ['access_token', 'refresh_token', 'client_secret', 'password']:
safe_data[key] = f"<present>" if value else f"<missing>"
elif isinstance(value, dict):
# Recursively handle nested dicts (for scoped tokens)
safe_data[key] = SessionManager._get_safe_representation(value)
else:
safe_data[key] = value
return safe_data
def debug_session_file(self) -> None:
"""
Debug method to log the current state of the session file
"""
"""Debug method to log the current state of the session file"""
try:
logger.info(f"=== SESSION FILE DEBUG INFO ===")
# Get VFS debug info
vfs_info = self.vfs.debug_info()
for key, value in vfs_info.items():
logger.info(f"VFS {key}: {value}")
@@ -476,7 +544,6 @@ class SessionManager:
for provider, provider_data in data.items():
if isinstance(provider_data, dict):
# Check if nested (has countries)
has_countries = any(
isinstance(v, dict) and len(k) <= 3
for k, v in provider_data.items()
@@ -486,12 +553,11 @@ class SessionManager:
logger.info(f" {provider} (country-aware):")
for country, session_data in provider_data.items():
if isinstance(session_data, dict):
safe_keys = self._get_safe_keys(session_data)
logger.info(f" {country}: {safe_keys}")
safe_repr = self._get_safe_representation(session_data)
logger.info(f" {country}: {safe_repr}")
else:
# Flat structure (no country)
safe_keys = self._get_safe_keys(provider_data)
logger.info(f" {provider} (no country): {safe_keys}")
safe_repr = self._get_safe_representation(provider_data)
logger.info(f" {provider} (no country): {safe_repr}")
else:
logger.error("Session file contains invalid JSON or is empty")
else:
@@ -502,15 +568,4 @@ class SessionManager:
logger.info(f"=== END SESSION FILE DEBUG ===")
except Exception as e:
logger.error(f"Error during session file debug: {e}")
@staticmethod
def _get_safe_keys(session_data: Dict[str, Any]) -> list:
"""Helper to get safe representation of session keys"""
safe_keys = []
for key in session_data.keys():
if key in ['access_token', 'refresh_token']:
safe_keys.append(f"{key}:present" if session_data[key] else f"{key}:missing")
else:
safe_keys.append(f"{key}:{session_data[key]}")
return safe_keys
logger.error(f"Error during session file debug: {e}")
lib/streaming_providers/base/ui/__init__.py
+31
View File
@@ -0,0 +1,31 @@
# ============================================================================
# FILE 1: streaming_providers/base/ui/__init__.py
# ============================================================================
"""
UI notification system for streaming providers
Provides adapters for different UI environments (Kodi, console, web, etc.)
"""
from .notification_interface import NotificationInterface, NotificationResult
from .notification_factory import NotificationFactory
# Conditional imports - only import if environment supports them
try:
from .kodi_notification_adapter import KodiNotificationAdapter
__all__ = [
'NotificationInterface',
'NotificationResult',
'NotificationFactory',
'KodiNotificationAdapter',
'ConsoleNotificationAdapter'
]
except ImportError:
# Kodi not available
__all__ = [
'NotificationInterface',
'NotificationResult',
'NotificationFactory',
'ConsoleNotificationAdapter'
]
from .console_notification_adapter import ConsoleNotificationAdapter
lib/streaming_providers/base/ui/console_notification_adapter.py
+167
View File
@@ -0,0 +1,167 @@
# ============================================================================
# FILE 3: streaming_providers/base/ui/console_notification_adapter.py
# ============================================================================
"""
Console notification adapter for non-Kodi environments
Displays remote login information as simple text output
"""
import time
from typing import Optional
from .notification_interface import NotificationInterface, NotificationResult
from ..utils.logger import logger
class ConsoleNotificationAdapter(NotificationInterface):
"""
Console-based notification adapter
Displays remote login information as text output
Suitable for:
- Standalone script execution
- Headless environments
- Development/testing
- CI/CD pipelines
"""
def __init__(self):
"""Initialize console notification adapter"""
super().__init__()
self._start_time = None
self._expires_in = 0
self._last_update = 0
@property
def supports_qr_display(self) -> bool:
"""Console cannot display QR codes"""
return False
@property
def supports_countdown(self) -> bool:
"""Console supports text-based countdown"""
return True
@property
def is_blocking(self) -> bool:
"""Console output is non-blocking"""
return False
def show_remote_login(
self,
login_code: str,
qr_url: str,
expires_in: int,
interval: int = 10
) -> NotificationResult:
"""
Show remote login information in console
Args:
login_code: Short login code
qr_url: URL to QR code
expires_in: Expiration time in seconds
interval: Update interval
Returns:
NotificationResult.CONTINUE (always, as console can't cancel)
"""
self._is_active = True
self._start_time = time.time()
self._expires_in = expires_in
self._last_update = 0
# Print header
print("\n" + "=" * 70)
print(" MAGENTATV REMOTE LOGIN REQUIRED")
print("=" * 70)
print()
# Print instructions
print("Please authenticate using your mobile device:")
print()
print(f" Option 1: Scan QR Code")
print(f" Visit this URL on your mobile device:")
print(f" {qr_url}")
print()
print(f" Option 2: Manual Entry")
print(f" Login Code: {login_code}")
print()
print(f" This code expires in {expires_in} seconds")
print()
print("=" * 70)
print()
logger.info(f"Remote login started: code={login_code}, expires_in={expires_in}s")
logger.info(f"QR code URL: {qr_url}")
return NotificationResult.CONTINUE
def update_countdown(self, remaining_seconds: int) -> bool:
"""
Update countdown in console
Only prints updates at reasonable intervals to avoid spam
Args:
remaining_seconds: Seconds remaining
Returns:
bool: Always True (console can't be cancelled by user)
"""
if not self._is_active:
return True
# Print milestone updates (every 30 seconds, or at key intervals)
if remaining_seconds <= 0:
print(f"⏰ Remote login expired")
return True
# Print at: 240s, 180s, 120s, 60s, 30s, 10s
milestones = [240, 180, 120, 60, 30, 10]
if remaining_seconds in milestones or remaining_seconds <= 10:
minutes = remaining_seconds // 60
seconds = remaining_seconds % 60
if minutes > 0:
time_str = f"{minutes}m {seconds}s"
else:
time_str = f"{seconds}s"
print(f"⏳ Waiting for authentication... {time_str} remaining")
return True
def close(self, success: bool = False, message: Optional[str] = None):
"""
Close console notification
Args:
success: Whether authentication succeeded
message: Optional message
"""
if not self._is_active:
return
self._is_active = False
print()
print("=" * 70)
if success:
print("✓ Remote login successful!")
elif message:
print(f"✗ Remote login failed: {message}")
else:
print("✗ Remote login failed")
print("=" * 70)
print()
def is_cancelled(self) -> bool:
"""
Check if cancelled (always False for console)
Returns:
bool: Always False (console can't be cancelled interactively)
"""
return False
lib/streaming_providers/base/ui/kodi_notification_adapter.py
+310
View File
@@ -0,0 +1,310 @@
# ============================================================================
# FILE 4: streaming_providers/base/ui/kodi_notification_adapter.py
# ============================================================================
"""
Kodi notification adapter using xbmcgui dialogs
Displays remote login QR code in a non-blocking progress dialog
"""
import time
import os
import tempfile
from typing import Optional
from .notification_interface import NotificationInterface, NotificationResult
from ..utils.logger import logger
class KodiNotificationAdapter(NotificationInterface):
"""
Kodi-based notification adapter
Uses xbmcgui.DialogProgress() for non-blocking display
Features:
- Downloads and displays QR code SVG
- Live countdown updates
- User can cancel
- Non-blocking operation
"""
def __init__(self, http_manager=None):
"""
Initialize Kodi notification adapter
Args:
http_manager: Optional HTTPManager instance for QR code download
"""
super().__init__()
# Import Kodi modules
try:
import xbmcgui
import xbmc
import xbmcvfs
self.xbmcgui = xbmcgui
self.xbmc = xbmc
self.xbmcvfs = xbmcvfs
self._kodi_available = True
except ImportError as e:
logger.error(f"Kodi modules not available: {e}")
self._kodi_available = False
raise RuntimeError("Kodi modules not available")
self._dialog = None
self._qr_image_path = None
self._start_time = None
self._expires_in = 0
self._interval = 10
self._http_manager = http_manager
@property
def supports_qr_display(self) -> bool:
"""Kodi can display QR code images"""
return True
@property
def supports_countdown(self) -> bool:
"""Kodi supports live countdown"""
return True
@property
def is_blocking(self) -> bool:
"""Kodi dialog is non-blocking"""
return False
def show_remote_login(
self,
login_code: str,
qr_url: str,
expires_in: int,
interval: int = 10
) -> NotificationResult:
"""
Show remote login dialog in Kodi
Args:
login_code: Short login code
qr_url: URL to QR code SVG
expires_in: Expiration time in seconds
interval: Update interval
Returns:
NotificationResult indicating outcome
"""
if not self._kodi_available:
return NotificationResult.ERROR
self._is_active = True
self._start_time = time.time()
self._expires_in = expires_in
self._interval = interval
self._is_cancelled = False
try:
# Create progress dialog
self._dialog = self.xbmcgui.DialogProgress()
# Download QR code SVG
qr_image_path = self._download_qr_code(qr_url)
# Build dialog heading and message
heading = "MagentaTV Remote Login"
# Initial message with instructions
line1 = f"Please scan the QR code with your mobile device"
line2 = f"Login Code: [B]{login_code}[/B]"
line3 = f"Expires in: {self._format_time(expires_in)}"
# Show dialog
self._dialog.create(heading, line1, line2, line3)
# Set QR code image if available
if qr_image_path and os.path.exists(qr_image_path):
# Note: DialogProgress doesn't support images directly
# We'll show the code prominently instead
logger.info(f"QR code downloaded to: {qr_image_path}")
# TODO: Consider using a custom window for better QR display
logger.info(f"Kodi dialog shown: code={login_code}, expires_in={expires_in}s")
return NotificationResult.CONTINUE
except Exception as e:
logger.error(f"Failed to show Kodi dialog: {e}")
self._is_active = False
return NotificationResult.ERROR
def update_countdown(self, remaining_seconds: int) -> bool:
"""
Update countdown in Kodi dialog
Args:
remaining_seconds: Seconds remaining
Returns:
bool: True to continue, False if user cancelled
"""
if not self._is_active or not self._dialog:
return False
try:
# Check if user cancelled
if self._dialog.iscanceled():
logger.info("User cancelled remote login in Kodi")
self._is_cancelled = True
return False
# Calculate percentage for progress bar
elapsed = self._expires_in - remaining_seconds
percentage = int((elapsed / self._expires_in) * 100)
# Update dialog
line1 = "Please scan the QR code with your mobile device"
line2 = "Or open the MagentaTV app and enter the code"
line3 = f"Time remaining: {self._format_time(remaining_seconds)}"
self._dialog.update(percentage, line1, line2, line3)
return True
except Exception as e:
logger.error(f"Failed to update Kodi dialog: {e}")
return False
def close(self, success: bool = False, message: Optional[str] = None):
"""
Close Kodi dialog
Args:
success: Whether authentication succeeded
message: Optional message
"""
if not self._is_active:
return
self._is_active = False
try:
# Close progress dialog
if self._dialog:
self._dialog.close()
self._dialog = None
# Show result notification
if success:
self.xbmcgui.Dialog().notification(
"MagentaTV",
"Remote login successful!",
self.xbmcgui.NOTIFICATION_INFO,
3000
)
elif message:
self.xbmcgui.Dialog().notification(
"MagentaTV",
f"Remote login failed: {message}",
self.xbmcgui.NOTIFICATION_ERROR,
5000
)
# Cleanup QR image
self._cleanup_qr_image()
except Exception as e:
logger.error(f"Failed to close Kodi dialog: {e}")
def is_cancelled(self) -> bool:
"""
Check if user cancelled
Returns:
bool: True if user clicked cancel
"""
if self._dialog and self._dialog.iscanceled():
self._is_cancelled = True
return self._is_cancelled
def _download_qr_code(self, qr_url: str) -> Optional[str]:
"""
Download QR code SVG from URL
Args:
qr_url: URL to QR code SVG
Returns:
str: Path to downloaded file, or None if failed
"""
try:
# Use http_manager if available, otherwise fall back to requests
if self._http_manager:
logger.debug(f"Downloading QR code via http_manager from: {qr_url}")
response = self._http_manager.get(
qr_url,
operation='qr_download',
timeout=10
)
else:
# Fallback to requests if http_manager not available
try:
import requests
logger.debug(f"Downloading QR code via requests from: {qr_url}")
response = requests.get(qr_url, timeout=10)
except ImportError:
logger.warning("Neither http_manager nor requests available, cannot download QR code")
return None
response.raise_for_status()
# Verify content type
content_type = response.headers.get('Content-Type', '')
if 'svg' not in content_type.lower():
logger.warning(f"Unexpected content type: {content_type}")
# Create temp file for QR code
temp_dir = tempfile.gettempdir()
qr_filename = f"magentatv_qr_{int(time.time())}.svg"
qr_path = os.path.join(temp_dir, qr_filename)
# Save to file
with open(qr_path, 'wb') as f:
f.write(response.content)
self._qr_image_path = qr_path
logger.info(f"QR code downloaded successfully: {qr_path}")
return qr_path
except Exception as e:
logger.error(f"Failed to download QR code: {e}")
return None
def _cleanup_qr_image(self):
"""Cleanup temporary QR image file"""
if self._qr_image_path and os.path.exists(self._qr_image_path):
try:
os.remove(self._qr_image_path)
logger.debug(f"Cleaned up QR image: {self._qr_image_path}")
except Exception as e:
logger.warning(f"Failed to cleanup QR image: {e}")
finally:
self._qr_image_path = None
@staticmethod
def _format_time(seconds: int) -> str:
"""
Format seconds as human-readable time
Args:
seconds: Time in seconds
Returns:
str: Formatted time string (e.g., "3m 45s")
"""
if seconds <= 0:
return "Expired"
minutes = seconds // 60
secs = seconds % 60
if minutes > 0:
return f"{minutes}m {secs}s"
else:
return f"{secs}s"
lib/streaming_providers/base/ui/notification_factory.py
+150
View File
@@ -0,0 +1,150 @@
# ============================================================================
# FILE 5: streaming_providers/base/ui/notification_factory.py
# ============================================================================
"""
Factory for creating appropriate notification adapters
Auto-detects environment (Kodi vs standalone) and creates correct adapter
"""
from typing import Optional
from .notification_interface import NotificationInterface
from ..utils.logger import logger
class NotificationFactory:
"""
Factory for creating notification adapters
Automatically detects the runtime environment and creates
the appropriate adapter:
- KodiNotificationAdapter if running in Kodi
- ConsoleNotificationAdapter if running standalone
"""
_cached_adapter: Optional[NotificationInterface] = None
_environment_detected: Optional[str] = None
@classmethod
def create(cls, force_environment: Optional[str] = None,
http_manager=None) -> NotificationInterface:
"""
Create appropriate notification adapter
Args:
force_environment: Force specific environment ('kodi' or 'console')
If None, auto-detects
http_manager: Optional HTTPManager instance for network requests
Returns:
NotificationInterface: Appropriate adapter for current environment
"""
# Return cached adapter if available and no http_manager change
if cls._cached_adapter is not None and force_environment is None and http_manager is None:
logger.debug(f"Using cached notification adapter: {cls._environment_detected}")
return cls._cached_adapter
# Detect environment
if force_environment:
environment = force_environment.lower()
logger.info(f"Forced notification environment: {environment}")
else:
environment = cls._detect_environment()
logger.info(f"Detected notification environment: {environment}")
# Create appropriate adapter
if environment == 'kodi':
adapter = cls._create_kodi_adapter(http_manager=http_manager)
else:
adapter = cls._create_console_adapter()
# Cache the adapter
cls._cached_adapter = adapter
cls._environment_detected = environment
return adapter
@classmethod
def _detect_environment(cls) -> str:
"""
Auto-detect runtime environment
Returns:
str: 'kodi' or 'console'
"""
try:
# Try to import xbmcgui
import xbmcgui
# If import succeeds, we're in Kodi
logger.debug("Kodi modules available - using Kodi notification adapter")
return 'kodi'
except ImportError:
# If import fails, we're standalone
logger.debug("Kodi modules not available - using console notification adapter")
return 'console'
@classmethod
def _create_kodi_adapter(cls, http_manager=None) -> NotificationInterface:
"""
Create Kodi notification adapter
Args:
http_manager: Optional HTTPManager for QR code download
Returns:
KodiNotificationAdapter
"""
try:
from .kodi_notification_adapter import KodiNotificationAdapter
adapter = KodiNotificationAdapter(http_manager=http_manager)
logger.info("✓ Kodi notification adapter created")
return adapter
except Exception as e:
logger.error(f"Failed to create Kodi adapter: {e}")
logger.warning("Falling back to console adapter")
return cls._create_console_adapter()
@classmethod
def _create_console_adapter(cls) -> NotificationInterface:
"""
Create console notification adapter
Returns:
ConsoleNotificationAdapter
"""
from .console_notification_adapter import ConsoleNotificationAdapter
adapter = ConsoleNotificationAdapter()
logger.info("✓ Console notification adapter created")
return adapter
@classmethod
def reset_cache(cls):
"""Reset cached adapter (useful for testing)"""
cls._cached_adapter = None
cls._environment_detected = None
logger.debug("Notification adapter cache reset")
@classmethod
def get_current_environment(cls) -> Optional[str]:
"""
Get currently detected environment
Returns:
str: 'kodi', 'console', or None if not yet detected
"""
return cls._environment_detected
@classmethod
def is_kodi_available(cls) -> bool:
"""
Check if Kodi environment is available
Returns:
bool: True if Kodi modules can be imported
"""
try:
import xbmcgui
return True
except ImportError:
return False
lib/streaming_providers/base/ui/notification_interface.py
+151
View File
@@ -0,0 +1,151 @@
# ============================================================================
# FILE 2: streaming_providers/base/ui/notification_interface.py
# ============================================================================
"""
Abstract notification interface for displaying authentication prompts
Supports multiple UI backends (Kodi, console, web, etc.)
"""
from abc import ABC, abstractmethod
from typing import Optional
from enum import Enum
class NotificationResult(Enum):
"""Result of notification display"""
CONTINUE = "continue" # User wants to continue
CANCELLED = "cancelled" # User cancelled
TIMEOUT = "timeout" # Notification timed out
ERROR = "error" # Error displaying notification
class NotificationInterface(ABC):
"""
Abstract interface for displaying remote login notifications
Implementations must handle:
- Displaying login information (code, QR, URL)
- Countdown updates during polling
- User cancellation
- Cleanup on completion
"""
def __init__(self):
"""Initialize notification interface"""
self._is_active = False
self._is_cancelled = False
@abstractmethod
def show_remote_login(
self,
login_code: str,
qr_url: str,
expires_in: int,
interval: int = 10
) -> NotificationResult:
"""
Show remote login notification to user
This method should:
1. Display the login code and QR code/URL
2. Show countdown timer
3. Allow user cancellation
4. Return when done or cancelled
Args:
login_code: Short code user can type (e.g., "PY48E62Q")
qr_url: Full URL to QR code SVG (e.g., "https://wcps.t-online.de/caas/default/v1/remoteLogin/PY48E62Q")
expires_in: Total seconds until expiration
interval: Polling interval in seconds (for countdown updates)
Returns:
NotificationResult indicating outcome
"""
pass
@abstractmethod
def update_countdown(self, remaining_seconds: int) -> bool:
"""
Update countdown display
Args:
remaining_seconds: Seconds remaining until expiration
Returns:
bool: True to continue, False if user cancelled
"""
pass
@abstractmethod
def close(self, success: bool = False, message: Optional[str] = None):
"""
Close/cleanup the notification
Args:
success: Whether authentication succeeded
message: Optional message to display
"""
pass
@abstractmethod
def is_cancelled(self) -> bool:
"""
Check if user has cancelled
Returns:
bool: True if user cancelled
"""
pass
def mark_cancelled(self):
"""Mark notification as cancelled"""
self._is_cancelled = True
@property
def is_active(self) -> bool:
"""Check if notification is currently active"""
return self._is_active
@property
def supports_qr_display(self) -> bool:
"""
Check if this notifier can display QR codes
Returns:
bool: True if QR code display is supported
"""
return False
@property
def supports_countdown(self) -> bool:
"""
Check if this notifier supports live countdown updates
Returns:
bool: True if countdown updates are supported
"""
return False
@property
def is_blocking(self) -> bool:
"""
Check if this notifier blocks the calling thread
Returns:
bool: True if blocking, False if non-blocking
"""
return True
def get_capabilities(self) -> dict:
"""
Get notifier capabilities
Returns:
dict: Capability information
"""
return {
'type': self.__class__.__name__,
'supports_qr_display': self.supports_qr_display,
'supports_countdown': self.supports_countdown,
'is_blocking': self.is_blocking,
'is_active': self.is_active
}
lib/streaming_providers/providers/magenta2/__init__.py
+45
View File
@@ -0,0 +1,45 @@
# streaming_providers/providers/magenta2/__init__.py
from .provider import Magenta2Provider
from .models import Magenta2Channel, Magenta2PlaybackRestrictedException, DeviceLimitExceededException
from .auth import Magenta2Authenticator, Magenta2AuthToken, Magenta2Credentials
from .discovery import DiscoveryService
from .endpoint_manager import EndpointManager, EndpointCategory
from .config_models import BootstrapConfig, ManifestConfig, OpenIDConfig, ProviderConfig, MpxConfig, DrmConfig, \
TvHubConfig
from .constants import (
SUPPORTED_COUNTRIES,
)
__all__ = [
# Core provider
'Magenta2Provider',
# Models
'Magenta2Channel',
'Magenta2PlaybackRestrictedException',
'DeviceLimitExceededException',
# Authentication
'Magenta2Authenticator',
'Magenta2AuthToken',
'Magenta2Credentials',
# New discovery system
'DiscoveryService',
'EndpointManager',
'EndpointCategory',
# Configuration models
'BootstrapConfig',
'ManifestConfig',
'OpenIDConfig',
'ProviderConfig',
'MpxConfig',
'DrmConfig',
'TvHubConfig',
# Constants
'SUPPORTED_COUNTRIES',
]
__version__ = '2.0.0' # Major version bump for architectural changes
lib/streaming_providers/providers/magenta2/auth.py
+1696
View File
File diff suppressed because it is too large Load Diff
lib/streaming_providers/providers/magenta2/config_models.py
+399
View File
@@ -0,0 +1,399 @@
# streaming_providers/providers/magenta2/config_models.py
from dataclasses import dataclass, field
from typing import Dict, Optional, Any
from datetime import datetime
@dataclass
class BootstrapConfig:
"""Configuration extracted from bootstrap response"""
client_model: str
device_model: str
sam3_client_id: Optional[str] = None
taa_url: Optional[str] = None
device_tokens_url: Optional[str] = None
line_auth_url: Optional[str] = None
remote_login_url: Optional[str] = None
openid_config_url: Optional[str] = None
account_base_url: Optional[str] = None
consumer_accounts_url: Optional[str] = None
raw_data: Dict[str, Any] = field(default_factory=dict)
@classmethod
def from_api_response(cls, bootstrap_data: Dict[str, Any], platform: str) -> 'BootstrapConfig':
"""Create BootstrapConfig from API response"""
base_settings = bootstrap_data.get('baseSettings', {})
return cls(
client_model=base_settings.get('clientModel', f"ftv-{platform}"),
device_model=base_settings.get('deviceModel', f"{platform.upper()}_FTV"),
sam3_client_id=base_settings.get('sam3ClientId'),
taa_url=base_settings.get('taaUrl'),
device_tokens_url=base_settings.get('deviceTokensUrl'),
line_auth_url=base_settings.get('lineAuthUrl'),
remote_login_url=base_settings.get('remoteLoginUrl'),
openid_config_url=base_settings.get('sam3Url'),
account_base_url=base_settings.get('accountBaseUrl'),
consumer_accounts_url=base_settings.get('consumerAccountsBaseUrl'),
raw_data=bootstrap_data
)
def update_from_manifest(self, manifest_config: 'ManifestConfig') -> None:
"""Update bootstrap config with data from manifest"""
from ...base.utils.logger import logger
# Get SAM3 client ID from manifest if not in bootstrap
if not self.sam3_client_id:
sam3_client_id = manifest_config.get_parameter_value('SAM3ClientId')
if sam3_client_id:
self.sam3_client_id = sam3_client_id
logger.info(f"✓ SAM3 Client ID from manifest: {sam3_client_id}")
# Get TAA URL from manifest if not in bootstrap
if not self.taa_url:
taa_url = manifest_config.get_parameter_value('TAA-URL')
if taa_url:
self.taa_url = taa_url
logger.debug(f"TAA URL from manifest: {taa_url}")
# Get Line Auth URL from manifest if not in bootstrap
if not self.line_auth_url:
line_auth_url = manifest_config.get_parameter_value('LineAuthURL')
if line_auth_url:
self.line_auth_url = line_auth_url
logger.debug(f"Line Auth URL from manifest: {line_auth_url}")
# Get Remote Login URL from manifest if not in bootstrap
if not self.remote_login_url:
remote_login_url = manifest_config.get_parameter_value('RemoteLoginURL')
if remote_login_url:
self.remote_login_url = remote_login_url
logger.debug(f"Remote Login URL from manifest: {remote_login_url}")
@dataclass
class MpxConfig:
"""MPX (ThePlatform) configuration from manifest"""
account_pid: str
license_service_url: str
selector_service_url: str
user_profile_url: Optional[str] = None
bookmark_base_url: Optional[str] = None
pvr_base_url: Optional[str] = None
feeds: Dict[str, str] = field(default_factory=dict)
# ADD THIS: Channel stations feed
channel_stations_feed: Optional[str] = None
@classmethod
def from_manifest_data(cls, manifest_data: Dict[str, Any]) -> 'MpxConfig':
"""Create MpxConfig from manifest data"""
def get_param(key: str) -> Optional[str]:
"""Helper to get value from parameters array"""
if 'settings' not in manifest_data:
return None
settings = manifest_data['settings']
if 'parameters' not in settings:
return None
for param in settings['parameters']:
if param.get('key') == key:
value = param.get('value')
return value if value and value != 'unused' else None
return None
mpx_data = manifest_data.get('mpx', {})
# Build feeds dict from parameters
feeds = {}
feed_keys = [
'mpxBasicUrlAllChannelSchedulesFeed',
'mpxBasicUrlEntitledChannelsFeed',
'mpxAllListingsFeedUrl',
'mpxAllProgramsFeedUrl'
]
for feed_key in feed_keys:
feed_url = get_param(feed_key)
if feed_url:
simple_key = feed_key.replace('mpx', '').replace('Url', '').replace('BasicUrl', '')
feeds[simple_key] = feed_url
# ADD THIS: Extract channel stations feed
channel_stations_feed = get_param('mpxDefaultUrlAllChannelStationsFeed')
return cls(
account_pid=get_param('mpxAccountPid') or mpx_data.get('accountPid', 'mdeprod'),
license_service_url=get_param('mpxBasicUrlGetApplicableDistributionRights') or mpx_data.get(
'licenseServiceUrl', ''),
selector_service_url=get_param('mpxBasicUrlSelectorService') or mpx_data.get('selectorServiceUrl', ''),
user_profile_url=get_param('mpxUserProfileUrl') or mpx_data.get('userProfileUrl'),
bookmark_base_url=get_param('mpxBookmarkBaseUrl') or mpx_data.get('bookmarkBaseUrl'),
pvr_base_url=get_param('mpxPvrBaseUrl') or mpx_data.get('pvrBaseUrl'),
feeds=feeds,
# ADD THIS:
channel_stations_feed=channel_stations_feed
)
def get_account_uri(self) -> str:
"""
NEW: Construct MPX account URI for persona token composition
Format: urn:theplatform:auth:root:{accountPid}
"""
return f"urn:theplatform:auth:root:{self.account_pid}"
@dataclass
class DrmConfig:
"""DRM configuration from manifest"""
widevine_license_url: str
vod_widevine_license_url: Optional[str] = None
fairplay_license_url: Optional[str] = None
@classmethod
def from_manifest_data(cls, manifest_data: Dict[str, Any]) -> 'DrmConfig':
"""Create DrmConfig from manifest data"""
from ...base.utils.logger import logger
# Create a temporary helper function for this method
def get_param(key: str) -> Optional[str]:
"""Helper to get value from parameters array"""
if 'settings' not in manifest_data:
return None
settings = manifest_data['settings']
if 'parameters' not in settings:
return None
for param in settings['parameters']:
if param.get('key') == key:
value = param.get('value')
return value if value and value != 'unused' else None
return None
# Try parameters array first (current structure)
widevine_url = get_param('widevineLicenseAcquisitionURL')
fairplay_url = get_param('fairplayLicenseAcquisitionURL')
# Fallback to legacy structure
if not widevine_url or not fairplay_url:
livetv_drm = manifest_data.get('livetv', {}).get('drm', {})
vod_drm = manifest_data.get('vod', {}).get('drm', {})
if not widevine_url:
widevine_url = livetv_drm.get('widevineLicenseAcquisitionUrl', '')
if not fairplay_url:
fairplay_url = livetv_drm.get('fairplayLicenseAcquisitionUrl', '')
vod_widevine = vod_drm.get('widevineLicenseAcquisitionUrl')
else:
vod_widevine = None # Not in parameters array
logger.debug(f"DRM config: widevine={bool(widevine_url)}, fairplay={bool(fairplay_url)}")
return cls(
widevine_license_url=widevine_url or '',
vod_widevine_license_url=vod_widevine,
fairplay_license_url=fairplay_url or ''
)
@dataclass
class TvHubConfig:
"""TV Hub URLs configuration from manifest"""
base_urls: Dict[str, str] = field(default_factory=dict)
@classmethod
def from_manifest_data(cls, manifest_data: Dict[str, Any]) -> 'TvHubConfig':
"""Create TvHubConfig from manifest data"""
# Create a temporary helper function for this method
def get_param(param_key: str) -> Optional[str]:
"""Helper to get value from parameters array"""
if 'settings' not in manifest_data:
return None
settings_data = manifest_data['settings']
if 'parameters' not in settings_data:
return None
for param in settings_data['parameters']:
if param.get('key') == param_key:
value = param.get('value')
return value if value and value != 'unused' else None
return None
base_urls = {}
# TV Hub URLs from parameters
tvhub_keys = [
'homeUrl',
'settingsMenu',
'broadcastDetailsURL',
'vodDetailsURL',
'searchUrl',
'kidsSearchURL',
'myWatchlistUrl',
'myMoviesURL'
]
# Check settings object first
if 'settings' in manifest_data:
settings_obj = manifest_data['settings']
for tvhub_key in tvhub_keys:
if tvhub_key in settings_obj:
url = settings_obj.get(tvhub_key)
if url and url != 'unused':
base_urls[tvhub_key] = url
# Also check parameters array
for tvhub_key in tvhub_keys:
if tvhub_key not in base_urls:
url = get_param(tvhub_key)
if url:
base_urls[tvhub_key] = url
# Legacy structure fallback
tv_hubs = manifest_data.get('tvHubUrls', {})
for hub_name, hub_url in tv_hubs.items():
if isinstance(hub_url, str) and hub_url.startswith('http') and hub_name not in base_urls:
base_urls[hub_name] = hub_url
return cls(base_urls=base_urls)
@dataclass
class ManifestConfig:
"""Complete configuration from manifest discovery"""
mpx: MpxConfig
drm: DrmConfig
tv_hubs: TvHubConfig
youbora_config: Dict[str, Any] = field(default_factory=dict)
npvr_config: Dict[str, Any] = field(default_factory=dict)
raw_data: Dict[str, Any] = field(default_factory=dict)
@classmethod
def from_api_response(cls, manifest_data: Dict[str, Any]) -> 'ManifestConfig':
"""Create ManifestConfig from API response"""
return cls(
mpx=MpxConfig.from_manifest_data(manifest_data),
drm=DrmConfig.from_manifest_data(manifest_data),
tv_hubs=TvHubConfig.from_manifest_data(manifest_data),
youbora_config=manifest_data.get('youbora', {}),
npvr_config=manifest_data.get('npvr', {}),
raw_data=manifest_data
)
def get_parameter_value(self, key: str) -> Optional[str]:
"""Get value from settings.parameters array by key"""
if 'settings' not in self.raw_data:
return None
settings = self.raw_data['settings']
if 'parameters' not in settings or not isinstance(settings['parameters'], list):
return None
for param in settings['parameters']:
if param.get('key') == key:
value = param.get('value')
# Empty string or "unused" means not available
return value if value and value != 'unused' else None
return None
def get_device_token(self) -> Optional[str]:
"""Extract device token from raw data"""
# Try direct path first (legacy)
if 'deviceToken' in self.raw_data:
return self.raw_data['deviceToken']
# Try nested in sts object (current structure)
if 'sts' in self.raw_data and isinstance(self.raw_data['sts'], dict):
if 'deviceToken' in self.raw_data['sts']:
return self.raw_data['sts']['deviceToken']
return None
def get_authorize_tokens_url(self) -> Optional[str]:
"""Extract authorize tokens URL from raw data"""
# Check in sts object first (current structure)
if 'sts' in self.raw_data and isinstance(self.raw_data['sts'], dict):
if 'authorizeTokensUrl' in self.raw_data['sts']:
return self.raw_data['sts']['authorizeTokensUrl']
# Check direct path (legacy)
if 'authorizeTokensUrl' in self.raw_data:
return self.raw_data['authorizeTokensUrl']
# Check in settings.parameters array as fallback
line_auth_url = self.get_parameter_value('LineAuthURL')
if line_auth_url:
return line_auth_url
return None
@dataclass
class OpenIDConfig:
"""OpenID Connect configuration"""
token_endpoint: str
authorization_endpoint: str
userinfo_endpoint: Optional[str] = None
revocation_endpoint: Optional[str] = None
raw_data: Dict[str, Any] = field(default_factory=dict)
@classmethod
def from_api_response(cls, openid_data: Dict[str, Any]) -> 'OpenIDConfig':
"""Create OpenIDConfig from API response"""
return cls(
token_endpoint=openid_data.get('token_endpoint', ''),
authorization_endpoint=openid_data.get('authorization_endpoint', ''),
userinfo_endpoint=openid_data.get('userinfo_endpoint'),
revocation_endpoint=openid_data.get('revocation_endpoint'),
raw_data=openid_data
)
@dataclass
class ProviderConfig:
"""Complete provider configuration assembled from all discovery sources"""
bootstrap: BootstrapConfig
manifest: Optional[ManifestConfig] = None
openid: Optional[OpenIDConfig] = None
discovered_at: datetime = field(default_factory=datetime.now)
@property
def is_complete(self) -> bool:
"""Check if configuration is complete enough for operation"""
return self.bootstrap is not None and self.manifest is not None
def get_resolved_feed_url(self, feed_name: str) -> Optional[str]:
"""Get resolved MPX feed URL with account PID substitution"""
if not self.manifest or not self.manifest.mpx:
return None
feed_template = self.manifest.mpx.feeds.get(feed_name)
if not feed_template:
return None
return feed_template.replace('{MpxAccountPid}', self.manifest.mpx.account_pid)
def get_resolved_tvhub_url(self, hub_name: str, client_model: Optional[str] = None) -> Optional[str]:
"""Get resolved TV Hub URL with client model substitution"""
if not self.manifest:
return None
hub_template = self.manifest.tv_hubs.base_urls.get(hub_name)
if not hub_template:
return None
resolved_client_model = client_model or self.bootstrap.client_model
return hub_template.replace('{clientModel}', resolved_client_model)
def get_mpx_account_uri(self) -> Optional[str]:
"""NEW: Get MPX account URI for persona token composition"""
if not self.manifest or not self.manifest.mpx:
return None
return self.manifest.mpx.get_account_uri()
def get_device_token(self) -> Optional[str]:
"""NEW: Get device token from manifest"""
if not self.manifest:
return None
return self.manifest.get_device_token()
def get_authorize_tokens_url(self) -> Optional[str]:
"""NEW: Get authorize tokens URL from manifest"""
if not self.manifest:
return None
return self.manifest.get_authorize_tokens_url()
lib/streaming_providers/providers/magenta2/constants.py
+263
View File
@@ -0,0 +1,263 @@
# streaming_providers/providers/magenta2/constants.py
# ============================================================================
# Magenta2 Configuration
# ============================================================================
# Supported countries (Magenta2 is Germany-specific)
SUPPORTED_COUNTRIES = ['de']
# Default country
DEFAULT_COUNTRY = 'de'
# Platform configuration
DEFAULT_PLATFORM = 'android-tv'
MAGENTA2_PLATFORMS = {
'web': {
'device_name': 'Web Browser',
'firmware': 'Chrome 120',
'user_agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'terminal_type': 'WEB'
},
'android-tv': {
'device_name': 'Android TV',
'firmware': 'Android 11',
'user_agent': 'Mozilla/5.0 (Linux; Android 11; Android TV) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36',
'terminal_type': 'ATV_ANDROIDTV'
},
'atv-launcher': {
'device_name': 'MagentaTV Stick',
'firmware': 'Android 11',
'user_agent': 'Mozilla/5.0 (Linux; Android 11; AFTS Build/PPR1.180610.011) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36',
'terminal_type': 'ATV_LAUNCHER'
},
'android-mobile': {
'device_name': 'Android Mobile',
'firmware': 'Android 13',
'user_agent': 'Mozilla/5.0 (Linux; Android 13; SM-G991B) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Mobile Safari/537.36',
'terminal_type': 'ANDROID_MOBILE'
},
'ios': {
'device_name': 'iPhone',
'firmware': 'iOS 15',
'user_agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 15_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/15.0 Mobile/15E148 Safari/604.1',
'terminal_type': 'IOS'
}
}
# ============================================================================
# Manifest Request Configuration
# ============================================================================
# App identification for manifest requests
MAGENTA2_APP_NAME = 'MagentaTV'
MAGENTA2_APP_VERSION = '104180'
MAGENTA2_RUNTIME_VERSION = '1'
# Model names for manifest requests (different from device models!)
MANIFEST_MODEL_MAPPINGS = {
'web': 'DT:WEB',
'android-tv': 'DT:ATV-AndroidTV',
'atv-launcher': 'DT:ATV-Launcher',
'android-mobile': 'DT:Android-Mobile',
'ios': 'DT:IOS'
}
# Firmware strings for manifest requests
MANIFEST_FIRMWARE_MAPPINGS = {
'web': 'Chrome 120',
'android-tv': 'API level 30',
'atv-launcher': 'API level 30',
'android-mobile': 'API level 33',
'ios': 'iOS 15.0'
}
# Also update the fallback mappings:
CLIENT_MODEL_MAPPINGS = {
'web': 'ftv-web',
'android-tv': 'ftv-androidtv',
'atv-launcher': 'ftv-androidtv',
'android-mobile': 'ftv-android',
'ios': 'ftv-ios'
}
DEVICE_MODEL_MAPPINGS = {
'web': 'WebBrowser_FTV',
'android-tv': 'AndroidTV_FTV',
'atv-launcher': 'AndroidTV_FTV',
'android-mobile': 'AndroidMobile_FTV',
'ios': 'iOS_FTV'
}
# And update subscriber types if needed:
SUBSCRIBER_TYPES = {
'web': 'WEB_OTT_DT',
'android-tv': 'FTV_OTT_DT',
'atv-launcher': 'FTV_OTT_DT', # Same as android-tv
'android-mobile': 'MOB_OTT_DT', # Different for mobile
'ios': 'IOS_OTT_DT'
}
# ============================================================================
# API Configuration - MINIMAL HARDCODING
# ============================================================================
# Only bootstrap endpoint is hardcoded - everything else discovered dynamically
MAGENTA2_BASE_URL = 'https://prod.dcm.telekom-dienste.de/v1'
MAGENTA2_BOOTSTRAP_URL = MAGENTA2_BASE_URL + '/settings/{terminal_type}/bootstrap'
MAGENTA2_MANIFEST_URL = MAGENTA2_BASE_URL + '/settings/{terminal_type}/manifest'
# Fallback endpoints if discovery fails
MAGENTA2_FALLBACK_ENDPOINTS = {
'OPENID_CONFIG': 'https://accounts.login.idm.telekom.com/.well-known/openid-configuration',
'TAA_AUTH': 'https://taa.p7s1.io/api/v1/taa',
'ENTITLEMENT': 'https://entitlement.p7s1.io/api/user/entitlement-token',
}
# ============================================================================
# Application Configuration
# ============================================================================
# Application identifiers
IDM = "TDGIDM"
APPVERSION2 = "3.134.4462"
SSO_URL = "https://ssom.magentatv.de/login"
# SSO User Agent
SSO_USER_AGENT = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
# ============================================================================
# OAuth2 Configuration
# ============================================================================
# Client IDs for different platforms
MAGENTA2_CLIENT_IDS = {
'web': '709115c2-f87e-4bad-9b94-28ac08d72cd9',
'android-tv': '05f5f3df-1130-4707-a761-c04d0c50b7f2',
'ios': '21218403-52ec-4a65-abf4-f36a0eadd631'
}
# OAuth2 scopes
MAGENTA2_OAUTH_SCOPE = "openid profile offline_access tvhubs"
# OAuth2 redirect URI
MAGENTA2_REDIRECT_URI = "https://web2.magentatv.de/authn/idm"
# ============================================================================
# Request Headers Configuration
# ============================================================================
# Headers for different API endpoints
MAGENTA2_HEADERS = {
'DEFAULT': {
'Content-Type': 'application/json',
'Accept': 'application/json'
},
'SSO': {
'User-Agent': SSO_USER_AGENT,
'Content-Type': 'application/json',
'origin': 'https://web2.magentatv.de',
'referer': 'https://web2.magentatv.de/'
},
'DCM': {
'Content-Type': 'application/json',
'Accept': 'application/json'
},
'OAUTH2': {
'Content-Type': 'application/x-www-form-urlencoded'
}
}
# ============================================================================
# Authentication Configuration
# ============================================================================
# Grant types
GRANT_TYPES = {
'LINE_AUTH': 'urn:com:telekom:ott-app-services:access-auth',
'AUTH_CODE': 'authorization_code',
'PASSWORD': 'password',
'REFRESH_TOKEN': 'refresh_token',
'REMOTE_LOGIN': 'urn:telekom:com:grant-type:remote-login',
'CLIENT_CREDENTIALS': 'client_credentials'
}
# ============================================================================
# Content Configuration
# ============================================================================
# Content types
CONTENT_TYPE_LIVE = 'LIVE'
CONTENT_TYPE_VOD = 'VOD'
# Stream modes
MODE_LIVE = 'live'
MODE_VOD = 'vod'
# ============================================================================
# DRM Configuration
# ============================================================================
# DRM system
DRM_SYSTEM_WIDEVINE = 'widevine'
# DRM request headers
DRM_REQUEST_HEADERS = {
'Content-Type': 'application/octet-stream'
}
# ============================================================================
# Request Configuration
# ============================================================================
# Default timeout for HTTP requests (seconds)
DEFAULT_REQUEST_TIMEOUT = 30
# Default maximum retries for failed requests
DEFAULT_MAX_RETRIES = 3
# Default time window for EPG queries (hours)
DEFAULT_EPG_WINDOW_HOURS = 3
# Cache durations (seconds)
BOOTSTRAP_CACHE_DURATION = 3600 # 1 hour
OPENID_CONFIG_CACHE_DURATION = 86400 # 24 hours
MANIFEST_CACHE_DURATION = 7200 # 2 hours
# ============================================================================
# Error Codes
# ============================================================================
# Known error codes from Magenta2 API
ERROR_CODES = {
'DEVICE_LIMIT_EXCEEDED': 'deviceLimitExceeded',
'PLAYBACK_RESTRICTED': 'ENT_RVOD_Playback_Restricted',
'UNAUTHORIZED': 'ENT_Unauthorized',
'INVALID_TOKEN': 'INVALID_TOKEN',
'SESSION_EXPIRED': 'SESSION_EXPIRED'
}
# ============================================================================
# TAA Configuration
# ============================================================================
# TAA request template
TAA_REQUEST_TEMPLATE = {
"accessTokenSource": IDM,
"appVersion": APPVERSION2,
"channel": {
"id": "Tv"
},
"natco": "DE",
"type": "telekom"
}
# ============================================================================
# Bootstrap Configuration
# ============================================================================
# Bootstrap parameters
BOOTSTRAP_PARAMS = {
'$redirect': 'false'
}
# REMOVED: Old BOOTSTRAP_KEYS - now handled in config_models.py
lib/streaming_providers/providers/magenta2/discovery.py
+379
View File
@@ -0,0 +1,379 @@
# streaming_providers/providers/magenta2/discovery.py
import time
from typing import Dict, Optional, Any
from ...base.network import HTTPManager
from ...base.models.proxy_models import ProxyConfig
from ...base.utils.logger import logger
from .config_models import BootstrapConfig, ManifestConfig, OpenIDConfig, ProviderConfig
from .constants import (
MAGENTA2_BOOTSTRAP_URL,
MAGENTA2_MANIFEST_URL,
SUBSCRIBER_TYPES,
DEFAULT_REQUEST_TIMEOUT,
BOOTSTRAP_CACHE_DURATION,
OPENID_CONFIG_CACHE_DURATION
)
class DiscoveryService:
"""
Service for dynamic discovery of Magenta2 endpoints and configuration
"""
def __init__(self, platform: str, terminal_type: str, device_id: str, session_id: str,
http_manager: HTTPManager, proxy_config: Optional[ProxyConfig] = None):
self.platform = platform
self.terminal_type = terminal_type
self.device_id = device_id
self.session_id = session_id
self.http_manager = http_manager
self.proxy_config = proxy_config
self.subscriber_type = SUBSCRIBER_TYPES.get(platform, 'FTV_OTT_DT')
# Cache storage
self._bootstrap_config: Optional[BootstrapConfig] = None
self._manifest_config: Optional[ManifestConfig] = None
self._openid_config: Optional[OpenIDConfig] = None
self._last_bootstrap: Optional[float] = None
self._last_manifest: Optional[float] = None
self._last_openid: Optional[float] = None
def discover_provider_config(self, force_refresh: bool = False) -> ProviderConfig:
"""
Perform complete provider configuration discovery
Returns:
ProviderConfig: Complete provider configuration
"""
logger.info("Starting Magenta2 provider configuration discovery")
try:
# Step 1: Bootstrap discovery
bootstrap_config = self.discover_bootstrap(force_refresh)
if not bootstrap_config:
raise Exception("Bootstrap discovery failed - cannot proceed")
# Step 2: Manifest discovery (includes device token)
manifest_config = self.discover_manifest(force_refresh)
# IMPORTANT: Update bootstrap with manifest data
if manifest_config:
bootstrap_config.update_from_manifest(manifest_config)
# Log device token status
device_token = manifest_config.get_device_token()
if device_token:
logger.info("✓ Device token obtained from manifest")
else:
logger.warning("⚠️ No device token found in manifest")
# Step 3: OpenID discovery (if bootstrap provided OpenID config URL)
openid_config = None
if bootstrap_config.openid_config_url:
try:
openid_config = self.discover_openid_config(force_refresh)
except Exception as e:
logger.warning(f"OpenID discovery failed: {e}")
provider_config = ProviderConfig(
bootstrap=bootstrap_config,
manifest=manifest_config,
openid=openid_config
)
# Log MPX account info for persona token composition
if manifest_config and manifest_config.mpx:
account_uri = manifest_config.mpx.get_account_uri()
logger.info(f"✓ MPX account URI for persona token: {account_uri}")
return provider_config
except Exception as e:
logger.error(f"Configuration discovery failed: {e}")
self._create_fallback_configuration()
raise
def discover_bootstrap(self, force_refresh: bool = False) -> Optional[BootstrapConfig]:
"""
Discover bootstrap configuration
Returns:
BootstrapConfig: Bootstrap configuration, None if failed
"""
# Check cache
if not force_refresh and self._bootstrap_config and self._last_bootstrap:
cache_age = time.time() - self._last_bootstrap
if cache_age < BOOTSTRAP_CACHE_DURATION:
logger.debug("Using cached bootstrap configuration")
return self._bootstrap_config
try:
logger.info("Discovering bootstrap configuration")
terminal_type = self.terminal_type.lower().replace('_', '-')
url = MAGENTA2_BOOTSTRAP_URL.format(terminal_type=terminal_type)
params = {
'deviceid': self.device_id,
'sid': self.session_id,
'$redirect': 'false'
}
headers = self._get_dcm_headers()
response = self.http_manager.get(
url,
operation='bootstrap',
headers=headers,
params=params,
timeout=DEFAULT_REQUEST_TIMEOUT
)
response.raise_for_status()
bootstrap_data = response.json()
self._bootstrap_config = BootstrapConfig.from_api_response(bootstrap_data, self.platform)
self._last_bootstrap = time.time()
logger.info(
f"Bootstrap discovery successful: "
f"clientModel={self._bootstrap_config.client_model}, "
f"deviceModel={self._bootstrap_config.device_model}"
)
# Log critical endpoints
if self._bootstrap_config.taa_url:
logger.debug(f"TAA URL: {self._bootstrap_config.taa_url}")
if self._bootstrap_config.device_tokens_url:
logger.debug(f"Device tokens URL: {self._bootstrap_config.device_tokens_url}")
return self._bootstrap_config
except Exception as e:
logger.error(f"Bootstrap discovery failed: {e}")
# Don't cache failed attempts
self._bootstrap_config = None
self._last_bootstrap = None
return None
def discover_manifest(self, force_refresh: bool = False) -> Optional[ManifestConfig]:
"""
ENHANCED: Discover manifest configuration including device token
Uses correct manifest endpoint parameters
Returns:
ManifestConfig: Manifest configuration, None if failed
"""
# Check cache
if not force_refresh and self._manifest_config and self._last_manifest:
cache_age = time.time() - self._last_manifest
if cache_age < BOOTSTRAP_CACHE_DURATION:
logger.debug("Using cached manifest configuration")
return self._manifest_config
try:
logger.info("Discovering manifest configuration")
# Determine manifest URL - prefer device_tokens_url from bootstrap
if self._bootstrap_config and self._bootstrap_config.device_tokens_url:
manifest_url = self._bootstrap_config.device_tokens_url
logger.debug(f"Using bootstrap device_tokens_url for manifest: {manifest_url}")
else:
terminal_type = self.terminal_type.lower().replace('_', '-')
manifest_url = MAGENTA2_MANIFEST_URL.format(terminal_type=terminal_type)
logger.debug(f"Using fallback manifest URL: {manifest_url}")
# Build correct manifest parameters
from .constants import (
MAGENTA2_APP_NAME,
MAGENTA2_APP_VERSION,
MAGENTA2_RUNTIME_VERSION,
MANIFEST_MODEL_MAPPINGS,
MANIFEST_FIRMWARE_MAPPINGS
)
params = {
'model': MANIFEST_MODEL_MAPPINGS.get(self.platform, 'DT:ATV-AndroidTV'),
'deviceId': self.device_id,
'appname': MAGENTA2_APP_NAME,
'appVersion': MAGENTA2_APP_VERSION,
'firmware': MANIFEST_FIRMWARE_MAPPINGS.get(self.platform, 'API level 30'),
'runtimeVersion': MAGENTA2_RUNTIME_VERSION,
'duid': self.device_id # Same as deviceId
}
logger.debug(f"Manifest request params: {params}")
headers = self._get_dcm_headers()
response = self.http_manager.get(
manifest_url,
operation='manifest_discovery',
headers=headers,
params=params,
timeout=DEFAULT_REQUEST_TIMEOUT
)
response.raise_for_status()
manifest_data = response.json()
self._manifest_config = ManifestConfig.from_api_response(manifest_data)
self._last_manifest = time.time()
# ENHANCED: Log device token and critical auth info
device_token = self._manifest_config.get_device_token()
authorize_tokens_url = self._manifest_config.get_authorize_tokens_url()
if device_token:
logger.info("✓ Device token found in manifest")
logger.debug(f"Device token preview: {device_token[:20]}...{device_token[-10:]}")
else:
logger.warning("⚠️ Device token NOT found in manifest response")
if authorize_tokens_url:
logger.debug(f"Authorize tokens URL: {authorize_tokens_url}")
# Log MPX account info (critical for persona token)
mpx_account_pid = self._manifest_config.mpx.account_pid
account_uri = self._manifest_config.mpx.get_account_uri()
logger.info(f"MPX account PID: {mpx_account_pid}")
logger.info(f"MPX account URI: {account_uri}")
logger.info(
f"Manifest discovery successful: "
f"MPX account={mpx_account_pid}, "
f"DRM endpoints={len([k for k in self._manifest_config.drm.__dict__.keys() if self._manifest_config.drm.__dict__[k]])}, "
f"TV hubs={len(self._manifest_config.tv_hubs.base_urls)}"
)
return self._manifest_config
except Exception as e:
logger.error(f"Manifest discovery failed: {e}")
# Don't cache failed attempts
self._manifest_config = None
self._last_manifest = None
return None
def discover_openid_config(self, force_refresh: bool = False) -> Optional[OpenIDConfig]:
"""
Discover OpenID Connect configuration
Returns:
OpenIDConfig: OpenID configuration, None if failed
"""
# Check cache
if not force_refresh and self._openid_config and self._last_openid:
cache_age = time.time() - self._last_openid
if cache_age < OPENID_CONFIG_CACHE_DURATION:
logger.debug("Using cached OpenID configuration")
return self._openid_config
try:
if not self._bootstrap_config or not self._bootstrap_config.openid_config_url:
logger.warning("No OpenID config URL available from bootstrap")
return None
logger.info("Discovering OpenID configuration")
openid_url = self._bootstrap_config.openid_config_url
response = self.http_manager.get(
openid_url,
operation='openid_discovery',
headers={'Accept': 'application/json'},
timeout=DEFAULT_REQUEST_TIMEOUT
)
response.raise_for_status()
openid_data = response.json()
self._openid_config = OpenIDConfig.from_api_response(openid_data)
self._last_openid = time.time()
logger.info(f"OpenID discovery successful: token_endpoint={self._openid_config.token_endpoint}")
return self._openid_config
except Exception as e:
logger.error(f"OpenID discovery failed: {e}")
# Don't cache failed attempts
self._openid_config = None
self._last_openid = None
return None
def _get_dcm_headers(self) -> Dict[str, str]:
"""Get headers for DCM requests"""
from .constants import MAGENTA2_PLATFORMS, DEFAULT_PLATFORM
platform_config = MAGENTA2_PLATFORMS.get(self.platform, MAGENTA2_PLATFORMS[DEFAULT_PLATFORM])
return {
'User-Agent': platform_config['user_agent'],
'Content-Type': 'application/json',
'Accept': 'application/json',
'x-dt-session-id': self.session_id,
'x-dt-call-id': self._generate_call_id()
}
@staticmethod
def _generate_call_id() -> str:
"""Generate call ID for requests"""
import uuid
return str(uuid.uuid4())
def get_discovery_status(self) -> Dict[str, Any]:
"""Get discovery service status"""
status = {
'bootstrap_available': self._bootstrap_config is not None,
'manifest_available': self._manifest_config is not None,
'openid_available': self._openid_config is not None,
}
if self._bootstrap_config:
status['bootstrap'] = {
'client_model': self._bootstrap_config.client_model,
'device_model': self._bootstrap_config.device_model,
'has_device_tokens_url': bool(self._bootstrap_config.device_tokens_url),
'has_openid_config_url': bool(self._bootstrap_config.openid_config_url),
'has_taa_url': bool(self._bootstrap_config.taa_url),
}
if self._manifest_config:
device_token = self._manifest_config.get_device_token()
status['manifest'] = {
'mpx_account_pid': self._manifest_config.mpx.account_pid,
'mpx_account_uri': self._manifest_config.mpx.get_account_uri(),
'feed_count': len(self._manifest_config.mpx.feeds),
'has_device_token': bool(device_token),
'device_token_preview': device_token[:20] + '...' if device_token else None,
'drm_endpoints': {
'widevine': bool(self._manifest_config.drm.widevine_license_url),
'vod_widevine': bool(self._manifest_config.drm.vod_widevine_license_url),
'fairplay': bool(self._manifest_config.drm.fairplay_license_url),
},
'tvhub_count': len(self._manifest_config.tv_hubs.base_urls),
}
if self._openid_config:
status['openid'] = {
'has_token_endpoint': bool(self._openid_config.token_endpoint),
'has_authorization_endpoint': bool(self._openid_config.authorization_endpoint),
}
# Cache status
now = time.time()
status['cache'] = {
'bootstrap_age': now - self._last_bootstrap if self._last_bootstrap else None,
'manifest_age': now - self._last_manifest if self._last_manifest else None,
'openid_age': now - self._last_openid if self._last_openid else None,
}
return status
def clear_cache(self) -> None:
"""Clear all discovery cache"""
self._bootstrap_config = None
self._manifest_config = None
self._openid_config = None
self._last_bootstrap = None
self._last_manifest = None
self._last_openid = None
logger.info("Discovery cache cleared")
lib/streaming_providers/providers/magenta2/endpoint_manager.py
+230
View File
@@ -0,0 +1,230 @@
# streaming_providers/providers/magenta2/endpoint_manager.py
from typing import Dict, Optional, Any, List
from dataclasses import dataclass
from enum import Enum
import logging
from .config_models import ProviderConfig
from .constants import MAGENTA2_FALLBACK_ENDPOINTS
logger = logging.getLogger(__name__)
class EndpointCategory(Enum):
"""Categories for different types of endpoints"""
AUTHENTICATION = "auth"
CONTENT = "content"
DRM = "drm"
EPG = "epg"
MPX = "mpx"
TVHUBS = "tvhubs"
USER = "user"
@dataclass
class EndpointInfo:
"""Information about a specific endpoint"""
category: EndpointCategory
url: str
requires_auth: bool = False
cache_duration: int = 3600 # seconds
last_verified: Optional[float] = None
is_fallback: bool = False
class EndpointManager:
"""
Manages and resolves all dynamic endpoints for Magenta2 provider
"""
def __init__(self, provider_config: ProviderConfig):
self.config = provider_config
self._endpoints: Dict[str, EndpointInfo] = {}
self._initialize_endpoints()
def _initialize_endpoints(self) -> None:
"""Initialize endpoints from provider configuration"""
self._add_bootstrap_endpoints()
self._add_manifest_endpoints()
self._add_openid_endpoints()
self._add_fallback_endpoints()
def _add_bootstrap_endpoints(self) -> None:
"""Add endpoints from bootstrap configuration"""
bootstrap = self.config.bootstrap
# Authentication endpoints
if bootstrap.taa_url:
self._add_endpoint('taa_auth', EndpointCategory.AUTHENTICATION, bootstrap.taa_url)
if bootstrap.openid_config_url:
self._add_endpoint('openid_config', EndpointCategory.AUTHENTICATION, bootstrap.openid_config_url)
if bootstrap.line_auth_url:
self._add_endpoint('line_auth', EndpointCategory.AUTHENTICATION, bootstrap.line_auth_url)
if bootstrap.remote_login_url:
self._add_endpoint('remote_login', EndpointCategory.AUTHENTICATION, bootstrap.remote_login_url)
# Content endpoints
if bootstrap.device_tokens_url:
self._add_endpoint('device_tokens', EndpointCategory.CONTENT, bootstrap.device_tokens_url)
if bootstrap.account_base_url:
self._add_endpoint('account_base', EndpointCategory.USER, bootstrap.account_base_url)
if bootstrap.consumer_accounts_url:
self._add_endpoint('consumer_accounts', EndpointCategory.USER, bootstrap.consumer_accounts_url)
def _add_manifest_endpoints(self) -> None:
"""Add endpoints from manifest configuration"""
if not self.config.manifest:
return
manifest = self.config.manifest
# MPX endpoints
if manifest.mpx.license_service_url:
self._add_endpoint('mpx_license', EndpointCategory.MPX, manifest.mpx.license_service_url)
if manifest.mpx.selector_service_url:
self._add_endpoint('mpx_selector', EndpointCategory.MPX, manifest.mpx.selector_service_url)
if manifest.mpx.channel_stations_feed:
self._add_endpoint('channel_stations', EndpointCategory.CONTENT, manifest.mpx.channel_stations_feed)
logger.info(f"Channel stations feed found: {manifest.mpx.channel_stations_feed}")
# DRM endpoints
if manifest.drm.widevine_license_url:
self._add_endpoint('widevine_license', EndpointCategory.DRM, manifest.drm.widevine_license_url)
if manifest.drm.vod_widevine_license_url:
self._add_endpoint('vod_widevine_license', EndpointCategory.DRM, manifest.drm.vod_widevine_license_url)
if manifest.drm.fairplay_license_url:
self._add_endpoint('fairplay_license', EndpointCategory.DRM, manifest.drm.fairplay_license_url)
# MPX feeds (resolved with account PID)
for feed_name, feed_template in manifest.mpx.feeds.items():
resolved_url = self.config.get_resolved_feed_url(feed_name)
if resolved_url:
self._add_endpoint(f'mpx_feed_{feed_name}', EndpointCategory.MPX, resolved_url)
# TV Hub URLs (resolved with client model)
for hub_name in manifest.tv_hubs.base_urls.keys():
resolved_url = self.config.get_resolved_tvhub_url(hub_name)
if resolved_url:
self._add_endpoint(f'tvhub_{hub_name}', EndpointCategory.TVHUBS, resolved_url)
def _add_openid_endpoints(self) -> None:
"""Add endpoints from OpenID configuration"""
if not self.config.openid:
return
openid = self.config.openid
if openid.token_endpoint:
self._add_endpoint('oauth_token', EndpointCategory.AUTHENTICATION, openid.token_endpoint,
requires_auth=False)
if openid.authorization_endpoint:
self._add_endpoint('oauth_authorize', EndpointCategory.AUTHENTICATION, openid.authorization_endpoint)
if openid.userinfo_endpoint:
self._add_endpoint('userinfo', EndpointCategory.USER, openid.userinfo_endpoint, requires_auth=True)
if openid.revocation_endpoint:
self._add_endpoint('oauth_revoke', EndpointCategory.AUTHENTICATION, openid.revocation_endpoint)
def _add_fallback_endpoints(self) -> None:
"""Add fallback endpoints for critical functionality"""
fallbacks = {
'openid_config': MAGENTA2_FALLBACK_ENDPOINTS['OPENID_CONFIG'],
'taa_auth': MAGENTA2_FALLBACK_ENDPOINTS['TAA_AUTH'],
'entitlement': MAGENTA2_FALLBACK_ENDPOINTS['ENTITLEMENT'],
'mpx_license': 'https://license.entitlement.theplatform.eu/license/web/ContentAccessRules/getApplicableDistributionRights',
'mpx_selector': 'https://link.api.eu.theplatform.com/s/',
'widevine_license': 'https://widevine.entitlement.theplatform.eu/wv/web/ModularDrm/getRawWidevineLicense',
}
for endpoint_name, url in fallbacks.items():
# Only add fallback if we don't already have this endpoint
if endpoint_name not in self._endpoints:
self._add_endpoint(endpoint_name, EndpointCategory.AUTHENTICATION, url, is_fallback=True)
def _add_endpoint(self, name: str, category: EndpointCategory, url: str,
requires_auth: bool = False, is_fallback: bool = False) -> None:
"""Add an endpoint to the manager"""
self._endpoints[name] = EndpointInfo(
category=category,
url=url,
requires_auth=requires_auth,
is_fallback=is_fallback
)
logger.debug(f"Added endpoint: {name} -> {url} (category: {category.value}, fallback: {is_fallback})")
def get_endpoint(self, name: str) -> Optional[str]:
"""Get endpoint URL by name"""
endpoint_info = self._endpoints.get(name)
return endpoint_info.url if endpoint_info else None
def get_endpoint_info(self, name: str) -> Optional[EndpointInfo]:
"""Get complete endpoint information by name"""
return self._endpoints.get(name)
def get_endpoints_by_category(self, category: EndpointCategory) -> Dict[str, str]:
"""Get all endpoints for a specific category"""
return {
name: info.url
for name, info in self._endpoints.items()
if info.category == category
}
def has_endpoint(self, name: str) -> bool:
"""Check if endpoint exists"""
return name in self._endpoints
def is_fallback_endpoint(self, name: str) -> bool:
"""Check if endpoint is a fallback"""
endpoint_info = self._endpoints.get(name)
return endpoint_info.is_fallback if endpoint_info else False
def get_all_endpoints(self) -> Dict[str, EndpointInfo]:
"""Get all endpoints"""
return self._endpoints.copy()
def get_stats(self) -> Dict[str, Any]:
"""Get endpoint manager statistics"""
total = len(self._endpoints)
by_category = {}
fallback_count = 0
for info in self._endpoints.values():
category = info.category.value
by_category[category] = by_category.get(category, 0) + 1
if info.is_fallback:
fallback_count += 1
return {
'total_endpoints': total,
'endpoints_by_category': by_category,
'fallback_endpoints': fallback_count,
'dynamic_endpoints': total - fallback_count,
'is_complete': self.config.is_complete
}
def validate_critical_endpoints(self) -> List[str]:
"""Validate that all critical endpoints are available"""
critical_endpoints = [
'taa_auth',
'openid_config',
'mpx_license',
'widevine_license'
]
missing = []
for endpoint in critical_endpoints:
if not self.has_endpoint(endpoint):
missing.append(endpoint)
return missing
lib/streaming_providers/providers/magenta2/models.py
+209
View File
@@ -0,0 +1,209 @@
# streaming_providers/providers/magenta2/models.py
from dataclasses import dataclass, field
from typing import Dict, Optional
import json
from ...base.models import StreamingChannel
# streaming_providers/providers/magenta2/models.py
# Add this exception if not already present
class DeviceLimitExceededException(Exception):
"""Exception raised when device limit is exceeded"""
pass
class Magenta2PlaybackRestrictedException(Exception):
"""Exception raised when playback is restricted for content"""
pass
@dataclass
class Magenta2Channel:
"""
Represents a Magenta2 channel with all necessary streaming data
"""
# Core identification
name: str
channel_id: str
# Visual elements
logo_url: Optional[str] = None
# Streaming configuration
mode: str = "live" # "live" or "vod"
session_manifest: bool = False
manifest: Optional[str] = None
manifest_script: Optional[str] = None
# CDM (Content Decryption Module) settings
cdm_type: Optional[str] = None
use_cdm: bool = True
cdm: Optional[str] = None # Usually "pid={pid}"
cdm_mode: str = 'external'
# Video settings
video: str = 'best'
on_demand: bool = True
speed_up: bool = True
# Additional metadata
content_type: str = 'LIVE' # 'LIVE' or 'VOD'
description: Optional[str] = None
genre: Optional[str] = None
language: str = 'de'
country: str = 'DE'
# Streaming data
license_url: Optional[str] = None
certificate_url: Optional[str] = None
streaming_format: Optional[str] = None
# Internal tracking
raw_data: Dict = field(default_factory=dict)
@classmethod
def from_api_data(cls, api_data: Dict, **kwargs) -> 'Magenta2Channel':
"""
Create Magenta2Channel from API response data
Args:
api_data: Raw API response data
**kwargs: Additional parameters to override defaults
Returns:
Magenta2Channel instance
"""
channel = cls(
name=api_data.get('title', api_data.get('name', 'Unknown Channel')),
channel_id=api_data.get('id', ''),
content_type=api_data.get('type', 'LIVE'),
raw_data=api_data.copy()
)
# Apply any additional parameters
for key, value in kwargs.items():
if hasattr(channel, key):
setattr(channel, key, value)
return channel
def set_streaming_data(self, manifest: str, cdm_type: str = None,
pid: str = None, license_url: str = None,
certificate_url: str = None, streaming_format: str = None) -> None:
"""
Configure streaming-specific data
Args:
manifest: Manifest URL for streaming
cdm_type: Content decryption module type
pid: Program ID for CDM
license_url: License URL for DRM
certificate_url: Certificate URL for DRM
streaming_format: Streaming format (e.g., 'dash')
"""
self.manifest = manifest
if cdm_type:
self.cdm_type = cdm_type
if pid:
self.cdm = f"pid={pid}"
if license_url:
self.license_url = license_url
if certificate_url:
self.certificate_url = certificate_url
if streaming_format:
self.streaming_format = streaming_format
def set_logo(self, logo_url: str) -> None:
"""Set channel logo URL"""
self.logo_url = logo_url
def set_metadata(self, description: str = None, genre: str = None) -> None:
"""Set additional metadata"""
if description:
self.description = description
if genre:
self.genre = genre
def is_live(self) -> bool:
"""Check if channel is live TV"""
return self.content_type == 'LIVE' and self.mode == 'live'
def is_vod(self) -> bool:
"""Check if channel is video on demand"""
return self.content_type == 'VOD' or self.mode == 'vod'
def to_streaming_channel(self, provider_name: str = 'magenta2') -> StreamingChannel:
"""
Convert to generic StreamingChannel object
Args:
provider_name: Provider name to set
Returns:
StreamingChannel instance
"""
return StreamingChannel(
name=self.name,
channel_id=self.channel_id,
provider=provider_name,
logo_url=self.logo_url,
mode=self.mode,
session_manifest=self.session_manifest,
manifest=self.manifest,
manifest_script=self.manifest_script,
cdm_type=self.cdm_type,
use_cdm=self.use_cdm,
cdm=self.cdm,
cdm_mode=self.cdm_mode,
video=self.video,
on_demand=self.on_demand,
speed_up=self.speed_up,
content_type=self.content_type,
description=self.description,
genre=self.genre,
language=self.language,
country=self.country,
license_url=self.license_url,
certificate_url=self.certificate_url,
streaming_format=self.streaming_format
)
def to_dict(self) -> Dict:
"""
Convert to dictionary format matching your output specification
Returns:
Dictionary in the format expected by your application
"""
return {
'Name': self.name,
'LogoUrl': self.logo_url,
'Mode': self.mode,
'SessionManifest': self.session_manifest,
'Manifest': self.manifest,
'ManifestScript': self.manifest_script,
'CdmType': self.cdm_type,
'UseCdm': self.use_cdm,
'Cdm': self.cdm,
'CdmMode': self.cdm_mode,
'Video': self.video,
'OnDemand': self.on_demand,
'SpeedUp': self.speed_up
}
def to_json(self, indent: int = 2) -> str:
"""Convert to JSON string"""
return json.dumps(self.to_dict(), indent=indent, ensure_ascii=False)
def __str__(self) -> str:
return f"Magenta2Channel(name='{self.name}', id='{self.channel_id}', type='{self.content_type}')"
def __repr__(self) -> str:
return self.__str__()
lib/streaming_providers/providers/magenta2/provider.py
+1041
View File
File diff suppressed because it is too large Load Diff
lib/streaming_providers/providers/magenta2/remote_login_handler.py
+352
View File
@@ -0,0 +1,352 @@
# streaming_providers/providers/magenta2/remote_login_handler.py
"""
Remote Login Handler for Magenta2 Backchannel Authentication
Implements QR code-based authentication flow as fallback when line auth fails
"""
import time
from typing import Dict, Optional, Any
from dataclasses import dataclass
from ...base.network import HTTPManager
from ...base.utils.logger import logger
from ...base.ui import NotificationFactory, NotificationInterface, NotificationResult
from .constants import (
SSO_USER_AGENT,
DEFAULT_REQUEST_TIMEOUT,
GRANT_TYPES,
)
@dataclass
class RemoteLoginSession:
"""Remote login session data"""
initial_login_code: str
auth_req_id: str
auth_req_sec: str
interval: int
expires_in: int
qr_code_url: str
started_at: float
class RemoteLoginHandler:
"""
Handles the complete backchannel authentication / remote login flow
Flow:
1. Start backchannel auth -> get login code and QR URL
2. Display QR code to user (via notification adapter)
3. Poll token endpoint until user completes mobile authentication
4. Handle countdown internally with notification updates
"""
def __init__(self, http_manager: HTTPManager, sam3_client_id: str,
backchannel_start_url: str, token_endpoint: str,
qr_code_url_template: str,
notifier: Optional[NotificationInterface] = None):
"""
Initialize remote login handler
Args:
http_manager: HTTP manager for requests
sam3_client_id: SAM3 client ID
backchannel_start_url: Backchannel auth start endpoint
token_endpoint: OAuth token endpoint for polling
qr_code_url_template: QR code URL template with {code} placeholder
notifier: Optional notification interface (auto-created if None)
"""
self.http_manager = http_manager
self.sam3_client_id = sam3_client_id
self.backchannel_start_url = backchannel_start_url
self.token_endpoint = token_endpoint
self.qr_code_url_template = qr_code_url_template
# Get or create notifier with http_manager
if notifier:
self._notifier = notifier
else:
self._notifier = NotificationFactory.create(http_manager=http_manager)
self._current_session: Optional[RemoteLoginSession] = None
logger.debug(f"RemoteLoginHandler initialized with {self._notifier.__class__.__name__}")
def set_notifier(self, notifier: NotificationInterface) -> None:
"""
Set custom notification interface
Args:
notifier: Notification interface to use
"""
self._notifier = notifier
logger.debug(f"Notifier set to: {notifier.__class__.__name__}")
def start_remote_login(self, scope: str = "tvhubs offline_access") -> RemoteLoginSession:
"""
Start backchannel authentication flow
Args:
scope: OAuth scopes to request
Returns:
RemoteLoginSession with login code and polling parameters
Raises:
Exception: If backchannel auth start fails
"""
try:
logger.info("Starting remote login (backchannel auth)")
# Build request payload
payload = {
'client_id': self.sam3_client_id,
'scope': scope
}
headers = {
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
'User-Agent': SSO_USER_AGENT
}
# Start backchannel auth
response = self.http_manager.post(
self.backchannel_start_url,
operation='backchannel_start',
headers=headers,
data=payload,
timeout=DEFAULT_REQUEST_TIMEOUT
)
response.raise_for_status()
data = response.json()
# Extract session data
initial_login_code = data.get('initial_login_code')
auth_req_id = data.get('auth_req_id')
auth_req_sec = data.get('auth_req_sec')
interval = int(data.get('interval', 10))
expires_in = int(data.get('expires_in', 300))
if not all([initial_login_code, auth_req_id, auth_req_sec]):
raise Exception("Incomplete backchannel auth response")
# Build QR code URL
qr_code_url = self.qr_code_url_template.format(code=initial_login_code)
# Create session
session = RemoteLoginSession(
initial_login_code=initial_login_code,
auth_req_id=auth_req_id,
auth_req_sec=auth_req_sec,
interval=interval,
expires_in=expires_in,
qr_code_url=qr_code_url,
started_at=time.time()
)
self._current_session = session
logger.info(
f"Remote login started: code={initial_login_code}, "
f"interval={interval}s, expires_in={expires_in}s"
)
return session
except Exception as e:
logger.error(f"Failed to start remote login: {e}")
raise Exception(f"Remote login start failed: {e}")
def poll_for_token(self, session: RemoteLoginSession) -> Optional[Dict[str, Any]]:
"""
Poll token endpoint until user completes authentication or timeout
NOW HANDLES COUNTDOWN UPDATES INTERNALLY
Args:
session: Active remote login session
Returns:
Token data dict if successful, None if timed out/cancelled
Raises:
Exception: If polling fails with unexpected error
"""
try:
logger.info("Starting token polling for remote login")
start_time = session.started_at
next_poll_time = start_time
last_countdown_update = start_time
headers = {
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
'User-Agent': SSO_USER_AGENT
}
payload = {
'client_id': self.sam3_client_id,
'grant_type': GRANT_TYPES['REMOTE_LOGIN'],
'auth_req_id': session.auth_req_id,
'auth_req_sec': session.auth_req_sec
}
poll_count = 0
max_polls = session.expires_in // session.interval + 1
while True:
current_time = time.time()
elapsed = current_time - start_time
remaining = max(0, session.expires_in - elapsed)
# Check if session expired
if elapsed >= session.expires_in:
logger.warning(
f"Remote login session expired after {elapsed:.1f}s "
f"(limit: {session.expires_in}s)"
)
self._notifier.close(success=False, message="Session expired")
return None
# Update countdown display (every second or as needed)
if current_time - last_countdown_update >= 1.0:
# Check if user cancelled
if self._notifier.is_cancelled():
logger.info("User cancelled remote login")
self._notifier.close(success=False, message="Cancelled by user")
return None
# Update countdown
if not self._notifier.update_countdown(int(remaining)):
logger.info("Countdown update returned False - user cancelled")
self._notifier.close(success=False, message="Cancelled by user")
return None
last_countdown_update = current_time
# Wait until next poll time
if current_time < next_poll_time:
sleep_time = min(1.0, next_poll_time - current_time) # Sleep max 1 second for countdown updates
time.sleep(sleep_time)
continue
# Perform poll
poll_count += 1
logger.debug(
f"Polling attempt {poll_count}/{max_polls} "
f"(remaining: {remaining:.0f}s)"
)
try:
response = self.http_manager.post(
self.token_endpoint,
operation='remote_login_poll',
headers=headers,
data=payload,
timeout=DEFAULT_REQUEST_TIMEOUT
)
# 202 = User hasn't completed authentication yet
if response.status_code == 202:
logger.debug("Authentication not yet completed (202)")
next_poll_time = time.time() + session.interval
continue
# Success - user completed authentication
if response.status_code == 200:
token_data = response.json()
logger.info(
f"✓ Remote login successful after {elapsed:.1f}s "
f"({poll_count} polls)"
)
self._notifier.close(success=True)
return token_data
# Other status codes = error
response.raise_for_status()
except Exception as e:
# If we get an error during polling, check if we should retry
if elapsed < session.expires_in:
logger.debug(f"Poll error (will retry): {e}")
next_poll_time = time.time() + session.interval
continue
else:
# Session expired, give up
raise
except Exception as e:
logger.error(f"Remote login polling failed: {e}")
self._notifier.close(success=False, message=str(e))
raise Exception(f"Remote login polling failed: {e}")
def perform_complete_flow(self, scope: str = "tvhubs offline_access") -> Optional[Dict[str, Any]]:
"""
Perform complete remote login flow:
1. Start session
2. Display QR code via notifier
3. Poll for completion with automatic countdown updates
Args:
scope: OAuth scopes to request
Returns:
Token data dict if successful, None if failed/timeout/cancelled
"""
try:
# Step 1: Start session
session = self.start_remote_login(scope)
# Step 2: Display QR code to user via notifier
result = self._notifier.show_remote_login(
login_code=session.initial_login_code,
qr_url=session.qr_code_url,
expires_in=session.expires_in,
interval=session.interval
)
if result != NotificationResult.CONTINUE:
logger.warning(f"Failed to show notification: {result}")
return None
# Step 3: Poll for completion (handles countdown internally)
token_data = self.poll_for_token(session)
if token_data:
logger.info("✓ Remote login flow completed successfully")
else:
logger.warning("Remote login flow timed out or was cancelled")
return token_data
except Exception as e:
logger.error(f"Remote login flow failed: {e}")
self._notifier.close(success=False, message=str(e))
return None
finally:
self._current_session = None
def cancel_current_session(self) -> None:
"""Cancel current remote login session"""
if self._current_session:
logger.info("Cancelling remote login session")
self._current_session = None
self._notifier.close(success=False, message="Cancelled")
def get_session_status(self) -> Optional[Dict[str, Any]]:
"""Get current session status"""
if not self._current_session:
return None
session = self._current_session
elapsed = time.time() - session.started_at
remaining = max(0, session.expires_in - elapsed)
return {
'login_code': session.initial_login_code,
'qr_code_url': session.qr_code_url,
'elapsed_seconds': elapsed,
'remaining_seconds': remaining,
'is_expired': remaining <= 0,
'interval': session.interval,
'is_cancelled': self._notifier.is_cancelled()
}
lib/streaming_providers/providers/magenta2/sam3_client.py
+635
View File
@@ -0,0 +1,635 @@
# streaming_providers/providers/magenta2/sam3_client.py
import re
from typing import Dict, Optional, Any, List
from dataclasses import dataclass
from urllib.parse import urlparse, parse_qs
from ...base.network import HTTPManager
from ...base.utils.logger import logger
from .constants import (
SSO_USER_AGENT,
DEFAULT_REQUEST_TIMEOUT,
GRANT_TYPES,
)
@dataclass
class Sam3AuthMethod:
"""SAM3 authentication method"""
name: str
enabled: bool = False
@dataclass
class Sam3AuthMethods:
"""Available SAM3 authentication methods"""
password: bool = False
code: bool = False
line: bool = False
@dataclass
class Sam3FormField:
"""HTML form field extracted from SAM3 login pages"""
name: str
value: str
type: str = "hidden"
class Sam3Client:
"""
SAM3 authentication client implementing the complete login flow from C++ code
"""
def __init__(self, http_manager: HTTPManager, session_id: str, device_id: str,
sam3_client_id: str, issuer_url: str = None,
oauth_token_endpoint: str = None, line_auth_endpoint: str = None,
backchannel_start_url: str = None, qr_code_url_template: str = None):
"""
Initialize SAM3 client with all required endpoints
"""
self.http_manager = http_manager
self.session_id = session_id
self.device_id = device_id
self.sam3_client_id = sam3_client_id
# STORE ALL ENDPOINTS FOR DIFFERENT AUTH FLOWS
self.issuer_url = issuer_url # For user auth flow (username/password)
self.oauth_token_endpoint = oauth_token_endpoint # For OAuth flows
self.line_auth_endpoint = line_auth_endpoint # For device line auth
self.token_endpoint = line_auth_endpoint # Backwards compatibility
# Remote login endpoints
self.backchannel_start_url = backchannel_start_url
self.qr_code_url_template = qr_code_url_template
# Other SAM3 endpoints (will be updated from OpenID config)
self.authorization_endpoint: Optional[str] = None
self.userinfo_endpoint: Optional[str] = None
# Authentication state
self.auth_methods = Sam3AuthMethods()
self.form_fields: List[Sam3FormField] = []
self.refresh_token: Optional[str] = None
self.access_tokens: Dict[str, str] = {} # scope -> token
self.__last_line_auth_response: Optional[Dict[str, Any]] = None
# Remote login handler (lazy initialized)
self._remote_login_handler: Optional['RemoteLoginHandler'] = None
logger.debug(
f"SAM3 client initialized - "
f"Issuer: {issuer_url}, "
f"OAuth: {oauth_token_endpoint}, "
f"Line: {line_auth_endpoint}, "
f"Backchannel: {backchannel_start_url}"
)
# ========================================================================
# Endpoint Resolution Helpers
# ========================================================================
def _get_token_endpoint(self) -> str:
"""Get the appropriate token endpoint with fallback logic"""
return self.oauth_token_endpoint or self.token_endpoint or self.line_auth_endpoint
def _get_line_auth_endpoint(self) -> str:
"""Get line auth endpoint with fallback"""
return self.line_auth_endpoint or self.token_endpoint
# ========================================================================
# Common Token Request Method
# ========================================================================
def _make_token_request(self, operation: str, payload: Dict[str, Any],
endpoint_override: str = None) -> Dict[str, Any]:
"""
Common method for all token requests
Args:
operation: Operation name for logging
payload: Request payload (will be form-encoded)
endpoint_override: Optional endpoint override
Returns:
Token response data
"""
endpoint = endpoint_override or self._get_token_endpoint()
if not endpoint:
raise Exception("No token endpoint available")
headers = {
'Content-Type': 'application/x-www-form-urlencoded',
'User-Agent': SSO_USER_AGENT
}
response = self.http_manager.post(
endpoint,
operation=operation,
headers=headers,
data=payload,
timeout=DEFAULT_REQUEST_TIMEOUT
)
response.raise_for_status()
data = response.json()
# Store refresh token if provided
if 'refresh_token' in data:
self.refresh_token = data['refresh_token']
logger.debug(f"Refresh token stored/updated from {operation}")
return data
# ========================================================================
# Authentication Methods Discovery
# ========================================================================
def discover_auth_methods(self, line_auth_url: str) -> bool:
"""
Discover available authentication methods
Matching C++ GetAuthMethods()
"""
try:
logger.debug("Discovering SAM3 authentication methods")
headers = {
'User-Agent': SSO_USER_AGENT,
'Accept': 'application/json'
}
response = self.http_manager.get(
line_auth_url,
operation='auth_methods',
headers=headers,
timeout=DEFAULT_REQUEST_TIMEOUT
)
response.raise_for_status()
data = response.json()
content = data.get('content', {})
if 'supportedAuthenticationKinds' in content:
auth_kinds = content['supportedAuthenticationKinds']
for method in auth_kinds:
if method == GRANT_TYPES['PASSWORD']:
self.auth_methods.password = True
elif method == GRANT_TYPES['AUTH_CODE']:
self.auth_methods.code = True
elif method == GRANT_TYPES['LINE_AUTH']:
self.auth_methods.line = True
logger.info(
f"SAM3 auth methods: password={self.auth_methods.password}, "
f"code={self.auth_methods.code}, line={self.auth_methods.line}"
)
return True
except Exception as e:
logger.error(f"Failed to discover SAM3 auth methods: {e}")
return False
# ========================================================================
# User Authentication Flow (Username/Password)
# ========================================================================
def sam3_login(self, username: str, password: str) -> Dict[str, str]:
"""
Complete SAM3 login flow matching C++ Sam3Client::Sam3Login()
"""
try:
logger.info("Starting SAM3 login flow")
# Step 1: Get initial authorization page
auth_url = self._build_authorization_url()
logger.debug(f"Step 1: Getting authorization page: {auth_url}")
initial_response = self.http_manager.get(
auth_url,
operation='sam3_auth',
headers=self._get_sso_headers(),
timeout=DEFAULT_REQUEST_TIMEOUT,
allow_redirects=False
)
# Step 2: Parse HTML and extract form fields
self._parse_html_form_fields(initial_response.text)
logger.debug(f"Step 2: Extracted {len(self.form_fields)} form fields")
# Step 3: Submit username to factorx endpoint
factorx_url = f"{self.issuer_url}/factorx" if self.issuer_url else "https://login.telekom-dienste.de/factorx"
username_data = self._build_username_payload(username)
logger.debug(f"Step 3: Submitting username to {factorx_url}")
username_response = self.http_manager.post(
factorx_url,
operation='sam3_username',
headers=self._get_sso_headers(),
data=username_data,
timeout=DEFAULT_REQUEST_TIMEOUT,
allow_redirects=False
)
# Step 4: Parse response and extract updated form fields
self._parse_html_form_fields(username_response.text)
logger.debug(f"Step 4: Updated form fields: {len(self.form_fields)}")
# Step 5: Submit password to factorx endpoint
password_data = self._build_password_payload(username, password)
logger.debug(f"Step 5: Submitting password to {factorx_url}")
password_response = self.http_manager.post(
factorx_url,
operation='sam3_password',
headers=self._get_sso_headers(),
data=password_data,
timeout=DEFAULT_REQUEST_TIMEOUT,
allow_redirects=False
)
# Step 6: Extract authorization code from redirect
redirect_url = password_response.headers.get('Location', '')
logger.debug(f"Step 6: Redirect URL: {redirect_url}")
if not redirect_url:
# Follow redirects manually if needed
final_response = self.http_manager.get(
factorx_url,
operation='sam3_final',
headers=self._get_sso_headers(),
timeout=DEFAULT_REQUEST_TIMEOUT,
allow_redirects=True
)
redirect_url = final_response.url
# Step 7: Extract code and state from redirect URL
code, state = self._extract_code_and_state(redirect_url)
if not code or not state:
raise Exception("Could not extract authorization code and state from redirect")
logger.info("SAM3 login completed successfully")
return {
'code': code,
'state': state,
'redirect_url': redirect_url
}
except Exception as e:
logger.error(f"SAM3 login failed: {e}")
raise Exception(f"SAM3 login failed: {e}")
# ========================================================================
# Line Authentication (Device Token)
# ========================================================================
def get_last_line_auth_response(self) -> Optional[Dict[str, Any]]:
"""Get the last line auth response data"""
return self.__last_line_auth_response
def line_auth(self, device_token: str) -> bool:
"""
Line authentication using device token
"""
try:
endpoint = self._get_line_auth_endpoint()
if not endpoint:
raise Exception("No line auth endpoint available")
logger.debug("Performing line authentication with device token")
payload = {
'grant_type': GRANT_TYPES['LINE_AUTH'],
'client_id': self.sam3_client_id,
'token': device_token,
'scope': 'tvhubs offline_access'
}
# Use common token request method
data = self._make_token_request('line_auth', payload, endpoint)
# Store the actual response data
self.__last_line_auth_response = data
if 'refresh_token' in data:
logger.info("Line authentication successful, refresh token obtained")
return True
logger.warning("Line authentication succeeded but no refresh token received")
return False
except Exception as e:
logger.error(f"Line authentication failed: {e}")
return False
# ========================================================================
# Remote Login (Backchannel Authentication)
# ========================================================================
def _get_remote_login_handler(self) -> Optional['RemoteLoginHandler']:
"""
Get or create remote login handler
Returns:
RemoteLoginHandler if endpoints available, None otherwise
"""
# Return existing handler if available
if self._remote_login_handler:
return self._remote_login_handler
# Check if we have required endpoints
if not all([
self.backchannel_start_url,
self._get_token_endpoint(),
self.qr_code_url_template
]):
logger.debug("Remote login not available - missing endpoints")
return None
# Import here to avoid circular dependency
from .remote_login_handler import RemoteLoginHandler
# Create handler
self._remote_login_handler = RemoteLoginHandler(
http_manager=self.http_manager,
sam3_client_id=self.sam3_client_id,
backchannel_start_url=self.backchannel_start_url,
token_endpoint=self._get_token_endpoint(),
qr_code_url_template=self.qr_code_url_template
)
logger.info("✓ Remote login handler initialized")
return self._remote_login_handler
def can_use_remote_login(self) -> bool:
"""Check if remote login is available"""
return self._get_remote_login_handler() is not None
def remote_login(self, scope: str = "tvhubs offline_access") -> Optional[Dict[str, Any]]:
"""
Perform complete remote login (backchannel auth) flow
Args:
scope: OAuth scopes to request
Returns:
Token data dict if successful, None if failed/timeout/cancelled
"""
handler = self._get_remote_login_handler()
if not handler:
logger.error("Remote login not available")
return None
# Perform complete flow (notifier is already set in handler)
token_data = handler.perform_complete_flow(scope)
# Store refresh token if we got one
if token_data and 'refresh_token' in token_data:
self.refresh_token = token_data['refresh_token']
logger.info("✓ Remote login successful, refresh token obtained")
return token_data
def get_remote_login_status(self) -> Optional[Dict[str, Any]]:
"""Get current remote login session status"""
handler = self._get_remote_login_handler()
if not handler:
return None
return handler.get_session_status()
# ========================================================================
# Generic Token Operations
# ========================================================================
def get_token(self, grant_type: str, scope: str, credential1: str = "",
credential2: str = "") -> str:
"""
Generic token acquisition
"""
try:
logger.debug(f"Getting token with grant_type: {grant_type}, scope: {scope}")
payload = {
'grant_type': grant_type,
'client_id': self.sam3_client_id
}
# Add credentials based on grant type
if grant_type == GRANT_TYPES['REFRESH_TOKEN']:
payload['refresh_token'] = credential1
payload['scope'] = f"{scope} offline_access"
elif grant_type == GRANT_TYPES['REMOTE_LOGIN']:
payload['auth_req_id'] = credential1
payload['auth_req_sec'] = credential2
# Use common token request method
data = self._make_token_request('get_token', payload)
access_token = data.get('access_token')
if not access_token:
raise Exception("No access token in token response")
# Store token by scope
self.access_tokens[scope] = access_token
logger.debug(f"Token obtained for scope: {scope}")
return access_token
except Exception as e:
logger.error(f"Token acquisition failed: {e}")
raise
def refresh_access_token(self, scope: str) -> str:
"""
Refresh access token
Matching C++ Sam3Client::RefreshToken()
"""
if not self.refresh_token:
raise Exception("No refresh token available")
return self.get_token(GRANT_TYPES['REFRESH_TOKEN'], scope, self.refresh_token)
def get_access_token(self, scope: str) -> str:
"""
Get access token for specific scope with line auth fallback
"""
# Return cached token if available
if scope in self.access_tokens:
return self.access_tokens[scope]
# Try to use line auth established session first
if self.refresh_token:
try:
token = self.refresh_access_token(scope)
if token:
return token
except Exception as e:
logger.warning(f"Token refresh failed for scope {scope}: {e}")
# If no refresh token available, we can't get an access token
logger.warning(f"No access token available for scope {scope} - line auth may be needed")
return ""
# ========================================================================
# HTML Form Parsing Helpers
# ========================================================================
def _build_authorization_url(self) -> str:
"""Build authorization URL matching C++ implementation"""
if not self.authorization_endpoint:
# Fallback to standard URL
return "https://login.telekom-dienste.de/oauth2/auth"
params = {
'client_id': self.sam3_client_id,
'redirect_uri': 'https://web2.magentatv.de/authn/idm',
'response_type': 'code',
'scope': 'openid offline_access'
}
param_string = '&'.join([f"{k}={v}" for k, v in params.items()])
return f"{self.authorization_endpoint}?{param_string}"
def _parse_html_form_fields(self, html_content: str) -> None:
"""
Parse HTML form fields matching C++ ParseHtml()
"""
self.form_fields.clear()
# Look for form with id="login" or similar
form_start = html_content.find('form id="login"')
if form_start == -1:
form_start = html_content.find('<form')
if form_start == -1:
logger.warning("No form found in HTML content")
return
form_end = html_content.find('</form>', form_start)
if form_end == -1:
logger.warning("No form end tag found")
return
form_html = html_content[form_start:form_end]
# Find all hidden input fields
pattern = r'<input[^>]*type="hidden"[^>]*name="([^"]*)"[^>]*value="([^"]*)"[^>]*>'
matches = re.findall(pattern, form_html, re.IGNORECASE)
for name, value in matches:
self.form_fields.append(Sam3FormField(name=name, value=value))
logger.debug(f"Found form field: {name} = {value}")
# Also look for input fields without explicit type="hidden"
pattern_all = r'<input[^>]*name="([^"]*)"[^>]*value="([^"]*)"[^>]*>'
all_matches = re.findall(pattern_all, form_html, re.IGNORECASE)
for name, value in all_matches:
# Skip duplicates
if not any(field.name == name for field in self.form_fields):
self.form_fields.append(Sam3FormField(name=name, value=value))
logger.debug(f"Found additional form field: {name} = {value}")
def _build_username_payload(self, username: str) -> str:
"""Build username submission payload"""
payload_parts = []
# Add all hidden fields
for field in self.form_fields:
payload_parts.append(f"{field.name}={field.value}")
# Add username field
payload_parts.append(f"pw_usr={self._url_encode(username)}")
payload_parts.append("pw_submit=")
payload_parts.append("hidden_pwd=")
return "&".join(payload_parts)
def _build_password_payload(self, username: str, password: str) -> str:
"""Build password submission payload"""
payload_parts = []
# Add all hidden fields (they might have changed)
for field in self.form_fields:
payload_parts.append(f"{field.name}={field.value}")
# Add password fields
payload_parts.append(f"hidden_usr={self._url_encode(username)}")
payload_parts.append(f"pw_pwd={self._url_encode(password)}")
payload_parts.append("pw_submit=")
return "&".join(payload_parts)
def _extract_code_and_state(self, redirect_url: str) -> tuple[str, str]:
"""Extract code and state from redirect URL"""
try:
parsed = urlparse(redirect_url)
query_params = parse_qs(parsed.query)
code = query_params.get('code', [None])[0]
state = query_params.get('state', [None])[0]
if code and state:
logger.debug(f"Extracted code: {code[:8]}..., state: {state}")
return code, state
# Also check fragment (#) for SPA redirects
if '#' in redirect_url:
fragment = redirect_url.split('#')[1]
fragment_params = parse_qs(fragment)
code = fragment_params.get('code', [None])[0]
state = fragment_params.get('state', [None])[0]
return code, state
except Exception as e:
logger.error(f"Failed to extract code and state from URL: {e}")
return None, None
@staticmethod
def _get_sso_headers() -> Dict[str, str]:
"""Get headers for SSO requests"""
return {
'User-Agent': SSO_USER_AGENT,
'Content-Type': 'application/x-www-form-urlencoded',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Accept-Language': 'en-US,en;q=0.5',
'Accept-Encoding': 'gzip, deflate, br',
'Origin': 'https://web2.magentatv.de',
'Referer': 'https://web2.magentatv.de/'
}
@staticmethod
def _url_encode(value: str) -> str:
"""URL encode a string"""
from urllib.parse import quote
return quote(value)
# ========================================================================
# Configuration Updates
# ========================================================================
def update_endpoints(self, openid_config: Dict[str, Any]) -> None:
"""Update endpoints from OpenID configuration"""
# Update issuer URL from OpenID config
if 'issuer' in openid_config:
self.issuer_url = openid_config['issuer']
# Update other endpoints for user authentication flows
self.authorization_endpoint = openid_config.get('authorization_endpoint')
self.userinfo_endpoint = openid_config.get('userinfo_endpoint')
# Get backchannel auth start endpoint
if 'backchannel_authentication_endpoint' in openid_config:
self.backchannel_start_url = openid_config['backchannel_authentication_endpoint']
logger.debug(f"Backchannel auth endpoint from OpenID: {self.backchannel_start_url}")
# OAuth token endpoint might be different from line auth endpoint
if 'token_endpoint' in openid_config:
self.oauth_token_endpoint = openid_config['token_endpoint']
logger.debug(f"Updated SAM3 endpoints: "
f"issuer={self.issuer_url}, "
f"auth={bool(self.authorization_endpoint)}, "
f"oauth_token={bool(self.oauth_token_endpoint)}, "
f"line_auth={bool(self.line_auth_endpoint)}, "
f"backchannel={bool(self.backchannel_start_url)}")
lib/streaming_providers/providers/magenta2/sso_client.py
+219
View File
@@ -0,0 +1,219 @@
# streaming_providers/providers/magenta2/sso_client.py
from typing import Dict, Optional, Any
from ...base.network import HTTPManager
from ...base.utils.logger import logger
from .constants import (
SSO_URL,
SSO_USER_AGENT,
DEFAULT_REQUEST_TIMEOUT
)
class SsoClient:
"""
SSO (Single Sign-On) client implementing the complete SSO flow from C++ code
"""
def __init__(self, http_manager: HTTPManager, session_id: str, device_id: str):
self.http_manager = http_manager
self.session_id = session_id
self.device_id = device_id
def sso_login(self) -> str:
"""
Get SSO login redirect URL
Matching C++ SsoClient::SSOLogin()
"""
try:
logger.debug("Getting SSO login redirect URL")
url = f"{SSO_URL}login"
headers = self._get_sso_headers()
response = self.http_manager.get(
url,
operation='sso_login',
headers=headers,
timeout=DEFAULT_REQUEST_TIMEOUT
)
response.raise_for_status()
data = response.json()
login_redirect_url = data.get('loginRedirectUrl')
if not login_redirect_url:
raise Exception("No loginRedirectUrl in SSO response")
logger.debug(f"SSO login redirect URL obtained: {login_redirect_url}")
return login_redirect_url
except Exception as e:
logger.error(f"SSO login failed: {e}")
raise Exception(f"SSO login failed: {e}")
def sso_authenticate(self, code: str = "", state: str = "") -> Dict[str, Any]:
"""
SSO authentication with code/state or refresh token
Matching C++ SsoClient::SSOAuthenticate()
Returns:
Dict with userInfo including userId, accountId, displayName, personaToken
"""
try:
logger.debug("Performing SSO authentication")
url = f"{SSO_URL}authenticate"
headers = self._get_sso_headers()
# Build request body matching C++ structure
if code and state:
# Authentication with authorization code
body = {
"checkRefreshToken": True,
"returnCode": {
"code": code,
"state": state
}
}
logger.debug(f"SSO auth with code: {code[:8]}..., state: {state}")
else:
# Authentication with refresh token only
body = {
"checkRefreshToken": True
}
logger.debug("SSO auth with refresh token only")
response = self.http_manager.post(
url,
operation='sso_authenticate',
headers=headers,
json_data=body,
timeout=DEFAULT_REQUEST_TIMEOUT
)
response.raise_for_status()
data = response.json()
# Check for userInfo in response
if 'userInfo' not in data:
raise Exception("No userInfo in SSO authentication response")
user_info = data['userInfo']
# Extract required fields with fallbacks
result = {
'userId': user_info.get('userId', ''),
'accountId': user_info.get('accountId', ''),
'displayName': user_info.get('displayName', ''),
'personaToken': user_info.get('personaToken', ''),
'raw_response': data
}
# Validate required fields
if not result['userId'] or not result['accountId']:
logger.warning("SSO authentication missing required user identifiers")
if not result['personaToken']:
logger.warning("SSO authentication did not return personaToken")
logger.info(
f"SSO authentication successful: "
f"userId={result['userId']}, "
f"accountId={result['accountId']}, "
f"displayName={result['displayName'][:20]}... "
f"personaToken={'YES' if result['personaToken'] else 'NO'}"
)
return result
except Exception as e:
logger.error(f"SSO authentication failed: {e}")
raise Exception(f"SSO authentication failed: {e}")
def sso_refresh(self) -> Dict[str, Any]:
"""
Refresh SSO session using refresh token
This is essentially sso_authenticate without code/state
"""
return self.sso_authenticate()
@staticmethod
def validate_persona_token(self, persona_token: str) -> bool:
"""
Validate persona token (optional enhancement)
Not in original C++ code but useful for token management
"""
try:
# Simple validation - check if token has expected structure
if not persona_token or len(persona_token) < 10:
return False
# Could add more sophisticated validation here
# For now, just check if it looks like a JWT or base64 token
parts = persona_token.split('.')
if len(parts) == 3:
# Looks like a JWT
return True
else:
# Might be base64 encoded
import base64
try:
decoded = base64.b64decode(persona_token + '==')
return len(decoded) > 0
except:
return False
except Exception as e:
logger.debug(f"Persona token validation failed: {e}")
return False
def get_user_profile(self) -> Optional[Dict[str, Any]]:
"""
Get additional user profile information (optional enhancement)
Not in original C++ code
"""
try:
url = f"{SSO_URL}profile"
headers = self._get_sso_headers()
response = self.http_manager.get(
url,
operation='sso_profile',
headers=headers,
timeout=DEFAULT_REQUEST_TIMEOUT
)
response.raise_for_status()
return response.json()
except Exception as e:
logger.debug(f"Failed to get user profile: {e}")
return None
def _get_sso_headers(self) -> Dict[str, str]:
"""Get headers for SSO requests matching C++ implementation"""
headers = {
'User-Agent': SSO_USER_AGENT,
'Content-Type': 'application/json',
'Accept': 'application/json',
'origin': 'https://web2.magentatv.de',
'referer': 'https://web2.magentatv.de/'
}
# Add session and device headers if available
if self.session_id:
headers['session-id'] = self.session_id
if self.device_id:
headers['device-id'] = self.device_id
return headers
def debug_sso_state(self) -> Dict[str, Any]:
"""Debug method to check SSO client state"""
return {
'session_id': self.session_id,
'device_id': self.device_id,
'sso_url': SSO_URL,
'headers_configured': bool(self.session_id and self.device_id)
}
lib/streaming_providers/providers/magenta2/taa_client.py
+374
View File
@@ -0,0 +1,374 @@
# streaming_providers/providers/magenta2/taa_client.py
import base64
import json
import time
from typing import Dict, Optional, Any
from dataclasses import dataclass
from ...base.network import HTTPManager
from ...base.utils.logger import logger
from .constants import (
IDM,
APPVERSION2,
MAGENTA2_PLATFORMS,
DEFAULT_PLATFORM
)
@dataclass
class TaaAuthResult:
"""Result of TAA authentication"""
access_token: str
refresh_token: Optional[str] = None
dc_cts_persona_token: Optional[str] = None
persona_id: Optional[str] = None
account_id: Optional[str] = None
consumer_id: Optional[str] = None
tv_account_id: Optional[str] = None
account_token: Optional[str] = None
account_uri: Optional[str] = None
token_exp: Optional[int] = None
raw_response: Optional[Dict[str, Any]] = None
device_limit_exceeded: bool = False
class TaaClient:
"""
Telekom Authentication and Authorization (TAA) client
Handles complete TAA authentication flow with proper JWT parsing
"""
def __init__(self, http_manager: HTTPManager, platform: str = DEFAULT_PLATFORM):
self.http_manager = http_manager
self.platform = platform
self.platform_config = MAGENTA2_PLATFORMS.get(platform, MAGENTA2_PLATFORMS[DEFAULT_PLATFORM])
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
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
taa_endpoint: TAA endpoint URL
Returns:
TaaAuthResult with complete authentication data
"""
try:
logger.debug("Starting TAA authentication")
# Build complete TAA payload
taa_payload = self._build_complete_taa_payload(
sam3_token=sam3_token,
device_id=device_id,
client_model=client_model,
device_model=device_model
)
# Build headers
headers = self._get_taa_headers(sam3_token)
# Use provided endpoint or fallback
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')}")
# Perform TAA request
response = self.http_manager.post(
endpoint,
operation='taa_auth',
headers=headers,
json_data=taa_payload
)
# Check for device limit exceeded
if response.status_code == 400:
try:
error_data = response.json()
if error_data.get('deviceLimitExceeded'):
logger.error("Device limit exceeded in TAA authentication")
return TaaAuthResult(
access_token="",
device_limit_exceeded=True
)
except (ValueError, KeyError):
pass # Will be handled by raise_for_status below
response.raise_for_status()
taa_data = response.json()
# Parse TAA response
result = self._parse_taa_response(taa_data)
logger.info("TAA authentication successful")
return result
except Exception as e:
logger.error(f"TAA authentication failed: {e}")
raise Exception(f"TAA authentication failed: {e}")
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]:
"""
Build complete TAA payload matching C++ structure exactly
C++ structure:
{
"keyValue": "IDM/APPVERSION2/TokenChannelParams(id=Tv)/TokenDeviceParams(id=...,model=...,os=...)/DE/telekom",
"accessToken": "...",
"accessTokenSource": "IDM",
"appVersion": "APPVERSION2",
"channel": {"id": "Tv"},
"device": {"id": "...", "model": "...", "os": "..."},
"natco": "DE",
"type": "telekom"
}
"""
# Use provided models or fallback to platform defaults
resolved_device_model = device_model or self.platform_config['device_name']
resolved_client_model = client_model or f"ftv-{self.platform}"
# Build keyValue string matching C++ format exactly
key_value_parts = [
IDM,
APPVERSION2,
"TokenChannelParams(id=Tv)",
f"TokenDeviceParams(id={device_id},model={resolved_device_model},os={self.platform_config['firmware']})",
"DE",
"telekom"
]
key_value = "/".join(key_value_parts)
# Build complete payload matching C++ structure
payload = {
"keyValue": key_value,
"accessToken": sam3_token,
"accessTokenSource": IDM,
"appVersion": APPVERSION2,
"channel": {
"id": "Tv"
},
"device": {
"id": device_id,
"model": resolved_device_model,
"os": self.platform_config['firmware']
},
"natco": "DE",
"type": "telekom"
}
# Add client model if available (not in original C++ but useful)
if resolved_client_model:
payload["client"] = {"model": resolved_client_model}
logger.debug(f"Built TAA payload with keyValue: {key_value}")
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
def _parse_taa_jwt_complete(self, jwt_token: str) -> Dict[str, Any]:
"""
Complete JWT parsing extracting ALL required fields from TAA token
"""
try:
parts = jwt_token.split('.')
if len(parts) != 3:
logger.warning("Invalid JWT format in TAA token")
return {}
# Decode payload
payload_b64 = parts[1]
padding = len(payload_b64) % 4
if padding:
payload_b64 += '=' * (4 - padding)
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
claim_mappings = {
# Core persona token (most important!)
'dc_cts_persona_token': [
'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',
'urn:telekom:ott:dc_cts_personaId'
],
'account_id': [
'dc_cts_accountId',
'accountId',
'urn:telekom:ott:dc_cts_accountId'
],
'consumer_id': [
'dc_cts_consumerId',
'consumerId',
'urn:telekom:ott:dc_cts_consumerId'
],
'tv_account_id': [
'dc_tvAccountId',
'tvAccountId',
'urn:telekom:ott:dc_tvAccountId'
],
# Account token
'account_token': [
'dc_cts_account_token',
'accountToken',
'urn:telekom:ott:dc_cts_account_token'
],
}
# Extract all claims
for target_key, source_keys in claim_mappings.items():
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}")
return {}
def _get_taa_headers(self, sam3_token: str) -> Dict[str, str]:
"""Get headers for TAA requests"""
return {
'User-Agent': self.platform_config['user_agent'],
'Accept': 'application/json',
'Content-Type': 'application/json',
'Authorization': f'Bearer {sam3_token}'
}
def validate_taa_token(self, taa_token: str) -> bool:
"""
Validate TAA token expiration and basic structure
"""
try:
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')
if token_exp and token_exp < time.time():
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
return False
except Exception as e:
logger.debug(f"TAA token validation failed: {e}")
return False
def debug_taa_token(self, taa_token: str) -> Dict[str, Any]:
"""
Debug method to analyze TAA token contents
"""
claims = self._parse_taa_jwt_complete(taa_token)
return {
'token_structure': 'VALID' if len(taa_token.split('.')) == 3 else 'INVALID',
'claims_available': list(claims.keys()),
'essential_claims': {
'dc_cts_persona_token': bool(claims.get('dc_cts_persona_token')),
'account_uri': bool(claims.get('account_uri')),
'persona_id': bool(claims.get('persona_id')),
'account_id': bool(claims.get('account_id'))
},
'token_expiration': {
'exp': claims.get('token_exp'),
'is_expired': claims.get('token_exp', 0) < time.time() if claims.get('token_exp') else None,
'current_time': time.time()
} if claims.get('token_exp') else None
}
resources/settings.xml
+21
View File
@@ -67,6 +67,27 @@
<setting id="rtlplus_proxy_port" type="number" label="30112" visible="eq(-2,true)" />
</category>
<category id="magenta2" label="Magenta TV 2.0 (DE)">
<setting id="magenta2_username" type="text" label="Username" default="">
<level>0</level>
<constraints>
<allowempty>false</allowempty>
</constraints>
</setting>
<setting id="magenta2_password" type="text" label="Password" default="" option="hidden">
<level>0</level>
<constraints>
<allowempty>false</allowempty>
</constraints>
</setting>
<setting id="magenta2_proxy_enabled" type="bool" label="Enable Proxy" default="false" />
<setting id="magenta2_proxy_host" type="text" label="Proxy Host" visible="eq(-1,true)" />
<setting id="magenta2_proxy_port" type="number" label="Proxy Port" visible="eq(-2,true)" />
</category>
<!-- Magenta TV Austria -->
<category label="Magenta TV (AT)">
<setting id="enable_magentaeu_at" type="bool" label="Enable Magenta TV (AT)" default="false" />