Add more readable property provider label

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'

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

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']

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()