diff --git a/lib/streaming_providers/providers/joyn/auth.py b/lib/streaming_providers/providers/joyn/auth.py index 2212275..58144c0 100644 --- a/lib/streaming_providers/providers/joyn/auth.py +++ b/lib/streaming_providers/providers/joyn/auth.py @@ -10,7 +10,7 @@ from typing import Any, Dict, Optional from urllib.parse import parse_qs, urlencode, urlparse, urlunparse from ...base.auth.base_auth import BaseAuthToken, TokenAuthLevel -from ...base.auth.base_oauth2_auth import BaseOAuth2Authenticator, OAuth2Error, WafBlockedException +from ...base.auth.base_oauth2_auth import BaseOAuth2Authenticator, WafBlockedException from ...base.auth.credentials import ClientCredentials, UserPasswordCredentials from ...base.models.proxy_models import ProxyConfig from ...base.utils.logger import logger @@ -94,10 +94,6 @@ class JoynAuthToken(BaseAuthToken): class JoynAuthenticator(BaseOAuth2Authenticator): """ Joyn authenticator based on actual network traffic logs. - - Joyn does NOT use standard OIDC discovery - they use a custom /sso/endpoints - call to get platform-specific endpoints. This implementation follows the - exact flow captured in production logs while leveraging base class extensibility. """ def __init__( @@ -110,28 +106,24 @@ class JoynAuthenticator(BaseOAuth2Authenticator): http_manager=None, proxy_config: Optional[ProxyConfig] = None, ): - # Validate inputs if country not in SUPPORTED_COUNTRIES: - raise ValueError(f"Unsupported country: {country}. Must be one of: {SUPPORTED_COUNTRIES}") + raise ValueError(f"Unsupported country: {country}") if http_manager is None: raise ValueError("http_manager is required for JoynAuthenticator") - # Store Joyn-specific attributes self.country = country self.platform = platform self.distribution_tenant = COUNTRY_TENANT_MAPPING.get(country, "JOYN") - # Endpoints discovered from /sso/endpoints call + # Cache for flow parameters self._sso_endpoints_cache = None self._sso_endpoints_timestamp = None - self._sso_cache_ttl = 3600 # 1 hour - - # Cache for persistent flow parameters + self._sso_cache_ttl = 3600 self._cmp_uc_id = None self._cmp_uc_instance = None self._auth_base_path = None - # Initialize base class first (this sets up settings_manager) + # Initialize base class super().__init__( provider_name="joyn", settings_manager=settings_manager, @@ -143,10 +135,8 @@ class JoynAuthenticator(BaseOAuth2Authenticator): proxy_config=proxy_config, ) - # NOW load or generate persistent device ID (after base class init) + # Load or generate persistent device ID self._device_id = self._load_or_generate_device_id() - - # PKCE is required for Joyn self._use_pkce = True # Create config object @@ -172,11 +162,8 @@ class JoynAuthenticator(BaseOAuth2Authenticator): } self._config = JoynConfig(country, platform) - - # Disable OIDC discovery - Joyn uses custom /sso/endpoints self._enable_oidc_discovery = False - # Register with settings manager if settings_manager is not None: settings_manager.register_provider( "joyn", @@ -184,55 +171,35 @@ class JoynAuthenticator(BaseOAuth2Authenticator): available_countries=SUPPORTED_COUNTRIES, ) - # Extract client ID - web client ID + # CRITICAL: Use the CORRECT web client ID self._client_id = DEVICE_IDS.get(self.platform, DEVICE_IDS[DEFAULT_PLATFORM]) logger.info(f"Using Joyn client_id: {self._client_id}") - # Set up fallback credentials if needed if self.credentials is None: logger.info(f"No credentials for joyn/{self.country}, using anonymous fallback") self.credentials = self.get_fallback_credentials() - if isinstance(self.credentials, UserPasswordCredentials): - logger.info( - f"JoynAuthenticator [{self.country}]: user credentials loaded for '{self.credentials.username}'") - else: - logger.info(f"JoynAuthenticator [{self.country}]: using {type(self.credentials).__name__}") - - # ======================================================================== - # Device ID Management - # ======================================================================== - def _load_or_generate_device_id(self) -> str: """Load existing device ID from settings or generate new one""" - # Check if settings_manager exists and has the method - if self.settings_manager is not None and hasattr(self.settings_manager, 'get_setting'): + if self.settings_manager and hasattr(self.settings_manager, 'get_setting'): try: device_id = self.settings_manager.get_setting("joyn_device_id") if device_id: logger.debug(f"Loaded existing device_id: {device_id}") return device_id except Exception as e: - logger.debug(f"Could not load device_id from settings: {e}") + logger.debug(f"Could not load device_id: {e}") - # Generate new device ID new_device_id = str(uuid.uuid4()) - - # Try to save it if settings_manager is available - if self.settings_manager is not None and hasattr(self.settings_manager, 'set_setting'): + if self.settings_manager and hasattr(self.settings_manager, 'set_setting'): try: self.settings_manager.set_setting("joyn_device_id", new_device_id) - logger.debug(f"Saved new device_id: {new_device_id}") except Exception as e: - logger.debug(f"Could not save device_id to settings: {e}") + logger.debug(f"Could not save device_id: {e}") logger.debug(f"Generated new device_id: {new_device_id}") return new_device_id - # ======================================================================== - # Required Abstract Properties - # ======================================================================== - @property def oauth_client_id(self) -> str: return self._client_id @@ -246,15 +213,8 @@ class JoynAuthenticator(BaseOAuth2Authenticator): from .constants import get_oauth_redirect_uri return get_oauth_redirect_uri(self.country) - # ======================================================================== - # SSO Endpoints Discovery (from logs) - # ======================================================================== - def _discover_sso_endpoints(self) -> Dict[str, str]: - """ - Discover Joyn's SSO endpoints via /sso/endpoints call. - Also extracts the base path for authorization endpoint. - """ + """Discover Joyn's SSO endpoints, but IGNORE any client_id from the response""" if self._sso_endpoints_cache and self._sso_endpoints_timestamp: if (time.time() - self._sso_endpoints_timestamp) < self._sso_cache_ttl: return self._sso_endpoints_cache @@ -273,40 +233,38 @@ class JoynAuthenticator(BaseOAuth2Authenticator): endpoints = response.json() - # Extract the endpoints + # Get the full auth endpoint from discovery auth_endpoint_full = endpoints.get("web-login", "") - token_endpoint = endpoints.get("redeem-token", "https://auth.joyn.de/auth/7pass/token") - # IMPORTANT: Extract ONLY the base path from the auth endpoint - # Strip all query parameters to avoid parameter duplication + # CRITICAL: Extract ONLY the base path, ignore all query parameters parsed_auth = urlparse(auth_endpoint_full) self._auth_base_path = urlunparse(( parsed_auth.scheme, parsed_auth.netloc, parsed_auth.path, "", # params - "", # query + "", # query - DISCARD any existing query params "" # fragment )) - # Extract cmpUcId and cmpUcInstance from the full URL if present + # Extract cmpUcId and cmpUcInstance for tracking (but NOT client_id) params = parse_qs(parsed_auth.query) self._cmp_uc_id = params.get("cmpUcId", [None])[0] self._cmp_uc_instance = params.get("cmpUcInstance", [None])[0] self._sso_endpoints_cache = { "authorization_base_path": self._auth_base_path, - "token_endpoint": token_endpoint, + "token_endpoint": endpoints.get("redeem-token", "https://auth.joyn.de/auth/7pass/token"), } self._sso_endpoints_timestamp = time.time() - logger.debug(f"Discovered Joyn SSO endpoints - auth base: {self._auth_base_path}") + logger.debug(f"Discovered auth base path: {self._auth_base_path}") + logger.debug(f"cmpUcId: {self._cmp_uc_id}, cmpUcInstance: {self._cmp_uc_instance}") return self._sso_endpoints_cache except Exception as e: logger.warning(f"Failed to discover SSO endpoints: {e}") - # Fallback to hardcoded endpoints from logs self._auth_base_path = "https://auth.7pass.de/authz-srv/authz" return { "authorization_base_path": self._auth_base_path, @@ -315,22 +273,16 @@ class JoynAuthenticator(BaseOAuth2Authenticator): @property def oauth_authorize_endpoint(self) -> str: - """Get clean authorization base path (no query parameters)""" - self._discover_sso_endpoints() # Ensure endpoints are discovered + """Get clean authorization base path""" + self._discover_sso_endpoints() return self._auth_base_path or "https://auth.7pass.de/authz-srv/authz" @property def oauth_token_endpoint(self) -> str: - """Get token endpoint from SSO discovery""" endpoints = self._discover_sso_endpoints() return endpoints.get("token_endpoint", "https://auth.joyn.de/auth/7pass/token") - # ======================================================================== - # Joyn-Specific Headers - # ======================================================================== - def _get_joyn_auth_headers(self) -> Dict[str, str]: - """Generate Joyn-specific authentication headers""" headers = JOYN_AUTH_HEADERS_BASE.copy() headers.update({ "Origin": JOYN_DOMAINS.get(self.country, JOYN_DOMAINS["de"]), @@ -343,22 +295,14 @@ class JoynAuthenticator(BaseOAuth2Authenticator): return headers def _get_auth_headers(self) -> Dict[str, str]: - """Base authentication headers""" return self._get_joyn_auth_headers() - # ======================================================================== - # Extensibility Hooks for Base Class Integration - # ======================================================================== - def _should_use_json_for_token_exchange(self, **kwargs) -> bool: - """Joyn always uses JSON for token endpoints.""" return True def _build_token_exchange_payload( self, authorization_code: str, code_verifier: str, state: str = None, **kwargs ) -> Dict[str, Any]: - """Build token exchange payload with Joyn-specific fields.""" - # Start with base payload payload = super()._build_token_exchange_payload( authorization_code=authorization_code, code_verifier=code_verifier, @@ -366,7 +310,6 @@ class JoynAuthenticator(BaseOAuth2Authenticator): **kwargs ) - # Add Joyn-specific tracking parameters cd1 = kwargs.get('cd1') if cd1 is None: cd1 = self._device_id @@ -378,108 +321,59 @@ class JoynAuthenticator(BaseOAuth2Authenticator): return payload def _get_token_exchange_endpoint(self, **kwargs) -> str: - """Get token exchange endpoint - use SSO-discovered endpoint.""" return self.oauth_token_endpoint def _get_token_exchange_headers(self, **kwargs) -> Dict[str, str]: - """Get token exchange headers with Joyn-specific additions.""" headers = super()._get_token_exchange_headers(**kwargs) - # Ensure Joyn-specific headers are included joyn_headers = self._get_joyn_auth_headers() for key, value in joyn_headers.items(): if key not in headers: headers[key] = value return headers - # ======================================================================== - # Token Exchange - # ======================================================================== - - def _exchange_authorization_code_for_token( - self, authorization_code: str, code_verifier: str, state: str = None, **kwargs - ) -> Dict[str, Any]: - """Exchange authorization code for tokens using base class hooks.""" - try: - logger.debug(f"Exchanging authorization code for token") - return super()._exchange_authorization_code_for_token( - authorization_code=authorization_code, - code_verifier=code_verifier, - state=state, - **kwargs - ) - except OAuth2Error: - raise - except Exception as e: - logger.error(f"Token exchange failed: {e}") - raise Exception(f"Token exchange failed: {e}") from e - - # ======================================================================== - # Token Refresh - # ======================================================================== - def _refresh_oauth_token(self) -> Optional[BaseAuthToken]: - """Refresh access token with proactive anonymous token detection.""" if not self._current_token or not self._current_token.refresh_token: - logger.debug(f"No refresh token available") return None try: - # Check if this is an anonymous token (no refresh capability) if hasattr(self._current_token, 'get_jwt_claims'): claims = self._current_token.get_jwt_claims() if claims and claims.get("jIdC", "").startswith("JNAA-"): - logger.debug("Anonymous token (JNAA-) cannot be refreshed") + logger.debug("Anonymous token cannot be refreshed") return None return super()._refresh_oauth_token() - - except OAuth2Error as e: - if "Anonymous refresh token" in str(e) or "422" in str(e): - logger.debug("Token cannot be refreshed (anonymous)") - return None + except Exception as e: logger.warning(f"Token refresh failed: {e}") return None - except Exception as e: - logger.warning(f"Token refresh failed unexpectedly: {e}") - return None - - # ======================================================================== - # Complete Login Flow - # ======================================================================== def _perform_oauth_authorization_code_flow(self, username: str, password: str) -> Dict[str, Any]: - """ - Complete Joyn login flow exactly as shown in production logs. - """ + """Complete Joyn login flow with CORRECT client_id""" try: logger.debug("Starting Joyn login flow") - # Step 0: Discover SSO endpoints and get clean base path + # Discover endpoints (to get base path and cmp params) self._discover_sso_endpoints() - # Ensure we have the required parameters if not self._auth_base_path: - raise Exception("Failed to get authorization endpoint base path") + raise Exception("Failed to get authorization endpoint") - # Step 1: Generate PKCE codes + # Generate PKCE codes state = self.generate_oauth_state() code_verifier = self.generate_pkce_verifier() code_challenge = self.generate_pkce_challenge(code_verifier) - # Use persistent device ID as cd1 cd1 = self._device_id - - # Get cmpUcId and cmpUcInstance from discovered endpoint cmp_uc_id = self._cmp_uc_id or str(uuid.uuid4()) cmp_uc_instance = self._cmp_uc_instance or 'WEB' - # Step 2: Build authorization URL from scratch with ONLY our parameters + # CRITICAL: Build URL with OUR client_id, NOT the one from discovery auth_params = { "response_type": "code", "scope": self.oauth_scope, "view_type": "login", "cd1": cd1, - "client_id": self.oauth_client_id, + "client_id": self.oauth_client_id, # OUR correct web client ID "prompt": "consent", "response_mode": "query", "cmpUcId": cmp_uc_id, @@ -491,15 +385,12 @@ class JoynAuthenticator(BaseOAuth2Authenticator): } auth_url = f"{self._auth_base_path}?{urlencode(auth_params)}" - logger.debug(f"Authorization URL built (length: {len(auth_url)})") + logger.debug(f"Auth URL built with client_id={self.oauth_client_id}") - # Create session for cookie management session = self._create_oauth_session() - # Helper for 7pass requests def _request(method, url, **kwargs): headers = kwargs.pop("headers", {}).copy() - # Clean Joyn headers for 7pass clean_headers = { k: v for k, v in headers.items() if not k.lower().startswith('joyn-') @@ -521,19 +412,14 @@ class JoynAuthenticator(BaseOAuth2Authenticator): return session.post(url, headers=clean_headers, timeout=timeout, allow_redirects=allow_redirects, **kwargs) - # Step 3: Initial authorization request response = _request("GET", auth_url, allow_redirects=True) - # WAF/CAPTCHA Detection if response.status_code in (403, 429) or "captcha" in response.text.lower(): - raise WafBlockedException("Joyn login blocked by WAF/CAPTCHA challenge") + raise WafBlockedException("Joyn login blocked by WAF/CAPTCHA") response.raise_for_status() final_url = response.url - logger.debug(f"Final URL after redirect: {final_url[:200]}...") - - # Check for error response if "error.html" in final_url or "error_code" in final_url: error_match = re.search(r'error_code=(\d+)', final_url) error_code = error_match.group(1) if error_match else "unknown" @@ -541,7 +427,6 @@ class JoynAuthenticator(BaseOAuth2Authenticator): error_desc = error_desc.group(1) if error_desc else "unknown" raise Exception(f"Authorization failed: error_code={error_code}, description={error_desc}") - # Check if already authenticated (direct callback) if self.oauth_redirect_uri in final_url: parsed = urlparse(final_url) query = parse_qs(parsed.query) @@ -555,33 +440,29 @@ class JoynAuthenticator(BaseOAuth2Authenticator): cd1=cd1, ) - # Step 4: Extract requestId - should be on signin.7pass.de parsed_url = urlparse(final_url) query_params = parse_qs(parsed_url.query) request_id = query_params.get("requestId", [None])[0] if not request_id: - # Try HTML extraction as fallback match = re.search(r'requestId["\']?\s*[=:]\s*["\']([^"\']+)', response.text) if match: request_id = match.group(1) if not request_id: - logger.error(f"Failed to extract request_id. Final URL: {final_url}") raise Exception("Could not extract request_id from response") logger.debug(f"Extracted request_id: {request_id}") - # Step 5: Public endpoint call + # Public endpoint try: - public_response = _request("GET", f"https://auth.7pass.de/public-srv/public/{request_id}") - public_response.raise_for_status() + _request("GET", f"https://auth.7pass.de/public-srv/public/{request_id}") except Exception as e: - logger.debug(f"Public endpoint call failed (non-fatal): {e}") + logger.debug(f"Public endpoint failed (non-fatal): {e}") - # Step 6: Check if user exists + # Check if user exists try: - check_response = _request( + _request( "POST", f"https://auth.7pass.de/users-srv/user/checkexists/{request_id}", json={"email": username, "requestId": request_id}, @@ -590,8 +471,7 @@ class JoynAuthenticator(BaseOAuth2Authenticator): except Exception as e: logger.debug(f"User check failed (non-fatal): {e}") - # Step 7: Submit login credentials - logger.debug(f"Submitting credentials for: {username}") + # Submit login login_response = _request( "POST", "https://auth.7pass.de/login-srv/verification/login", @@ -604,28 +484,24 @@ class JoynAuthenticator(BaseOAuth2Authenticator): allow_redirects=True, ) login_response.raise_for_status() - final_url = login_response.url - # Check for error if "error.html" in final_url or "error_code" in final_url: error_match = re.search(r'error_code=(\d+)', final_url) error_code = error_match.group(1) if error_match else "unknown" raise Exception(f"Login failed with error_code={error_code}") - # Step 8: Parse response parsed = urlparse(final_url) params = parse_qs(parsed.query) - # Step 9: Handle consent if needed + # Handle consent if params.get("code") is None: sub = params.get("sub", [None])[0] track_id = params.get("track_id", [None])[0] if sub and track_id: logger.debug(f"Accepting consent for sub={sub}") - - consent_response = _request( + _request( "POST", "https://auth.7pass.de/login-srv/consent/accept", json={ @@ -635,7 +511,6 @@ class JoynAuthenticator(BaseOAuth2Authenticator): }, content_type="application/json" ) - consent_response.raise_for_status() continue_response = _request( "POST", @@ -645,70 +520,46 @@ class JoynAuthenticator(BaseOAuth2Authenticator): allow_redirects=True, ) continue_response.raise_for_status() - final_url = continue_response.url parsed = urlparse(final_url) params = parse_qs(parsed.query) - # Step 10: Extract authorization code auth_code = params.get("code", [None])[0] if not auth_code: - raise Exception(f"No authorization code in response: {final_url}") + raise Exception(f"No authorization code in response") - logger.debug("Authorization code obtained successfully") + logger.debug("Authorization code obtained") - # Step 11: Exchange code for tokens - token_data = self._exchange_authorization_code_for_token( + return self._exchange_authorization_code_for_token( authorization_code=auth_code, code_verifier=code_verifier, state=state, cd1=cd1, ) - logger.info("Joyn login flow completed successfully") - return token_data - except WafBlockedException: raise except Exception as e: logger.error(f"Joyn login flow failed: {e}") raise - # ======================================================================== - # Authentication with Fallback Chain - # ======================================================================== - def authenticate_with_fallback(self, username: str, password: str) -> Dict[str, Any]: - """ - Authenticate with username/password, with automatic fallback chain: - 1. Try user credentials flow - 2. If user flow fails, fall back to client credentials (anonymous) - """ try: - # Try user authentication first return self._perform_oauth_authorization_code_flow(username, password) except Exception as e: logger.warning(f"User authentication failed: {e}, falling back to client credentials") - # Fall back to anonymous client credentials return self._perform_oauth_client_credentials_flow() - # ======================================================================== - # Client Credentials Flow (Anonymous) - CORRECTED ENDPOINT - # ======================================================================== - def _perform_oauth_client_credentials_flow(self) -> Dict[str, Any]: - """Client credentials flow for anonymous access using /auth/anonymous endpoint""" try: - logger.info(f"Starting client credentials flow for anonymous access") + logger.info(f"Starting client credentials flow") - # Build payload matching browser log payload = { "client_id": self.oauth_client_id, "client_name": self.platform, "anon_device_id": self._device_id } - # Use the correct anonymous endpoint (NOT /auth/7pass/token) anonymous_token_url = "https://auth.joyn.de/auth/anonymous" headers = { @@ -732,19 +583,14 @@ class JoynAuthenticator(BaseOAuth2Authenticator): response.raise_for_status() token_data = response.json() - logger.info(f"Client credentials flow successful - anonymous access granted") + logger.info(f"Client credentials flow successful") return token_data except Exception as e: logger.error(f"Client credentials flow failed: {e}") raise - # ======================================================================== - # Credentials & Token Management - # ======================================================================== - def get_fallback_credentials(self) -> JoynCredentials: - """Get fallback credentials for anonymous access""" return JoynCredentials( client_id=self._client_id, client_secret="", @@ -752,13 +598,11 @@ class JoynAuthenticator(BaseOAuth2Authenticator): ) def _build_auth_payload(self) -> Dict[str, Any]: - """Build payload for client credentials flow""" if not self.credentials: raise Exception("No credentials available") return self.credentials.to_auth_payload() def _create_token_from_response(self, response_data: Dict[str, Any]) -> BaseAuthToken: - """Create token from API response""" token = JoynAuthToken( access_token=response_data["access_token"], refresh_token=response_data.get("refresh_token", ""), @@ -770,7 +614,6 @@ class JoynAuthenticator(BaseOAuth2Authenticator): return token def _classify_token(self, token: BaseAuthToken) -> TokenAuthLevel: - """Classify token based on JWT claims""" try: if not token or not token.access_token: return TokenAuthLevel.UNKNOWN @@ -794,7 +637,6 @@ class JoynAuthenticator(BaseOAuth2Authenticator): return TokenAuthLevel.UNKNOWN def _perform_authentication(self) -> BaseAuthToken: - """Complete authentication based on credential type""" if isinstance(self.credentials, UserPasswordCredentials): token_data = self.authenticate_with_fallback( self.credentials.username, self.credentials.password @@ -804,20 +646,13 @@ class JoynAuthenticator(BaseOAuth2Authenticator): return self._create_token_from_response(token_data) - # ======================================================================== - # Public Methods - # ======================================================================== - 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) 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""" self._current_token = None try: self.settings_manager.clear_token(self.provider_name) @@ -825,7 +660,6 @@ class JoynAuthenticator(BaseOAuth2Authenticator): pass 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"}