From 32b24833973e842137c813adc7d8b1e244f5aad5 Mon Sep 17 00:00:00 2001 From: Nirvana Date: Tue, 4 Nov 2025 17:58:37 +0100 Subject: [PATCH] Add more readable property provider label --- .../providers/magenta_eu/__init__.py | 22 + .../providers/magenta_eu/auth.py | 562 ++++++++++++++++++ .../providers/magenta_eu/constants.py | 214 +++++++ .../providers/magenta_eu/provider.py | 312 ++++++++++ 4 files changed, 1110 insertions(+) create mode 100644 lib/streaming_providers/providers/magenta_eu/__init__.py create mode 100644 lib/streaming_providers/providers/magenta_eu/auth.py create mode 100644 lib/streaming_providers/providers/magenta_eu/constants.py create mode 100644 lib/streaming_providers/providers/magenta_eu/provider.py diff --git a/lib/streaming_providers/providers/magenta_eu/__init__.py b/lib/streaming_providers/providers/magenta_eu/__init__.py new file mode 100644 index 0000000..170cb4b --- /dev/null +++ b/lib/streaming_providers/providers/magenta_eu/__init__.py @@ -0,0 +1,22 @@ +# streaming_providers/providers/magenta_eu/__init__.py +from .provider import MagentaProvider +from .auth import MagentaAuthenticator, MagentaAuthToken, MagentaCredentials +from .constants import ( + SUPPORTED_COUNTRIES, + DEFAULT_COUNTRY, + COUNTRY_CONFIG, + API_ENDPOINTS +) + +__all__ = [ + 'MagentaProvider', + 'MagentaAuthenticator', + 'MagentaAuthToken', + 'MagentaCredentials', + 'SUPPORTED_COUNTRIES', + 'DEFAULT_COUNTRY', + 'COUNTRY_CONFIG', + 'API_ENDPOINTS' +] + +__version__ = '1.0.0' \ No newline at end of file diff --git a/lib/streaming_providers/providers/magenta_eu/auth.py b/lib/streaming_providers/providers/magenta_eu/auth.py new file mode 100644 index 0000000..00c7d20 --- /dev/null +++ b/lib/streaming_providers/providers/magenta_eu/auth.py @@ -0,0 +1,562 @@ +# streaming_providers/providers/magenta_eu/auth.py +# -*- coding: utf-8 -*- +import uuid +import json +import base64 +import time +from typing import Dict, Optional, Any +from dataclasses import dataclass, field +from Crypto.Cipher import PKCS1_OAEP +from Crypto.PublicKey import RSA + +from ...base.auth.base_auth import BaseAuthenticator, BaseAuthToken, TokenAuthLevel +from ...base.auth.credentials import UserPasswordCredentials +from ...base.models.proxy_models import ProxyConfig +from ...base.utils.logger import logger +from .constants import ( + SUPPORTED_COUNTRIES, + DEFAULT_COUNTRY, + COUNTRY_CONFIG, + APP_VERSION, + DEVICE_NAME, + USER_AGENT, + X_USER_AGENT, + API_ENDPOINTS, + DEFAULT_REQUEST_TIMEOUT, + DEVICE_MODEL, + DEVICE_TYPE, + DEVICE_OS, + DEVICE_MANUFACTURER, + DEVICE_CONCURRENCY_PARAM, + LOGIN_CONTEXT, + LOGIN_TYPE, + CHANNEL_ID, + AUTH_FLOWS, + AUTH_STEPS, + CALL_TYPES, + MANAGE_DEVICE, + BROADCASTING_STREAM_LIMITATION_APPLIES, + get_base_url, + get_bifrost_url, + get_natco_key, + get_app_key, + get_language +) + + +class InvalidTokenError(Exception): + """Exception for invalid JWT tokens""" + pass + + +def base64url_decode(input: str) -> bytes: + """Base64 URL decode""" + padding = '=' * (4 - (len(input) % 4)) + return base64.urlsafe_b64decode(input + padding) + + +def decode_jwt(token: str, verify: bool = True) -> Dict[str, Any]: + """Decode JWT token""" + try: + header_b64, payload_b64, signature = token.split('.') + payload_json = base64url_decode(payload_b64).decode('utf-8') + payload = json.loads(payload_json) + + if verify and 'exp' in payload: + if payload['exp'] < time.time(): + raise InvalidTokenError("Token has expired") + + return payload + except (ValueError, json.JSONDecodeError, UnicodeDecodeError): + raise InvalidTokenError("Invalid token format") + + +def is_token_valid(token: str) -> bool: + """Check if token is valid""" + if not token: + return False + try: + decode_jwt(token) + return True + except InvalidTokenError: + return False + + +@dataclass +class MagentaCredentials(UserPasswordCredentials): + """Magenta TV credentials with country support""" + country: str = DEFAULT_COUNTRY + + def validate(self) -> bool: + """Validate credentials""" + return bool(self.username and self.password and self.country in SUPPORTED_COUNTRIES) + + @property + def credential_type(self) -> str: + return "magenta_user_password" + + +@dataclass +class MagentaAuthToken(BaseAuthToken): + """Magenta TV authentication token""" + refresh_token: Optional[str] = field(default="") + device_id: Optional[str] = field(default="") + session_id: Optional[str] = field(default="") + channel_map_id: Optional[str] = field(default="") + + def to_dict(self) -> Dict[str, Any]: + """Convert token to dictionary""" + data = { + 'access_token': self.access_token, + 'refresh_token': self.refresh_token or "", + 'token_type': self.token_type, + 'expires_in': self.expires_in, + 'issued_at': self.issued_at, + 'auth_level': self.auth_level.value if self.auth_level else TokenAuthLevel.UNKNOWN.value, + 'credential_type': self.credential_type or "" + } + # Include session data + if self.device_id: + data['device_id'] = self.device_id + if self.session_id: + data['session_id'] = self.session_id + if self.channel_map_id: + data['channel_map_id'] = self.channel_map_id + return data + + def get_jwt_claims(self) -> Optional[Dict[str, Any]]: + """Extract JWT claims from access token""" + try: + if not self.access_token: + return None + return decode_jwt(self.access_token, verify=False) + except Exception as e: + logger.debug(f"Failed to extract JWT claims: {e}") + return None + + +class MagentaAuthConfig: + """Configuration for Magenta TV authentication""" + + def __init__(self, country: str, http_manager): + self.country = country + self.http_manager = http_manager + self.country_config = COUNTRY_CONFIG[country] + + # Application configuration + self.app_version = APP_VERSION + self.device_name = DEVICE_NAME + self.user_agent = USER_AGENT + self.x_user_agent = X_USER_AGENT + self.timeout = DEFAULT_REQUEST_TIMEOUT + + def get_base_headers(self) -> Dict[str, str]: + """Get base headers for requests""" + return { + 'User-Agent': self.user_agent, + 'Accept': 'application/json', + 'Content-Type': 'application/json', + } + + def get_auth_headers(self, call_type: str = CALL_TYPES['GUEST_USER'], + flow: str = AUTH_FLOWS['START_UP'], + step: str = AUTH_STEPS['GET_ACCESS_TOKEN'], + device_id: str = None, session_id: str = None) -> Dict[str, str]: + """Get authentication headers""" + headers = self.get_base_headers() + headers.update({ + 'X-User-Agent': self.x_user_agent, + 'X-Call-Type': call_type, + 'X-Tv-Flow': flow, + 'X-Tv-Step': step, + 'x-request-session-id': session_id or str(uuid.uuid4()), + 'x-request-tracking-id': str(uuid.uuid4()), + 'requestid': str(uuid.uuid4()), + 'Tenant': 'tv', + 'Origin': self.country_config['base_url'], + 'App_key': self.country_config['app_key'], + 'App_version': self.app_version, + 'Device-Id': device_id or str(uuid.uuid4()), + 'Device-Name': self.device_name, + }) + return headers + + def encrypt_password(self, password: str) -> str: + """Encrypt password using RSA public key""" + try: + rsa_key = self.country_config['rsa_key'], + if not rsa_key: + logger.error(f"No RSA public key configured for country: {self.country}") + return password + + key = RSA.import_key(rsa_key) + cipher = PKCS1_OAEP.new(key) + ciphertext = cipher.encrypt(password.encode("utf-8")) + return base64.b64encode(ciphertext).decode() + except Exception as e: + logger.error(f"Error encrypting password: {e}") + return password + + +class MagentaAuthenticator(BaseAuthenticator): + """Magenta TV authenticator - directly extends BaseAuthenticator""" + + def __init__(self, country: str = DEFAULT_COUNTRY, + settings_manager=None, + credentials=None, + config_dir: Optional[str] = None, + http_manager=None, + proxy_config: Optional[ProxyConfig] = None): + """ + Initialize Magenta authenticator + """ + if country not in SUPPORTED_COUNTRIES: + raise ValueError(f"Unsupported country: {country}. Must be one of: {SUPPORTED_COUNTRIES}") + + if http_manager is None: + raise ValueError("http_manager is required for MagentaAuthenticator") + + self.country = country + self._http_manager = http_manager + self._proxy_config = proxy_config + + # Setup config + self._config = MagentaAuthConfig(self.country, self._http_manager) + + # NOW call parent __init__ - this will setup settings_manager and load session + super().__init__( + provider_name='magenta_eu', + settings_manager=settings_manager, + credentials=credentials, + country=country, + config_dir=config_dir, + enable_kodi_integration=True + ) + + # After parent init, we can extract session data from the loaded token + self._extract_session_data_from_token() + + def _extract_session_data_from_token(self) -> None: + """Extract session data from the loaded token (if any)""" + if self._current_token and isinstance(self._current_token, MagentaAuthToken): + # Session data is already stored in the token + logger.debug("Session data extracted from loaded token") + else: + # No token loaded, initialize empty session data + logger.debug("No token loaded, session data will be initialized on first auth") + + @property + def auth_endpoint(self) -> str: + """Authentication endpoint - required by BaseAuthenticator""" + return API_ENDPOINTS['LOGIN'].format(natco=self.country) + + def _get_auth_headers(self) -> Dict[str, str]: + """Get headers for authentication request - required by BaseAuthenticator""" + device_id = "" + session_id = "" + + # Get session data from current token if available + if self._current_token and isinstance(self._current_token, MagentaAuthToken): + device_id = self._current_token.device_id or "" + session_id = self._current_token.session_id or "" + + return self._config.get_auth_headers( + call_type=CALL_TYPES['GUEST_USER'], + flow=AUTH_FLOWS['USERNAME_PASSWORD_LOGIN'], + step=AUTH_STEPS['GET_ACCESS_TOKEN'], + device_id=device_id, + session_id=session_id + ) + + def _build_auth_payload(self) -> Dict[str, Any]: + """Build authentication payload - required by BaseAuthenticator""" + if not self.credentials or not isinstance(self.credentials, MagentaCredentials): + raise Exception("No valid Magenta credentials available") + + device_id = "" + if self._current_token and isinstance(self._current_token, MagentaAuthToken): + device_id = self._current_token.device_id or "" + + # If no device_id, initialize session + if not device_id: + device_id, _ = self._initialize_session() + + encrypted_password = self._config.encrypt_password(self.credentials.password) + + return { + "appVersion": self._config.app_version, + "channel": {"id": CHANNEL_ID}, + "natco": self.country, + "type": LOGIN_TYPE, + "forceRegister": False, + "context": LOGIN_CONTEXT, + "device": { + "id": device_id, + "model": DEVICE_MODEL, + "os": DEVICE_OS, + "deviceName": DEVICE_MODEL, + "manageDevice": MANAGE_DEVICE, + "deviceType": DEVICE_TYPE, + "deviceOS": DEVICE_OS, + "deviceModel": self._config.x_user_agent, + "deviceManufacturer": DEVICE_MANUFACTURER, + "concurrencyLimitParam": DEVICE_CONCURRENCY_PARAM, + "broadcastingStreamLimitationApplies": BROADCASTING_STREAM_LIMITATION_APPLIES + }, + "telekomLogin": { + "username": self.credentials.username, + "password": encrypted_password + } + } + + def _create_token_from_response(self, response_data: Dict[str, Any]) -> BaseAuthToken: + """Create token from API response - required by BaseAuthenticator""" + # Get existing session data from current token + device_id = "" + session_id = "" + channel_map_id = "" + + if self._current_token and isinstance(self._current_token, MagentaAuthToken): + device_id = self._current_token.device_id or "" + session_id = self._current_token.session_id or "" + channel_map_id = self._current_token.channel_map_id or "" + + # If no device_id/session_id, initialize session + if not device_id or not session_id: + device_id, session_id = self._initialize_session() + + token = MagentaAuthToken( + access_token=response_data['accessToken'], + refresh_token=response_data.get('refreshToken', ''), + token_type='Bearer', + expires_in=response_data.get('expiresIn', 3600), + issued_at=time.time(), + device_id=device_id, + session_id=session_id, + channel_map_id=channel_map_id + ) + + # Classify token + token.auth_level = self._classify_token(token) + logger.debug(f"Token created and classified as: {token.auth_level.value}") + + return token + + def get_fallback_credentials(self): + """Get fallback credentials - required by BaseAuthenticator""" + # Return empty credentials as fallback + return MagentaCredentials(username="", password="", country=self.country) + + def _initialize_session(self) -> tuple: + """Initialize session by visiting startup page, returns (device_id, session_id)""" + try: + startup_url = API_ENDPOINTS['STARTUP_PAGE'].format( + base_url=self._config.country_config['base_url'] + ) + + headers = self._config.get_base_headers() + response = self._http_manager.get( + startup_url, + operation='session_init', + headers=headers, + timeout=self._config.timeout + ) + + # Extract cookies + device_id = "" + session_id = "" + if hasattr(response, 'cookies'): + cookies = response.cookies.get_dict() + device_id = cookies.get("deviceId", str(uuid.uuid4())) + session_id = cookies.get("sessionId", str(uuid.uuid4())) + + logger.debug(f"Session initialized - device_id: {device_id}, session_id: {session_id}") + + return device_id, session_id + + except Exception as e: + logger.warning(f"Session initialization failed: {e}") + # Generate fallback IDs + return str(uuid.uuid4()), str(uuid.uuid4()) + + def _perform_authentication(self) -> BaseAuthToken: + """Perform Magenta TV authentication - required by BaseAuthenticator""" + if not self.credentials or not isinstance(self.credentials, MagentaCredentials): + raise Exception("No valid Magenta credentials available") + + logger.info(f"Performing Magenta TV authentication for country: {self.country}") + + # Perform login + headers = self._get_auth_headers() + payload = self._build_auth_payload() + + response = self._http_manager.post( + self.auth_endpoint, + operation='auth', + headers=headers, + json_data=payload, + timeout=self._config.timeout + ) + + response.raise_for_status() + token_data = response.json() + + # Handle device limit exceeded + if token_data.get("deviceLimitExceed", False): + logger.info("Device limit exceeded, attempting token upgrade") + token_data = self._upgrade_token(token_data['refreshToken']) + + return self._create_token_from_response(token_data) + + def _upgrade_token(self, refresh_token: str) -> Dict[str, Any]: + """Upgrade token when device limit is exceeded""" + upgrade_url = API_ENDPOINTS['UPGRADE_TOKEN'].format(natco=self.country) + + headers = self._config.get_auth_headers( + call_type=CALL_TYPES['GUEST_USER'], + flow=AUTH_FLOWS['USERNAME_PASSWORD_LOGIN'], + step=AUTH_STEPS['UPGRADE_TOKEN'] + ) + headers['Refresh_token'] = refresh_token + + payload = self._build_auth_payload() + + response = self._http_manager.post( + upgrade_url, + operation='auth_upgrade', + headers=headers, + json_data=payload, + timeout=self._config.timeout + ) + + response.raise_for_status() + return response.json() + + def _refresh_token(self) -> Optional[BaseAuthToken]: + """Refresh Magenta TV token - override base method""" + if not self._current_token or not self._current_token.refresh_token: + logger.debug("No valid refresh token available") + return None + + try: + logger.debug(f"Refreshing Magenta TV token for country: {self.country}") + + refresh_url = API_ENDPOINTS['REFRESH_TOKEN'].format(natco=self.country) + + headers = self._config.get_auth_headers( + call_type=CALL_TYPES['AUTH_USER'], + flow=AUTH_FLOWS['START_UP'], + step=AUTH_STEPS['REFRESH_TOKEN'] + ) + headers['Refresh_token'] = self._current_token.refresh_token + + payload = { + "clientVersion": self._config.app_version, + "concurrencyLimitParam": DEVICE_CONCURRENCY_PARAM, + "deviceId": self._current_token.device_id or "" + } + + response = self._http_manager.post( + refresh_url, + operation='auth_refresh', + headers=headers, + json_data=payload, + timeout=self._config.timeout + ) + + response.raise_for_status() + token_data = response.json() + + # Create new token but preserve session data + new_token = MagentaAuthToken( + access_token=token_data['accessToken'], + refresh_token=token_data.get('refreshToken', ''), + token_type='Bearer', + expires_in=token_data.get('expiresIn', 3600), + issued_at=time.time(), + device_id=self._current_token.device_id, + session_id=self._current_token.session_id, + channel_map_id=self._current_token.channel_map_id + ) + + # Classify token + new_token.auth_level = self._classify_token(new_token) + logger.info("Token refresh successful") + return new_token + + except Exception as e: + logger.warning(f"Token refresh failed: {e}") + return None + + def _classify_token(self, token: BaseAuthToken) -> TokenAuthLevel: + """Classify Magenta TV token - required by BaseAuthenticator""" + try: + if not token or not token.access_token: + return TokenAuthLevel.UNKNOWN + + claims = token.get_jwt_claims() + if not claims: + return TokenAuthLevel.UNKNOWN + + # Magenta tokens with username in claims indicate user authentication + if 'username' in claims or 'preferred_username' in claims: + return TokenAuthLevel.USER_AUTHENTICATED + + # Anonymous tokens typically have limited claims + if len(claims) <= 3: # Basic claims like exp, iat, iss + return TokenAuthLevel.ANONYMOUS + + return TokenAuthLevel.USER_AUTHENTICATED + + except Exception as e: + logger.debug(f"Error classifying token: {e}") + return TokenAuthLevel.UNKNOWN + + def get_user_account(self) -> Dict[str, Any]: + """Get user account information""" + access_token = self.get_bearer_token() + + account_url = API_ENDPOINTS['USER_ACCOUNT'].format( + bifrost_url=get_bifrost_url(self.country) + ) + + params = { + 'fresh_login': 'false', + 'app_language': get_language(self.country), + 'natco_code': self.country + } + + device_id = "" + session_id = "" + if self._current_token and isinstance(self._current_token, MagentaAuthToken): + device_id = self._current_token.device_id or "" + session_id = self._current_token.session_id or "" + + headers = self._config.get_auth_headers( + call_type=CALL_TYPES['AUTH_USER'], + flow=AUTH_FLOWS['START_UP'], + step=AUTH_STEPS['GET_USER_ACCOUNT'], + device_id=device_id, + session_id=session_id + ) + headers['Bff_token'] = access_token + + response = self._http_manager.get( + account_url, + operation='user_account', + headers=headers, + params=params, + timeout=self._config.timeout + ) + + response.raise_for_status() + account_data = response.json() + + # Save channel map ID to current token + if 'channelMap_id' in account_data and self._current_token and isinstance(self._current_token, + MagentaAuthToken): + self._current_token.channel_map_id = account_data['channelMap_id'] + # Save the updated token with channel_map_id + self._save_session() + + return account_data \ No newline at end of file diff --git a/lib/streaming_providers/providers/magenta_eu/constants.py b/lib/streaming_providers/providers/magenta_eu/constants.py new file mode 100644 index 0000000..0f9358f --- /dev/null +++ b/lib/streaming_providers/providers/magenta_eu/constants.py @@ -0,0 +1,214 @@ +# streaming_providers/providers/magenta_eu/constants.py +# ============================================================================ +# Magenta TV Configuration +# ============================================================================ + +# Supported countries +SUPPORTED_COUNTRIES = ['hr', 'pl', 'me', 'at', 'hu'] + +# Default country +DEFAULT_COUNTRY = 'at' + +# Country-specific configuration +COUNTRY_CONFIG = { + 'hr': { + 'base_url': 'https://mojmaxtv.hrvatskitelekom.hr', + 'bifrost_url': 'https://tv-hr-prod.yo-digital.com/hr-bifrost', + 'natco_key': 'l2lyvGVbUm2EKJE96ImQgcc8PKMZWtbE', + 'app_key': 'GWaBW4RTloLwpUgYVzOiW5zUxFLmoMj5', + 'language': 'hr', + 'rsa_key': """-----BEGIN PUBLIC KEY----- +MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDMmKReuzuaCk10Wa6vv4ybcqjVN3cruj27IRp9YhdgEw9jcG728Aj9s60mY8B/czzW5ntKJQktyBRBZ98BKznRWBrVN/n9JR/m1UDc38PW4BPe4z5VtBe99dyFcJQ1VJij6HG0BFtw3isPR5NAUAAyGnXpNWKCat5TtBckqVatBQIDAQAB +-----END PUBLIC KEY-----""" + }, + 'pl': { + 'base_url': 'https://magentatv.pl', + 'bifrost_url': 'https://tv-pl-prod.yo-digital.com/pl-bifrost', + 'natco_key': 'ovINYLVrQsLj8wPDQzNYotQMkFha9PFF', + 'app_key': '2zjAjpJ6dRfHR1nIwF47jz75g7Qf4F6z', + 'language': 'pl', + 'rsa_key': """-----BEGIN PUBLIC KEY----- +MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAxTkEaSufgfBKbzBAHRzSQreMYYCAg4wtjk44A31qVrt0/IaBNbAlvDNEWmB9kwZXK6s41XiPg1k/xsWNoug0nIU1eTEeTsRDwrMzxwYyn5ObO1jylBK0mEAmcl/cEk8A+DqW+raxoWSNq6IH4pciO3QVxeAzIP73FSBrcj81LEeDQsJY6ySbZ1mcDD1/axuCrUm2masJq3nifNOEZG7mAiy5pvoN54grldvXuc8nBne9tdMnCFRisRIO7f2KG5pQVqtLh41aeRTD0eg9c7SWYDhTcRb5MA+PqluZrtwQ3+J7kBGGlxkYHuQ10slR6YUqEELdd85CE7DtMx+PcGLxWwIDAQAB +-----END PUBLIC KEY-----""" + }, + 'me': { + 'base_url': 'https://magentatv.me', + 'bifrost_url': 'https://tv-me-prod.yo-digital.com/me-bifrost', + 'natco_key': 'ANKB5xVVywklLUd9WtEOh8eyLnlAypTM', + 'app_key': 'erYJuNj5fnVXtRgjkr4scxbr3oEkM4I4', + 'language': 'me', + 'rsa_key': """-----BEGIN PUBLIC KEY----- +MIICIjANBgkqhkiG9w0BAQEFAAOCAg8AMIICCgKCAgEApoovXHgusvwv7S0cdg+xQYuY3Kb/J50n3GKpW4UmH4xG8NCAD6PBHPKGlVYl6sAnLXofJeq0DsXZJ9xFGH1wTQdd3hxwhGrigstluu4PbpiVRjrFe9AuZto+/Hb0+Ljr5BKFgNMYhpZOaBWgjkqtYWFyMzB1v3VtgUURqxyzMhgiPoCcqYynnqK1weJLGEMSZNlHeWwX46VlK/JqIwPw2ajl5IwvQFzY6BEQvfwXVJxIcvQwkcmeR1A3ACE1E1tvrATb1EPNeKh/E6Kukb8EGTB1O3WJa4IrdX6HI3BJLXftLZ9Y55HuHa+GrGkJRJvchxYCugxgHKSqNOnsT8FWYm1gdZZ6/i/jPo7/tpEdd2U6FxG7/I4rcpk2FzSSd3LKEHGf5v1V982Idbrlm+1LfsjoTx5Go2M8+uBR92pV0qSQlB4YXgQQgqpDUq9f9TwN67nj2jfRECR2hBITkK3+sC9ZxSHYg4Cp3ycHqetxUJ21kVmO6jMG+JsBHLuDKCIc3yrHXHBwhZt4XOhH6K564pUKBOfbe9mZlGrttEla0FK9B/WM315QQ9WgHA5pFyDFmjsB4MvFqKrbsrVUxDZM1COHVC7QbyEte45BDDO/amaB1aa6st5mNIm22m0s0jW8flHarGuulCDNBt9jHI6MnYLb3wpVj00b62Vmx1FX9qkCAwEAAQ== +-----END PUBLIC KEY-----""" + }, + 'at': { + 'base_url': 'https://tv.magenta.at', + 'bifrost_url': 'https://tv-at-prod.yo-digital.com/at-bifrost', + 'natco_key': 'NZu7aIg1vFTNLwHcb0Kjhqk54ql9RJj5', + 'app_key': 'CTnKA63ruKM0JM1doxAXwwyQLLmQiEiy', + 'language': 'de', + 'rsa_key': """-----BEGIN PUBLIC KEY----- +MIIBIDANBgkqhkiG9w0BAQEFAAOCAQ0AMIIBCAKCAQEA0EsuGonW8+y5Oq2Hopru5oFCaXEDMBrWPDkzIeY1Vvot3z/l9Q3/bcUQV3Yo6DvbhlARQnMeqiKJ5dutZmdxuj+nhZq9FcL20RUywqGnxTCHaGAEb05Qlovu7Rbld2GeJa4nFP1RY5glUlr/DYVB+tIHqPfZVUSc1PS+l5QkB9TJir57ALxERBJGjT5vhQixXGf6IqmLkxm1okIbuGJa2ttmSWNq0OVi2cF40ZsV64ly7a3m6n2WBYmhqd3ghSprNHXwJBwYwu1L+9CF9oLQXHs9cUDhqyQB+3iDU2Ro/rtZsGcnvnIiDHRIWZ94zcOTOpdUH4pBujc8jF3qdw99UwIBJQ== +-----END PUBLIC KEY-----""" + }, + 'hu': { + 'base_url': 'https://player.telekomtvgo.hu', + 'bifrost_url': 'https://tv-hu-prod.yo-digital.com/bifrost', + 'natco_key': 'Tydx7H7fJO6HxgjvJok0ZhVWFmX3om0P', + 'app_key': 'exSJHBiSAN6wAAeqdWLdTUfdTi2PNark', + 'language': 'hu', + 'rsa_key': """-----BEGIN PUBLIC KEY----- +MIIBIDANBgkqhkiG9w0BAQEFAAOCAQ0AMIIBCAKCAQEA0EsuGonW8+y5Oq2Hopru5oFCaXEDMBrWPDkzIeY1Vvot3z/l9Q3/bcUQV3Yo6DvbhlARQnMeqiKJ5dutZmdxuj+nhZq9FcL20RUywqGnxTCHaGAEb05Qlovu7Rbld2GeJa4nFP1RY5glUlr/DYVB+tIHqPfZVUSc1PS+l5QkB9TJir57ALxERBJGjT5vhQixXGf6IqmLkxm1okIbuGJa2ttmSWNq0OVi2cF40ZsV64ly7a3m6n2WBYmhqd3ghSprNHXwJBwYwu1L+9CF9oLQXHs9cUDhqyQB+3iDU2Ro/rtZsGcnvnIiDHRIWZ94zcOTOpdUH4pBujc8jF3qdw99UwIBJQ== +-----END PUBLIC KEY-----""" + } +} + +# ============================================================================ +# Application Configuration +# ============================================================================ + +APP_VERSION = "02.0.1080" +OS = "Linux" +BROWSER = "Chrome" +BROWSER_VERSION = "132" +DEVICE_NAME = f"{OS} - {BROWSER}" + +# ============================================================================ +# Device Configuration +# ============================================================================ + +DEVICE_MODEL = "WEB" +DEVICE_TYPE = "WEB" +DEVICE_OS = OS +DEVICE_MANUFACTURER = f"{OS} x86_64" +DEVICE_CONCURRENCY_PARAM = "TVSOA-restriction-unmanagedDeviceStreamLimit" + +# User agent configuration +USER_AGENT = f"Mozilla/5.0 (X11; {OS} x86_64) AppleWebKit/537.36 (KHTML, like Gecko) {BROWSER}/{BROWSER_VERSION}.0.0.0 Safari/537.36" +X_USER_AGENT = f"{DEVICE_MODEL.lower()}|{DEVICE_TYPE.lower()}|{BROWSER}-{BROWSER_VERSION}|{APP_VERSION}|1" + +# ============================================================================ +# API Endpoints +# ============================================================================ + +# Base API URLs +GATEWAY_BASE_URL = "https://gateway-{natco}-proxy.tv.yo-digital.com" + +# API endpoints +API_ENDPOINTS = { + 'LOGIN': GATEWAY_BASE_URL + "/{natco}-idm/P/onboarding/login", + 'REFRESH_TOKEN': GATEWAY_BASE_URL + "/{natco}-idm/P/onboarding/refresh-token", + 'UPGRADE_TOKEN': GATEWAY_BASE_URL + "/{natco}-idm/P/onboarding/upgrade-token", + 'USER_ACCOUNT': "{bifrost_url}/user/account", + 'EPG_CHANNELS': "{bifrost_url}/epg/channel", + 'STARTUP_PAGE': "{base_url}/epg" +} + +# Widevine configuration +WV_URL = "https://widevine.entitlement.theplatform.eu/wv/web/ModularDrm/getRawWidevineLicense?schema=1.0&form=json&releasePid=" +ACC_URL = "http://access.auth.theplatform.com/data/Account" + +# ============================================================================ +# Authentication Configuration +# ============================================================================ + +# Login context and type +LOGIN_CONTEXT = "login" +LOGIN_TYPE = "telekom" +CHANNEL_ID = "Tv" + +# Authentication flows and steps +AUTH_FLOWS = { + 'USERNAME_PASSWORD_LOGIN': 'USERNAME_PASSWORD_LOGIN', + 'START_UP': 'START_UP' +} + +AUTH_STEPS = { + 'GET_ACCESS_TOKEN': 'GET_ACCESS_TOKEN', + 'REFRESH_TOKEN': 'REFRESH_TOKEN', + 'UPGRADE_TOKEN': 'UPGRADE_TOKEN', + 'GET_USER_ACCOUNT': 'GET_USER_ACCOUNT', + 'EPG_CHANNEL': 'EPG_CHANNEL' +} + +CALL_TYPES = { + 'GUEST_USER': 'GUEST_USER', + 'AUTH_USER': 'AUTH_USER' +} + +# ============================================================================ +# DRM Configuration +# ============================================================================ + +DRM_SYSTEM_WIDEVINE = 'widevine' + +# DRM request headers +DRM_REQUEST_HEADERS = { + 'Content-Type': 'application/octet-stream', + 'User-Agent': USER_AGENT +} + +# ============================================================================ +# Request Configuration +# ============================================================================ + +DEFAULT_REQUEST_TIMEOUT = 30 +DEFAULT_MAX_RETRIES = 3 + +# ============================================================================ +# Headers Configuration +# ============================================================================ + +BASE_HEADERS = { + 'User-Agent': USER_AGENT, +} + +AUTH_HEADERS_BASE = { + 'User-Agent': USER_AGENT, + 'Accept': 'application/json', + 'Content-Type': 'application/json', + 'X-User-Agent': X_USER_AGENT, + 'Tenant': 'tv', +} + +# ============================================================================ +# Content Configuration +# ============================================================================ + +CONTENT_TYPE_LIVE = 'LIVE' +STREAMING_FORMAT_DASH = 'dash' + +# ============================================================================ +# Device Management +# ============================================================================ + +MANAGE_DEVICE = False +BROADCASTING_STREAM_LIMITATION_APPLIES = False + +def get_country_config(country: str) -> dict: + """Get configuration for specific country""" + return COUNTRY_CONFIG.get(country, COUNTRY_CONFIG[DEFAULT_COUNTRY]) + +def get_base_url(country: str) -> str: + """Get base URL for country""" + return get_country_config(country)['base_url'] + +def get_bifrost_url(country: str) -> str: + """Get bifrost URL for country""" + return get_country_config(country)['bifrost_url'] + +def get_natco_key(country: str) -> str: + """Get natco key for country""" + return get_country_config(country)['natco_key'] + +def get_app_key(country: str) -> str: + """Get app key for country""" + return get_country_config(country)['app_key'] + +def get_language(country: str) -> str: + """Get language for country""" + return get_country_config(country)['language'] + +def get_rsa_key(country: str) -> str: + """Get RSA public key for country""" + return get_country_config(country)['rsa_key'] \ No newline at end of file diff --git a/lib/streaming_providers/providers/magenta_eu/provider.py b/lib/streaming_providers/providers/magenta_eu/provider.py new file mode 100644 index 0000000..734f7ad --- /dev/null +++ b/lib/streaming_providers/providers/magenta_eu/provider.py @@ -0,0 +1,312 @@ +# streaming_providers/providers/magenta_eu/provider.py +# -*- coding: utf-8 -*- +from typing import Dict, Optional, List +import json +import time +from datetime import datetime, timedelta + +from ...base.provider import StreamingProvider +from ...base.models import DRMConfig, LicenseConfig, DRMSystem +from ...base.models.streaming_channel import StreamingChannel +from ...base.network import HTTPManagerFactory, ProxyConfigManager +from ...base.models.proxy_models import ProxyConfig +from ...base.utils.logger import logger +from .auth import MagentaAuthenticator, MagentaCredentials +from .constants import ( + SUPPORTED_COUNTRIES, + DEFAULT_COUNTRY, + USER_AGENT, + API_ENDPOINTS, + DEFAULT_REQUEST_TIMEOUT, + DEFAULT_MAX_RETRIES, + DRM_SYSTEM_WIDEVINE, + DRM_REQUEST_HEADERS, + WV_URL, + CONTENT_TYPE_LIVE, + STREAMING_FORMAT_DASH, + get_base_url, + get_bifrost_url, + get_natco_key, + get_app_key, + get_language +) + + +class MagentaProvider(StreamingProvider): + """Magenta TV streaming provider implementation""" + + def __init__(self, country: str = DEFAULT_COUNTRY, + config_dir: Optional[str] = None, + proxy_config: Optional[ProxyConfig] = None, + proxy_url: Optional[str] = None): + """ + Initialize Magenta provider + """ + super().__init__(country=country) + + if country not in SUPPORTED_COUNTRIES: + raise ValueError(f"Unsupported country: {country}. Must be one of: {SUPPORTED_COUNTRIES}") + + # Setup proxy configuration + self.proxy_config = ( + proxy_config or + (ProxyConfig.from_url(proxy_url) if proxy_url else None) or + self._load_proxy_from_manager(config_dir) + ) + + if self.proxy_config: + logger.info("Using proxy configuration for Magenta TV") + else: + logger.debug("No proxy configuration found for Magenta TV") + + # Create HTTP manager + self.http_manager = HTTPManagerFactory.create_for_provider( + provider_name='magenta_eu', + proxy_config=self.proxy_config, + user_agent=USER_AGENT, + timeout=DEFAULT_REQUEST_TIMEOUT, + max_retries=DEFAULT_MAX_RETRIES + ) + + # Create authenticator + self.authenticator = MagentaAuthenticator( + country=country, + config_dir=config_dir, + http_manager=self.http_manager, + proxy_config=self.proxy_config + ) + + # Authenticate + try: + self.bearer_token = self.authenticator.get_bearer_token() + except Exception as e: + logger.warning(f"Could not authenticate during initialization: {e}") + self.bearer_token = None + + def _load_proxy_from_manager(self, config_dir: Optional[str]) -> Optional[ProxyConfig]: + """Load proxy configuration from ProxyConfigManager""" + try: + proxy_manager = ProxyConfigManager(config_dir) + return proxy_manager.get_proxy_config('magenta_eu', self.country) + except Exception as e: + logger.warning(f"Could not load proxy from ProxyConfigManager: {e}") + return None + + @property + def provider_name(self) -> str: + return 'magenta_eu' + + @property + def provider_label(self) -> str: + return f'Magenta TV ({self.country.upper()})' + + @property + def uses_dynamic_manifests(self) -> bool: + return False + + def authenticate(self, **kwargs) -> str: + """Authenticate and return bearer token""" + self.bearer_token = self.authenticator.get_bearer_token(force_refresh=kwargs.get('force_refresh', False)) + return self.bearer_token + + def get_dynamic_manifest_params(self, channel: StreamingChannel, **kwargs) -> Optional[str]: + return None + + def refresh_authentication(self) -> str: + """Force refresh authentication""" + self.bearer_token = self.authenticator.get_bearer_token(force_refresh=True) + return self.bearer_token + + def fetch_channels(self, **kwargs) -> List[StreamingChannel]: + """Fetch available channels from Magenta TV""" + try: + # Get user account to ensure we have channel map ID + self.authenticator.get_user_account() + + channels_url = API_ENDPOINTS['EPG_CHANNELS'].format( + bifrost_url=get_bifrost_url(self.country) + ) + + # Get channel map ID from authenticator + channel_map_id = "" + if (self.authenticator._current_token and + isinstance(self.authenticator._current_token, + self.authenticator.__class__.__bases__[0].MagentaAuthToken)): + channel_map_id = self.authenticator._current_token.channel_map_id or "" + + params = { + 'channelMap_id': channel_map_id, + 'includeVirtualChannels': 'true', + 'natco_key': get_natco_key(self.country), + 'app_language': get_language(self.country), + 'natco_code': self.country + } + + headers = self.authenticator._config.get_auth_headers( + call_type="GUEST_USER", + flow="START_UP", + step="EPG_CHANNEL" + ) + + response = self.http_manager.get( + channels_url, + operation='channels', + headers=headers, + params=params, + timeout=DEFAULT_REQUEST_TIMEOUT + ) + response.raise_for_status() + + channels_data = response.json() + channels = self._process_channels_response(channels_data) + + logger.info(f"Successfully fetched {len(channels)} channels for country {self.country}") + return channels + + except Exception as e: + raise Exception(f"Error fetching channels from Magenta TV: {e}") + + def _process_channels_response(self, response_data: Dict) -> List[StreamingChannel]: + """Process channels response and convert to StreamingChannel objects""" + if 'channels' not in response_data: + raise Exception("Invalid channels response structure") + + channels = [] + for channel_data in response_data['channels']: + try: + # Extract channel information + title = channel_data.get('title', 'Unknown Channel') + logo = channel_data.get('channel_logo', '') + manifest = channel_data.get('video_src_dash', '') + pid = channel_data.get('pid_dash', '') + station_id = channel_data.get('station_id', '') + channel_number = channel_data.get('channel_number', '') + media_pid = channel_data.get('media_pid', '') + + # Build manifest script + manifest_script_parts = [] + if channel_number: + manifest_script_parts.append(f"chno={channel_number}") + if station_id: + manifest_script_parts.append(f"epgid={station_id}") + if media_pid: + manifest_script_parts.append(f"media={media_pid}") + + manifest_script = " ".join(manifest_script_parts) if manifest_script_parts else "" + + # Create streaming channel + streaming_channel = StreamingChannel( + name=title, + channel_id=station_id or pid or title, + provider=self.provider_name, + logo_url=logo, + mode="live", + session_manifest=False, + manifest=manifest, + manifest_script=manifest_script, + cdm_type=DRM_SYSTEM_WIDEVINE, + use_cdm=True, + cdm=f"pid={pid}" if pid else "", + cdm_mode='external', + video='best', + on_demand=True, + speed_up=True, + content_type=CONTENT_TYPE_LIVE, + country=self.country.upper(), + language=get_language(self.country), + streaming_format=STREAMING_FORMAT_DASH + ) + + channels.append(streaming_channel) + + except Exception as e: + logger.warning(f"Error processing channel data: {e}") + + return channels + + def enrich_channel_data(self, channel: StreamingChannel, **kwargs) -> Optional[StreamingChannel]: + """ + Enrich channel with streaming data + Magenta TV provides manifest URLs directly in channel data, + so this mainly ensures DRM configuration is set up + """ + try: + if not channel.manifest: + return None + + # Get DRM config + drm_config = self.get_drm_config(channel) + if drm_config: + channel.drm_config = drm_config + + return channel + + except Exception as e: + logger.warning(f"Error enriching channel data for {channel.name}: {e}") + return None + + def get_manifest(self, channel_id: str, **kwargs) -> Optional[str]: + """ + Get manifest URL for a specific channel by ID + For Magenta TV, manifests are already provided in channel data + """ + # Since manifests are provided directly in channel data, + # this would need to fetch channel data again or use cached data + return None + + def get_drm_configs(self, channel: StreamingChannel, **kwargs) -> List[DRMConfig]: + """Get DRM configurations for channel""" + drm_config = self.get_drm_config(channel) + return [drm_config] if drm_config else [] + + def get_drm_config(self, channel: StreamingChannel, **kwargs) -> Optional[DRMConfig]: + """Get DRM configuration for channel""" + try: + pid = channel.cdm.replace("pid=", "") if channel.cdm else "" + if not pid: + return None + + license_url = f"{WV_URL}{pid}" + + headers = DRM_REQUEST_HEADERS.copy() + headers.update({ + 'Authorization': f'Bearer {self.bearer_token}', + 'Origin': get_base_url(self.country), + 'Referer': f"{get_base_url(self.country)}/", + }) + + return DRMConfig( + system=DRMSystem.WIDEVINE, + license_url=license_url, + headers=headers, + challenge_data=b'', + session_id=str(int(time.time())) + ) + + except Exception as e: + logger.warning(f"Error creating DRM config for {channel.name}: {e}") + return None + + def validate_credentials(self, credentials: MagentaCredentials) -> bool: + """Validate Magenta TV credentials""" + try: + # Test authentication with provided credentials + temp_authenticator = MagentaAuthenticator( + country=credentials.country, + config_dir=self.authenticator.settings_manager.config_dir if hasattr( + self.authenticator.settings_manager, 'config_dir') else None, + http_manager=self.http_manager, + proxy_config=self.proxy_config, + credentials=credentials + ) + + token = temp_authenticator.get_bearer_token() + return bool(token and len(token) > 0) + + except Exception as e: + logger.debug(f"Credential validation failed: {e}") + return False + + def get_supported_countries(self) -> List[str]: + """Get list of supported countries""" + return SUPPORTED_COUNTRIES.copy() \ No newline at end of file