From 44f9d0645b3fc5f49c779d9a50bc6fa8fbb208b6 Mon Sep 17 00:00:00 2001 From: Nirvana Date: Tue, 19 May 2026 13:09:08 +0200 Subject: [PATCH] Joyn change auth --- .../providers/joyn/auth.py | 971 +++++++++++++----- 1 file changed, 714 insertions(+), 257 deletions(-) diff --git a/lib/streaming_providers/providers/joyn/auth.py b/lib/streaming_providers/providers/joyn/auth.py index 51d43ba..e538cf9 100644 --- a/lib/streaming_providers/providers/joyn/auth.py +++ b/lib/streaming_providers/providers/joyn/auth.py @@ -5,10 +5,11 @@ import time import uuid from dataclasses import dataclass, field from typing import Any, Dict, Optional -from urllib.parse import parse_qs, urlparse +from urllib.parse import parse_qs, urlencode, urlparse -from ...base.auth.base_auth import BaseAuthenticator, BaseAuthToken, TokenAuthLevel -from ...base.auth.credentials import ClientCredentials, UserPasswordCredentials +from ...base.auth.base_auth import BaseAuthToken, TokenAuthLevel +from ...base.auth.base_oauth2_auth import BaseOAuth2Authenticator +from ...base.auth.credentials import ClientCredentials from ...base.models.proxy_models import ProxyConfig from ...base.utils.logger import logger from .constants import ( @@ -20,6 +21,7 @@ from .constants import ( JOYN_AUTH_ENDPOINTS, JOYN_CLIENT_VERSION, JOYN_DOMAINS, + JOYN_OAUTH_SCOPE, JOYN_SSO_DISCOVERY_URL, JOYN_USER_AGENT, SUPPORTED_COUNTRIES, @@ -40,10 +42,11 @@ class JoynSSODiscovery: self.platform = platform self._endpoints_cache = None self._cache_timestamp = None - self._cache_ttl = 3600 + 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", @@ -52,7 +55,12 @@ class JoynSSODiscovery: } def get_endpoints(self, force_refresh: bool = False) -> Dict[str, str]: - if self._endpoints_cache and not force_refresh and time.time() - self._cache_timestamp < self._cache_ttl: + """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: @@ -60,39 +68,64 @@ class JoynSSODiscovery: "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 = 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() - endpoint = fallback.get(auth_type) or fallback.get(f"{self.platform}-login") or fallback.get("web-login", "") + # 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) + """ + 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: @@ -100,6 +133,7 @@ class JoynCredentials(ClientCredentials): return True 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, @@ -113,9 +147,14 @@ class JoynCredentials(ClientCredentials): @dataclass class JoynAuthToken(BaseAuthToken): + """ + 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 "", @@ -125,50 +164,72 @@ 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 + try: if not self.access_token: return None + parts = self.access_token.split(".") if len(parts) != 3: return None - import base64 - import json + 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: - def __init__(self, country: str, distribution_tenant: str, http_manager, platform: str = DEFAULT_PLATFORM): + """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", @@ -177,18 +238,25 @@ class JoynAuthConfig: } def get_auth_headers(self) -> Dict[str, str]: + """Get headers for authentication requests""" headers = self.get_base_headers() - headers.update({ - "joyn-client-version": JOYN_CLIENT_VERSION, - "joyn-country": self.country.upper(), - "joyn-distribution-tenant": self.distribution_tenant, - "joyn-platform": self.platform, - "joyn-request-id": str(uuid.uuid4()), - }) + headers.update( + { + "joyn-client-version": JOYN_CLIENT_VERSION, + "joyn-country": self.country.upper(), + "joyn-distribution-tenant": self.distribution_tenant, + "joyn-platform": self.platform, + "joyn-request-id": str(uuid.uuid4()), + } + ) return headers -class JoynAuthenticator(BaseAuthenticator): +class JoynAuthenticator(BaseOAuth2Authenticator): + """ + Joyn authenticator using OAuth2 client credentials flow with dynamic endpoints + """ + def __init__( self, country: str = DEFAULT_COUNTRY, @@ -199,19 +267,38 @@ class JoynAuthenticator(BaseAuthenticator): http_manager=None, proxy_config: Optional[ProxyConfig] = None, ): + """ + Initialize authenticator for specific country + """ if country not in SUPPORTED_COUNTRIES: - raise ValueError(f"Unsupported country: {country}") - if http_manager is None: - raise ValueError("http_manager is required for JoynAuthenticator") + raise ValueError( + f"Unsupported country: {country}. Must be one of: {SUPPORTED_COUNTRIES}" + ) + # Validate that http_manager is provided + if http_manager is None: + raise ValueError( + "http_manager is required for JoynAuthenticator. " + "It should be created in JoynProvider and passed to the authenticator." + ) + + # Set country-specific attributes FIRST 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 - self._proxy_config = proxy_config - self._config = JoynAuthConfig(self.country, self.distribution_tenant, self._http_manager, self.platform) + + # 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() + # NOW call parent __init__ - config, http_manager AND country are ready super().__init__( provider_name="joyn", settings_manager=settings_manager, @@ -219,40 +306,108 @@ class JoynAuthenticator(BaseAuthenticator): country=country, config_dir=config_dir, enable_kodi_integration=True, + http_manager=self._http_manager, + proxy_config=proxy_config, ) + def _get_joyn_auth_headers(self) -> Dict[str, str]: + """Get standardized Joyn authentication headers""" + from .constants import JOYN_AUTH_HEADERS_BASE + + headers = JOYN_AUTH_HEADERS_BASE.copy() + headers["Origin"] = f"https://www.joyn.{self.country.lower()}" + headers.update( + { + "joyn-country": self.country.upper(), + "joyn-distribution-tenant": self.distribution_tenant, + "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 http_manager(self): - return self._http_manager + 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() - return JOYN_AUTH_ENDPOINTS["ANONYMOUS"] + 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() - def _get_joyn_auth_headers(self) -> Dict[str, str]: - from .constants import JOYN_AUTH_HEADERS_BASE - headers = JOYN_AUTH_HEADERS_BASE.copy() - headers["Origin"] = f"https://www.joyn.{self.country.lower()}" - headers.update({ - "joyn-country": self.country.upper(), - "joyn-distribution-tenant": self.distribution_tenant, - "joyn-platform": self.platform, - "joyn-request-id": str(uuid.uuid4()), - }) - return headers - def _build_auth_payload(self) -> Dict[str, Any]: + """Build authentication payload - only used 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 object from API response""" token = JoynAuthToken( access_token=response_data["access_token"], refresh_token=response_data.get("refresh_token", ""), @@ -260,14 +415,18 @@ class JoynAuthenticator(BaseAuthenticator): 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="", + client_secret="", # Joyn doesn't use client_secret country=self.country, ) @@ -275,264 +434,562 @@ class JoynAuthenticator(BaseAuthenticator): """Get token redemption URL for OAuth flows""" return self._config.get_token_redeem_endpoint() - def _extract_client_id_from_endpoints(self) -> str: - try: - endpoints = self._config.sso_discovery.get_endpoints() - platform_key = f"{self.platform}-login" - login_url = endpoints.get(platform_key) - if not login_url: - login_url = endpoints.get("web-login") - if not login_url: - raise Exception("No login endpoint found") - 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") - return client_id - except Exception as e: - logger.error(f"Error extracting client_id: {e}, using fallback") - return DEVICE_IDS.get(self.platform, DEVICE_IDS[DEFAULT_PLATFORM]) - - def _get_sso_endpoints(self) -> Dict[str, str]: - 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() - return response.json() - - def _perform_oauth_client_credentials_flow(self) -> Dict[str, Any]: - try: - headers = self._get_auth_headers() - payload = self._build_auth_payload() - response = self.http_manager.post(self.auth_endpoint, operation="auth", headers=headers, json_data=payload) - response.raise_for_status() - return response.json() - except Exception as e: - logger.error(f"Client credentials flow failed: {e}") - raise - - def _perform_oauth_authorization_code_flow(self, username: str, password: str) -> Dict[str, Any]: - try: - logger.debug("Starting Joyn form-based login flow") - endpoints = self._get_sso_endpoints() - web_login_url = endpoints.get('web-login') - parsed = urlparse(web_login_url) - query_params = parse_qs(parsed.query) - client_id = query_params.get('client_id', [None])[0] - if not client_id: - raise Exception("Could not extract client_id from web-login URL") - - # Clear cookies - if hasattr(self.http_manager, 'clear_cookies'): - self.http_manager.clear_cookies() - - # Add browser headers - if hasattr(self.http_manager, '_session'): - self.http_manager._session.headers.update({ - "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8", - "Accept-Language": "de-DE,de;q=0.9,en;q=0.8", - "Upgrade-Insecure-Requests": "1", - "Sec-Fetch-Dest": "document", - "Sec-Fetch-Mode": "navigate", - "Sec-Fetch-Site": "none", - "Sec-Fetch-User": "?1", - }) - - # Step 1: Get request_id - response = self.http_manager.get(web_login_url, operation="oauth", allow_redirects=True) - final_url = response.url - parsed_url = urlparse(final_url) - request_id = parse_qs(parsed_url.query).get('requestId', [None])[0] - if not request_id: - request_match = re.search(r'requestId["\']?\s*:\s*["\']([^"\']+)', response.text) - if request_match: - request_id = request_match.group(1) - else: - raise Exception("Could not extract request_id") - - # Step 2-4: Setup flow - lang_url = f'https://auth.7pass.de/registration-setup-srv/public/list?acceptlanguage=undefined&requestId={request_id}' - self.http_manager.get(lang_url, operation="oauth").raise_for_status() - check_url = f'https://auth.7pass.de/users-srv/user/checkexists/{request_id}' - self.http_manager.post(check_url, operation="oauth", json_data={"email": username, "requestId": request_id}).raise_for_status() - verify_url = 'https://auth.7pass.de/verification-srv/v2/setup/public/configured/list' - self.http_manager.post(verify_url, operation="oauth", json_data={"email": username, "request_id": request_id}).raise_for_status() - - # Step 5: Submit password - login_url = 'https://auth.7pass.de/login-srv/login' - login_data = {"username": username, "password": password, "requestId": request_id} - response = self.http_manager._session.post( - login_url, data=login_data, - headers={"Content-Type": "application/x-www-form-urlencoded", "Referer": final_url, "Origin": "https://signin.7pass.de"}, - timeout=self.http_manager.config.timeout, allow_redirects=False - ) - if response.status_code in [302, 303]: - redirect_url = response.headers.get('Location') - response = self.http_manager.get(redirect_url, operation="oauth") - redirect_url = response.url - else: - redirect_url = response.url - - # Step 6-7: Extract codes and handle consent - parsed = urlparse(redirect_url) - id_dict = parse_qs(parsed.query) - if 'code' not in id_dict: - sub = id_dict.get('sub', [None])[0] - if sub: - consent_url = 'https://auth.7pass.de/consent-management-srv/consent/scope/accept' - self.http_manager.post(consent_url, operation="oauth", json_data={"sub": sub, "client_id": client_id, "scopes": [{"offline_access": "denied"}]}).raise_for_status() - track_id = id_dict.get('track_id', [None])[0] or id_dict.get('cd1', [None])[0] - if track_id: - continue_url = f'https://auth.7pass.de/login-srv/precheck/continue/{track_id}' - response = self.http_manager.get(continue_url, operation="oauth", allow_redirects=False) - if response.status_code in [302, 303] and response.headers.get('Location'): - response = self.http_manager.get(response.headers['Location'], operation="oauth") - parsed = urlparse(response.url) - id_dict = parse_qs(parsed.query) - - # Step 8: Exchange code for token - code = id_dict.get('code', [None])[0] - if not code: - raise Exception("No authorization code found") - tracking_id = id_dict.get('cd1', [None])[0] or id_dict.get('track_id', [None])[0] - token_data = { - "client_id": client_id, "code": code, "code_verifier": "", - "redirect_uri": f"https://www.joyn.{self.country}/oauth", - "tracking_id": tracking_id, "tracking_name": "web" - } - token_endpoint = endpoints.get('redeem-token') - response = self.http_manager.post(token_endpoint, operation="auth", json_data=token_data) - response.raise_for_status() - auth_token = response.json() - auth_token['has_account'] = True - auth_token['issued_at'] = time.time() - return auth_token - except Exception as e: - logger.error(f"Joyn form-based login failed: {e}") - raise - + # MAIN AUTHENTICATION METHOD def _perform_authentication(self) -> BaseAuthToken: + """ + Perform authentication using appropriate flow based on credential type + """ + from ...base.auth.credentials import ClientCredentials, UserPasswordCredentials + if isinstance(self.credentials, UserPasswordCredentials): - token_data = self._perform_oauth_authorization_code_flow(self.credentials.username, self.credentials.password) - else: + # Use OAuth2 authorization code flow with PKCE + 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 + ) + elif isinstance(self.credentials, ClientCredentials): + # Use client credentials flow (anonymous auth) + logger.info(f"Using OAuth2 client credentials flow for {self.provider_name}") token_data = self._perform_oauth_client_credentials_flow() + else: + raise Exception( + f"Unsupported credential type for {self.provider_name}: {type(self.credentials)}" + ) + return self._create_token_from_response(token_data) - def _refresh_token(self) -> Optional[BaseAuthToken]: - """Refresh the current token using refresh token""" + def _perform_oauth_client_credentials_flow(self) -> Dict[str, Any]: + """ + Client credentials flow — Joyn uses JSON instead of form data. + """ + try: + logger.debug(f"Starting Joyn-specific OAuth2 client credentials flow") + + headers = self._get_auth_headers() + payload = self._build_auth_payload() + + logger.debug(str(headers)) + logger.debug(str(payload)) + + response = self.http_manager.post( + self.auth_endpoint, + operation="auth", + headers=headers, + json_data=payload, + ) + + 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 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}") + + def _perform_two_step_verification( + self, session, username: str, password: str, request_id: str, referer_url: str + ) -> tuple[str, str]: + """ + Perform the two-step verification process using constants. + Returns: (final_sub, final_status_id) + """ + from .constants import JOYN_CIDAAS_ENDPOINTS, JOYN_USER_AGENT + + # Common headers for verification requests + verification_headers = { + "User-Agent": JOYN_USER_AGENT, + "Accept": "application/json", + "Content-Type": "application/json", + "Origin": "https://signin.7pass.de", + "Referer": referer_url, + } + + # Step 1: Initiate password verification + initiate_data = { + "request_id": request_id, + "email": username, + "medium_id": "PASSWORD", + "usage_type": "PASSWORDLESS_AUTHENTICATION", + "type": "PASSWORD", + } + + logger.debug("Step 1: Initiating password verification") + initiate_response = session.post( + JOYN_CIDAAS_ENDPOINTS["VERIFICATION_INITIATE"], + json_data=initiate_data, + headers=verification_headers, + timeout=self._config.timeout, + ) + + logger.debug(f"Initiate response status: {initiate_response.status_code}") + initiate_response.raise_for_status() + initiate_result = initiate_response.json() + + if not initiate_result.get("success"): + raise Exception(f"Password verification initiation failed: {initiate_result}") + + exchange_data = initiate_result["data"] + exchange_id = exchange_data["exchange_id"]["exchange_id"] + sub = exchange_data["sub"] + status_id = exchange_data["status_id"] + + logger.debug(f"Got exchange_id: {exchange_id}, sub: {sub}, status_id: {status_id}") + + # Step 2: Authenticate with password + authenticate_data = { + "exchange_id": exchange_id, + "pass_code": password, + "sub": sub, + "type": "PASSWORD", + "password": password, + } + + logger.debug("Step 2: Authenticating with password") + authenticate_response = session.post( + JOYN_CIDAAS_ENDPOINTS["VERIFICATION_AUTHENTICATE"], + json_data=authenticate_data, + headers=verification_headers, + timeout=self._config.timeout, + ) + + logger.debug(f"Authenticate response status: {authenticate_response.status_code}") + authenticate_response.raise_for_status() + authenticate_result = authenticate_response.json() + + if not authenticate_result.get("success"): + raise Exception(f"Password authentication failed: {authenticate_result}") + + auth_exchange_data = authenticate_result["data"] + final_sub = auth_exchange_data["sub"] + final_status_id = auth_exchange_data["status_id"] + + logger.debug(f"Got final sub: {final_sub}, final_status_id: {final_status_id}") + return final_sub, final_status_id + + def _perform_final_login( + self, session, final_sub: str, final_status_id: str, request_id: str + ) -> Optional[str]: + """ + Perform the final login with verification results. + + Returns the redirect Location URL containing the authorization code, + or None if no redirect was returned. + + FIX: Previously returned bool and discarded the redirect URL, which + made authorization code extraction impossible. Now returns the Location + header so the caller can extract the code directly. + """ + from .constants import JOYN_CIDAAS_ENDPOINTS, JOYN_USER_AGENT + + login_data = { + "sub": final_sub, + "status_id": final_status_id, + "verificationType": "PASSWORD", + "requestId": request_id, + "remember_me": "true", + } + + login_headers = { + "User-Agent": JOYN_USER_AGENT, + "Accept": "application/json", + "Content-Type": "application/x-www-form-urlencoded", + "Origin": "https://signin.7pass.de", + } + + logger.debug("Step 3: Final login with verification results") + login_response = session.post( + JOYN_CIDAAS_ENDPOINTS["LOGIN_VERIFICATION"], + data=urlencode(login_data).encode(), + headers=login_headers, + timeout=self._config.timeout, + allow_redirects=False, + ) + + logger.debug(f"Login response status: {login_response.status_code}") + login_response.raise_for_status() + + if login_response.status_code in (302, 303): + location = login_response.headers.get("Location") + if location: + logger.debug(f"Final login redirected to: {location}") + return location + logger.warning("Final login returned redirect status but no Location header") + return None + + # 200 response — some flows return the callback URL in the body + if login_response.status_code == 200: + try: + body = login_response.json() + callback_url = body.get("redirect_uri") or body.get("location") or body.get("url") + if callback_url: + logger.debug(f"Final login returned callback URL in body: {callback_url}") + return callback_url + except (ValueError, KeyError, AttributeError): + pass + logger.warning("Final login returned 200 but no redirect URL could be extracted") + + return None + + @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: + return None + + parsed = urlparse(url) + + # 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]: + """ + Joyn OAuth2 authorization code flow with two-step verification and PKCE. + Handles both cases: existing session (direct redirect) and new login (form-based). + + FIX: Authorization code extraction now works by: + 1. Capturing the Location header returned by _perform_final_login + 2. Parsing the code directly from that URL via _extract_code_from_url + The old _extract_authorization_code stub is replaced by these two methods. + """ + try: + logger.debug("Starting optimized Joyn OAuth2 authorization code flow") + + # Step 1: Get authorization endpoint and build URL + web_login_url = self._config.get_authorize_endpoint() + logger.debug(f"Using authorize endpoint: {web_login_url}") + + # Build authorization URL with PKCE + code_verifier = self.generate_pkce_verifier() + code_challenge = self.generate_pkce_challenge(code_verifier) + 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", + "response_mode": "query", + "view_type": "login", + "prompt": "consent", + "cd1": str(uuid.uuid4()), + } + + authorization_url = f"{web_login_url}?{urlencode(params)}" + logger.debug(f"Built authorization URL: {authorization_url}") + + session = self._create_oauth_session() + + # Step 2: Get authorization page + logger.debug("Fetching authorization page") + auth_response = session.get(authorization_url, timeout=self._config.timeout) + + 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) + 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) + + auth_code = query_params.get("code", [None])[0] + received_state = query_params.get("state", [None])[0] + + if not auth_code: + raise Exception("Redirect to callback but no authorization code found") + + 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") + + token_data = self._exchange_authorization_code_for_token( + authorization_code=auth_code, + code_verifier=code_verifier, + state=state, + ) + + logger.debug("Joyn OAuth2 authorization code flow successful (existing session)") + return token_data + + # Step 2b: No existing session — perform form-based login + logger.info("No existing session - performing form-based login") + + # Extract request_id from the response URL + parsed_url = urlparse(auth_response.url) + query_params = parse_qs(parsed_url.query) + request_id = query_params.get("requestId", [None])[0] + + if not request_id: + # Fallback: extract from response body + request_id_match = re.search( + r'requestId["\']?\s*:\s*["\']([^"\']+)', auth_response.text + ) + if request_id_match: + request_id = request_id_match.group(1) + else: + raise Exception("Could not extract request_id from authorization page") + + logger.debug(f"Extracted request_id: {request_id}") + + # Step 3: Two-step verification + final_sub, final_status_id = self._perform_two_step_verification( + session, username, password, request_id, auth_response.url + ) + + # Step 4: Final login — returns the redirect Location URL + redirect_url = self._perform_final_login( + session, final_sub, final_status_id, request_id + ) + + if not redirect_url: + raise Exception( + "Final login did not return a redirect URL. " + "Cannot extract authorization code." + ) + + # Step 5: Extract authorization code from the redirect URL + auth_code = self._extract_code_from_url(redirect_url) + + if not auth_code: + raise Exception( + f"Could not extract authorization code from redirect URL: {redirect_url}" + ) + + logger.debug(f"Extracted authorization code from redirect") + + # Step 6: Exchange authorization code for tokens + logger.debug("Exchanging authorization code for tokens") + token_data = self._exchange_authorization_code_for_token( + authorization_code=auth_code, code_verifier=code_verifier, state=state + ) + + logger.debug("Joyn OAuth2 authorization code flow successful (form login)") + return token_data + + except Exception as e: + logger.error(f"Joyn OAuth2 authorization code flow failed: {e}") + raise Exception(f"OAuth2 authorization code flow failed: {e}") + + 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""" + return { + "code": authorization_code, + "client_id": self.oauth_client_id, + "redirect_uri": self.oauth_redirect_uri, + "tracking_id": str(uuid.uuid4()), + "tracking_name": self.platform, + "code_verifier": code_verifier, + # 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""" 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}") + logger.debug(f"Refreshing OAuth2 token for {self.provider_name}") + + # Joyn refresh uses the platform device ID, not the OAuth client ID payload = { "client_id": DEVICE_IDS.get(self.platform, DEVICE_IDS[DEFAULT_PLATFORM]), "client_name": self.platform, - "grant_type": "Bearer", + "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"] - response = self.http_manager.post(refresh_endpoint, operation="auth", headers=headers, json_data=payload, timeout=self._config.timeout) + + 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}") + if response.status_code >= 400: + logger.error(f"Refresh error response: {response.text}") + response.raise_for_status() + 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}") + logger.info(f"OAuth2 token refresh successful for {self.provider_name}") return refreshed_token + except Exception as e: - logger.warning(f"Token refresh failed for {self.provider_name}: {e}") + logger.warning(f"OAuth2 token refresh failed for {self.provider_name}: {e}") return None - def _classify_token(self, token: BaseAuthToken) -> TokenAuthLevel: - """Classify Joyn token based on JWT claims""" - try: - if not token or not token.access_token: - return TokenAuthLevel.UNKNOWN - - # Parse JWT token - try: - parts = token.access_token.split(".") - if len(parts) != 3: - return TokenAuthLevel.UNKNOWN - import base64 - import json - payload_b64 = parts[1] - padding = len(payload_b64) % 4 - if padding: - payload_b64 += "=" * (4 - padding) - payload_json = base64.b64decode(payload_b64).decode("utf-8") - claims = json.loads(payload_json) - except Exception as e: - logger.warning(f"Failed to parse JWT: {e}") - return TokenAuthLevel.UNKNOWN - - # Classification logic - 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["web"], DEVICE_IDS["android"], DEVICE_IDS["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 - - scope = claims.get("scope", "") - if scope: - scopes = scope.split() - if "offline_access" in scopes and "profile" in scopes: - return TokenAuthLevel.USER_AUTHENTICATED - elif "openid" in scopes and len(scopes) <= 2: - return TokenAuthLevel.CLIENT_CREDENTIALS - - return TokenAuthLevel.UNKNOWN - except Exception as e: - logger.error(f"Error classifying token: {e}") - return TokenAuthLevel.UNKNOWN - + # 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""" + """Invalidate current token (forces re-authentication on next request)""" self._current_token = None try: - self.settings_manager.clear_token(self.provider_name, self.country) - except (AttributeError, KeyError, TypeError): - try: - self.settings_manager.clear_token(self.provider_name) - except (AttributeError, KeyError): - pass + 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__, "auth_level": self._current_token.auth_level.value, "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