From 39eb953190288f6d6ab16929ea820f04f26f7ccc Mon Sep 17 00:00:00 2001 From: Nirvana Date: Fri, 13 Mar 2026 14:05:30 +0100 Subject: [PATCH] Add discovery --- .../providers/magenta2/provider.py | 38 +++++++++++++++++- .../providers/magenta2/vod_manager.py | 39 +++++++++++++++++-- 2 files changed, 72 insertions(+), 5 deletions(-) diff --git a/lib/streaming_providers/providers/magenta2/provider.py b/lib/streaming_providers/providers/magenta2/provider.py index 11bc7c9..1a4bc22 100644 --- a/lib/streaming_providers/providers/magenta2/provider.py +++ b/lib/streaming_providers/providers/magenta2/provider.py @@ -277,6 +277,7 @@ class Magenta2Provider(StreamingProvider): bootstrap=self.endpoint_manager.config.bootstrap, provider_config=self.endpoint_manager.config, session_id=self.session_id, + auth_headers_callback=self._vod_auth_headers, ) logger.info("✓ VodManager initialized with discovered config") @@ -1113,6 +1114,41 @@ class Magenta2Provider(StreamingProvider): return successful_channels + def _vod_auth_headers(self) -> Dict[str, str]: + """ + Build authentication headers required by authenticated VOD endpoints: + vodproductinformation (wcps.t-online.de/uspip/...) + VodPlayer (wcps.t-online.de/vops/...) + + Required headers (observed from browser traffic): + Authorization: Bearer + x-mpx-authorization: Basic (account_uri:jwt b64) + x-dt-session-id: + x-dt-call-id: + origin / referer: magenta.tv + """ + persona_token = self._ensure_authenticated() + # The persona_token is already Base64(account_uri:jwt) — used as-is + # for x-mpx-authorization. For Authorization we need just the raw JWT. + persona_jwt = self._extract_persona_jwt_from_token(persona_token) + + headers = { + "x-mpx-authorization": f"Basic {persona_token}", + "x-dt-session-id": self.session_id, + "x-dt-call-id": self._generate_call_id(), + "origin": "https://www.magenta.tv", + "referer": "https://www.magenta.tv/", + "User-Agent": self.platform_config["user_agent"], + "Accept": "application/json", + "x-permissionflagpersonalizeduireco": "false", + } + if persona_jwt: + headers["Authorization"] = f"Bearer {persona_jwt}" + else: + # Fallback: some endpoints accept Basic auth too + headers["Authorization"] = f"Basic {persona_token}" + return headers + def get_vod_category(self, category_path, **kwargs): if not self._vod_manager: raise RuntimeError("VodManager not available - configuration discovery may have failed") @@ -1876,4 +1912,4 @@ class Magenta2Provider(StreamingProvider): if hasattr(self.authenticator, "get_sam3_client_status"): result["sam3_client"] = self.authenticator.get_sam3_client_status() - return result + return result \ No newline at end of file diff --git a/lib/streaming_providers/providers/magenta2/vod_manager.py b/lib/streaming_providers/providers/magenta2/vod_manager.py index 25cd184..cd5885f 100644 --- a/lib/streaming_providers/providers/magenta2/vod_manager.py +++ b/lib/streaming_providers/providers/magenta2/vod_manager.py @@ -98,6 +98,7 @@ class VodManager: provider_config=None, session_id: Optional[str] = None, preferred_quality: str = "HD", + auth_headers_callback=None, ): self._http = http_manager self._provider = provider_name @@ -106,6 +107,10 @@ class VodManager: # Normalise to uppercase; fall back to "HD" for unknown values. _q = (preferred_quality or "HD").upper() self._preferred_quality: str = _q if _q in self._QUALITY_FALLBACK else "HD" + # Optional callable() -> Dict[str, str] returning auth headers for + # authenticated VOD endpoints (vodproductinformation, VodPlayer). + # When None, requests are sent without auth (browsing/catalogue only). + self._auth_headers_callback = auth_headers_callback # All content values resolved from BootstrapConfig -- no platform needed. self._home_url: Optional[str] = getattr(bootstrap, "home_url", None) @@ -209,10 +214,10 @@ class VodManager: return TVHUBS_BASE_URL.format(client_model=self._client_model) - def _get(self, url: str, params: Dict) -> Optional[Dict]: + def _get(self, url: str, params: Dict, headers: Optional[Dict] = None) -> Optional[Dict]: """Perform a GET request and return the parsed JSON body, or None on error.""" try: - response = self._http.get(url, params=params) + response = self._http.get(url, params=params, headers=headers) if response and response.status_code == 200: return response.json() logger.warning( @@ -223,6 +228,32 @@ class VodManager: logger.error(f"{self._provider}: VOD request exception for {url}: {exc}") return None + def _auth_headers(self) -> Optional[Dict]: + """ + Return authentication headers for protected VOD endpoints, or None. + + The callback is supplied by the provider at VodManager construction time + and encapsulates persona-token retrieval so VodManager stays decoupled + from the authenticator. + + Required headers for wcps vodproductinformation / VodPlayer: + Authorization: Bearer + x-mpx-authorization: Basic (account_uri:jwt b64) + x-dt-session-id: + x-dt-call-id: + """ + if not self._auth_headers_callback: + return None + try: + return self._auth_headers_callback() + except Exception as exc: + logger.warning(f"{self._provider}: auth_headers_callback failed: {exc}") + return None + + def _get_auth(self, url: str, params: Dict) -> Optional[Dict]: + """Like _get but injects auth headers from the callback.""" + return self._get(url, params, headers=self._auth_headers()) + # ========================================================================= # Private helpers – Personal Bar Discovery # ========================================================================= @@ -965,7 +996,7 @@ class VodManager: Returns: (theplatform_href, media_id) on success, or (None, None) on any failure. """ - prod_data = self._get(product_url, params) + prod_data = self._get_auth(product_url, params) if not prod_data: return None, None @@ -987,7 +1018,7 @@ class VodManager: ) return None, None - vod_player_data = self._get(vod_player_url, params) + vod_player_data = self._get_auth(vod_player_url, params) if not vod_player_data: return None, None