From 0843f816c2b538d893a017eef8bcb81a1e3ece2f Mon Sep 17 00:00:00 2001 From: Nirvana Date: Fri, 29 May 2026 10:02:47 +0200 Subject: [PATCH] Oauth2 additions --- lib/streaming_providers/base/auth/__init__.py | 17 + .../base/auth/base_oauth2_auth.py | 196 +++++- .../base/auth/remote_login_extension.py | 197 ++++++ .../base/auth/remote_login_manager.py | 568 ++++++++++++++++++ .../providers/rtlplus/auth.py | 5 + .../providers/rtlplus/constants.py | 4 +- 6 files changed, 964 insertions(+), 23 deletions(-) create mode 100644 lib/streaming_providers/base/auth/remote_login_extension.py create mode 100644 lib/streaming_providers/base/auth/remote_login_manager.py diff --git a/lib/streaming_providers/base/auth/__init__.py b/lib/streaming_providers/base/auth/__init__.py index 7e93ee9..4686699 100644 --- a/lib/streaming_providers/base/auth/__init__.py +++ b/lib/streaming_providers/base/auth/__init__.py @@ -1,8 +1,16 @@ # streaming_providers/base/auth/__init__.py from .base_auth import BaseAuthenticator, BaseAuthToken +from .base_oauth2_auth import ( + BaseOAuth2Authenticator, + OAuth2Error, + OIDCConfiguration, + SessionAwareHTTPManager, +) from .credential_manager import CredentialManager from .credentials import BaseCredentials, ClientCredentials, UserPasswordCredentials from .session_manager import SessionManager +from .remote_login_extension import OAuth2RemoteLoginMixin +from .remote_login_manager import RemoteLoginManager, RemoteLoginSession # Only export what consumers should use __all__ = [ @@ -13,4 +21,13 @@ __all__ = [ "ClientCredentials", "SessionManager", "CredentialManager", + # OAuth2 base + "BaseOAuth2Authenticator", + "OAuth2Error", + "OIDCConfiguration", + "SessionAwareHTTPManager", + # Remote login + "RemoteLoginManager", + "RemoteLoginSession", + "OAuth2RemoteLoginMixin", ] diff --git a/lib/streaming_providers/base/auth/base_oauth2_auth.py b/lib/streaming_providers/base/auth/base_oauth2_auth.py index ea556c8..67089b0 100644 --- a/lib/streaming_providers/base/auth/base_oauth2_auth.py +++ b/lib/streaming_providers/base/auth/base_oauth2_auth.py @@ -11,11 +11,13 @@ import uuid from abc import abstractmethod from dataclasses import dataclass, field from typing import Any, Callable, Dict, List, Optional -from urllib.parse import parse_qs, urlencode, urlparse +from urllib.parse import parse_qs, quote_plus, urlencode, urlparse from ..models.proxy_models import ProxyConfig from ..utils.logger import logger from .base_auth import BaseAuthenticator, BaseAuthToken, TokenAuthLevel +from .remote_login_extension import OAuth2RemoteLoginMixin +from .remote_login_manager import RemoteLoginManager @dataclass @@ -120,7 +122,9 @@ class SessionAwareHTTPManager: self.cookies[cookie.name] = cookie.value -class BaseOAuth2Authenticator(BaseAuthenticator): +# OAuth2RemoteLoginMixin comes first so its methods take precedence over any +# same-named stubs that might exist in BaseAuthenticator. +class BaseOAuth2Authenticator(OAuth2RemoteLoginMixin, BaseAuthenticator): """ Base class for OAuth2/OIDC authentication with dynamic endpoint discovery. @@ -168,6 +172,9 @@ class BaseOAuth2Authenticator(BaseAuthenticator): self._oidc_discovery_failures: int = 0 self._oidc_discovery_successes: int = 0 + # Remote login manager (device/QR code flows) + self._remote_login_manager = RemoteLoginManager(self) + @property def http_manager(self): """Safe access to http_manager""" @@ -300,7 +307,6 @@ class BaseOAuth2Authenticator(BaseAuthenticator): 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}") @@ -517,7 +523,7 @@ class BaseOAuth2Authenticator(BaseAuthenticator): return session # ======================================================================== - # Client Credentials Flow - Fix: Use oauth_token_endpoint + # Client Credentials Flow # ======================================================================== def _perform_oauth_client_credentials_flow(self) -> Dict[str, Any]: @@ -531,7 +537,6 @@ class BaseOAuth2Authenticator(BaseAuthenticator): headers = self._get_auth_headers() data = self._build_auth_payload() - # Use oauth_token_endpoint instead of auth_endpoint response = self.http_manager.post( self.oauth_token_endpoint, operation="auth", headers=headers, data=data ) @@ -551,7 +556,14 @@ class BaseOAuth2Authenticator(BaseAuthenticator): # ======================================================================== def _build_authorization_url(self, extra_params: Dict[str, Any] = None) -> tuple[str, str, str]: - """Build authorization URL with optional PKCE""" + """ + Build authorization URL with optional PKCE. + + Note: extra_params are appended directly to the query string and sent + to the authorization server. Only pass parameters that the target + provider explicitly supports (e.g. login_hint, prompt, acr_values). + Do NOT pass internal implementation keys here. + """ state = self.generate_oauth_state() params = { "response_type": "code", @@ -594,7 +606,6 @@ class BaseOAuth2Authenticator(BaseAuthenticator): headers = self._get_token_exchange_headers(**kwargs) endpoint = kwargs.get('token_endpoint') or self._get_token_exchange_endpoint(**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) @@ -656,7 +667,7 @@ class BaseOAuth2Authenticator(BaseAuthenticator): return self.oauth_token_endpoint # ======================================================================== - # Generic Form-Based Login Flow - Fix: Preserve exception chain + # Generic Form-Based Login Flow # ======================================================================== def _perform_generic_form_login( @@ -719,7 +730,6 @@ class BaseOAuth2Authenticator(BaseAuthenticator): raise Exception(f"Authentication response validation failed: {error_msg}") # 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, @@ -732,7 +742,7 @@ class BaseOAuth2Authenticator(BaseAuthenticator): raise Exception(f"OAuth2 form-based login failed: {e}") from e # ======================================================================== - # Token Refresh - Fix: Consistent payload encoding + correct endpoint + # Token Refresh # ======================================================================== def _build_refresh_payload(self) -> Dict[str, Any]: @@ -763,7 +773,6 @@ class BaseOAuth2Authenticator(BaseAuthenticator): data = self._build_refresh_payload() headers = self._get_auth_headers() - # Apply consistent Content-Type logic if self._should_use_json_for_token_exchange(): headers["Content-Type"] = "application/json" request_kwargs = {"json_data": data} @@ -771,7 +780,6 @@ class BaseOAuth2Authenticator(BaseAuthenticator): 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.oauth_token_endpoint, operation="auth", headers=headers, **request_kwargs ) @@ -789,7 +797,7 @@ class BaseOAuth2Authenticator(BaseAuthenticator): return None # ======================================================================== - # Error Response Handling - Fix: Clean error handling + # Error Response Handling # ======================================================================== @staticmethod @@ -809,7 +817,7 @@ class BaseOAuth2Authenticator(BaseAuthenticator): ) # ======================================================================== - # JS Extraction Helpers - Fix: Selective exception handling + # JS Extraction Helpers # ======================================================================== def _extract_from_js( @@ -844,16 +852,13 @@ class BaseOAuth2Authenticator(BaseAuthenticator): 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 @@ -915,7 +920,6 @@ class BaseOAuth2Authenticator(BaseAuthenticator): current_token.auth_level = self._classify_token(current_token) logger.debug(f"Token classified as: {current_token.auth_level.value}") - # 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: @@ -1008,15 +1012,13 @@ class BaseOAuth2Authenticator(BaseAuthenticator): return self._refresh_oauth_token() # ======================================================================== - # Status and Diagnostics - Fix: Defensive endpoint access + # Status and Diagnostics # ======================================================================== 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: @@ -1041,6 +1043,7 @@ class BaseOAuth2Authenticator(BaseAuthenticator): "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, + "remote_login_active_sessions": self._remote_login_manager.get_active_session_count(), } if self._enable_oidc_discovery: @@ -1099,6 +1102,157 @@ class BaseOAuth2Authenticator(BaseAuthenticator): except Exception as e: return False, f"Error processing authentication response: {e}", None + # ======================================================================== + # Remote Login Flow Methods + # ======================================================================== + + def _perform_remote_login_flow(self) -> Dict[str, Any]: + """ + Perform complete remote login flow with QR code + polling. + + Called when normal form login is unavailable (e.g. WAF blocks it). + Subclasses may call this directly or override _perform_oauth_authorization_code_flow + to fall through to it automatically. + + Only passes standard OAuth2 parameters to the authorization URL. + Provider-specific extras (e.g. login_hint, prompt) should be added + by overriding this method or by passing them via extra_params in a + subclass-level _build_authorization_url call. + """ + # Build auth URL with PKCE using no extra_params; see docstring above + auth_url, state, code_verifier = self._build_authorization_url() + + # Append login_hint as a properly URL-encoded query parameter if available. + # This is done post-build to avoid coupling _build_authorization_url to + # credential-specific logic. + username = getattr(self.credentials, "username", None) + if username: + auth_url += f"&login_hint={quote_plus(username)}" + + # Create session via manager + session = self._remote_login_manager.create_session( + auth_url=auth_url, + state=state, + code_verifier=code_verifier, + expires_in=300, # 5 minutes + ) + + try: + # Create polling callback that delegates token exchange to authenticator + poll_callback = self._remote_login_manager.create_polling_callback( + session_id=session.session_id, + token_exchange_func=self._exchange_authorization_code_for_token, + ) + + logger.info( + f"Starting remote login for {self.provider_name} with code {session.login_code}" + ) + + token_data = self.show_remote_login_and_wait_for_auth( + login_code=session.login_code, + qr_target_url=session.auth_url, + expires_in=session.expires_in, + interval=2, + auth_callback=poll_callback, + ) + + logger.info(f"Remote login successful for {self.provider_name}") + return token_data + + except Exception: + # Enrich the exception with session-level error details if available + completed_session = self._remote_login_manager.get_session(session.session_id) + if completed_session and completed_session.error: + raise RuntimeError( + f"Remote login failed for {self.provider_name}: {completed_session.error}" + ) + raise + + def complete_remote_login(self, callback_url: str, session_id: Optional[str] = None) -> bool: + """ + Complete a remote login session using a full OAuth2 callback URL. + + Called by external components (file watcher, HTTP server, or user + input handler) when the callback URL becomes available. + + Args: + callback_url: The full callback URL containing ?code=xxx&state=yyy + session_id: Optional specific session ID to target + + Returns: + True if a pending session was successfully updated, False otherwise + """ + return self._remote_login_manager.complete_from_callback_url(callback_url, session_id) + + def complete_remote_login_with_code( + self, auth_code: str, session_id: Optional[str] = None + ) -> bool: + """ + Complete a remote login session with a raw authorization code. + + Use this when the user manually types or pastes the code. + + Args: + auth_code: The raw authorization code from the provider + session_id: Optional specific session ID to target + + Returns: + True if a pending session was successfully updated, False otherwise + """ + return self._remote_login_manager.complete_with_authorization_code( + auth_code=auth_code, + session_id=session_id, + ) + + def cancel_remote_login(self, session_id: Optional[str] = None) -> bool: + """ + Cancel a pending remote login session. + + Args: + session_id: Specific session ID to cancel, or None to cancel all + + Returns: + True if at least one session was cancelled, False otherwise + """ + return self._remote_login_manager.cancel_session(session_id) > 0 + + def start_remote_login_callback_server( + self, port: int = 8080, host: str = "127.0.0.1" + ) -> bool: + """ + Start an optional HTTP callback server for phone-to-device communication. + + When running, phones can send the authorization code back to the device + automatically by calling: + http://{host}:{port}/callback?code=xxx&token=yyy + + Retrieve the required token value via get_callback_server_token(). + + Args: + port: Port to listen on (default 8080) + host: Bind address. Use "127.0.0.1" (default) to restrict to + localhost, or "0.0.0.0" to accept connections from the + local network (e.g. from a phone on the same Wi-Fi). + + Returns: + True if the server started successfully, False otherwise + """ + return self._remote_login_manager.start_callback_server(port=port, host=host) + + def stop_remote_login_callback_server(self) -> None: + """Gracefully stop the HTTP callback server if running.""" + self._remote_login_manager.stop_callback_server() + + def get_callback_server_token(self) -> Optional[str]: + """ + Return the callback server's shared secret token. + + Embed this in QR URLs so that only the device that started the server + can accept completions: + http://192.168.1.x:8080/callback?code=xxx&token= + """ + return self._remote_login_manager.get_callback_server_token() + # ======================================================================== # Abstract Method # ======================================================================== diff --git a/lib/streaming_providers/base/auth/remote_login_extension.py b/lib/streaming_providers/base/auth/remote_login_extension.py new file mode 100644 index 0000000..4878fe0 --- /dev/null +++ b/lib/streaming_providers/base/auth/remote_login_extension.py @@ -0,0 +1,197 @@ +# streaming_providers/base/auth/remote_login_extension.py +""" +Remote Login Extension for BaseOAuth2Authenticator +Provides a production-ready method to handle device/remote login flows +with QR display, background polling, and proper timeout/cancellation handling. +Encapsulated to avoid modifying core OAuth2 or UI files. +""" +import threading +import time +from typing import Any, Callable, Dict, Optional + +from ..ui.notification_factory import NotificationFactory +from ..utils.logger import logger + + +class _RemoteLoginPollingWorker(threading.Thread): + """ + Internal polling worker that loops at the specified interval until + authentication succeeds, times out, or is cancelled. + + Designed so the main thread drives cancellation by calling stop(); the + worker never blocks longer than `interval` seconds at a time, so + cancellation latency is bounded. + """ + + def __init__( + self, + callback: Callable[[], Optional[Dict[str, Any]]], + expires_in: int, + interval: int, + ): + super().__init__(daemon=True, name="RemoteLoginPollingWorker") + self.callback = callback + self.expires_in = max(1, expires_in) + self.interval = max(1, interval) + + # Results — read only after join() + self.result: Optional[Dict[str, Any]] = None + self.error: Optional[str] = None + self.completed: bool = False + self.timed_out: bool = False + + self._stop_event = threading.Event() + self._start_time: float = 0.0 # Set in run() for accuracy + + def run(self) -> None: + self._start_time = time.monotonic() + + while not self._stop_event.is_set(): + elapsed = time.monotonic() - self._start_time + remaining = self.expires_in - elapsed + + if remaining <= 0: + self.timed_out = True + break + + try: + token_data = self.callback() + except Exception as e: + self.error = str(e) + break + + if token_data is not None: + self.result = token_data + self.completed = True + break + + # Sleep in small increments so stop() cancels promptly. + # Cap at `remaining` so we don't sleep past the deadline. + sleep_time = min(self.interval, max(0.5, remaining)) + self._stop_event.wait(timeout=sleep_time) + + def stop(self, timeout: float = 2.0) -> None: + """Signal the worker to stop and wait for it to exit.""" + self._stop_event.set() + self.join(timeout=timeout) + + +class OAuth2RemoteLoginMixin: + """ + Mixin that adds remote/device login capabilities to BaseOAuth2Authenticator. + Handles QR display, polling lifecycle, cancellation, and timeout gracefully. + + Usage: + class BaseOAuth2Authenticator(OAuth2RemoteLoginMixin, BaseAuthenticator): + ... + """ + + def show_remote_login_and_wait_for_auth( + self, + login_code: str, + qr_target_url: str, + expires_in: int, + interval: int = 5, + auth_callback: Optional[Callable[[], Optional[Dict[str, Any]]]] = None, + ) -> Dict[str, Any]: + """ + Show remote login UI and block until authentication completes, + times out, or is cancelled. + + Args: + login_code: Short code for manual entry fallback. + qr_target_url: URL to encode in the QR code. + expires_in: Total timeout in seconds before the session expires. + interval: Polling interval in seconds (minimum 1). + auth_callback: Callable that checks authentication status. + Must return token data (dict) if complete, or None + if still pending. + + Returns: + Dict containing token data on successful authentication. + + Raises: + ValueError: If input parameters are invalid. + TypeError: If auth_callback is not callable. + TimeoutError: If the authentication session expires. + RuntimeError: If the user cancels or a polling error occurs. + """ + # --- Input Validation --- + if not login_code or not isinstance(login_code, str): + raise ValueError("login_code must be a non-empty string") + if not qr_target_url or not isinstance(qr_target_url, str): + raise ValueError("qr_target_url must be a non-empty string") + if not isinstance(expires_in, int) or expires_in <= 0: + raise ValueError("expires_in must be a positive integer") + if not callable(auth_callback): + raise TypeError("auth_callback must be a callable returning Dict or None") + + logger.info( + f"Initializing remote login flow: code={login_code}, expires_in={expires_in}s" + ) + + # --- Get environment-appropriate UI adapter --- + adapter = NotificationFactory.create() + + worker: Optional[_RemoteLoginPollingWorker] = None + + try: + # Show UI (handles QR generation & display internally) + adapter.show_remote_login( + login_code=login_code, + qr_target_url=qr_target_url, + expires_in=expires_in, + interval=interval, + ) + + # --- Start polling worker --- + worker = _RemoteLoginPollingWorker(auth_callback, expires_in, interval) + worker.start() + + # --- Main loop: wait for completion, timeout, or cancellation --- + while worker.is_alive(): + if adapter.is_cancelled(): + worker.stop() + adapter.close(success=False, message="Cancelled by user") + raise RuntimeError("Remote login cancelled by user") + + if worker.error: + worker.stop() + adapter.close(success=False, message=worker.error) + raise RuntimeError(f"Polling failed: {worker.error}") + + # Small sleep to avoid busy-waiting while staying UI-responsive + time.sleep(0.2) + + # Worker has exited — inspect its terminal state + + if worker.completed and worker.result is not None: + adapter.close(success=True) + logger.info("Remote login authentication completed successfully") + return worker.result + + if worker.timed_out: + adapter.close(success=False, message="Session expired") + raise TimeoutError( + "Remote login session expired before authentication completed" + ) + + # Worker exited cleanly but without a result and without timing out — + # this means an error was set (handled above) or the stop event fired + # (cancellation already handled above). Reaching here is unexpected. + adapter.close(success=False, message="Authentication failed") + raise RuntimeError("Remote login failed unexpectedly") + + except BaseException as exc: + # Ensure the worker is always stopped and the adapter is always + # closed, even on KeyboardInterrupt or SystemExit. + # We close the adapter only if it hasn't been closed yet; any + # exception raised by adapter.close() is suppressed so the + # original exception propagates cleanly. + if worker is not None and worker.is_alive(): + worker.stop() + try: + adapter.close(success=False) + except Exception: + pass + raise \ No newline at end of file diff --git a/lib/streaming_providers/base/auth/remote_login_manager.py b/lib/streaming_providers/base/auth/remote_login_manager.py new file mode 100644 index 0000000..4c5a4ec --- /dev/null +++ b/lib/streaming_providers/base/auth/remote_login_manager.py @@ -0,0 +1,568 @@ +# streaming_providers/base/auth/remote_login_manager.py +""" +Remote Login Manager for OAuth2 Device/QR Code Flows +Encapsulates session state, polling logic, and external completion handlers. +Thread-safe, production-hardened, and environment-agnostic. +""" +import secrets +import string +import threading +import time +from dataclasses import dataclass, field +from typing import Any, Callable, Dict, Optional +from urllib.parse import parse_qs, urlparse + +from ..utils.logger import logger + + +@dataclass +class RemoteLoginSession: + """ + Mutable state for an active remote login session. + + All mutations must be performed while holding the manager's lock. + The fields `authorization_code`, `completed`, `cancelled`, and `error` + are intentionally mutable; callers are responsible for synchronisation. + """ + session_id: str + state: str + code_verifier: str + auth_url: str + login_code: str + created_at: float = field(default_factory=time.time) + expires_in: int = 300 # 5 minutes default + + # Mutable completion fields (protected by external lock) + authorization_code: Optional[str] = None + completed: bool = False + cancelled: bool = False + error: Optional[str] = None + + def is_expired(self) -> bool: + return time.time() > self.created_at + self.expires_in + + def get_remaining(self) -> int: + return max(0, int((self.created_at + self.expires_in) - time.time())) + + def set_authorization_code(self, auth_code: str) -> bool: + """ + Store the authorization code so the polling callback can exchange it. + + Intentionally does NOT set `completed`; that happens only after the + token exchange succeeds (inside the polling callback, under lock). + + Caller must hold the manager lock. + + Returns: + True if the code was stored, False if the session is ineligible. + """ + if self.completed or self.cancelled or self.is_expired(): + return False + self.authorization_code = auth_code + return True + + def mark_cancelled(self) -> None: + """Thread-safe cancellation marker (caller must hold lock).""" + if not self.completed and not self.cancelled: + self.cancelled = True + + +class RemoteLoginManager: + """ + Manages remote login sessions for OAuth2 authenticators. + + Features: + - Thread-safe session lifecycle management + - Polling callback with proper lock release before network I/O + - External completion via URL or raw code + - Optional HTTP callback server with per-server session token auth + - Automatic cleanup of expired sessions + """ + + # Readable alphabet excluding visually ambiguous characters (0/O/I/1) + _LOGIN_CODE_ALPHABET = ''.join( + c for c in (string.ascii_uppercase + string.digits) + if c not in frozenset('0OI1') + ) + + def __init__(self, authenticator_ref: Any): + """ + Initialize manager with reference to authenticator. + + Args: + authenticator_ref: The BaseOAuth2Authenticator instance (or subclass) + """ + self._authenticator = authenticator_ref + self._sessions: Dict[str, RemoteLoginSession] = {} + self._lock = threading.RLock() # Reentrant for nested calls + self._callback_server = None + self._callback_thread = None + self._server_token: Optional[str] = None + + # ======================================================================== + # Session Creation & Management + # ======================================================================== + + def create_session( + self, + auth_url: str, + state: str, + code_verifier: str, + login_code: Optional[str] = None, + expires_in: int = 300, + ) -> RemoteLoginSession: + """ + Create a new remote login session. + + Args: + auth_url: Full authorization URL with PKCE params + state: OAuth2 state parameter for CSRF protection + code_verifier: PKCE code verifier + login_code: Human-readable code for manual entry (auto-generated if None) + expires_in: Session timeout in seconds + + Returns: + RemoteLoginSession instance + """ + session_id = secrets.token_urlsafe(16) + code = login_code or self._generate_login_code() + + session = RemoteLoginSession( + session_id=session_id, + state=state, + code_verifier=code_verifier, + auth_url=auth_url, + login_code=code, + expires_in=expires_in, + ) + + with self._lock: + self._cleanup_expired_sessions() + self._sessions[session_id] = session + + logger.debug(f"Created remote login session {session_id[:8]}... with code {code}") + return session + + def get_session(self, session_id: str) -> Optional[RemoteLoginSession]: + """Get session by ID (thread-safe).""" + with self._lock: + session = self._sessions.get(session_id) + if session and session.is_expired(): + self._sessions.pop(session_id, None) + return None + return session + + def cancel_session(self, session_id: Optional[str] = None) -> int: + """ + Cancel one or all pending sessions. + + Args: + session_id: Specific session to cancel, or None for all + + Returns: + Number of sessions cancelled + """ + with self._lock: + if session_id: + session = self._sessions.get(session_id) + if session and not session.completed: + session.mark_cancelled() + logger.info(f"Cancelled remote login session {session_id[:8]}...") + return 1 + return 0 + else: + count = 0 + for sess in self._sessions.values(): + if not sess.completed: + sess.mark_cancelled() + count += 1 + logger.info(f"Cancelled {count} pending remote login sessions") + return count + + # ======================================================================== + # Polling Callback Factory + # ======================================================================== + + def create_polling_callback( + self, + session_id: str, + token_exchange_func: Callable[[str, str, str], Dict[str, Any]], + ) -> Callable[[], Optional[Dict[str, Any]]]: + """ + Create a thread-safe polling callback for the UI adapter. + + The returned callback: + 1. Checks session status under lock + 2. Releases lock BEFORE calling network-sensitive token_exchange_func + 3. Marks the session as completed (not merely code-received) only after + a successful token exchange + 4. Handles errors gracefully without crashing the polling thread + + Args: + session_id: The session to monitor + token_exchange_func: Function(auth_code, verifier, state) -> token_data + + Returns: + Callable returning token_data on success, None if still waiting + """ + def poll_callback() -> Optional[Dict[str, Any]]: + auth_code: Optional[str] = None + verifier: Optional[str] = None + state: Optional[str] = None + session: Optional[RemoteLoginSession] = None + + # Step 1: Read session state under lock (fast, no I/O) + with self._lock: + sess = self._sessions.get(session_id) + if not sess: + logger.debug(f"Poll: Session {session_id[:8]}... not found") + return None + + if sess.completed: + return None # Already done; shouldn't be called again + if sess.cancelled: + logger.debug(f"Poll: Session {session_id[:8]}... cancelled") + return None + if sess.is_expired(): + logger.warning(f"Poll: Session {session_id[:8]}... expired") + return None + + if not sess.authorization_code: + return None # Still waiting for the user to authorise + + # Capture all data needed for the exchange, then release lock + auth_code = sess.authorization_code + verifier = sess.code_verifier + state = sess.state + session = sess + + # Step 2: Exchange token OUTSIDE lock (network I/O — must not hold lock) + try: + logger.info(f"Exchanging authorization code for session {session_id[:8]}...") + token_data = token_exchange_func(auth_code, verifier, state) + + # Step 3: Mark as fully completed under lock + with self._lock: + # Re-check that the session wasn't cancelled/expired during exchange + live = self._sessions.get(session_id) + if live is session and not session.cancelled and not session.is_expired(): + session.completed = True + else: + logger.warning( + f"Session {session_id[:8]}... state changed during token exchange; " + "discarding result" + ) + return None + + logger.info(f"Token exchange successful for session {session_id[:8]}...") + return token_data + + except Exception as e: + logger.error(f"Token exchange failed for session {session_id[:8]}...: {e}") + with self._lock: + live = self._sessions.get(session_id) + if live is session: + session.error = str(e) + return None + + return poll_callback + + # ======================================================================== + # External Completion Handlers + # ======================================================================== + + def complete_from_callback_url(self, callback_url: str, session_id: Optional[str] = None) -> bool: + """ + Complete a session using a full OAuth2 callback URL. + + Args: + callback_url: URL containing ?code=xxx&state=yyy + session_id: Optional specific session (auto-selects if None) + + Returns: + True if completion succeeded, False otherwise + """ + try: + parsed = urlparse(callback_url) + query = parse_qs(parsed.query) + auth_code = query.get('code', [None])[0] + received_state = query.get('state', [None])[0] + + if not auth_code: + logger.warning("No authorization code in callback URL") + return False + + return self.complete_with_authorization_code(auth_code, received_state, session_id) + + except Exception as e: + logger.error(f"Failed to parse callback URL: {e}") + return False + + def complete_with_authorization_code( + self, + auth_code: str, + received_state: Optional[str] = None, + session_id: Optional[str] = None, + ) -> bool: + """ + Complete a session using raw authorization code. + + Stores the code on the session so the next poll_callback invocation + can perform the token exchange (outside the lock). + + Args: + auth_code: The authorization code from provider + received_state: Optional state param for validation + session_id: Optional specific session (auto-selects first pending if None) + + Returns: + True if completion succeeded, False otherwise + """ + with self._lock: + # Find target session + target: Optional[RemoteLoginSession] = None + target_id: Optional[str] = session_id + + if session_id: + target = self._sessions.get(session_id) + else: + # Auto-select first valid pending session + for sid, sess in self._sessions.items(): + if sess.completed or sess.cancelled or sess.is_expired(): + continue + if received_state and sess.state != received_state: + continue # State mismatch, keep looking + target = sess + target_id = sid + break + + if not target: + logger.warning("No eligible remote login session found for completion") + return False + + if target.is_expired(): + logger.warning(f"Session {target_id[:8]}... expired before completion") + return False + + if target.cancelled: + logger.debug(f"Session {target_id[:8]}... was cancelled, ignoring completion") + return False + + # Validate state if provided + if received_state and target.state != received_state: + logger.warning( + f"State mismatch for session {target_id[:8]}...: " + f"expected {target.state!r}, got {received_state!r}" + ) + return False + + # Store the code; token exchange happens on the next poll_callback invocation + stored = target.set_authorization_code(auth_code) + if stored: + logger.info(f"Session {target_id[:8]}... received authorization code; awaiting poll") + else: + logger.warning(f"Session {target_id[:8]}... could not accept authorization code") + return stored + + # ======================================================================== + # Optional: Secure HTTP Callback Server + # ======================================================================== + + def start_callback_server( + self, + port: int = 8080, + host: str = "127.0.0.1", # Default to localhost for security + session_token: Optional[str] = None, + ) -> bool: + """ + Start a minimal HTTP server to receive callbacks from mobile devices. + + SECURITY: + - Defaults to localhost only to prevent external access + - Requires session_token in URL query to prevent spoofing + - Token is per-server-instance and cryptographically random + - Handler class is created fresh per call to avoid class-level state sharing + + Phone would call: http://{host}:{port}/callback?code=xxx&token=yyy + + Args: + port: Port to bind + host: Host to bind (use "127.0.0.1" for security, "0.0.0.0" for network) + session_token: Optional shared secret for auth (auto-generated if None) + + Returns: + True if server started successfully, False otherwise + """ + if self._callback_server: + logger.warning("Callback server already running") + return False + + try: + from http.server import BaseHTTPRequestHandler, HTTPServer + import urllib.parse + + # Generate secure token if not provided + effective_token = session_token or secrets.token_urlsafe(32) + self._server_token = effective_token + + # Capture manager and token in a closure rather than as class-level + # attributes to avoid state sharing between server instances. + manager_ref = self + auth_token_ref = effective_token + + class _CallbackHandler(BaseHTTPRequestHandler): + _manager = manager_ref + _auth_token = auth_token_ref + + def log_message(self, format, *args): # noqa: A002 + logger.debug(f"Callback: {format % args}") + + def _send_response(self, code: int, html: bytes, content_type: str = "text/html; charset=utf-8"): + self.send_response(code) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(html))) + self.send_header("Cache-Control", "no-store") + self.send_header("X-Content-Type-Options", "nosniff") + self.end_headers() + self.wfile.write(html) + + def _handle_callback(self, query: Dict[str, list]): + # Validate token first + token = query.get('token', [None])[0] + if not token or token != self._auth_token: + logger.warning("Callback rejected: invalid or missing token") + self._send_response(403, b"Unauthorized") + return + + code = query.get('code', [None])[0] + session_id = query.get('session', [None])[0] + + if not code: + self._send_response(400, b"Missing 'code' parameter") + return + + success = self._manager.complete_with_authorization_code( + auth_code=code, + session_id=session_id, + ) + + if success: + self._send_response(200, ( + b"" + b"

\xe2\x9c\x93 Login Successful

" + b"

You may close this window and return to your device.

" + b"" + )) + logger.info("Remote login completed via HTTP callback") + else: + self._send_response(400, ( + b"" + b"

\xe2\x9c\x97 Login Failed

" + b"

No active session found. Please restart the login process.

" + b"" + )) + logger.warning("HTTP callback completion failed") + + def do_GET(self): + parsed = urllib.parse.urlparse(self.path) + if parsed.path == '/callback': + query = urllib.parse.parse_qs(parsed.query) + self._handle_callback(query) + else: + self._send_response(404, b"Not Found") + + def do_POST(self): + content_len = int(self.headers.get('Content-Length', 0)) + body = self.rfile.read(content_len).decode(errors='replace') + query = urllib.parse.parse_qs(body) if body else {} + if self.path == '/callback': + self._handle_callback(query) + else: + self._send_response(404, b"Not Found") + + self._callback_server = HTTPServer((host, port), _CallbackHandler) + self._callback_thread = threading.Thread( + target=self._callback_server.serve_forever, + daemon=True, + name="RemoteLoginCallbackServer", + ) + self._callback_thread.start() + + token_param = f"&token={effective_token}" if effective_token else "" + logger.info( + f"Callback server started on {host}:{port} - " + f"phones can POST to http://{host}:{port}/callback?code=xxx{token_param}" + ) + return True + + except Exception as e: + logger.error(f"Failed to start callback server: {e}") + return False + + def stop_callback_server(self) -> None: + """Gracefully stop the callback server.""" + if self._callback_server: + try: + self._callback_server.shutdown() + if self._callback_thread and self._callback_thread.is_alive(): + self._callback_thread.join(timeout=2.0) + logger.info("Callback server stopped") + except Exception as e: + logger.warning(f"Error stopping callback server: {e}") + finally: + self._callback_server = None + self._callback_thread = None + self._server_token = None + + def get_callback_server_token(self) -> Optional[str]: + """Get the current callback server token for embedding in QR URLs.""" + return self._server_token + + # ======================================================================== + # Utilities & Cleanup + # ======================================================================== + + @classmethod + def _generate_login_code(cls, length: int = 8) -> str: + """Generate human-readable login code excluding ambiguous characters.""" + return ''.join(secrets.choice(cls._LOGIN_CODE_ALPHABET) for _ in range(length)) + + def _cleanup_expired_sessions(self) -> int: + """ + Remove truly expired or completed sessions (caller must hold lock). + + Note: only sessions that are *both* completed AND expired are removed + eagerly so that a just-completed session is not cleaned up before the + caller can inspect it. Purely expired (never completed) sessions are + always removed. + + Returns: + Number of sessions removed + """ + to_remove = [ + sid for sid, sess in self._sessions.items() + if sess.is_expired() or (sess.completed and sess.is_expired()) + ] + for sid in to_remove: + self._sessions.pop(sid, None) + if to_remove: + logger.debug(f"Cleaned up {len(to_remove)} expired/completed sessions") + return len(to_remove) + + def get_active_session_count(self) -> int: + """Get count of non-expired, non-completed, non-cancelled sessions.""" + with self._lock: + return sum( + 1 for s in self._sessions.values() + if not s.completed and not s.cancelled and not s.is_expired() + ) + + def __del__(self): + """Best-effort cleanup on GC collection. Does not acquire the lock.""" + try: + self.stop_callback_server() + except Exception: + pass + # Do not acquire self._lock here: the GC may run during interpreter + # shutdown while the lock is held, causing a deadlock. + self._sessions.clear() \ No newline at end of file diff --git a/lib/streaming_providers/providers/rtlplus/auth.py b/lib/streaming_providers/providers/rtlplus/auth.py index e4d061b..07a66b9 100644 --- a/lib/streaming_providers/providers/rtlplus/auth.py +++ b/lib/streaming_providers/providers/rtlplus/auth.py @@ -51,6 +51,11 @@ class RTLPlusAuthenticator(BaseOAuth2Authenticator): http_manager=http_manager, ) + self.enable_oidc_discovery( + discovery_url=RTLPlusDefaults.AUTH_REALM_BASE, + cache_ttl=86400 # Cache for 24 hours + ) + if self.credentials is None: self.credentials = self._get_default_credentials() diff --git a/lib/streaming_providers/providers/rtlplus/constants.py b/lib/streaming_providers/providers/rtlplus/constants.py index a7c60cb..0858687 100644 --- a/lib/streaming_providers/providers/rtlplus/constants.py +++ b/lib/streaming_providers/providers/rtlplus/constants.py @@ -40,9 +40,9 @@ class RTLPlusDefaults: BEDROCK_CLIENT_ID = "bedrock-m6group_web" # API endpoints - AUTH_BASE_URL = "https://auth.rtl.de/auth/realms/rtlplus/protocol/openid-connect" + AUTH_REALM_BASE = "https://auth.rtl.de/auth/realms/rtlplus" + AUTH_BASE_URL = f"{AUTH_REALM_BASE}/protocol/openid-connect" AUTH_ENDPOINT = f"{AUTH_BASE_URL}/token" - AUTH_AUTHORIZE_ENDPOINT = f"{AUTH_BASE_URL}/auth" GRAPHQL_ENDPOINT = "https://cdn.gateway.now-plus-prod.aws-cbc.cloud/graphql" BASE_WEBSITE = "https://plus.rtl.de/" CONFIG_ENDPOINT = "https://plus.rtl.de/assets/config/config.json"