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

magentaeu: renew session and device id handling

Browse Source
This commit is contained in:
Nirvana
2026-07-10 10:22:47 +02:00
parent 2fe519db01
commit f162848742
4 changed files with 155 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/providers/magentaeu/auth.py
+130 -66
View File
@@ -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
lib/streaming_providers/providers/magentaeu/constants.py
+12 -1
View File
@@ -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
lib/streaming_providers/providers/magentaeu/epg_manager.py
+12 -2
View File
@@ -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(
lib/streaming_providers/providers/magentaeu/provider.py
+1 -10
View File
@@ -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)