diff --git a/lib/streaming_providers/base/auth/base_oauth2_auth.py b/lib/streaming_providers/base/auth/base_oauth2_auth.py index 02cce7d..ea556c8 100644 --- a/lib/streaming_providers/base/auth/base_oauth2_auth.py +++ b/lib/streaming_providers/base/auth/base_oauth2_auth.py @@ -1,12 +1,16 @@ # streaming_providers/base/auth/base_oauth2_auth.py import base64 +import dataclasses import hashlib import html import re import secrets +import threading +import time import uuid from abc import abstractmethod -from typing import Any, Callable, Dict, Optional +from dataclasses import dataclass, field +from typing import Any, Callable, Dict, List, Optional from urllib.parse import parse_qs, urlencode, urlparse from ..models.proxy_models import ProxyConfig @@ -14,8 +18,58 @@ from ..utils.logger import logger from .base_auth import BaseAuthenticator, BaseAuthToken, TokenAuthLevel +@dataclass +class OIDCConfiguration: + """ + Stores OIDC discovery configuration from .well-known/openid-configuration. + + Fields follow OpenID Connect Discovery 1.0 specification: + https://openid.net/specs/openid-connect-discovery-1_0.html#ProviderMetadata + """ + issuer: str = "" + authorization_endpoint: str = "" + token_endpoint: str = "" + userinfo_endpoint: str = "" + jwks_uri: str = "" + scopes_supported: List[str] = field(default_factory=list) + grant_types_supported: List[str] = field(default_factory=list) + response_types_supported: List[str] = field(default_factory=list) + response_modes_supported: List[str] = field(default_factory=list) + token_endpoint_auth_methods_supported: List[str] = field(default_factory=list) + code_challenge_methods_supported: List[str] = field(default_factory=list) + revocation_endpoint: Optional[str] = None + end_session_endpoint: Optional[str] = None + device_authorization_endpoint: Optional[str] = None + registration_endpoint: Optional[str] = None + introspection_endpoint: Optional[str] = None + + @classmethod + def from_discovery_response(cls, data: Dict[str, Any]) -> 'OIDCConfiguration': + """ + Create OIDCConfiguration from .well-known/openid-configuration response. + + Uses dataclasses.fields() (public API) for forward-compatibility. + Unknown fields are silently ignored per OIDC spec extensibility. + """ + valid_fields = {f.name for f in dataclasses.fields(cls)} + filtered_data = {k: v for k, v in data.items() if k in valid_fields} + return cls(**filtered_data) + + def is_complete(self) -> bool: + """Check if essential endpoints are populated.""" + return bool( + self.authorization_endpoint and + self.token_endpoint and + self.issuer + ) + + class OAuth2Error(Exception): - """OAuth2-specific error with structured error information""" + """ + OAuth2-specific error with structured error information. + + Intentionally separate from HTTP transport errors for clearer error handling. + """ def __init__(self, error: str, error_description: str = None, error_uri: str = None): self.error = error @@ -32,23 +86,18 @@ class SessionAwareHTTPManager: def __init__(self, http_manager): self.http_manager = http_manager - self.cookies = {} - self.headers = {} + self.cookies: Dict[str, str] = {} + self.headers: Dict[str, str] = {} def get(self, url: str, **kwargs): """GET request with cookie handling""" headers = kwargs.get("headers", {}).copy() headers.update(self.headers) - - # Add cookies if self.cookies: cookie_str = "; ".join([f"{k}={v}" for k, v in self.cookies.items()]) headers["Cookie"] = cookie_str - kwargs["headers"] = headers response = self.http_manager.get(url, operation="oauth", **kwargs) - - # Update cookies from response self._update_cookies_from_response(response) return response @@ -56,16 +105,11 @@ class SessionAwareHTTPManager: """POST request with cookie handling""" headers = kwargs.get("headers", {}).copy() headers.update(self.headers) - - # Add cookies if self.cookies: cookie_str = "; ".join([f"{k}={v}" for k, v in self.cookies.items()]) headers["Cookie"] = cookie_str - kwargs["headers"] = headers response = self.http_manager.post(url, operation="oauth", **kwargs) - - # Update cookies from response self._update_cookies_from_response(response) return response @@ -77,49 +121,61 @@ class SessionAwareHTTPManager: class BaseOAuth2Authenticator(BaseAuthenticator): + """ + Base class for OAuth2/OIDC authentication with dynamic endpoint discovery. + + Production-hardened: no silent failures, consistent endpoint handling, + proper exception chaining, and defensive diagnostics. + """ + def __init__( - self, - provider_name: str, - settings_manager=None, - credentials=None, - country: Optional[str] = None, # ADD THIS PARAMETER - config_dir: Optional[str] = None, - enable_kodi_integration: bool = True, - proxy_config: Optional[ProxyConfig] = None, - http_manager=None, + self, + provider_name: str, + settings_manager=None, + credentials=None, + country: Optional[str] = None, + config_dir: Optional[str] = None, + enable_kodi_integration: bool = True, + proxy_config: Optional[ProxyConfig] = None, + http_manager=None, ): - # Pass country to parent BaseAuthenticator super().__init__( provider_name, settings_manager, credentials, - country=country, # ADD THIS LINE + country=country, config_dir=config_dir, enable_kodi_integration=enable_kodi_integration, ) + self._oauth_state = None self._pkce_verifier = None - - # Preserve _config if subclass already set it, otherwise initialize to None - if not hasattr(self, "_config"): - self._config = None - + self._config = None self._proxy_config = proxy_config self._auth_endpoint = None self._http_manager = http_manager self._token_expiry_buffer = 300 + # OIDC Discovery support (backward compatible - disabled by default) + self._oidc_config: Optional[OIDCConfiguration] = None + self._oidc_discovery_url: Optional[str] = None + self._enable_oidc_discovery: bool = False + self._oidc_discovery_lock = threading.Lock() + self._oidc_discovery_timestamp: Optional[float] = None + self._oidc_cache_ttl: int = 86400 # 24 hours + + # Telemetry + self._oidc_discovery_failures: int = 0 + self._oidc_discovery_successes: int = 0 + @property def http_manager(self): - """Safe access to http_manager - use provided one or create fallback""" + """Safe access to http_manager""" if self._http_manager is not None: return self._http_manager - logger.warning(f"No HTTP manager available for {self.provider_name}, creating one") - try: from ...base.network import HTTPManagerFactory - self._http_manager = HTTPManagerFactory.create_for_provider( self.provider_name, proxy_config=self._proxy_config, @@ -129,90 +185,262 @@ class BaseOAuth2Authenticator(BaseAuthenticator): except Exception as e: logger.warning(f"Error creating HTTP manager via factory: {e}, using minimal fallback") self._http_manager = self._create_minimal_http_manager() - return self._http_manager @http_manager.setter def http_manager(self, value): - """Allow setting http_manager""" self._http_manager = value @property def config(self): - """Safe access to config with fallback""" - import traceback - + """Config accessor - fails loudly if not initialized by subclass""" if self._config is not None: return self._config - - # Log who's calling this before config is set - logger.warning(f"Config accessed before initialization for {self.provider_name}") - logger.debug(f"Call stack:\n{''.join(traceback.format_stack()[-5:])}") - - # Only create minimal config if absolutely necessary - class MinimalConfig: - def __init__(self): - self.timeout = 30 - self.user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" - self.base_website = "https://example.com" - self.auth_endpoint = "https://auth.example.com" - - def get_base_headers(self): - return { - "User-Agent": self.user_agent, - "Accept": "application/json", - } - - def get_auth_headers(self): - return self.get_base_headers() - - self._config = MinimalConfig() - logger.warning( - f"Using minimal config for {self.provider_name} - subclass should set config" + raise RuntimeError( + f"Config not initialized for {self.provider_name}. " + "Subclass must set self._config in __init__ before accessing config." ) - return self._config @config.setter def config(self, value): - """Allow subclasses to set config""" self._config = value @staticmethod def _create_minimal_http_manager(): - """Create absolute minimal HTTP manager fallback""" + """Minimal HTTP manager fallback for testing only""" class MinimalHTTPManager: @staticmethod def get(url, operation=None, headers=None, **kwargs): import requests - return requests.get(url, headers=headers, **kwargs) @staticmethod def post(url, operation=None, headers=None, data=None, **kwargs): import requests - return requests.post(url, headers=headers, data=data, **kwargs) return MinimalHTTPManager() @property def auth_endpoint(self) -> str: - """Get authentication endpoint - subclasses can override""" + """Get authentication endpoint""" if hasattr(self, "_auth_endpoint") and self._auth_endpoint: return self._auth_endpoint - if hasattr(self.config, "auth_endpoint"): return self.config.auth_endpoint - raise NotImplementedError("Subclass must implement auth_endpoint or set _auth_endpoint") @auth_endpoint.setter def auth_endpoint(self, value): - """Allow setting auth_endpoint directly""" self._auth_endpoint = value - # Abstract properties + # ======================================================================== + # OIDC Discovery Support + # ======================================================================== + + def enable_oidc_discovery(self, discovery_url: str, cache_ttl: int = 86400) -> None: + """Enable OIDC discovery with smart URL normalization""" + normalized_url = discovery_url.rstrip('/') + if '/.well-known/' not in normalized_url: + normalized_url += '/.well-known/openid-configuration' + self._oidc_discovery_url = normalized_url + self._enable_oidc_discovery = True + self._oidc_cache_ttl = cache_ttl + self._oidc_config = None + self._oidc_discovery_timestamp = None + logger.debug(f"OIDC discovery enabled for {self.provider_name}: {normalized_url}") + + def _is_oidc_cache_valid(self) -> bool: + """Check if cached OIDC config is still valid""" + if not self._oidc_config or not self._oidc_discovery_timestamp: + return False + return (time.time() - self._oidc_discovery_timestamp) < self._oidc_cache_ttl + + def _validate_oidc_metadata(self, metadata: Dict[str, Any]) -> bool: + """Validate required OIDC metadata fields""" + required = ["issuer", "authorization_endpoint", "token_endpoint"] + missing = [f for f in required if not metadata.get(f)] + if missing: + logger.error(f"OIDC metadata missing required fields for {self.provider_name}: {missing}") + return False + issuer = metadata.get("issuer", "") + if issuer and not issuer.startswith("https://"): + logger.warning(f"OIDC issuer uses HTTP (insecure) for {self.provider_name}: {issuer}") + return True + + def discover_oidc_endpoints(self, force_refresh: bool = False) -> Optional[OIDCConfiguration]: + """Discover OIDC endpoints with consolidated error handling""" + if not self._enable_oidc_discovery or not self._oidc_discovery_url: + return None + + if not force_refresh and self._is_oidc_cache_valid(): + return self._oidc_config + + with self._oidc_discovery_lock: + if not force_refresh and self._is_oidc_cache_valid(): + return self._oidc_config + + try: + headers = self.config.get_base_headers() + response = self.http_manager.get( + self._oidc_discovery_url, + operation="oidc_discovery", + headers=headers, + timeout=getattr(self.config, "timeout", 30) + ) + + # Consolidated status logging inside single error handling block + if response.status_code >= 400: + status_msg = f"OIDC discovery HTTP {response.status_code} for {self.provider_name}" + if response.status_code == 404: + logger.error(f"{status_msg} - endpoint not found: {self._oidc_discovery_url}") + elif response.status_code == 429: + retry_after = response.headers.get("Retry-After", "unknown") + logger.warning(f"{status_msg} - rate limited, Retry-After: {retry_after}") + elif response.status_code >= 500: + logger.warning(f"{status_msg} - transient server error, will retry") + # Increment failure counter and return cached if available + self._oidc_discovery_failures += 1 + if self._oidc_config: + logger.debug(f"Using cached OIDC config for {self.provider_name}") + return self._oidc_config + return None + + discovery_data = response.json() + + if not self._validate_oidc_metadata(discovery_data): + logger.warning(f"OIDC metadata validation failed for {self.provider_name}") + self._oidc_discovery_failures += 1 + if self._oidc_config: + return self._oidc_config + return None + + self._oidc_config = OIDCConfiguration.from_discovery_response(discovery_data) + self._oidc_discovery_timestamp = time.time() + self._oidc_discovery_successes += 1 + + logger.info(f"OIDC discovery successful for {self.provider_name}") + return self._oidc_config + + except Exception as e: + exc_type = type(e).__name__ + logger.warning(f"OIDC discovery failed for {self.provider_name} ({exc_type}): {e}") + self._oidc_discovery_failures += 1 + if self._oidc_config: + return self._oidc_config + return None + + def reload_oidc_configuration(self) -> Optional[OIDCConfiguration]: + """Force reload OIDC configuration""" + logger.info(f"Reloading OIDC configuration for {self.provider_name}") + self._oidc_config = None + self._oidc_discovery_timestamp = None + return self.discover_oidc_endpoints(force_refresh=True) + + def get_oidc_discovery_stats(self) -> Dict[str, Any]: + """Get OIDC discovery telemetry""" + total = self._oidc_discovery_successes + self._oidc_discovery_failures + return { + "successes": self._oidc_discovery_successes, + "failures": self._oidc_discovery_failures, + "total_attempts": total, + "success_rate": self._oidc_discovery_successes / max(1, total), + } + + # ======================================================================== + # Endpoint Properties + # ======================================================================== + + @property + def oauth_authorize_endpoint(self) -> str: + """Get OAuth2 authorization endpoint with priority resolution""" + if self._enable_oidc_discovery: + config = self.discover_oidc_endpoints() + if config and config.authorization_endpoint: + return config.authorization_endpoint + if hasattr(self, "_authorization_endpoint") and self._authorization_endpoint: + return self._authorization_endpoint + if hasattr(self, "auth_endpoint") and self.auth_endpoint: + auth_endpoint = self.auth_endpoint + if auth_endpoint.endswith("/token"): + return auth_endpoint.replace("/token", "/auth") + elif "/protocol/openid-connect/token" in auth_endpoint: + return auth_endpoint.replace("/protocol/openid-connect/token", "/protocol/openid-connect/auth") + else: + return "/".join(auth_endpoint.split("/")[:-1]) + "/auth" + raise NotImplementedError( + f"Subclass must implement oauth_authorize_endpoint or enable OIDC discovery for {self.provider_name}" + ) + + @property + def oauth_token_endpoint(self) -> str: + """Get OAuth2 token endpoint with priority resolution""" + if self._enable_oidc_discovery: + config = self.discover_oidc_endpoints() + if config and config.token_endpoint: + return config.token_endpoint + if hasattr(self, "_token_endpoint") and self._token_endpoint: + return self._token_endpoint + return self.auth_endpoint + + @property + def oauth_userinfo_endpoint(self) -> Optional[str]: + """Get OIDC userinfo endpoint if available""" + if self._enable_oidc_discovery: + config = self.discover_oidc_endpoints() + if config and config.userinfo_endpoint: + return config.userinfo_endpoint + return None + + # ======================================================================== + # Capability Detection + # ======================================================================== + + def supports_pkce(self) -> bool: + """Check if PKCE S256 is supported via OIDC discovery""" + if self._enable_oidc_discovery: + config = self.discover_oidc_endpoints() + if config and config.code_challenge_methods_supported: + return "S256" in config.code_challenge_methods_supported + return True + + @property + def use_pkce(self) -> bool: + """Allow subclasses to disable PKCE for legacy providers""" + if hasattr(self, "_use_pkce"): + return self._use_pkce + return self.supports_pkce() + + def get_supported_grant_types(self) -> List[str]: + """Get supported grant types from OIDC discovery""" + if self._enable_oidc_discovery: + config = self.discover_oidc_endpoints() + if config and config.grant_types_supported: + return config.grant_types_supported + return ["authorization_code", "refresh_token"] + + def is_grant_type_supported(self, grant_type: str) -> bool: + """Check if a specific grant type is supported""" + supported = self.get_supported_grant_types() + return grant_type in supported if supported else True + + def _should_use_json_for_token_exchange(self, **kwargs) -> bool: + """Determine if token exchange should use JSON payload""" + if kwargs.get("use_json") is not None: + return bool(kwargs["use_json"]) + if self._enable_oidc_discovery: + config = self.discover_oidc_endpoints() + if config and config.token_endpoint_auth_methods_supported: + if 'client_secret_post' not in config.token_endpoint_auth_methods_supported: + if 'application/json' in config.token_endpoint_auth_methods_supported: + return True + return False + + # ======================================================================== + # Abstract Properties + # ======================================================================== + @property @abstractmethod def oauth_client_id(self) -> str: @@ -228,29 +456,16 @@ class BaseOAuth2Authenticator(BaseAuthenticator): def oauth_redirect_uri(self) -> str: pass - @property - def oauth_authorize_endpoint(self) -> str: - """Get OAuth2 authorization endpoint""" - if hasattr(self, "auth_endpoint"): - auth_endpoint = self.auth_endpoint - else: - logger.warning(f"auth_endpoint not defined for {self.provider_name}, using default") - return "https://auth.example.com/oauth2/auth" - - if auth_endpoint.endswith("/token"): - return auth_endpoint.replace("/token", "/auth") - elif "/protocol/openid-connect/token" in auth_endpoint: - return auth_endpoint.replace("/token", "/auth") - else: - return "/".join(auth_endpoint.split("/")[:-1]) + "/auth" - + # ======================================================================== # PKCE Implementation + # ======================================================================== + @staticmethod def generate_pkce_verifier() -> str: """Generate PKCE code verifier (RFC 7636)""" token = secrets.token_bytes(32) verifier = base64.urlsafe_b64encode(token).rstrip(b"=").decode("ascii") - logger.debug(f"Generated PKCE verifier: {verifier}") + logger.debug(f"Generated PKCE verifier: {verifier[:10]}...") return verifier @staticmethod @@ -258,19 +473,22 @@ class BaseOAuth2Authenticator(BaseAuthenticator): """Generate PKCE code challenge from verifier""" challenge = hashlib.sha256(verifier.encode("ascii")).digest() challenge_b64 = base64.urlsafe_b64encode(challenge).rstrip(b"=").decode("ascii") - logger.debug(f"Generated PKCE challenge: {challenge_b64}") + logger.debug(f"Generated PKCE challenge: {challenge_b64[:10]}...") return challenge_b64 + # ======================================================================== # OAuth2 State Management + # ======================================================================== + def generate_oauth_state(self) -> str: - """Generate secure state parameter for OAuth2 flow""" + """Generate secure state parameter""" state = str(uuid.uuid4()) self._oauth_state = state return state @staticmethod def generate_oauth_nonce() -> str: - """Generate secure nonce parameter for OAuth2 flow""" + """Generate secure nonce parameter""" return str(uuid.uuid4()) @staticmethod @@ -279,290 +497,290 @@ class BaseOAuth2Authenticator(BaseAuthenticator): if not received_state or not original_state: logger.warning("OAuth2 state validation failed: missing state parameters") return False - is_valid = received_state == original_state if not is_valid: logger.warning("OAuth2 state validation failed: state mismatch") - return is_valid + # ======================================================================== # Session Management + # ======================================================================== + def _create_oauth_session(self) -> SessionAwareHTTPManager: """Create a session-aware HTTP manager for OAuth flows""" session = SessionAwareHTTPManager(self.http_manager) - session.headers.update( - { - "User-Agent": self.config.user_agent, - "Referer": getattr(self.config, "base_website", ""), - "Origin": getattr(self.config, "base_website", ""), - } - ) + session.headers.update({ + "User-Agent": self.config.user_agent, + "Referer": getattr(self.config, "base_website", ""), + "Origin": getattr(self.config, "base_website", ""), + }) return session - # Complete Client Credentials Flow + # ======================================================================== + # Client Credentials Flow - Fix: Use oauth_token_endpoint + # ======================================================================== + def _perform_oauth_client_credentials_flow(self) -> Dict[str, Any]: """ - Complete manual implementation of OAuth2 client credentials flow - Uses provider-specific headers and payload formatting + OAuth2 client credentials flow. + + Uses oauth_token_endpoint to respect OIDC discovery. """ try: logger.debug(f"Starting OAuth2 client credentials flow for {self.provider_name}") - headers = self._get_auth_headers() data = self._build_auth_payload() + # Use oauth_token_endpoint instead of auth_endpoint response = self.http_manager.post( - self.auth_endpoint, operation="auth", headers=headers, data=data + self.oauth_token_endpoint, operation="auth", headers=headers, data=data ) - self._check_oauth_error_response(response) response.raise_for_status() - token_data = response.json() logger.debug(f"OAuth2 client credentials flow successful for {self.provider_name}") return token_data - except OAuth2Error: raise except Exception as e: logger.error(f"OAuth2 client credentials flow failed for {self.provider_name}: {e}") raise Exception(f"OAuth2 client credentials flow failed: {e}") + # ======================================================================== # Authorization URL Building - def _build_authorization_url(self, extra_params: Dict[str, Any] = None) -> tuple[str, str, str]: - """Build authorization URL with PKCE for authorization code flow""" - code_verifier = self.generate_pkce_verifier() - code_challenge = self.generate_pkce_challenge(code_verifier) - state = self.generate_oauth_state() + # ======================================================================== + def _build_authorization_url(self, extra_params: Dict[str, Any] = None) -> tuple[str, str, str]: + """Build authorization URL with optional PKCE""" + state = self.generate_oauth_state() params = { "response_type": "code", "client_id": self.oauth_client_id, "redirect_uri": self.oauth_redirect_uri, "scope": self.oauth_scope, "state": state, - "code_challenge": code_challenge, - "code_challenge_method": "S256", } - + code_verifier = "" + if self.use_pkce: + code_verifier = self.generate_pkce_verifier() + code_challenge = self.generate_pkce_challenge(code_verifier) + params.update({ + "code_challenge": code_challenge, + "code_challenge_method": "S256", + }) + else: + logger.warning(f"PKCE disabled for {self.provider_name} - ensure provider supports secure flows") if extra_params: params.update(extra_params) - authorization_url = f"{self.oauth_authorize_endpoint}?{urlencode(params)}" - return authorization_url, state, code_verifier + # ======================================================================== # Authorization Code Exchange + # ======================================================================== + def _exchange_authorization_code_for_token( - self, authorization_code: str, code_verifier: str, state: str = None, **kwargs + self, authorization_code: str, code_verifier: str, state: str = None, **kwargs ) -> Dict[str, Any]: - """ - Exchange authorization code for access token (PKCE flow) - Enhanced to support provider-specific customizations - """ + """Exchange authorization code for access token with flexible payload format""" try: logger.debug(f"Exchanging authorization code for token for {self.provider_name}") - - # Allow subclasses to override the default payload data = self._build_token_exchange_payload( authorization_code=authorization_code, code_verifier=code_verifier, state=state, **kwargs, ) - - # Allow subclasses to override headers headers = self._get_token_exchange_headers(**kwargs) + endpoint = kwargs.get('token_endpoint') or self._get_token_exchange_endpoint(**kwargs) - # Allow subclasses to override data format and endpoint - endpoint = self._get_token_exchange_endpoint(**kwargs) - use_json = self._should_use_json_for_token_exchange(**kwargs) + # Fix: Explicit None check for boolean use_json parameter + use_json = kwargs.get('use_json') + if use_json is None: + use_json = self._should_use_json_for_token_exchange(**kwargs) request_kwargs = { "operation": "auth", "headers": headers, "timeout": getattr(self.config, "timeout", 30), } - if use_json: request_kwargs["json_data"] = data else: request_kwargs["data"] = urlencode(data).encode() response = self.http_manager.post(endpoint, **request_kwargs) - self._check_oauth_error_response(response) response.raise_for_status() - token_data = response.json() logger.debug(f"Authorization code exchange successful for {self.provider_name}") return token_data - except OAuth2Error: raise except Exception as e: logger.error(f"Authorization code exchange failed for {self.provider_name}: {e}") raise Exception(f"Authorization code exchange failed: {e}") - # New flexible methods that subclasses can override + # ======================================================================== + # Flexible Methods for Subclass Override + # ======================================================================== + def _build_token_exchange_payload( - self, authorization_code: str, code_verifier: str, state: str = None, **kwargs + self, authorization_code: str, code_verifier: str, state: str = None, **kwargs ) -> Dict[str, Any]: - """Build token exchange payload - subclasses can override for custom parameters""" + """Build token exchange payload""" data = { "grant_type": "authorization_code", "client_id": self.oauth_client_id, "code": authorization_code, "redirect_uri": self.oauth_redirect_uri, - "code_verifier": code_verifier, } - + if self.use_pkce and code_verifier: + data["code_verifier"] = code_verifier client_secret = getattr(self.credentials, "client_secret", None) if client_secret: data["client_secret"] = client_secret - return data def _get_token_exchange_headers(self, **kwargs) -> Dict[str, str]: - """Get token exchange headers - subclasses can override for custom headers""" + """Get token exchange headers with format-aware Content-Type""" headers = self._get_auth_headers() - - # Ensure Content-Type is appropriate - if kwargs.get("use_json", False) or self._should_use_json_for_token_exchange(**kwargs): - headers["Content-Type"] = "application/json" - else: - headers["Content-Type"] = "application/x-www-form-urlencoded" - + use_json = kwargs.get("use_json") + if use_json is None: + use_json = self._should_use_json_for_token_exchange(**kwargs) + headers["Content-Type"] = "application/json" if use_json else "application/x-www-form-urlencoded" return headers def _get_token_exchange_endpoint(self, **kwargs) -> str: - """Get token exchange endpoint - subclasses can override for custom endpoints""" - return self.auth_endpoint + """Get token exchange endpoint""" + return self.oauth_token_endpoint - @staticmethod - def _should_use_json_for_token_exchange(**kwargs) -> bool: - """Determine if token exchange should use JSON - subclasses can override""" - return False # Default to form-encoded for OAuth2 compliance + # ======================================================================== + # Generic Form-Based Login Flow - Fix: Preserve exception chain + # ======================================================================== - # Generic Form-Based Login Flow def _perform_generic_form_login( - self, - username: str, - password: str, - form_selector_pattern: str, - login_fields: Dict[str, str], - extra_params: Dict[str, Any] = None, - additional_form_data: Dict[str, str] = None, + self, + username: str, + password: str, + form_selector_pattern: str, + login_fields: Dict[str, str], + extra_params: Dict[str, Any] = None, + additional_form_data: Dict[str, str] = None, ) -> Dict[str, Any]: - """ - Generic OAuth2 form-based login flow + """Generic OAuth2 form-based login flow with proper exception handling""" + auth_url, state, code_verifier = self._build_authorization_url(extra_params) + session = self._create_oauth_session() - Args: - username: User's username - password: User's password - form_selector_pattern: Regex to find login form action URL - login_fields: Field names mapping (e.g., {'username': 'email', 'password': 'pass'}) - extra_params: Additional authorization URL parameters - additional_form_data: Additional form fields to submit + # Step 1: Get login form + auth_response = session.get(auth_url, timeout=self.config.timeout) + auth_response.raise_for_status() - Returns: - Token data dictionary - """ - try: - auth_url, state, code_verifier = self._build_authorization_url(extra_params) + # Step 2: Extract login form action URL + form_matches = re.findall(form_selector_pattern, auth_response.text) + if not form_matches: + raise Exception(f"Could not find login form using pattern: {form_selector_pattern}") + login_url = html.unescape(form_matches[0]) - session = self._create_oauth_session() + # Step 3: Build login data + login_data = {} + if additional_form_data: + login_data.update(additional_form_data) + login_data[login_fields.get("username", "username")] = username + login_data[login_fields.get("password", "password")] = password - # Step 1: Get login form - auth_response = session.get(auth_url, timeout=self.config.timeout) - auth_response.raise_for_status() + # Step 4: Submit login credentials + login_response = session.post( + login_url, + data=login_data, + timeout=self.config.timeout, + allow_redirects=False, + ) - # Step 2: Extract login form action URL - form_matches = re.findall(form_selector_pattern, auth_response.text) - if not form_matches: - raise Exception(f"Could not find login form using pattern: {form_selector_pattern}") - - login_url = html.unescape(form_matches[0]) - - # Step 3: Build login data - login_data = {} - if additional_form_data: - login_data.update(additional_form_data) - - login_data[login_fields.get("username", "username")] = username - login_data[login_fields.get("password", "password")] = password - - # Step 4: Submit login credentials - login_response = session.post( - login_url, - data=login_data, - timeout=self.config.timeout, - allow_redirects=False, - ) - - # Step 5: Handle redirect and extract authorization code - if login_response.status_code in [302, 303]: - redirect_url = login_response.headers.get("Location") - if not redirect_url: - raise Exception("No redirect URL found after login") + # Step 5: Handle redirect + if login_response.status_code in [302, 303]: + redirect_url = login_response.headers.get("Location") + if not redirect_url: + raise Exception("No redirect URL found after login") + else: + if "code=" in login_response.url: + redirect_url = login_response.url else: - redirect_response = session.get(login_response.url, timeout=self.config.timeout) - redirect_url = redirect_response.url + raise Exception( + f"Login did not produce expected redirect. Status: {login_response.status_code}. " + f"Check provider login flow implementation or credentials." + ) - # Step 6: Validate and extract authorization code - is_valid, error_msg, authorization_code = self.validate_authentication_response( - redirect_url, state - ) - if not is_valid: - raise Exception(f"Authentication response validation failed: {error_msg}") + # Step 6: Validate and extract authorization code + is_valid, error_msg, authorization_code = self.validate_authentication_response( + redirect_url, state + ) + if not is_valid: + raise Exception(f"Authentication response validation failed: {error_msg}") - # Step 7: Exchange code for token - token_data = self._exchange_authorization_code_for_token( + # Step 7: Exchange code for token + # Fix: Let OAuth2Error propagate; wrap other exceptions with cause chain + try: + return self._exchange_authorization_code_for_token( authorization_code=authorization_code, code_verifier=code_verifier, state=state, ) - - return token_data - + except OAuth2Error: + raise except Exception as e: - raise Exception(f"OAuth2 form-based login failed: {e}") + raise Exception(f"OAuth2 form-based login failed: {e}") from e + + # ======================================================================== + # Token Refresh - Fix: Consistent payload encoding + correct endpoint + # ======================================================================== + + def _build_refresh_payload(self) -> Dict[str, Any]: + """Build refresh token payload - consistent with token exchange""" + data = { + "grant_type": "refresh_token", + "refresh_token": self._current_token.refresh_token, + "client_id": self.oauth_client_id, + } + client_secret = getattr(self.credentials, "client_secret", None) + if client_secret: + data["client_secret"] = client_secret + return data - # Token Refresh def _refresh_oauth_token(self) -> Optional[BaseAuthToken]: - """Complete manual token refresh implementation""" + """ + Token refresh with consistent endpoint and payload handling. + + Uses oauth_token_endpoint to respect OIDC discovery. + Uses consistent Content-Type logic via _should_use_json_for_token_exchange. + """ if not self._current_token or not self._current_token.refresh_token: logger.debug(f"No refresh token available for {self.provider_name}") return None try: logger.debug(f"Refreshing OAuth2 token for {self.provider_name}") - - data = { - "grant_type": "refresh_token", - "refresh_token": self._current_token.refresh_token, - "client_id": self.oauth_client_id, - } - - client_secret = getattr(self.credentials, "client_secret", None) - if client_secret: - data["client_secret"] = client_secret - + data = self._build_refresh_payload() headers = self._get_auth_headers() - encoded_data = urlencode(data).encode() + # Apply consistent Content-Type logic + if self._should_use_json_for_token_exchange(): + headers["Content-Type"] = "application/json" + request_kwargs = {"json_data": data} + else: + headers["Content-Type"] = "application/x-www-form-urlencoded" + request_kwargs = {"data": urlencode(data).encode()} + + # Use oauth_token_endpoint to respect OIDC discovery response = self.http_manager.post( - self.auth_endpoint, operation="auth", headers=headers, data=encoded_data + self.oauth_token_endpoint, operation="auth", headers=headers, **request_kwargs ) - self._check_oauth_error_response(response) response.raise_for_status() - new_token_data = response.json() refreshed_token = self._create_token_from_response(new_token_data) logger.info(f"OAuth2 token refresh successful for {self.provider_name}") return refreshed_token - except OAuth2Error as e: logger.warning(f"OAuth2 token refresh failed for {self.provider_name}: {e}") return None @@ -570,197 +788,148 @@ class BaseOAuth2Authenticator(BaseAuthenticator): logger.warning(f"OAuth2 token refresh failed for {self.provider_name}: {e}") return None - # Error Response Handling + # ======================================================================== + # Error Response Handling - Fix: Clean error handling + # ======================================================================== + @staticmethod def _check_oauth_error_response(response): - """Check response for OAuth2 error and raise OAuth2Error if found""" - try: - if response.status_code >= 400: - try: - error_data = response.json() - if "error" in error_data: - raise OAuth2Error( - error=error_data.get("error"), - error_description=error_data.get("error_description"), - error_uri=error_data.get("error_uri"), - ) - except (ValueError, KeyError): - pass - except OAuth2Error: - raise - except Exception: - pass + """Check response for OAuth2 error - no silent swallowing""" + if response.status_code >= 400: + try: + error_data = response.json() + except ValueError: + # Not JSON; let raise_for_status() handle it + return + if "error" in error_data: + raise OAuth2Error( + error=error_data.get("error"), + error_description=error_data.get("error_description"), + error_uri=error_data.get("error_uri"), + ) - # Dynamic Client ID Extraction - def _extract_client_id_from_js( - self, main_page_url: str, js_file_pattern: str, client_id_pattern: str - ) -> Optional[str]: + # ======================================================================== + # JS Extraction Helpers - Fix: Selective exception handling + # ======================================================================== + + def _extract_from_js( + self, + main_page_url: str, + js_file_pattern: str, + content_pattern: str, + parse_function: Optional[Callable[[str], Any]] = None, + extract_type: str = "value", + ) -> Optional[Any]: """ - Extract client ID from provider's JavaScript + Generic helper to extract content from provider's JavaScript. - Args: - main_page_url: URL of the main page containing script references - js_file_pattern: Regex pattern to find the JS file URL - client_id_pattern: Regex pattern to extract client ID from JS content - - Returns: - Extracted client ID or None + Re-raises network/transport errors; only suppresses parsing-related errors. """ try: headers = self.config.get_base_headers() - response = self.http_manager.get(main_page_url, operation="api", headers=headers) response.raise_for_status() - js_matches = re.findall(js_file_pattern, response.text) if not js_matches: logger.warning(f"Could not find JS file using pattern: {js_file_pattern}") return None - js_url = main_page_url.rstrip("/") + "/" + js_matches[-1].lstrip("/") - js_response = self.http_manager.get(js_url, operation="api", headers=headers) js_response.raise_for_status() - - client_id_match = re.search(client_id_pattern, js_response.text) - if not client_id_match: - logger.warning(f"Could not find client ID using pattern: {client_id_pattern}") + content_match = re.search(content_pattern, js_response.text) + if not content_match: + logger.warning(f"Could not find content using pattern: {content_pattern}") return None - - return client_id_match.group(1) - - except Exception as e: - logger.error(f"Error extracting client ID from JS: {e}") + extracted = content_match.group(1) + if parse_function: + return parse_function(extracted) + return extracted + # Re-raise network/transport errors; only suppress parsing errors + except (ConnectionError, TimeoutError, OSError) as e: + logger.error(f"Network error extracting {extract_type} from JS for {self.provider_name}: {e}") + raise + except (AttributeError, ValueError, re.error) as e: + # Parsing/regex errors are recoverable - log and return None + logger.warning(f"Parse error extracting {extract_type} from JS for {self.provider_name}: {e}") return None + except Exception as e: + # Log unexpected errors but re-raise to avoid silent failures + logger.error(f"Unexpected error extracting {extract_type} from JS for {self.provider_name}: {e}") + raise + + def _extract_client_id_from_js( + self, main_page_url: str, js_file_pattern: str, client_id_pattern: str + ) -> Optional[str]: + """Extract client ID from JavaScript (wrapper)""" + return self._extract_from_js( + main_page_url=main_page_url, + js_file_pattern=js_file_pattern, + content_pattern=client_id_pattern, + extract_type="client_id", + ) - # Generic Config Extraction from JS def _extract_config_from_js( - self, - main_page_url: str, - js_file_pattern: str, - config_pattern: str, - parse_function: Callable[[str], Dict[str, Any]], + self, + main_page_url: str, + js_file_pattern: str, + config_pattern: str, + parse_function: Callable[[str], Dict[str, Any]], ) -> Optional[Dict[str, Any]]: - """ - Generic JS config extraction - - Args: - main_page_url: URL of the main page - js_file_pattern: Regex to find JS file - config_pattern: Regex to extract config section - parse_function: Function to parse the config string into a dict - - Returns: - Parsed configuration dictionary or None - """ - try: - headers = self.config.get_base_headers() - - response = self.http_manager.get(main_page_url, operation="api", headers=headers) - response.raise_for_status() - - js_matches = re.findall(js_file_pattern, response.text) - if not js_matches: - return None - - js_url = main_page_url.rstrip("/") + "/" + js_matches[-1].lstrip("/") - - js_response = self.http_manager.get(js_url, operation="api", headers=headers) - js_response.raise_for_status() - - config_match = re.search(config_pattern, js_response.text) - if not config_match: - return None - - config_str = config_match.group(1) - return parse_function(config_str) - - except Exception as e: - logger.error(f"Error extracting config from JS: {e}") - return None + """Extract config from JavaScript (wrapper)""" + return self._extract_from_js( + main_page_url=main_page_url, + js_file_pattern=js_file_pattern, + content_pattern=config_pattern, + parse_function=parse_function, + extract_type="config", + ) + # ======================================================================== # Token Upgrade Support - def _should_upgrade_to_user_token(self, token: BaseAuthToken) -> bool: - """ - Check if token should be upgraded - now uses the base class logic + # ======================================================================== - Override in subclass only if provider has specific upgrade rules - """ + def _should_upgrade_to_user_token(self, token: BaseAuthToken) -> bool: + """Check if token should be upgraded""" return self.should_upgrade_token(token) def _get_effective_credentials(self): - """ - Get effective credentials with priority: - 1. Stored user credentials (if available) - 2. Current credentials if valid - 3. Fallback credentials - """ + """Get effective credentials with priority resolution""" from ...base.auth.credentials import UserPasswordCredentials - - # ALWAYS check stored credentials first for user credentials stored_creds = self.settings_manager.get_provider_credentials(self.provider_name) if stored_creds and isinstance(stored_creds, UserPasswordCredentials): if stored_creds.validate(): return stored_creds - - # Then use current credentials if self.credentials and self.credentials.validate(): return self.credentials - - # Finally fallback fallback = self.get_fallback_credentials() self.credentials = fallback return fallback def get_bearer_token(self, force_refresh: bool = False, force_upgrade: bool = False) -> str: - """ - Get bearer token with automatic upgrade support - - Args: - force_refresh: Force token refresh even if not expired - force_upgrade: Force token upgrade attempt regardless of current level - - Returns: - Bearer token string - """ + """Get bearer token with automatic upgrade support""" logger.debug( f"get_bearer_token called: force_refresh={force_refresh}, force_upgrade={force_upgrade}" ) - - # Get current token (authenticate if needed) current_token = self.authenticate(force_refresh=force_refresh) - - # Classify token if needed if current_token.auth_level == TokenAuthLevel.UNKNOWN: current_token.auth_level = self._classify_token(current_token) logger.debug(f"Token classified as: {current_token.auth_level.value}") - # Check if upgrade is needed/requested + # Cache upgrade check result to avoid duplicate calls should_upgrade = force_upgrade or self._should_upgrade_to_user_token(current_token) if should_upgrade and not force_refresh: - logger.info( - f"Token upgrade triggered (force={force_upgrade}, auto={self._should_upgrade_to_user_token(current_token)})" - ) - + upgrade_reason = "forced" if force_upgrade else "auto" + logger.info(f"Token upgrade triggered ({upgrade_reason}) for {self.provider_name}") original_credentials = self.credentials - try: - # Get effective credentials (prioritizes stored user credentials) self.credentials = self._get_effective_credentials() - if not self.credentials or not self.credentials.validate(): logger.debug("No valid credentials for upgrade") return current_token.bearer_token - - # Perform authentication with new credentials user_token = self._perform_authentication() - if user_token and not user_token.is_expired: - # Classify the new token user_token.auth_level = self._classify_token(user_token) - - # Verify it's actually an upgrade if user_token.is_user_authenticated(): logger.info("Successfully upgraded to user token") self._current_token = user_token @@ -776,25 +945,23 @@ class BaseOAuth2Authenticator(BaseAuthenticator): logger.warning("User authentication failed, keeping current token") self.credentials = original_credentials return current_token.bearer_token - except Exception as e: logger.error(f"Token upgrade failed: {e}") self.credentials = original_credentials return current_token.bearer_token - return current_token.bearer_token if current_token else "" + # ======================================================================== # Main Authentication Flow + # ======================================================================== + def _perform_authentication(self) -> BaseAuthToken: """Complete OAuth2 authentication based on credential type""" from ...base.auth.credentials import ClientCredentials, UserPasswordCredentials - logger.debug( f"Starting OAuth2 authentication for {self.provider_name} with credential type: {type(self.credentials)}" ) - original_credentials = self.credentials - try: if isinstance(self.credentials, UserPasswordCredentials): logger.info(f"Attempting OAuth2 user authentication for {self.provider_name}") @@ -802,42 +969,35 @@ class BaseOAuth2Authenticator(BaseAuthenticator): self.credentials.username, self.credentials.password ) elif isinstance(self.credentials, ClientCredentials): - logger.info( - f"Attempting OAuth2 client credentials authentication for {self.provider_name}" - ) + logger.info(f"Attempting OAuth2 client credentials authentication for {self.provider_name}") token_data = self._perform_oauth_client_credentials_flow() else: raise Exception(f"Unsupported credential type for OAuth2: {type(self.credentials)}") - token = self._create_token_from_response(token_data) logger.info(f"OAuth2 authentication successful for {self.provider_name}") return token - except Exception as e: logger.error(f"Primary OAuth2 authentication failed for {self.provider_name}: {e}") - if isinstance(original_credentials, UserPasswordCredentials): - logger.info( - f"User authentication failed, falling back to client credentials for {self.provider_name}" - ) + logger.info(f"User authentication failed, falling back to client credentials for {self.provider_name}") try: self.credentials = self.get_fallback_credentials() token_data = self._perform_oauth_client_credentials_flow() result = self._create_token_from_response(token_data) - logger.info( - f"Successfully fell back to client credentials for {self.provider_name}" - ) + logger.info(f"Successfully fell back to client credentials for {self.provider_name}") return result except Exception as fallback_error: self.credentials = original_credentials logger.error( - f"Fallback to client credentials also failed for {self.provider_name}: {fallback_error}" - ) - raise e + f"Fallback to client credentials also failed for {self.provider_name}: {fallback_error}") + raise else: raise e + # ======================================================================== # Token Management + # ======================================================================== + @abstractmethod def _create_token_from_response(self, response_data: Dict[str, Any]) -> BaseAuthToken: """Create provider-specific token from OAuth2 response""" @@ -847,37 +1007,66 @@ class BaseOAuth2Authenticator(BaseAuthenticator): """Override base refresh to use manual OAuth2 refresh flow""" return self._refresh_oauth_token() - # Status and Diagnostics + # ======================================================================== + # Status and Diagnostics - Fix: Defensive endpoint access + # ======================================================================== + def get_authentication_status(self) -> Dict[str, Any]: """Get comprehensive OAuth2 authentication status information""" status = super().get_authentication_status() + # Wrap endpoint property access defensively to avoid triggering + # OIDC discovery during diagnostic calls, which could hang or throw + oauth_authorize_ep = "" + oauth_token_ep = "" + try: + oauth_authorize_ep = self.oauth_authorize_endpoint + except (NotImplementedError, RuntimeError) as e: + oauth_authorize_ep = f"" + try: + oauth_token_ep = self.oauth_token_endpoint + except (NotImplementedError, RuntimeError) as e: + oauth_token_ep = f"" + oauth_status = { "oauth_client_id": self.oauth_client_id, "oauth_scope": self.oauth_scope, "oauth_redirect_uri": self.oauth_redirect_uri, - "oauth_authorize_endpoint": self.oauth_authorize_endpoint, + "oauth_authorize_endpoint": oauth_authorize_ep, + "oauth_token_endpoint": oauth_token_ep, "authentication_flow": "oauth2", - "pkce_support": True, + "pkce_support": self.supports_pkce(), + "pkce_enabled": self.use_pkce, "proxy_support": hasattr(self, "http_manager"), "credential_type": type(self.credentials).__name__, "has_refresh_token": bool(self._current_token and self._current_token.refresh_token), + "oidc_discovery_enabled": self._enable_oidc_discovery, } + if self._enable_oidc_discovery: + config = self._oidc_config + oauth_status["oidc_config_cached"] = config is not None + if config: + oauth_status["oidc_issuer"] = config.issuer + oauth_status["oidc_grant_types"] = config.grant_types_supported + oauth_status["oidc_pkce_methods"] = config.code_challenge_methods_supported + oauth_status["oidc_discovery_stats"] = self.get_oidc_discovery_stats() + if self._current_token: - oauth_status.update( - { - "token_expires_in": self._current_token.expires_in, - "token_issued_at": self._current_token.issued_at, - "token_is_expired": self._current_token.is_expired, - "token_needs_refresh": self._current_token.needs_refresh(), - } - ) + oauth_status.update({ + "token_expires_in": self._current_token.expires_in, + "token_issued_at": self._current_token.issued_at, + "token_is_expired": self._current_token.is_expired, + "token_needs_refresh": self._current_token.needs_refresh(), + }) status.update(oauth_status) return status + # ======================================================================== # Utility Methods + # ======================================================================== + @staticmethod def extract_authorization_code_from_url(url: str) -> Optional[str]: """Extract authorization code from callback URL""" @@ -890,41 +1079,33 @@ class BaseOAuth2Authenticator(BaseAuthenticator): return None def validate_authentication_response( - self, url: str, original_state: str + self, url: str, original_state: str ) -> tuple[bool, Optional[str], Optional[str]]: - """ - Validate OAuth2 authentication response - Returns: (is_valid, error_message, authorization_code) - """ + """Validate OAuth2 authentication response""" try: parsed = urlparse(url) query_params = parse_qs(parsed.query) - if "error" in query_params: error = query_params["error"][0] error_description = query_params.get("error_description", [""])[0] return False, f"{error}: {error_description}", None - received_state = query_params.get("state", [None])[0] if not self.validate_oauth_state(received_state, original_state): return False, "State validation failed", None - authorization_code = query_params.get("code", [None])[0] if not authorization_code: return False, "No authorization code in response", None - return True, None, authorization_code - except Exception as e: return False, f"Error processing authentication response: {e}", None - # Abstract method for provider-specific authorization code flow + # ======================================================================== + # Abstract Method + # ======================================================================== + @abstractmethod def _perform_oauth_authorization_code_flow( - self, username: str, password: str + self, username: str, password: str ) -> Dict[str, Any]: - """ - Perform OAuth2 authorization code flow with PKCE for user login - Must be implemented by subclasses for provider-specific login forms - """ - pass + """Perform OAuth2 authorization code flow with PKCE for user login""" + pass \ No newline at end of file diff --git a/lib/streaming_providers/providers/joyn/auth.py b/lib/streaming_providers/providers/joyn/auth.py index 44d35cc..d17c110 100644 --- a/lib/streaming_providers/providers/joyn/auth.py +++ b/lib/streaming_providers/providers/joyn/auth.py @@ -1,5 +1,7 @@ # streaming_providers/providers/joyn/auth.py # -*- coding: utf-8 -*- +import base64 +import json import re import time import uuid @@ -16,124 +18,36 @@ from .constants import ( COUNTRY_TENANT_MAPPING, DEFAULT_COUNTRY, DEFAULT_PLATFORM, - DEFAULT_REQUEST_TIMEOUT, DEVICE_IDS, + JOYN_7PASS_BASE_URL, + JOYN_7PASS_ENDPOINTS, JOYN_AUTH_ENDPOINTS, + JOYN_AUTH_HEADERS_BASE, JOYN_CLIENT_VERSION, JOYN_DOMAINS, JOYN_OAUTH_SCOPE, - JOYN_SSO_DISCOVERY_URL, JOYN_USER_AGENT, SUPPORTED_COUNTRIES, ) -class JoynSSODiscovery: - """Service to discover SSO endpoints dynamically""" - - def __init__( - self, - http_manager, - country: str = DEFAULT_COUNTRY, - platform: str = DEFAULT_PLATFORM, - ): - self.http_manager = http_manager - self.country = country - self.platform = platform - self._endpoints_cache = None - self._cache_timestamp = None - self._cache_ttl = 3600 # 1 hour cache - - @staticmethod - def get_fallback_endpoints() -> Dict[str, str]: - """Fallback endpoints if discovery fails""" - return { - "device-login": "https://sso.joyn.de/ci", - "device-register": "https://sso.joyn.de/cr", - "web-login": "https://auth.7pass.de/authz-srv/authz", - "redeem-token": "https://auth.joyn.de/auth/7pass/token", - } - - def get_endpoints(self, force_refresh: bool = False) -> Dict[str, str]: - """Get SSO endpoints, with caching""" - if ( - self._endpoints_cache - and not force_refresh - and time.time() - self._cache_timestamp < self._cache_ttl - ): - return self._endpoints_cache - - try: - params = { - "client_id": DEVICE_IDS[self.platform], - "client_name": self.platform, - } - - response = self.http_manager.get( - JOYN_SSO_DISCOVERY_URL, operation="sso_discovery", params=params - ) - response.raise_for_status() - - self._endpoints_cache = response.json() - self._cache_timestamp = time.time() - logger.debug( - f"SSO discovery successful, endpoints: {list(self._endpoints_cache.keys())}" - ) - return self._endpoints_cache - - except Exception as e: - # Fallback to hardcoded endpoints if discovery fails - logger.warning(f"SSO discovery failed, using fallback: {e}") - return self.get_fallback_endpoints() - - def get_auth_endpoint(self, auth_type: str = None) -> str: - """Get specific auth endpoint by type""" - # If no auth_type specified, use platform-specific login endpoint - if auth_type is None: - auth_type = f"{self.platform}-login" - - endpoints = self.get_endpoints() - endpoint = endpoints.get(auth_type) - if not endpoint: - logger.warning(f"Auth endpoint '{auth_type}' not found, using fallback") - fallback = self.get_fallback_endpoints() - # Try platform-specific first, then generic web-login - endpoint = ( - fallback.get(auth_type) - or fallback.get(f"{self.platform}-login") - or fallback.get("web-login", "") - ) - return endpoint - - @dataclass class JoynCredentials(ClientCredentials): - """ - Joyn-specific credentials for client credentials flow (anonymous auth) - """ - + """Joyn-specific credentials for client credentials flow (anonymous auth)""" client_name: str = DEFAULT_PLATFORM country: str = DEFAULT_COUNTRY distribution_tenant: Optional[str] = field(default=None) def __post_init__(self): - # Set client_id from constant if not provided if not self.client_id: self.client_id = DEVICE_IDS.get(self.client_name, DEVICE_IDS[DEFAULT_PLATFORM]) - if not self.distribution_tenant and self.country in COUNTRY_TENANT_MAPPING: self.distribution_tenant = COUNTRY_TENANT_MAPPING[self.country] def validate(self) -> bool: - """Validate Joyn credentials""" - if not self.client_id or not self.client_name: - return False - if self.country not in SUPPORTED_COUNTRIES: - return False - return True + return bool(self.client_id and self.client_name and self.country in SUPPORTED_COUNTRIES) def to_auth_payload(self) -> Dict[str, Any]: - """Convert to authentication payload for Joyn's anonymous auth endpoint""" return { "client_id": self.client_id, "client_name": self.client_name, @@ -147,14 +61,10 @@ class JoynCredentials(ClientCredentials): @dataclass class JoynAuthToken(BaseAuthToken): - """ - Joyn-specific authentication token - """ - + """Joyn-specific authentication token""" refresh_token: Optional[str] = field(default="") def to_dict(self) -> Dict[str, Any]: - """Convert token to dictionary""" return { "access_token": self.access_token, "refresh_token": self.refresh_token or "", @@ -164,148 +74,54 @@ class JoynAuthToken(BaseAuthToken): } def get_jwt_claims(self) -> Optional[Dict[str, Any]]: - """Extract JWT claims from access token for debugging and classification""" - import base64 - import json - + """Extract JWT claims from access token for classification""" try: if not self.access_token: return None - parts = self.access_token.split(".") if len(parts) != 3: return None - payload_b64 = parts[1] padding = len(payload_b64) % 4 if padding: payload_b64 += "=" * (4 - padding) - payload_json = base64.b64decode(payload_b64).decode("utf-8") return json.loads(payload_json) - except Exception as e: logger.debug(f"Failed to extract JWT claims: {e}") return None -class JoynAuthConfig: - """Configuration object for Joyn authentication with dynamic endpoints""" - - def __init__( - self, - country: str, - distribution_tenant: str, - http_manager, - platform: str = DEFAULT_PLATFORM, - ): - self.country = country - self.distribution_tenant = distribution_tenant - self.platform = platform - self.user_agent = JOYN_USER_AGENT - self.timeout = DEFAULT_REQUEST_TIMEOUT - self.http_manager = http_manager - - # Only create SSO discovery if we have http_manager - if http_manager is not None: - self.sso_discovery = JoynSSODiscovery(http_manager, country, platform) - else: - self.sso_discovery = None - - def get_token_redeem_endpoint(self) -> str: - """Get token redemption endpoint for user login flows""" - if self.sso_discovery: - return self.sso_discovery.get_auth_endpoint("redeem-token") - # Fallback if SSO discovery not available - return JoynSSODiscovery.get_fallback_endpoints()["redeem-token"] - - def get_authorize_endpoint(self) -> str: - """Get authorization endpoint for OAuth2 flow""" - if self.sso_discovery: - # Try platform-specific login endpoint first - return self.sso_discovery.get_auth_endpoint(f"{self.platform}-login") - # Fallback if SSO discovery not available - try platform-specific, then web-login - fallback = JoynSSODiscovery.get_fallback_endpoints() - return fallback.get(f"{self.platform}-login") or fallback.get("web-login", "") - - def get_base_headers(self) -> Dict[str, str]: - """Get base headers for all requests""" - return { - "User-Agent": self.user_agent, - "Accept": "application/json", - "Content-Type": "application/json", - "Origin": JOYN_DOMAINS.get(self.country, JOYN_DOMAINS["de"]), - } - - def get_auth_headers(self) -> Dict[str, str]: - headers = self.get_base_headers() - headers.update({ - "joyn-client-version": JOYN_CLIENT_VERSION, - "joyn-country": self.country.upper(), - # FIX: Add country suffix to distribution tenant - "joyn-distribution-tenant": f"JOYN_{self.country.upper()}", # "JOYN_DE" - "joyn-platform": self.platform, - "joyn-request-id": str(uuid.uuid4()), - }) - return headers - - class JoynAuthenticator(BaseOAuth2Authenticator): """ - Joyn authenticator using OAuth2 client credentials flow with dynamic endpoints + Joyn authenticator using OIDC discovery + custom 7pass login flow. + + Production-hardened: respects base class abstractions, proper exception chains, + consistent endpoint handling, and clean session management. """ def __init__( - self, - country: str = DEFAULT_COUNTRY, - platform: str = DEFAULT_PLATFORM, - settings_manager=None, - credentials=None, - config_dir: Optional[str] = None, - http_manager=None, - proxy_config: Optional[ProxyConfig] = None, + self, + country: str = DEFAULT_COUNTRY, + platform: str = DEFAULT_PLATFORM, + settings_manager=None, + credentials=None, + config_dir: Optional[str] = None, + http_manager=None, + proxy_config: Optional[ProxyConfig] = None, ): - """ - Initialize authenticator for specific country - """ + # Validate inputs if country not in SUPPORTED_COUNTRIES: - raise ValueError( - f"Unsupported country: {country}. Must be one of: {SUPPORTED_COUNTRIES}" - ) - - # Validate that http_manager is provided + raise ValueError(f"Unsupported country: {country}. Must be one of: {SUPPORTED_COUNTRIES}") if http_manager is None: - raise ValueError( - "http_manager is required for JoynAuthenticator. " - "It should be created in JoynProvider and passed to the authenticator." - ) + raise ValueError("http_manager is required for JoynAuthenticator") - # Set country-specific attributes FIRST + # Store Joyn-specific attributes self.country = country self.platform = platform self.distribution_tenant = COUNTRY_TENANT_MAPPING[country] - # Store http_manager reference (provided by JoynProvider) - self._http_manager = http_manager - - # Setup Joyn-specific config BEFORE super().__init__ - self._config = JoynAuthConfig( - self.country, self.distribution_tenant, self._http_manager, self.platform - ) - - # Extract and cache client_id during initialization - self._client_id = self._extract_client_id_from_endpoints() - - # Register BEFORE super().__init__ so the Kodi sync in SettingsManager - # knows joyn is country-aware and syncs with the country parameter - if settings_manager is not None: - settings_manager.register_provider( - "joyn", - supports_countries=True, - available_countries=["de", "at", "ch"], - ) - - # NOW call parent __init__ - config, http_manager AND country are ready + # Initialize base class FIRST to ensure proper config/http_manager setup super().__init__( provider_name="joyn", settings_manager=settings_manager, @@ -313,120 +129,148 @@ class JoynAuthenticator(BaseOAuth2Authenticator): country=country, config_dir=config_dir, enable_kodi_integration=True, - http_manager=self._http_manager, + http_manager=http_manager, proxy_config=proxy_config, ) - # Guaranteed non-None credentials after init + # Enable OIDC discovery using 7pass base URL + self.enable_oidc_discovery(JOYN_7PASS_BASE_URL) + + # Joyn's 7pass flow doesn't use PKCE + self._use_pkce = False + + # Register with settings manager + if settings_manager is not None: + settings_manager.register_provider( + "joyn", + supports_countries=True, + available_countries=SUPPORTED_COUNTRIES, + ) + + # Extract client ID (7pass OIDC doesn't expose app client_id; fallback to platform IDs) + self._client_id = self._extract_or_fallback_client_id() + + # Set up fallback credentials if needed if self.credentials is None: - logger.info(f"No credentials resolved for joyn/{self.country}, using anonymous fallback") + logger.info(f"No credentials for joyn/{self.country}, using anonymous fallback") self.credentials = self.get_fallback_credentials() - # Safe credential log + # Log credential type from ...base.auth.credentials import UserPasswordCredentials if isinstance(self.credentials, UserPasswordCredentials): logger.info( - f"JoynAuthenticator [{self.country}]: user credentials loaded, " - f"username='{self.credentials.username}', password=[REDACTED]" - ) + f"JoynAuthenticator [{self.country}]: user credentials loaded for '{self.credentials.username}'") else: - logger.info( - f"JoynAuthenticator [{self.country}]: using {type(self.credentials).__name__}" - ) + logger.info(f"JoynAuthenticator [{self.country}]: using {type(self.credentials).__name__}") + + # ======================================================================== + # Required Abstract Properties + # ======================================================================== + + @property + def oauth_client_id(self) -> str: + return self._client_id + + @property + def oauth_scope(self) -> str: + return JOYN_OAUTH_SCOPE + + @property + def oauth_redirect_uri(self) -> str: + from .constants import get_oauth_redirect_uri + return get_oauth_redirect_uri(self.country) + + # ======================================================================== + # Token Configuration + # ======================================================================== + + def _should_use_json_for_token_exchange(self, **kwargs) -> bool: + """Joyn uses JSON instead of form-encoded""" + return True + + def _build_token_exchange_payload( + self, authorization_code: str, code_verifier: str, state: Optional[str] = None, **kwargs + ) -> Dict[str, Any]: + """Joyn-specific token exchange payload with tracking_id""" + tracking_id = kwargs.get("cd1") or str(uuid.uuid4()) + return { + "code": authorization_code, + "client_id": self.oauth_client_id, + "redirect_uri": self.oauth_redirect_uri, + "tracking_id": tracking_id, + "tracking_name": self.platform, + "code_verifier": "", # PKCE explicitly disabled for Joyn + } + + def _get_token_exchange_headers(self, **kwargs) -> Dict[str, str]: + """Joyn-specific headers for token exchange""" + return self._get_joyn_auth_headers() + + # ======================================================================== + # Joyn-Specific Headers + # ======================================================================== def _get_joyn_auth_headers(self) -> Dict[str, str]: - from .constants import JOYN_AUTH_HEADERS_BASE + """Generate Joyn-specific authentication headers""" headers = JOYN_AUTH_HEADERS_BASE.copy() - headers["Origin"] = f"https://www.joyn.{self.country.lower()}" headers.update({ + "Origin": JOYN_DOMAINS.get(self.country, JOYN_DOMAINS["de"]), "joyn-country": self.country.upper(), - # FIX: Add country suffix here too "joyn-distribution-tenant": f"JOYN_{self.country.upper()}", "joyn-platform": self.platform, "joyn-request-id": str(uuid.uuid4()), }) return headers - def _extract_client_id_from_endpoints(self) -> str: - """Extract client_id from SSO endpoints during initialization""" - try: - # Get endpoints from SSO discovery - if self._config.sso_discovery is None: - raise ValueError("SSO discovery not available — http_manager was not set") - endpoints = self._config.sso_discovery.get_endpoints() - - # Get the platform-specific login endpoint - platform_key = f"{self.platform}-login" - login_url = endpoints.get(platform_key) - - if not login_url: - logger.warning( - f"No {platform_key} endpoint found, trying generic web-login as fallback" - ) - login_url = endpoints.get("web-login") - - if not login_url: - raise Exception( - f"No login endpoint found for platform '{self.platform}' or generic 'web-login' in SSO discovery" - ) - - # Extract client_id from the URL parameters - parsed_url = urlparse(login_url) - query_params = parse_qs(parsed_url.query) - - client_id = query_params.get("client_id", [None])[0] - - if not client_id: - raise Exception("No client_id found in login endpoint") - - logger.debug(f"Extracted and cached client_id for {self.platform}: {client_id}") - return client_id - - except Exception as e: - logger.error( - f"Error extracting client_id from endpoints: {e}, using fallback: {DEVICE_IDS.get(self.platform, DEVICE_IDS[DEFAULT_PLATFORM])}" - ) - return DEVICE_IDS.get(self.platform, DEVICE_IDS[DEFAULT_PLATFORM]) - - @property - def oauth_client_id(self) -> str: - """Get OAuth2 client ID - uses cached value from initialization""" - return self._client_id - - @property - def oauth_scope(self) -> str: - """OAuth2 scopes for authorization code flow""" - return JOYN_OAUTH_SCOPE - - @property - def oauth_redirect_uri(self) -> str: - """OAuth2 redirect URI - country-specific""" - from .constants import get_oauth_redirect_uri - - return get_oauth_redirect_uri(self.country) - - @property - def auth_endpoint(self) -> str: - """Authentication endpoint URL - dynamic based on flow""" - from ...base.auth.credentials import UserPasswordCredentials - - if isinstance(self.credentials, UserPasswordCredentials): - # For authorization code flow - use token endpoint from SSO discovery - return self._config.get_token_redeem_endpoint() - else: - # For client credentials flow - use anonymous endpoint - return JOYN_AUTH_ENDPOINTS["ANONYMOUS"] - def _get_auth_headers(self) -> Dict[str, str]: - """Get headers for authentication request""" - return self._config.get_auth_headers() + """Base authentication headers""" + return { + "User-Agent": JOYN_USER_AGENT, + "Accept": "application/json", + "Content-Type": "application/json", + "Origin": JOYN_DOMAINS.get(self.country, JOYN_DOMAINS["de"]), + "joyn-client-version": JOYN_CLIENT_VERSION, + "joyn-country": self.country.upper(), + "joyn-distribution-tenant": f"JOYN_{self.country.upper()}", + "joyn-platform": self.platform, + "joyn-request-id": str(uuid.uuid4()), + } + + # ======================================================================== + # Client ID Extraction + # ======================================================================== + + def _extract_or_fallback_client_id(self) -> str: + """ + Extract client ID. Note: 7pass OIDC discovery returns IdP metadata, + not app-specific client_id. Safe fallback to platform device IDs. + """ + client_id = DEVICE_IDS.get(self.platform, DEVICE_IDS[DEFAULT_PLATFORM]) + logger.debug(f"Using platform client_id: {client_id}") + return client_id + + # ======================================================================== + # Credentials + # ======================================================================== + + def get_fallback_credentials(self) -> JoynCredentials: + """Get fallback credentials when no user credentials are available""" + return JoynCredentials( + client_id=self._client_id, + client_secret="", + country=self.country, + ) def _build_auth_payload(self) -> Dict[str, Any]: - """Build authentication payload - only used for client credentials flow""" + """Build authentication payload for client credentials flow""" if not self.credentials: raise Exception("No credentials available") return self.credentials.to_auth_payload() + # ======================================================================== + # Token Creation & Classification + # ======================================================================== + def _create_token_from_response(self, response_data: Dict[str, Any]) -> BaseAuthToken: """Create token object from API response""" token = JoynAuthToken( @@ -436,145 +280,135 @@ class JoynAuthenticator(BaseOAuth2Authenticator): expires_in=response_data.get("expires_in", 86400), issued_at=response_data.get("issued_at", time.time()), ) - - # ALWAYS classify immediately when creating from response 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) -> JoynCredentials: - """Get fallback credentials when no user credentials are available""" - return JoynCredentials( - client_id=self._client_id, - client_secret="", # Joyn doesn't use client_secret - country=self.country, - ) - - def get_token_redeem_url(self) -> str: - """Get token redemption URL for OAuth flows""" - return self._config.get_token_redeem_endpoint() - - # MAIN AUTHENTICATION METHOD - def _perform_authentication(self) -> BaseAuthToken: - """ - Perform authentication using appropriate flow based on credential type. - Falls back to anonymous auth if user authentication fails. - """ - from ...base.auth.credentials import ClientCredentials, UserPasswordCredentials - - if isinstance(self.credentials, UserPasswordCredentials): - try: - logger.info(f"Using OAuth2 authorization code flow for {self.provider_name}") - token_data = self._perform_oauth_authorization_code_flow( - self.credentials.username, self.credentials.password - ) - return self._create_token_from_response(token_data) - - except Exception as e: - logger.warning( - f"User auth failed for {self.provider_name}, falling back to anonymous: {e}" - ) - # Fall back to client credentials (anonymous auth) - self.credentials = self.get_fallback_credentials() - # Fall through to client credentials flow below - - if isinstance(self.credentials, ClientCredentials): - logger.info(f"Using OAuth2 client credentials flow for {self.provider_name}") - token_data = self._perform_oauth_client_credentials_flow() - return self._create_token_from_response(token_data) - - raise Exception(f"Unsupported credential type: {type(self.credentials)}") - - def _perform_oauth_client_credentials_flow(self) -> Dict[str, Any]: - """ - Client credentials flow — Joyn uses JSON instead of form data. - """ + def _classify_token(self, token: BaseAuthToken) -> TokenAuthLevel: + """Classify Joyn token based on JWT claims""" try: - logger.debug(f"Starting Joyn-specific OAuth2 client credentials flow") + if not token or not token.access_token: + return TokenAuthLevel.UNKNOWN - headers = self._get_auth_headers() - payload = self._build_auth_payload() + if not isinstance(token, JoynAuthToken): + logger.warning("Token is not a JoynAuthToken") + return TokenAuthLevel.UNKNOWN - logger.debug(str(headers)) - logger.debug(str(payload)) + claims = token.get_jwt_claims() + if claims is None: + logger.warning("Invalid JWT format") + return TokenAuthLevel.UNKNOWN + + jidc = claims.get("jIdC", "") + if jidc.startswith("JNAA-"): + return TokenAuthLevel.CLIENT_CREDENTIALS + elif jidc.startswith("JNDE-"): + return TokenAuthLevel.USER_AUTHENTICATED + + if "social_id" in claims: + return TokenAuthLevel.USER_AUTHENTICATED + + client_id = claims.get("cId", "") + known_client_ids = {DEVICE_IDS.get("web"), DEVICE_IDS.get("android"), DEVICE_IDS.get("ios")} + if client_id in known_client_ids: + return TokenAuthLevel.CLIENT_CREDENTIALS + + subject = claims.get("sub", "") + if subject and len(subject) == 36: + return TokenAuthLevel.CLIENT_CREDENTIALS + + return TokenAuthLevel.UNKNOWN + except Exception as e: + logger.error(f"Error classifying token: {e}") + return TokenAuthLevel.UNKNOWN + + # ======================================================================== + # Token Refresh + # ======================================================================== + + def _refresh_oauth_token(self) -> Optional[BaseAuthToken]: + """ + Joyn-specific token refresh with custom endpoint and grant_type. + + Note: Joyn uses a non-standard refresh flow (grant_type: "Bearer") + on a dedicated endpoint not exposed in OIDC discovery. + """ + if not self._current_token or not self._current_token.refresh_token: + logger.debug(f"No refresh token available for {self.provider_name}") + return None + + try: + logger.debug(f"Refreshing token for {self.provider_name}") + + payload = { + "client_id": DEVICE_IDS.get(self.platform, DEVICE_IDS[DEFAULT_PLATFORM]), + "client_name": self.platform, + "grant_type": "Bearer", # Joyn non-standard grant type + "refresh_token": self._current_token.refresh_token, + } + + headers = self._get_joyn_auth_headers() + refresh_endpoint = JOYN_AUTH_ENDPOINTS["REFRESH"] response = self.http_manager.post( - self.auth_endpoint, + refresh_endpoint, operation="auth", headers=headers, json_data=payload, + timeout=getattr(self.config, "timeout", 30), ) - self._check_oauth_error_response(response) + # Handle 422 "Anonymous refresh token" error + if response.status_code == 422: + try: + error_body = response.json() + if error_body.get("data") == "Anonymous refresh token": + logger.debug("Refresh failed - token type mismatch, forcing re-auth") + return None + except Exception: + pass + response.raise_for_status() - - token_data = response.json() - logger.debug(f"OAuth2 client credentials flow successful for {self.provider_name}") - return token_data - + new_token_data = response.json() + refreshed_token = self._create_token_from_response(new_token_data) + logger.info(f"Token refresh successful for {self.provider_name}") + return refreshed_token except Exception as e: - logger.error( - f"OAuth2 client credentials flow on endpoint {self.auth_endpoint} failed for {self.provider_name}: {e}" - ) - raise Exception(f"OAuth2 client credentials flow failed: {e}") - - @staticmethod - def _extract_code_from_url(url: str) -> Optional[str]: - """ - Extract the authorization code from a URL's query string or fragment. - Returns the code string, or None if not found. - """ - if not url: + logger.warning(f"Token refresh failed for {self.provider_name}: {e}") return None - parsed = urlparse(url) + # ======================================================================== + # The Complex 7pass Login Flow + # ======================================================================== - # Check query string first (response_mode=query) - query_params = parse_qs(parsed.query) - code = query_params.get("code", [None])[0] - if code: - logger.debug(f"Found authorization code in query string") - return code - - # Check fragment (response_mode=fragment, SPA flows) - if parsed.fragment: - fragment_params = parse_qs(parsed.fragment) - code = fragment_params.get("code", [None])[0] - if code: - logger.debug(f"Found authorization code in URL fragment") - return code - - return None - - # Authorization code flow - def _perform_oauth_authorization_code_flow( - self, username: str, password: str - ) -> Dict[str, Any]: + def _perform_oauth_authorization_code_flow(self, username: str, password: str) -> Dict[str, Any]: """ - Joyn OAuth2 authorization code flow with two-step verification and PKCE. - Uses http_manager's session exclusively for proper cookie persistence. + Joyn's complex 7pass OAuth2 flow with multi-step verification. + + Uses constants for all endpoints for better maintainability. """ try: - logger.debug("Starting optimized Joyn OAuth2 authorization code flow") + logger.debug("Starting Joyn OAuth2 authorization code flow") - # Step 1: Get authorization endpoint and build URL - web_login_url = self._config.get_authorize_endpoint() + # Use OIDC-discovered authorize endpoint + web_login_url = self.oauth_authorize_endpoint logger.debug(f"Using authorize endpoint: {web_login_url}") - self.http_manager.clear_cookies() - self.http_manager.reset_referer() + # Create fresh session for clean cookie/referer state + session = self._create_oauth_session() + original_headers = dict(session.headers) + # Generate OAuth state and store for later validation state = self.generate_oauth_state() - # Parse discovery URL but preserve existing params (especially cd1) + # Parse URL and preserve existing params (especially cd1) parsed_login_url = urlparse(web_login_url) base_login_url = parsed_login_url._replace(query="", fragment="").geturl() existing_params = { k: v[0] for k, v in parse_qs(parsed_login_url.query).items() } - # 2. Build URL without PKCE (working implementation does not use it) + # Build authorization URL (PKCE disabled for Joyn) params = { **existing_params, "response_type": "code", @@ -585,42 +419,49 @@ class JoynAuthenticator(BaseOAuth2Authenticator): "response_mode": "query", "view_type": "login", "prompt": "consent", - # cd1 is preserved from existing_params if present } authorization_url = f"{base_login_url}?{urlencode(params)}" - logger.debug(f"Built authorization URL: {authorization_url}") + logger.debug("Built authorization URL") - # Use http_manager's session directly - this ensures cookie persistence - session = self.http_manager._session + # Helper for 7pass requests using managed session + def _make_7pass_request(method: str, url: str, **kwargs): + request_headers = kwargs.pop("headers", {}).copy() + # Strip joyn-* headers and standard origin/referer for 7pass endpoints + clean_headers = {k: v for k, v in request_headers.items() if not k.lower().startswith('joyn-')} + clean_headers.update({ + "User-Agent": JOYN_USER_AGENT, + "Accept": "*/*", + "Accept-Encoding": "gzip, deflate", + }) + clean_headers.pop("Referer", None) + clean_headers.pop("Origin", None) - # Save original headers to restore later - original_headers = dict(session.headers) + # Handle content-type if provided + content_type = kwargs.pop("content_type", None) + if content_type: + clean_headers["Content-Type"] = content_type - # Step 2: Get authorization page - logger.debug("Fetching authorization page") + allow_redirects = kwargs.pop("allow_redirects", True) - # Temporarily strip joyn-* headers for 7pass endpoints to avoid 431 errors - self._strip_joyn_headers_for_7pass(session) + if method.upper() == "GET": + return session.get(url, headers=clean_headers, timeout=30, allow_redirects=allow_redirects, + **kwargs) + else: + return session.post(url, headers=clean_headers, timeout=30, allow_redirects=allow_redirects, + **kwargs) - auth_response = session.get( - authorization_url, - timeout=self._config.timeout, - allow_redirects=True - ) + # Get authorization page + auth_response = _make_7pass_request("GET", authorization_url, allow_redirects=True) - # Restore headers after getting auth page + # Restore Joyn headers for subsequent calls session.headers.clear() session.headers.update(original_headers) - - logger.debug(f"Authorization page response status: {auth_response.status_code}") - logger.debug(f"Authorization page response URL: {auth_response.url}") auth_response.raise_for_status() - # Step 2a: Check if we got redirected directly to callback (existing session) + # Check for existing session (direct callback) if self.oauth_redirect_uri in auth_response.url: logger.info("User already authenticated - extracting code from redirect") - parsed_url = urlparse(auth_response.url) query_params = parse_qs(parsed_url.query) @@ -633,18 +474,16 @@ class JoynAuthenticator(BaseOAuth2Authenticator): if not self.validate_oauth_state(received_state, state): raise Exception("State validation failed on direct redirect") - logger.debug(f"Extracted authorization code from direct redirect") + logger.debug("Extracted authorization code from direct redirect") - token_data = self._exchange_authorization_code_for_token( + # Exchange code for token using base class method + return self._exchange_authorization_code_for_token( authorization_code=auth_code, - code_verifier="", # PKCE disabled for this flow + code_verifier="", # PKCE disabled for Joyn state=state, ) - logger.debug("Joyn OAuth2 authorization code flow successful (existing session)") - return token_data - - # Step 2b: No existing session — use login-srv/login flow + # No existing session - perform login flow logger.info("No existing session - performing login-srv/login flow") # Extract request_id and cd1 from the signin redirect URL @@ -664,84 +503,32 @@ class JoynAuthenticator(BaseOAuth2Authenticator): logger.debug(f"Extracted request_id: {request_id}, cd1: {cd1}") - # Helper to make 7pass requests with clean headers while preserving cookies - def _make_7pass_request(method: str, url: str, **kwargs): - from requests import Request as RawRequest + # Pre-login checks (non-fatal - continue even if they fail) + pre_login_checks = [ + ("GET", + f"{JOYN_7PASS_ENDPOINTS['REGISTRATION_SETUP']}?acceptlanguage=undefined&requestId={request_id}", + None), + ("POST", f"{JOYN_7PASS_ENDPOINTS['USER_CHECK_EXISTS']}/{request_id}", + {"email": username, "requestId": request_id}), + ("POST", JOYN_7PASS_ENDPOINTS['VERIFICATION_CONFIGURED'], + {"email": username, "request_id": request_id}), + ] - request_headers = { - "User-Agent": JOYN_USER_AGENT, - "Accept": "*/*", - "Accept-Encoding": "gzip, deflate", - } + for method, endpoint, data in pre_login_checks: + try: + if method == "GET": + _make_7pass_request("GET", endpoint, content_type="application/json") + else: + _make_7pass_request("POST", endpoint, json=data, content_type="application/json") + except Exception as e: + logger.debug(f"Pre-login check non-fatal error (continuing): {e}") - allow_redirects = kwargs.pop("allow_redirects", True) - content_type = kwargs.pop("content_type", None) - if content_type: - request_headers["Content-Type"] = content_type - - req = RawRequest( - method=method.upper(), - url=url, - headers=request_headers, - cookies=session.cookies, - **kwargs - ) - - prepared = req.prepare() - # ↓ Explicitly ensure Referer and Origin are absent — the long signin.7pass.de - # redirect URL in Referer is what causes the 431 on auth.7pass.de endpoints - prepared.headers.pop("Referer", None) - prepared.headers.pop("Origin", None) - - return session.send(prepared, timeout=self._config.timeout, allow_redirects=allow_redirects) - - logger.debug("Step 3a: Language / registration setup check") - try: - _make_7pass_request( - "GET", - f"https://auth.7pass.de/registration-setup-srv/public/list" - f"?acceptlanguage=undefined&requestId={request_id}", - content_type="application/json" - ) - except Exception as e: - logger.debug(f"Step 3a non-fatal error (continuing): {e}") - - logger.debug("Step 3b: Check user exists") - try: - _make_7pass_request( - "POST", - f"https://auth.7pass.de/users-srv/user/checkexists/{request_id}", - json={"email": username, "requestId": request_id}, - content_type="application/json" - ) - except Exception as e: - logger.debug(f"Step 3b non-fatal error (continuing): {e}") - - logger.debug("Step 3c: Get configured verification methods") - try: - _make_7pass_request( - "POST", - "https://auth.7pass.de/verification-srv/v2/setup/public/configured/list", - json={"email": username, "request_id": request_id}, - content_type="application/json" - ) - except Exception as e: - logger.debug(f"Step 3c non-fatal error (continuing): {e}") - - # Step 4: POST credentials to login-srv/login - logger.debug(f"Step 4: POST credentials (username: {username} requestid: {request_id}) to login-srv/login") - login_payload = {"username": username, "password": password, "requestId": request_id} - - logger.debug(f"Login POST URL: https://auth.7pass.de/login-srv/login") - logger.debug(f"Login POST payload type: {type(login_payload)}") - logger.debug( - f"Login POST payload keys: {list(login_payload.keys()) if isinstance(login_payload, dict) else 'N/A'}") - logger.debug(f"Session cookies count: {len(session.cookies)}") - logger.debug(f"Session cookies: {[(c.name, c.domain) for c in session.cookies]}") + # Submit credentials to login-srv/login + logger.debug(f"POST credentials to login-srv/login for user: {username}") login_response = _make_7pass_request( "POST", - "https://auth.7pass.de/login-srv/login", + JOYN_7PASS_ENDPOINTS["LOGIN"], data=urlencode({"username": username, "password": password, "requestId": request_id}), headers={"Content-Type": "application/x-www-form-urlencoded"}, allow_redirects=True, @@ -758,7 +545,7 @@ class JoynAuthenticator(BaseOAuth2Authenticator): error_code = error_match.group(1) if error_match else "unknown" raise Exception(f"Login failed with error_code={error_code}: {final_url}") - # Step 5: Consent flow if no code yet + # Handle consent flow if no code yet if id_dict.get("code") is None: sub = id_dict.get("sub", [None])[0] track_id = id_dict.get("track_id", [None])[0] @@ -768,10 +555,10 @@ class JoynAuthenticator(BaseOAuth2Authenticator): f"login-srv/login returned neither code nor sub/track_id. URL: {final_url}" ) - logger.debug(f"Step 5a: Consent scope accept for sub={sub}") + logger.debug(f"Accepting consent scopes for sub={sub}") _make_7pass_request( "POST", - "https://auth.7pass.de/consent-management-srv/consent/scope/accept", + JOYN_7PASS_ENDPOINTS["CONSENT_ACCEPT"], json={ "sub": sub, "client_id": self.oauth_client_id, @@ -780,10 +567,10 @@ class JoynAuthenticator(BaseOAuth2Authenticator): content_type="application/json" ) - logger.debug(f"Step 5b: precheck/continue for track_id={track_id}") + logger.debug(f"Continuing flow with track_id={track_id}") continue_response = _make_7pass_request( "POST", - f"https://auth.7pass.de/login-srv/precheck/continue/{track_id}", + f"{JOYN_7PASS_ENDPOINTS['PRECHECK_CONTINUE']}/{track_id}", data=b"", content_type="application/x-www-form-urlencoded", allow_redirects=True @@ -794,6 +581,7 @@ class JoynAuthenticator(BaseOAuth2Authenticator): id_dict = parse_qs(urlparse(final_url).query) logger.debug(f"precheck/continue landed on: {final_url}") + # Extract authorization code auth_code = id_dict.get("code", [None])[0] if not auth_code: raise Exception( @@ -806,10 +594,12 @@ class JoynAuthenticator(BaseOAuth2Authenticator): if not cd1: raise Exception("cd1 tracking ID missing from login flow response") - logger.debug("Step 6: Exchanging authorization code for tokens") + logger.debug("Exchanging authorization code for tokens") + + # Exchange code for token using base class method token_data = self._exchange_authorization_code_for_token( authorization_code=auth_code, - code_verifier="", # PKCE disabled for this flow + code_verifier="", # PKCE disabled for Joyn state=state, cd1=cd1, ) @@ -818,199 +608,36 @@ class JoynAuthenticator(BaseOAuth2Authenticator): return token_data except Exception as e: - logger.error(f"Joyn OAuth2 authorization code flow failed: {e}") - raise + # Preserve original exception chain for debugging + logger.error(f"Joyn OAuth2 authorization flow failed: {e}") + raise Exception(f"Joyn OAuth2 authorization flow failed: {e}") from e - @staticmethod - def _strip_joyn_headers_for_7pass(session) -> None: - """Remove joyn-* headers that can cause 431 errors on 7pass endpoints""" - for key in list(session.headers.keys()): - if key.lower().startswith('joyn-'): - del session.headers[key] + # ======================================================================== + # Public Methods + # ======================================================================== - def _build_token_exchange_payload( - self, authorization_code: str, code_verifier: str, state: Optional[str] = None, **kwargs - ) -> Dict[str, Any]: - """Joyn-specific token exchange payload. + def get_bearer_token(self, force_refresh: bool = False, force_upgrade: bool = False) -> str: + """Get bearer token with automatic upgrade support""" + return super().get_bearer_token(force_refresh=force_refresh, force_upgrade=force_upgrade) - cd1 is the tracking UUID that 7pass threads through the whole login flow. - When available (login-srv path) we pass it as tracking_id; otherwise we - generate a fresh UUID as a fallback. - """ - tracking_id = kwargs.get("cd1") or str(uuid.uuid4()) - return { - "code": authorization_code, - "client_id": self.oauth_client_id, - "redirect_uri": self.oauth_redirect_uri, - "tracking_id": tracking_id, - "tracking_name": self.platform, - "code_verifier": "", # Joyn/7pass expects empty string - # No grant_type for Joyn - } - - def _get_token_exchange_endpoint(self, **kwargs) -> str: - """Joyn-specific token exchange endpoint""" - token_endpoint = self._config.get_token_redeem_endpoint() - logger.debug(f"Using dynamic token endpoint: {token_endpoint}") - return token_endpoint - - def _get_token_exchange_headers(self, **kwargs) -> Dict[str, str]: - return self._get_joyn_auth_headers() - - def _should_use_json_for_token_exchange(self, **kwargs) -> bool: - """Joyn uses JSON instead of form-encoded""" - return True - - def _refresh_oauth_token(self) -> Optional[BaseAuthToken]: - """Joyn-specific token refresh implementation with error handling""" - if not self._current_token or not self._current_token.refresh_token: - logger.debug(f"No refresh token available for {self.provider_name}") - return None - - try: - logger.debug(f"Refreshing OAuth2 token for {self.provider_name}") - - payload = { - "client_id": DEVICE_IDS.get(self.platform, DEVICE_IDS[DEFAULT_PLATFORM]), - "client_name": self.platform, - "grant_type": "Bearer", # Joyn uses 'Bearer' instead of 'refresh_token' - "refresh_token": self._current_token.refresh_token, - } - - headers = self._get_joyn_auth_headers() - refresh_endpoint = JOYN_AUTH_ENDPOINTS["REFRESH"] - - logger.debug(f"Refresh payload: {payload}") - logger.debug(f"Refresh headers: {headers}") - - response = self.http_manager.post( - refresh_endpoint, - operation="auth", - headers=headers, - json_data=payload, - timeout=self._config.timeout, - ) - - logger.debug(f"Refresh response status: {response.status_code}") - - # Handle 422 "Anonymous refresh token" error - force re-auth - if response.status_code == 422: - try: - error_body = response.json() - if error_body.get("data") == "Anonymous refresh token": - logger.debug("Refresh failed - token type mismatch, forcing re-auth") - return None # Return None to trigger full re-authentication - except Exception: - pass # Continue to raise below - - response.raise_for_status() - - new_token_data = response.json() - refreshed_token = self._create_token_from_response(new_token_data) - logger.info(f"OAuth2 token refresh successful for {self.provider_name}") - return refreshed_token - - except Exception as e: - logger.warning(f"OAuth2 token refresh failed for {self.provider_name}: {e}") - return None # Return None to trigger full re-authentication - - # Backward compatibility methods def is_authenticated(self) -> bool: """Check if currently authenticated with valid token""" return self._current_token is not None and not self._current_token.is_expired def invalidate_token(self) -> None: - """Invalidate current token (forces re-authentication on next request)""" + """Invalidate current token""" self._current_token = None try: self.settings_manager.clear_token(self.provider_name) except (AttributeError, KeyError, IOError, OSError): pass - def _classify_token(self, token: BaseAuthToken) -> TokenAuthLevel: - """ - Classify Joyn token based on JWT claims and token structure. - - FIX: Delegates JWT decoding to token.get_jwt_claims() instead of - duplicating the base64 decode logic that already lives there. - """ - try: - if not token or not token.access_token: - return TokenAuthLevel.UNKNOWN - - if not isinstance(token, JoynAuthToken): - logger.warning("Token is not a JoynAuthToken — cannot extract JWT claims") - return TokenAuthLevel.UNKNOWN - - claims = token.get_jwt_claims() - if claims is None: - logger.warning("Invalid JWT format or decode failure — cannot classify token") - return TokenAuthLevel.UNKNOWN - - logger.debug( - f"JWT claims for classification: " - f"{ {k: v for k, v in claims.items() if k not in ['access_token', 'refresh_token']} }" - ) - - # 1. Check jIdC prefix — most reliable indicator - jidc = claims.get("jIdC", "") - if jidc.startswith("JNAA-"): - logger.debug("Token classified as CLIENT_CREDENTIALS (JNAA prefix)") - return TokenAuthLevel.CLIENT_CREDENTIALS - elif jidc.startswith("JNDE-"): - logger.debug("Token classified as USER_AUTHENTICATED (JNDE prefix)") - return TokenAuthLevel.USER_AUTHENTICATED - - # 2. social_id presence — clear indicator of user authentication - if "social_id" in claims: - logger.debug("Token classified as USER_AUTHENTICATED (social_id present)") - return TokenAuthLevel.USER_AUTHENTICATED - - # 3. Check client ID (cId) against known client IDs - client_id = claims.get("cId", "") - known_client_ids = { - DEVICE_IDS["web"], - DEVICE_IDS["android"], - DEVICE_IDS["ios"], - } - if client_id in known_client_ids: - logger.debug("Token classified as CLIENT_CREDENTIALS (known client ID)") - return TokenAuthLevel.CLIENT_CREDENTIALS - - # 4. UUID-format subject — typical of anonymous/client-credentials tokens - subject = claims.get("sub", "") - if subject and len(subject) == 36: - logger.debug("Token classified as CLIENT_CREDENTIALS (UUID subject pattern)") - return TokenAuthLevel.CLIENT_CREDENTIALS - - # 5. Fallback: scope analysis - scope = claims.get("scope", "") - if scope: - scopes = scope.split() - if "offline_access" in scopes and "profile" in scopes: - logger.debug("Token classified as USER_AUTHENTICATED (user scopes present)") - return TokenAuthLevel.USER_AUTHENTICATED - elif "openid" in scopes and len(scopes) <= 2: - logger.debug("Token classified as CLIENT_CREDENTIALS (minimal scopes)") - return TokenAuthLevel.CLIENT_CREDENTIALS - - logger.warning("Could not definitively classify token, using UNKNOWN") - return TokenAuthLevel.UNKNOWN - - except Exception as e: - logger.error(f"Error classifying token: {e}") - return TokenAuthLevel.UNKNOWN - def debug_token_classification(self) -> Dict[str, Any]: """Debug method to analyze current token classification""" if not self._current_token: return {"error": "No current token"} - claims = ( - self._current_token.get_jwt_claims() - if hasattr(self._current_token, "get_jwt_claims") - else {} - ) + claims = self._current_token.get_jwt_claims() if hasattr(self._current_token, "get_jwt_claims") else {} return { "token_type": type(self._current_token).__name__, @@ -1018,17 +645,11 @@ class JoynAuthenticator(BaseOAuth2Authenticator): "is_expired": self._current_token.is_expired, "has_refresh": bool(self._current_token.refresh_token), "jwt_claims_available": bool(claims), - "key_claims": ( - { - "jIdC": claims.get("jIdC", "MISSING"), - "cId": claims.get("cId", "MISSING"), - "social_id": "PRESENT" if "social_id" in claims else "MISSING", - "sub": ( - claims.get("sub", "MISSING")[:8] + "..." if claims.get("sub") else "MISSING" - ), - "scope": claims.get("scope", "MISSING"), - } - if claims - else {} - ), + "key_claims": { + "jIdC": claims.get("jIdC", "MISSING"), + "cId": claims.get("cId", "MISSING"), + "social_id": "PRESENT" if "social_id" in claims else "MISSING", + "sub": claims.get("sub", "MISSING")[:8] + "..." if claims.get("sub") else "MISSING", + "scope": claims.get("scope", "MISSING"), + } if claims else {}, } \ No newline at end of file diff --git a/lib/streaming_providers/providers/joyn/constants.py b/lib/streaming_providers/providers/joyn/constants.py index b1891ef..fd3fee6 100644 --- a/lib/streaming_providers/providers/joyn/constants.py +++ b/lib/streaming_providers/providers/joyn/constants.py @@ -1,89 +1,67 @@ # streaming_providers/providers/joyn/constants.py -# ============================================================================ -# SSO Discovery Configuration -# ============================================================================ +# -*- coding: utf-8 -*- +""" +Joyn provider constants - Cleaned and organized +""" -# SSO endpoints discovery URL -JOYN_SSO_DISCOVERY_URL = "https://auth.joyn.de/sso/endpoints" +# ============================================================================ +# Provider Metadata +# ============================================================================ JOYN_LOGO = "https://upload.wikimedia.org/wikipedia/de/thumb/7/74/Joyn_%28Streaminganbieter%29_logo.svg/2560px-Joyn_%28Streaminganbieter%29_logo.svg.png" -# Default client IDs for different platforms (fallback) +# ============================================================================ +# Authentication - 7pass OIDC +# ============================================================================ + +# 7pass base URL (OIDC provider) +JOYN_7PASS_BASE_URL = "https://auth.7pass.de" + +# 7pass OIDC endpoints (discovered via OIDC discovery) +JOYN_7PASS_ENDPOINTS = { + "AUTHORIZE": f"{JOYN_7PASS_BASE_URL}/authorize", + "TOKEN": f"{JOYN_7PASS_BASE_URL}/token", + "LOGIN": f"{JOYN_7PASS_BASE_URL}/login-srv/login", + "CONSENT_ACCEPT": f"{JOYN_7PASS_BASE_URL}/consent-management-srv/consent/scope/accept", + "PRECHECK_CONTINUE": f"{JOYN_7PASS_BASE_URL}/login-srv/precheck/continue", + "USER_CHECK_EXISTS": f"{JOYN_7PASS_BASE_URL}/users-srv/user/checkexists", + "REGISTRATION_SETUP": f"{JOYN_7PASS_BASE_URL}/registration-setup-srv/public/list", + "VERIFICATION_CONFIGURED": f"{JOYN_7PASS_BASE_URL}/verification-srv/v2/setup/public/configured/list", +} + +# Joyn auth endpoints (non-OIDC) +JOYN_AUTH_ENDPOINTS = { + "REFRESH": "https://auth.joyn.de/auth/refresh", # Token refresh endpoint +} + +# OAuth2 Configuration +JOYN_OAUTH_SCOPE = "openid email profile offline_access" + +# Device IDs for different platforms (fallback for client identification) DEVICE_IDS = { "web": "709115c2-f87e-4bad-9b94-28ac08d72cd9", "android": "05f5f3df-1130-4707-a761-c04d0c50b7f2", "ios": "21218403-52ec-4a65-abf4-f36a0eadd631", } -# OAuth2 Configuration -JOYN_OAUTH_SCOPE = "openid email profile offline_access" - # ============================================================================ -# Authentication Configuration +# HTTP Headers & User Agent # ============================================================================ -# Base authentication URL -JOYN_AUTH_BASE_URL = "https://auth.joyn.de/auth" - -# Authentication endpoints -JOYN_AUTH_ENDPOINTS = { - "ANONYMOUS": f"{JOYN_AUTH_BASE_URL}/anonymous", # Client credentials flow - "REFRESH": f"{JOYN_AUTH_BASE_URL}/refresh", # Token refresh - "LOGOUT": f"{JOYN_AUTH_BASE_URL}/logout", # Logout -} - -# ============================================================================ -# Cidaas/7pass Configuration -# ============================================================================ - -# Cidaas base URL (7pass authentication service) -JOYN_CIDAAS_BASE_URL = "https://auth.7pass.de" - -# Cidaas API endpoints -JOYN_CIDAAS_ENDPOINTS = { - "VERIFICATION_INITIATE": f"{JOYN_CIDAAS_BASE_URL}/verification-srv/v2/authenticate/initiate/PASSWORD", - "VERIFICATION_AUTHENTICATE": f"{JOYN_CIDAAS_BASE_URL}/verification-srv/v2/authenticate/authenticate/PASSWORD", - "LOGIN_VERIFICATION": f"{JOYN_CIDAAS_BASE_URL}/login-srv/verification/login", - "REGISTRATION_SETUP": f"{JOYN_CIDAAS_BASE_URL}/registration-setup-srv/public/list", - "USER_CHECK_EXISTS": f"{JOYN_CIDAAS_BASE_URL}/users-srv/user/checkexists", - "VERIFICATION_LIST": f"{JOYN_CIDAAS_BASE_URL}/verification-srv/v2/setup/public/configured/list", - "CONSENT_ACCEPT": f"{JOYN_CIDAAS_BASE_URL}/consent-management-srv/consent/scope/accept", - "LOGIN_CONTINUE": f"{JOYN_CIDAAS_BASE_URL}/login-srv/precheck/continue", -} - -# Base URLs -JOYN_BASE_URLS = { - "ORIGIN": "https://www.joyn.de", - "REFERER": "https://www.joyn.de/", - "SIGNIN_BASE": "https://signin.7pass.de", -} - -# ============================================================================ -# API Configuration -# ============================================================================ - -# Client version used in API requests +JOYN_USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36" JOYN_CLIENT_VERSION = "5.1344.1" - -# Platform identifier DEFAULT_PLATFORM = "web" -# Default user agent for all requests -JOYN_USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36" - -# Base64 encoded secret key for signature generation -SIGNATURE_SECRET_KEY = "MzU0MzM3MzgzMzM4MzMzNjM1NDMzNzM4MzYzNDM2MzYzNTQzMzczODM2MzYzMzM4MzIzNjM1NDMzNzM4MzMzMDM2MzQzNTM5MzU0MzM3MzgzMzM5MzMzNTMyMzQzNTQzMzczODM2MzUzMzM5MzU0MzM3MzgzMzM4MzMzMjMzNDYzNTQzMzczODM2MzYzMzMzMzM0NDMzNDIzNTQzMzczODMzMzgzNjM2MzMzNQ==" - +# Base authentication headers (without dynamic values) JOYN_AUTH_HEADERS_BASE = { "User-Agent": JOYN_USER_AGENT, "Accept": "application/json", "Content-Type": "application/json", - "Origin": JOYN_BASE_URLS["ORIGIN"], + "Origin": "https://www.joyn.de", # Base origin, overridden per country "joyn-client-version": JOYN_CLIENT_VERSION, - # Note: 'joyn-platform' is added dynamically in auth.py and provider.py } -# Base API headers (without dynamic auth tokens) +# Base API headers (without auth token) JOYN_API_BASE_HEADERS = { "Accept": "application/json", "Content-Type": "application/json", @@ -94,7 +72,6 @@ JOYN_API_BASE_HEADERS = { # GraphQL Configuration # ============================================================================ -# GraphQL base URL JOYN_GRAPHQL_BASE_URL = "https://api.joyn.de/graphql" # GraphQL persisted query hashes @@ -109,7 +86,7 @@ JOYN_GRAPHQL_ENDPOINTS = { "LIVE_CHANNELS": f"{JOYN_GRAPHQL_BASE_URL}?operationName=LiveChannelsAndEpg&enable_user_location=true&watch_assistant_variant=true", } -# Base GraphQL headers (without country-specific ones) +# Base GraphQL headers JOYN_GRAPHQL_BASE_HEADERS = { "X-Api-Key": "4f0fd9f18abbe3cf0e87fdb556bc39c8", "Accept": "application/json", @@ -117,10 +94,8 @@ JOYN_GRAPHQL_BASE_HEADERS = { "User-Agent": JOYN_USER_AGENT, } -# GraphQL persisted query version -GRAPHQL_PERSISTED_QUERY_VERSION = 1 - # GraphQL query defaults +GRAPHQL_PERSISTED_QUERY_VERSION = 1 GRAPHQL_LIVE_CHANNELS_FILTER = "DEFAULT" GRAPHQL_MAX_RESULTS = 5000 GRAPHQL_OFFSET = 0 @@ -129,26 +104,12 @@ GRAPHQL_OFFSET = 0 # Streaming Configuration # ============================================================================ -# Streaming API endpoints JOYN_STREAMING_ENDPOINTS = { "ENTITLEMENT": "https://entitlement.p7s1.io/api/user/entitlement-token", "PLAYLIST": "https://api.vod-prd.s.joyn.de/v1/channel/{channel_id}/playlist", } -# Default video data payload configuration -""" -DEFAULT_VIDEO_CONFIG = { - 'manufacturer': 'unknown', - 'platform': 'browser', - 'maxSecurityLevel': 1, - 'model': 'unknown', - 'protectionSystem': 'widevine', - 'streamingFormat': 'dash', - 'enableSubtitles': True, - 'maxResolution': 1080, - 'version': 'v1', -} -""" +# Default video configuration for playlist requests DEFAULT_VIDEO_CONFIG = { "enableDolbyAtmos": True, "enableSubtitles": True, @@ -163,23 +124,22 @@ DEFAULT_VIDEO_CONFIG = { "maxSecurityLevel": 5, } +# Signature secret key (base64 encoded) +SIGNATURE_SECRET_KEY = "MzU0MzM3MzgzMzM4MzMzNjM1NDMzNzM4MzYzNDM2MzYzNTQzMzczODM2MzYzMzM4MzIzNjM1NDMzNzM4MzMzMDM2MzQzNTM5MzU0MzM3MzgzMzM5MzMzNTMyMzQzNTQzMzczODM2MzUzMzM5MzU0MzM3MzgzMzM4MzMzMjMzNDYzNTQzMzczODM2MzYzMzMzMzM0NDMzNDIzNTQzMzczODMzMzgzNjM2MzMzNQ==" + # ============================================================================ -# Content Configuration +# Content Types & Modes # ============================================================================ -# Content types CONTENT_TYPE_LIVE = "LIVE" CONTENT_TYPE_VOD = "VOD" -# Stream types STREAM_TYPE_LINEAR = "LINEAR" STREAM_TYPE_EVENT = "EVENT" STREAM_TYPE_ON_DEMAND = "ON_DEMAND" -# Livestream types for GraphQL queries DEFAULT_LIVESTREAM_TYPES = ["EVENT", "LINEAR", "ON_DEMAND"] -# Stream modes MODE_LIVE = "live" MODE_VOD = "vod" @@ -187,22 +147,27 @@ MODE_VOD = "vod" # Error Codes # ============================================================================ -# Known error codes from Joyn API ERROR_CODES = { "PLAYBACK_RESTRICTED": "ENT_RVOD_Playback_Restricted", "UNAUTHORIZED": "ENT_Unauthorized", "NOT_FOUND": "ENT_Not_Found", "GEOBLOCKED": "ENT_Geoblocked", - "VALIDATION_ERROR": "VALIDATION_ERROR", # Added for token refresh - "INVALID_JWT": "INVALID_JWT", # Added for expired tokens + "VALIDATION_ERROR": "VALIDATION_ERROR", + "INVALID_JWT": "INVALID_JWT", } # ============================================================================ -# Country/Region Configuration +# Country Configuration # ============================================================================ -# Country to distribution tenant mapping -COUNTRY_TENANT_MAPPING = {"de": "JOYN", "at": "JOYN_AT", "ch": "JOYN_CH"} +SUPPORTED_COUNTRIES = ["de", "at", "ch"] +DEFAULT_COUNTRY = "de" + +COUNTRY_TENANT_MAPPING = { + "de": "JOYN", + "at": "JOYN_AT", + "ch": "JOYN_CH", +} JOYN_DOMAINS = { "de": "https://www.joyn.de", @@ -210,32 +175,22 @@ JOYN_DOMAINS = { "ch": "https://www.joyn.ch", } - def get_oauth_redirect_uri(country: str) -> str: """Get country-specific OAuth redirect URI""" + # Joyn uses same redirect URI for all countries return "https://www.joyn.de/oauth" - -# Supported countries -SUPPORTED_COUNTRIES = list(COUNTRY_TENANT_MAPPING.keys()) - -# Default country -DEFAULT_COUNTRY = "de" - # ============================================================================ # DRM Configuration # ============================================================================ -# DRM system DRM_SYSTEM_WIDEVINE = "widevine" -# DRM request headers DRM_REQUEST_HEADERS = { "Content-Type": "application/octet-stream", "User-Agent": JOYN_USER_AGENT, } -# DRM license request template (without bearer token) DRM_LICENSE_HEADERS_BASE = { "Content-Type": "application/octet-stream", "User-Agent": JOYN_USER_AGENT, @@ -245,20 +200,14 @@ DRM_LICENSE_HEADERS_BASE = { # Request Configuration # ============================================================================ -# Default timeout for HTTP requests (seconds) DEFAULT_REQUEST_TIMEOUT = 30 - -# Default maximum retries for failed requests DEFAULT_MAX_RETRIES = 3 - -# Default time window for EPG queries (hours) DEFAULT_EPG_WINDOW_HOURS = 3 # ============================================================================ # Channel Configuration # ============================================================================ -# Default channel settings DEFAULT_CHANNEL_CONFIG = { "video": "best", "on_demand": True, @@ -268,5 +217,4 @@ DEFAULT_CHANNEL_CONFIG = { "session_manifest": False, } -# Default language -DEFAULT_LANGUAGE = "de" +DEFAULT_LANGUAGE = "de" \ No newline at end of file diff --git a/lib/streaming_providers/providers/joyn/provider.py b/lib/streaming_providers/providers/joyn/provider.py index 01c60b6..6946ff9 100644 --- a/lib/streaming_providers/providers/joyn/provider.py +++ b/lib/streaming_providers/providers/joyn/provider.py @@ -3,7 +3,6 @@ import hashlib import json import time -import datetime import urllib.parse from base64 import b64decode from datetime import datetime, timedelta @@ -583,7 +582,7 @@ class JoynProvider(StreamingProvider): Get manifest URL for a specific channel by ID Args: - channel_id: ID of the channel to get manifest for + content_id: ID of the channel to get manifest for content_type: Content type ('LIVE' or 'VOD') video_config: Optional video configuration dictionary **kwargs: Additional parameters @@ -615,7 +614,7 @@ class JoynProvider(StreamingProvider): Get all DRM configurations for a channel by ID Args: - channel_id: ID of the channel to get DRM configs + content_id: ID of the channel to get DRM configs content_type: Content type ('LIVE' or 'VOD') video_config: Optional video configuration dictionary **kwargs: Additional parameters