From a0e730aa076b6346cf7d09dc4ef933755810a028 Mon Sep 17 00:00:00 2001 From: Nirvana Date: Fri, 5 Jun 2026 19:41:37 +0200 Subject: [PATCH] magenta2: use OIDC discovery --- .../providers/magenta2/auth.py | 40 +++++++++++++++++-- .../providers/magenta2/sam3_client.py | 6 ++- .../providers/magenta2/token_flow_manager.py | 21 ++++------ 3 files changed, 48 insertions(+), 19 deletions(-) diff --git a/lib/streaming_providers/providers/magenta2/auth.py b/lib/streaming_providers/providers/magenta2/auth.py index 1dc6c1b..3dfad05 100644 --- a/lib/streaming_providers/providers/magenta2/auth.py +++ b/lib/streaming_providers/providers/magenta2/auth.py @@ -19,6 +19,7 @@ from dataclasses import dataclass, field from typing import Any, Dict, Optional from ...base.auth.base_auth import BaseAuthenticator, BaseAuthToken, TokenAuthLevel +from ...base.auth.base_oauth2_auth import OIDCConfiguration from ...base.auth.credentials import ClientCredentials from ...base.utils.logger import logger from .constants import ( @@ -656,8 +657,13 @@ class Magenta2Authenticator(BaseAuthenticator): if self._openid_config: issuer_url = self._openid_config.get("issuer") - oauth_endpoint = self._openid_config.get("token_endpoint") backchannel_start_url = self._openid_config.get("backchannel_auth_start") + # Use base class property — resolves from OIDC cache seeded by + # set_openid_config(), avoiding a redundant raw dict lookup. + try: + oauth_endpoint = self.oauth_token_endpoint + except (NotImplementedError, AttributeError): + oauth_endpoint = self._openid_config.get("token_endpoint") self._sam3_client = Sam3Client( http_manager=self._http_manager, @@ -669,6 +675,7 @@ class Magenta2Authenticator(BaseAuthenticator): line_auth_endpoint=line_auth_endpoint, backchannel_start_url=backchannel_start_url, qr_code_url_template=qr_code_url_template, + provider_name="MagentaTV", ) logger.info( @@ -827,14 +834,39 @@ class Magenta2Authenticator(BaseAuthenticator): def set_openid_config(self, openid_config: Dict[str, Any]) -> None: """ - Set OpenID configuration for SAM3 client + Set OpenID configuration for SAM3 client and seed the base class OIDC cache. + + The discovery document is already in hand at this point (fetched by + DiscoveryService), so we populate the base class cache directly rather + than triggering a redundant HTTP round-trip via enable_oidc_discovery(). + This makes oauth_token_endpoint and related base class properties work + without any further network activity. Args: - openid_config: OpenID configuration dictionary + openid_config: OpenID configuration dictionary (well-known document) """ self._openid_config = openid_config if self._sam3_client: self._sam3_client.update_endpoints(openid_config) + + # Seed the base class OIDC cache directly from the fetched document. + try: + oidc_cfg = OIDCConfiguration.from_discovery_response(openid_config) + if oidc_cfg.is_complete(): + self._oidc_config = oidc_cfg + self._oidc_discovery_timestamp = time.time() + self._enable_oidc_discovery = True + issuer = openid_config.get("issuer", "").rstrip("/") + self._oidc_discovery_url = f"{issuer}/.well-known/openid-configuration" + logger.debug( + f"Base class OIDC cache seeded " + f"(token_endpoint={oidc_cfg.token_endpoint})" + ) + else: + logger.warning("OpenID config incomplete — base class OIDC cache not seeded") + except Exception as e: + logger.warning(f"Could not seed base class OIDC cache: {e}") + logger.debug("OpenID configuration updated") def set_remote_login_urls( @@ -1254,4 +1286,4 @@ class Magenta2Authenticator(BaseAuthenticator): try: self.settings_manager.clear_token(self.provider_name, self.country) except Exception: - pass + pass \ No newline at end of file diff --git a/lib/streaming_providers/providers/magenta2/sam3_client.py b/lib/streaming_providers/providers/magenta2/sam3_client.py index cff7294..cc526a0 100644 --- a/lib/streaming_providers/providers/magenta2/sam3_client.py +++ b/lib/streaming_providers/providers/magenta2/sam3_client.py @@ -51,6 +51,7 @@ class Sam3Client: line_auth_endpoint: str = None, backchannel_start_url: str = None, qr_code_url_template: str = None, + provider_name: str = "MagentaTV", ): """ Initialize SAM3 client with all required endpoints @@ -85,6 +86,8 @@ class Sam3Client: # Remote login handler (lazy initialized) self._remote_login_handler: Optional["RemoteLoginHandler"] = None + self.provider_name = provider_name + logger.debug( f"SAM3 client initialized - " f"Issuer: {issuer_url}, " @@ -459,10 +462,11 @@ class Sam3Client: # Create handler with CURRENT client ID self._remote_login_handler = RemoteLoginHandler( http_manager=self.http_manager, - sam3_client_id=self.sam3_client_id, # Use current value + 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, + provider_name=self.provider_name, # ← add this ) logger.info( diff --git a/lib/streaming_providers/providers/magenta2/token_flow_manager.py b/lib/streaming_providers/providers/magenta2/token_flow_manager.py index f458a73..294697e 100644 --- a/lib/streaming_providers/providers/magenta2/token_flow_manager.py +++ b/lib/streaming_providers/providers/magenta2/token_flow_manager.py @@ -631,26 +631,19 @@ class TokenFlowManager: def _get_yo_digital_via_remote_login(self) -> TokenFlowResult: """Try remote_login to get tvhubs + refresh_token, then chain to yo_digital""" try: - # Check if remote_login is available - if not hasattr(self.sam3_client, "can_use_remote_login"): - return TokenFlowResult( - success=False, - error="remote_login not available", - flow_path="yo_digital_via_remote_login", - ) - - if not self.sam3_client.can_use_remote_login(): + if self._remote_login_callback: + logger.info("Attempting remote_login flow via callback") + remote_token_data = self._remote_login_callback() + elif self.sam3_client.can_use_remote_login(): + logger.info("Attempting remote_login flow via sam3_client (no callback)") + remote_token_data = self.sam3_client.remote_login(scope="tvhubs offline_access") + else: return TokenFlowResult( success=False, error="remote_login not configured", flow_path="yo_digital_via_remote_login", ) - logger.info("Attempting remote_login flow") - - # Perform remote login - remote_token_data = self.sam3_client.remote_login(scope="tvhubs offline_access") - if not remote_token_data: return TokenFlowResult( success=False,