nirvana-7777/script.service.ultimate
1
0
Fork 0
mirror of https://github.com/nirvana-7777/script.service.ultimate.git synced 2026-10-10 18:02:23 +02:00
Code Issues Packages Projects Releases Wiki Activity

joyn auth

Browse Source
This commit is contained in:
Nirvana
2026-05-29 16:12:58 +02:00
parent 1782a109a2
commit 580572f840
1 changed files with 44 additions and 210 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/providers/joyn/auth.py
+44 -210
View File
@@ -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"}