From f1628487420edb8f8ceb5bd4332e63e31a1d1a70 Mon Sep 17 00:00:00 2001 From: Nirvana Date: Fri, 10 Jul 2026 10:22:47 +0200 Subject: [PATCH] magentaeu: renew session and device id handling --- .../providers/magentaeu/auth.py | 196 ++++++++++++------ .../providers/magentaeu/constants.py | 13 +- .../providers/magentaeu/epg_manager.py | 14 +- .../providers/magentaeu/provider.py | 11 +- 4 files changed, 155 insertions(+), 79 deletions(-) diff --git a/lib/streaming_providers/providers/magentaeu/auth.py b/lib/streaming_providers/providers/magentaeu/auth.py index 0b9dd8b..01f9f1a 100644 --- a/lib/streaming_providers/providers/magentaeu/auth.py +++ b/lib/streaming_providers/providers/magentaeu/auth.py @@ -37,6 +37,7 @@ from .constants import ( DEVICE_NAME, DEVICE_OS, DEVICE_TYPE, + GUEST_SESSION_TTL_SECONDS, LOGIN_CONTEXT, LOGIN_TYPE, MANAGE_DEVICE, @@ -97,6 +98,10 @@ class MagentaAuthToken(BaseAuthToken): device_id: Optional[str] = field(default="") session_id: Optional[str] = field(default="") channel_map_id: Optional[str] = field(default="") + # Epoch time device_id/session_id were last confirmed via real cookies + # from the startup page. 0 means "never validated" -- treated as stale + # regardless of GUEST_SESSION_TTL_SECONDS. + session_id_updated_at: float = field(default=0.0) def to_dict(self) -> Dict[str, Any]: """Convert token to dictionary""" @@ -118,6 +123,8 @@ class MagentaAuthToken(BaseAuthToken): data["session_id"] = self.session_id if self.channel_map_id: data["channel_map_id"] = self.channel_map_id + if self.session_id_updated_at: + data["session_id_updated_at"] = self.session_id_updated_at return data def get_jwt_claims(self) -> Optional[Dict[str, Any]]: @@ -237,48 +244,13 @@ class MagentaAuthenticator(BaseAuthenticator): logger.info(f"=== MagentaAuthenticator.__init__ AFTER super().__init__ ===") - # CORRECT LOGIC: Use stored IDs first, then cookie-based, then random - final_device_id = None - final_session_id = None - - # 1. FIRST PRIORITY: Use stored session IDs from loaded token - if self._current_token and isinstance(self._current_token, MagentaAuthToken): - stored_device_id = self._current_token.device_id - stored_session_id = self._current_token.session_id - - if stored_device_id and stored_session_id: - final_device_id = stored_device_id - final_session_id = stored_session_id - logger.debug( - f"Using stored session IDs - device_id: {final_device_id}, session_id: {final_session_id}" - ) - - # 2. SECOND PRIORITY: Only get cookie-based IDs if no stored IDs available - if not final_device_id or not final_session_id: - # Get cookie-based IDs (second priority) - cookie_device_id, cookie_session_id = device_id, session_id - - # If no cookie IDs provided as parameters, initialize guest session - if not cookie_device_id or not cookie_session_id: - cookie_device_id, cookie_session_id = self._initialize_guest_session() - logger.debug( - f"Initialized guest session from cookies - device_id: {cookie_device_id}, session_id: {cookie_session_id}" - ) - - if cookie_device_id and cookie_session_id: - final_device_id = cookie_device_id - final_session_id = cookie_session_id - logger.debug( - f"Using cookie-based session IDs - device_id: {final_device_id}, session_id: {final_session_id}" - ) - - # 3. LAST RESORT: Generate random IDs (shouldn't happen) - if not final_device_id or not final_session_id: - final_device_id = str(uuid.uuid4()) - final_session_id = str(uuid.uuid4()) - logger.warning(f"CRITICAL: No session IDs available, using random fallback") - - # Ensure current token has the correct IDs + # device_id/session_id are no longer decided once here and then + # trusted for the token's entire lifetime -- a persisted pair could + # be months old with no way to know it ever went invalid server-side. + # We just make sure a token object exists to read/write into; actual + # validation and refresh happens lazily in get_guest_session_ids(), + # called uniformly by every guest-flow call site (channel list, EPG, + # etc), independent of whether the user is authenticated. if not self._current_token or not isinstance(self._current_token, MagentaAuthToken): self._current_token = MagentaAuthToken( access_token="", @@ -286,15 +258,20 @@ class MagentaAuthenticator(BaseAuthenticator): token_type="Bearer", expires_in=0, issued_at=time.time(), - device_id=final_device_id, - session_id=final_session_id, ) - else: - self._current_token.device_id = final_device_id - self._current_token.session_id = final_session_id + + # Explicit device_id/session_id constructor args (if the caller + # already knows good values) still take precedence, but they don't + # get treated as pre-validated -- session_id_updated_at is left at + # 0 so the first get_guest_session_ids() call still checks them. + if device_id: + self._current_token.device_id = device_id + if session_id: + self._current_token.session_id = session_id logger.info( - f"Final session IDs - device_id: {final_device_id}, session_id: {final_session_id}" + f"=== MagentaAuthenticator.__init__ COMPLETE - device_id/session_id " + f"will be validated on first guest request ===" ) @property @@ -323,10 +300,25 @@ class MagentaAuthenticator(BaseAuthenticator): """Public access to HTTP manager""" return self._http_manager - def _initialize_guest_session(self) -> tuple[str, str]: - """Initialize guest session and get device_id/session_id from cookies""" + def _initialize_guest_session(self) -> tuple[str, str, bool]: + """ + Visit the provider's startup page to obtain a real deviceId/sessionId + pair from Set-Cookie, the same way a browser would. + + Uses API_ENDPOINTS["STARTUP_PAGE"] rather than a hardcoded "/epg" + path, since not every MagentaEU natco is guaranteed to serve the + startup page at that path -- a country override only needs to change + constants.py, not this method. + + Returns (device_id, session_id, obtained) where obtained=False means + we had to fall back to random UUIDs. Callers must NOT treat an + obtained=False result as a validated, cacheable session -- that's + exactly how a permanently-invalid session got persisted before. + """ try: - startup_url = f"{get_base_url(self.country)}/epg" + startup_url = API_ENDPOINTS["STARTUP_PAGE"].format( + base_url=get_base_url(self.country) + ) headers = get_base_headers() response = self._http_manager.get( @@ -336,7 +328,6 @@ class MagentaAuthenticator(BaseAuthenticator): timeout=DEFAULT_REQUEST_TIMEOUT, ) - # Extract cookies from response device_id = "" session_id = "" @@ -345,20 +336,86 @@ class MagentaAuthenticator(BaseAuthenticator): device_id = cookies.get("deviceId", "") session_id = cookies.get("sessionId", "") - # If no cookies found, generate UUIDs - if not device_id or not session_id: - device_id = str(uuid.uuid4()) - session_id = str(uuid.uuid4()) + if device_id and session_id: + logger.debug( + f"[{self.country}] Guest session established from cookies - " + f"device_id: {device_id}, session_id: {session_id}" + ) + return device_id, session_id, True - logger.debug( - f"Initialized guest session from cookies - device_id: {device_id}, session_id: {session_id}" + logger.warning( + f"[{self.country}] Startup page returned no deviceId/sessionId " + f"cookies; using unverified random fallback" ) - return device_id, session_id + return str(uuid.uuid4()), str(uuid.uuid4()), False except Exception as e: - logger.warning(f"Session initialization failed: {e}") - # Generate fallback IDs - return str(uuid.uuid4()), str(uuid.uuid4()) + logger.warning(f"[{self.country}] Guest session initialization failed: {e}") + return str(uuid.uuid4()), str(uuid.uuid4()), False + + def get_guest_session_ids(self, force_refresh: bool = False) -> tuple[str, str]: + """ + Return (device_id, session_id) for guest-flow requests (channel + list, EPG, etc). This is the single source of truth for every guest + call site -- previously each call site (get_channels(), the EPG + manager) read current_token.device_id/session_id directly, which + were set once at first-ever init and never re-validated, so a pair + that went stale server-side stayed stale forever. + + Values are re-validated whenever older than GUEST_SESSION_TTL_SECONDS + (or never validated at all -- session_id_updated_at == 0), regardless + of whether the user is authenticated; the bearer-token lifecycle + (_refresh_token) is separate from this guest session. + """ + token = self._current_token + device_id = "" + session_id = "" + updated_at = 0.0 + if isinstance(token, MagentaAuthToken): + device_id = token.device_id or "" + session_id = token.session_id or "" + updated_at = token.session_id_updated_at or 0.0 + + have_ids = bool(device_id and session_id) + age = time.time() - updated_at if have_ids else float("inf") + is_stale = age > GUEST_SESSION_TTL_SECONDS + + if have_ids and not is_stale and not force_refresh: + return device_id, session_id + + logger.info( + f"[{self.country}] Guest session " + f"{'forced refresh' if force_refresh else ('stale (age=%.0fs)' % age if have_ids else 'not yet established')}" + f" -- fetching fresh device_id/session_id" + ) + device_id, session_id, obtained = self._initialize_guest_session() + + if not self._current_token or not isinstance(self._current_token, MagentaAuthToken): + self._current_token = MagentaAuthToken( + access_token="", + refresh_token="", + token_type="Bearer", + expires_in=0, + issued_at=time.time(), + ) + + self._current_token.device_id = device_id + self._current_token.session_id = session_id + + if obtained: + self._current_token.session_id_updated_at = time.time() + self._save_session() + else: + # Don't stamp session_id_updated_at -- an unverified fallback + # must not be cached as if it were a real, validated session. + # The next call will retry instead of trusting a guess for + # GUEST_SESSION_TTL_SECONDS. + logger.warning( + f"[{self.country}] Guest session unverified; will retry on " + f"next call rather than caching this pair" + ) + + return device_id, session_id def _get_auth_headers(self) -> Dict[str, str]: """Get headers for authentication request - required by BaseAuthenticator""" @@ -432,6 +489,7 @@ class MagentaAuthenticator(BaseAuthenticator): device_id = "" session_id = "" channel_map_id = "" + session_id_updated_at = 0.0 # Try to get session IDs from multiple sources in priority order: @@ -440,6 +498,8 @@ class MagentaAuthenticator(BaseAuthenticator): device_id = response_data.get("device_id", "") if "session_id" in response_data: session_id = response_data.get("session_id", "") + if "session_id_updated_at" in response_data: + session_id_updated_at = response_data.get("session_id_updated_at", 0.0) or 0.0 # 2. Then from current token (for new authentications) if ( @@ -450,6 +510,7 @@ class MagentaAuthenticator(BaseAuthenticator): device_id = self._current_token.device_id or "" session_id = self._current_token.session_id or "" channel_map_id = self._current_token.channel_map_id or "" + session_id_updated_at = self._current_token.session_id_updated_at or 0.0 # 3. If we found session IDs, log it if device_id and session_id: @@ -490,6 +551,7 @@ class MagentaAuthenticator(BaseAuthenticator): device_id=device_id, session_id=session_id, channel_map_id=channel_map_id, + session_id_updated_at=session_id_updated_at, ) # Classify token @@ -577,15 +639,16 @@ class MagentaAuthenticator(BaseAuthenticator): response.raise_for_status() return response.json() - def _get_session_data(self) -> tuple[str, str, str]: + def _get_session_data(self) -> tuple[str, str, str, float]: """Safely get session data from current token""" if isinstance(self._current_token, MagentaAuthToken): return ( self._current_token.device_id or "", self._current_token.session_id or "", self._current_token.channel_map_id or "", + self._current_token.session_id_updated_at or 0.0, ) - return "", "", "" + return "", "", "", 0.0 def _refresh_token(self) -> Optional[BaseAuthToken]: """Refresh Magenta TV token - override base method""" @@ -598,7 +661,7 @@ class MagentaAuthenticator(BaseAuthenticator): refresh_url = API_ENDPOINTS["REFRESH_TOKEN"].format(natco=self.country) - device_id, session_id, channel_map_id = self._get_session_data() + device_id, session_id, channel_map_id, session_id_updated_at = self._get_session_data() # Build headers according to your working example headers = self._config.get_auth_headers( @@ -649,6 +712,7 @@ class MagentaAuthenticator(BaseAuthenticator): device_id=device_id, session_id=session_id, channel_map_id=channel_map_id, + session_id_updated_at=session_id_updated_at, ) # Classify token @@ -748,4 +812,4 @@ class MagentaAuthenticator(BaseAuthenticator): # Save the updated token with channel_map_id self._save_session() - return account_data + return account_data \ No newline at end of file diff --git a/lib/streaming_providers/providers/magentaeu/constants.py b/lib/streaming_providers/providers/magentaeu/constants.py index 0ca5f45..8c5a8cc 100644 --- a/lib/streaming_providers/providers/magentaeu/constants.py +++ b/lib/streaming_providers/providers/magentaeu/constants.py @@ -160,6 +160,17 @@ DRM_REQUEST_HEADERS = { DEFAULT_REQUEST_TIMEOUT = 30 DEFAULT_MAX_RETRIES = 3 +# ============================================================================ +# Guest Session Configuration +# ============================================================================ + +# How long a device_id/session_id pair (obtained from the startup-page +# cookies) is trusted before it gets re-validated against the server again. +# The real server-side TTL is unknown/unmeasured -- this is a conservative +# starting point, not an observed value. Tighten or loosen once we have +# data on how long a pair actually survives server-side. +GUEST_SESSION_TTL_SECONDS = 6 * 60 * 60 # 6 hours + # ============================================================================ # Headers Configuration # ============================================================================ @@ -268,4 +279,4 @@ def get_guest_headers(country: str, device_id: str, session_id: str) -> Dict[str "x-request-tracking-id": str(uuid.uuid4()), "x-user-agent": X_USER_AGENT, } - return headers + return headers \ No newline at end of file diff --git a/lib/streaming_providers/providers/magentaeu/epg_manager.py b/lib/streaming_providers/providers/magentaeu/epg_manager.py index 4191fdb..1124c04 100644 --- a/lib/streaming_providers/providers/magentaeu/epg_manager.py +++ b/lib/streaming_providers/providers/magentaeu/epg_manager.py @@ -368,7 +368,17 @@ class MagentaEUEpgManager: # ------------------------------------------------------------------ def _current_ids(self) -> Tuple[str, str]: - """Return (device_id, session_id) from the authenticator token.""" + """Return (device_id, session_id) for guest EPG requests. + + Sourced from the authenticator's guest session (get_guest_session_ids), + which is validated and refreshed on its own TTL -- not read directly + off current_token, since those values could otherwise go stale + without ever being re-checked for the lifetime of the process. + """ + if hasattr(self._auth, "get_guest_session_ids"): + return self._auth.get_guest_session_ids() + + # Fallback for authenticators that don't implement guest sessions token = getattr(self._auth, "current_token", None) device_id = getattr(token, "device_id", "") or "" session_id = getattr(token, "session_id", "") or "" @@ -399,7 +409,7 @@ class MagentaEUEpgManager: conversion belongs here; do not reintroduce it. """ merged: Dict[str, Any] = {} - headers = self._guest_headers(flow="START_UP", step="EPG_SCHEDULES") + headers = self._guest_headers(flow="EPG", step="EPG_SCHEDULES") # ✅ LOG THE FULL HEADERS (or at least the critical ones) logger.info( diff --git a/lib/streaming_providers/providers/magentaeu/provider.py b/lib/streaming_providers/providers/magentaeu/provider.py index 0b200b7..aaf6f36 100644 --- a/lib/streaming_providers/providers/magentaeu/provider.py +++ b/lib/streaming_providers/providers/magentaeu/provider.py @@ -165,16 +165,7 @@ class MagentaEUProvider(StreamingProvider): def get_channels(self, **kwargs) -> List[StreamingChannel]: """Fetch available channels from Magenta TV - no authentication required""" try: - device_id = ( - self.authenticator.current_token.device_id - if self.authenticator.current_token - else "" - ) - session_id = ( - self.authenticator.current_token.session_id - if self.authenticator.current_token - else "" - ) + device_id, session_id = self.authenticator.get_guest_session_ids() channels_url = API_ENDPOINTS["EPG_CHANNELS"].format( bifrost_url=get_bifrost_url(self.country)