From 6e364a182939d83fde47068e1b70aa4cea38d473 Mon Sep 17 00:00:00 2001 From: Nirvana Date: Mon, 20 Jul 2026 19:41:42 +0200 Subject: [PATCH] magentaeu: add transaction id to auth.py --- .../providers/magentaeu/auth.py | 58 ++++--- .../providers/magentaeu/constants.py | 142 +++++++++++++++++- .../providers/magentaeu/epg_manager.py | 2 +- .../providers/magentaeu/provider.py | 2 +- .../providers/magentaeu/utils.py | 47 ++---- 5 files changed, 191 insertions(+), 60 deletions(-) diff --git a/lib/streaming_providers/providers/magentaeu/auth.py b/lib/streaming_providers/providers/magentaeu/auth.py index 01f9f1a..697c7b9 100644 --- a/lib/streaming_providers/providers/magentaeu/auth.py +++ b/lib/streaming_providers/providers/magentaeu/auth.py @@ -44,6 +44,7 @@ from .constants import ( SUPPORTED_COUNTRIES, USER_AGENT, X_USER_AGENT, + build_auth_headers, get_base_headers, get_base_url, get_bifrost_url, @@ -160,27 +161,20 @@ class MagentaAuthConfig: step: str = AUTH_STEPS["GET_ACCESS_TOKEN"], device_id: str = None, session_id: str = None, + tracking_id: str = None, + call_time: str = None, ) -> Dict[str, str]: - """Get authentication headers""" - headers = get_base_headers() - headers.update( - { - "X-User-Agent": self.x_user_agent, - "X-Call-Type": call_type, - "X-Tv-Flow": flow, - "X-Tv-Step": step, - "x-request-session-id": session_id or str(uuid.uuid4()), - "x-request-tracking-id": str(uuid.uuid4()), - "requestid": str(uuid.uuid4()), - "Tenant": "tv", - "Origin": self.country_config["base_url"], - "App_key": self.country_config["app_key"], - "App_version": self.app_version, - "Device-Id": device_id or str(uuid.uuid4()), - "Device-Name": self.device_name, - } + """Get authentication headers, including x-txn-id""" + return build_auth_headers( + country=self.country, + device_id=device_id, + session_id=session_id, + flow=flow, + step=step, + call_type=call_type, + tracking_id=tracking_id, + call_time=call_time, ) - return headers def encrypt_password(self, password: str) -> str: """Encrypt password using RSA public key""" @@ -427,12 +421,17 @@ class MagentaAuthenticator(BaseAuthenticator): device_id = self._current_token.device_id or "" session_id = self._current_token.session_id or "" + tracking_id = str(uuid.uuid4()) + call_time = str(int(time.time() * 1000)) + return self._config.get_auth_headers( call_type=CALL_TYPES["GUEST_USER"], flow=AUTH_FLOWS["USERNAME_PASSWORD_LOGIN"], step=AUTH_STEPS["GET_ACCESS_TOKEN"], device_id=device_id, session_id=session_id, + tracking_id=tracking_id, + call_time=call_time, ) def _build_auth_payload(self) -> Dict[str, Any]: @@ -619,10 +618,19 @@ class MagentaAuthenticator(BaseAuthenticator): """Upgrade token when device limit is exceeded""" upgrade_url = API_ENDPOINTS["UPGRADE_TOKEN"].format(natco=self.country) + device_id, session_id, _, _ = self._get_session_data() + + tracking_id = str(uuid.uuid4()) + call_time = str(int(time.time() * 1000)) + headers = self._config.get_auth_headers( call_type=CALL_TYPES["GUEST_USER"], flow=AUTH_FLOWS["USERNAME_PASSWORD_LOGIN"], step=AUTH_STEPS["UPGRADE_TOKEN"], + device_id=device_id, + session_id=session_id, + tracking_id=tracking_id, + call_time=call_time, ) headers["Refresh_token"] = refresh_token @@ -663,11 +671,18 @@ class MagentaAuthenticator(BaseAuthenticator): device_id, session_id, channel_map_id, session_id_updated_at = self._get_session_data() + tracking_id = str(uuid.uuid4()) + call_time = str(int(time.time() * 1000)) + # Build headers according to your working example headers = self._config.get_auth_headers( call_type=CALL_TYPES["AUTH_USER"], flow=AUTH_FLOWS["START_UP"], step=AUTH_STEPS["REFRESH_TOKEN"], + device_id=device_id, + session_id=session_id, + tracking_id=tracking_id, + call_time=call_time, ) # Add the specific headers from your working example @@ -782,12 +797,17 @@ class MagentaAuthenticator(BaseAuthenticator): device_id = self._current_token.device_id or "" session_id = self._current_token.session_id or "" + tracking_id = str(uuid.uuid4()) + call_time = str(int(time.time() * 1000)) + headers = self._config.get_auth_headers( call_type=CALL_TYPES["AUTH_USER"], flow=AUTH_FLOWS["START_UP"], step=AUTH_STEPS["GET_USER_ACCOUNT"], device_id=device_id, session_id=session_id, + tracking_id=tracking_id, + call_time=call_time, ) headers["Bff_token"] = access_token diff --git a/lib/streaming_providers/providers/magentaeu/constants.py b/lib/streaming_providers/providers/magentaeu/constants.py index 1889339..cf205da 100644 --- a/lib/streaming_providers/providers/magentaeu/constants.py +++ b/lib/streaming_providers/providers/magentaeu/constants.py @@ -3,7 +3,10 @@ # Magenta TV Configuration # ============================================================================ -from typing import Dict +import hashlib +import time +import uuid +from typing import Dict, Optional # Supported countries SUPPORTED_COUNTRIES = ["hr", "pl", "me", "at", "hu"] @@ -279,4 +282,139 @@ 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 \ No newline at end of file + return headers + + +# ============================================================================ +# Header Generation with x-txn-id +# ============================================================================ + + +def _generate_txn_id( + tracking_id: str, session_id: str, device_id: str, call_time: str +) -> str: + """SHA-256(trackingId + sessionId + deviceId + callTime)[:32]""" + raw = tracking_id + session_id + device_id + call_time + return hashlib.sha256(raw.encode()).hexdigest()[:32] + + +def build_guest_headers( + country: str, + device_id: str, + session_id: str, + flow: str, + step: Optional[str] = None, + tracking_id: Optional[str] = None, +) -> Dict[str, str]: + """ + Build complete guest headers with x-txn-id. + + Note: device_id and session_id must already be resolved by the caller + (via get_guest_session_ids() or similar) before calling this. No + fallback UUIDs are generated here -- that's intentional, to avoid a + mismatch between the header value actually sent and the value hashed + into x-txn-id. + + (Relocated verbatim from utils.py; utils.py now re-exports this.) + """ + if tracking_id is None: + tracking_id = str(uuid.uuid4()) + + # Snapshot call_time once -- x-txn-id is derived from it, so both + # headers must use the same value. + call_time = str(int(time.time() * 1000)) + + headers = get_guest_headers(country, device_id, session_id) + headers["x-call-time"] = call_time + headers["x-tv-flow"] = flow + headers["x-request-tracking-id"] = tracking_id + headers["x-txn-id"] = _generate_txn_id(tracking_id, session_id, device_id, call_time) + if step is not None: + headers["x-tv-step"] = step + return headers + + +def build_headers_with_txn( + base_headers: Dict[str, str], + device_id: str, + session_id: str, + tracking_id: Optional[str] = None, + call_time: Optional[str] = None, +) -> Dict[str, str]: + """ + Add x-txn-id and related headers to any headers dict. + + Args: + base_headers: Existing headers to add to + device_id: Device ID for x-txn-id calculation -- must be the same + value already present as the Device-Id header in base_headers + session_id: Session ID for x-txn-id calculation -- must be the same + value already present as the x-request-session-id header + tracking_id: Optional tracking ID (generated if not provided) + call_time: Optional call time (generated if not provided) + + Returns: + Headers dict with x-txn-id, x-call-time, x-request-tracking-id added + """ + if tracking_id is None: + tracking_id = str(uuid.uuid4()) + + if call_time is None: + call_time = str(int(time.time() * 1000)) + + headers = base_headers.copy() + headers.update({ + "x-request-tracking-id": tracking_id, + "x-call-time": call_time, + "x-txn-id": _generate_txn_id(tracking_id, session_id, device_id, call_time), + }) + return headers + + +def build_auth_headers( + country: str, + device_id: str, + session_id: str, + flow: str, + step: str, + call_type: str = CALL_TYPES["GUEST_USER"], + tracking_id: Optional[str] = None, + call_time: Optional[str] = None, +) -> Dict[str, str]: + """ + Build complete authentication headers with x-txn-id. + + CRITICAL: device_id and session_id fallbacks are resolved ONCE here, + then reused for both the Device-Id / x-request-session-id headers and + the x-txn-id hash. Resolving `device_id or str(uuid.uuid4())` more than + once produces a different random value each time, which would send a + Device-Id header that doesn't match what's hashed into x-txn-id. + """ + resolved_device_id = device_id if device_id else str(uuid.uuid4()) + resolved_session_id = session_id if session_id else str(uuid.uuid4()) + + headers = { + "User-Agent": USER_AGENT, + "Accept": "application/json", + "Content-Type": "application/json", + "X-User-Agent": X_USER_AGENT, + "X-Call-Type": call_type, + "X-Tv-Flow": flow, + "X-Tv-Step": step, + "x-request-session-id": resolved_session_id, + "requestid": str(uuid.uuid4()), + "Tenant": "tv", + "Origin": get_base_url(country), + "App_key": get_app_key(country), + "App_version": APP_VERSION, + "Device-Id": resolved_device_id, + "Device-Name": DEVICE_NAME, + } + + return build_headers_with_txn( + base_headers=headers, + device_id=resolved_device_id, + session_id=resolved_session_id, + tracking_id=tracking_id, + call_time=call_time, + ) \ 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 2373ea6..a0f0348 100644 --- a/lib/streaming_providers/providers/magentaeu/epg_manager.py +++ b/lib/streaming_providers/providers/magentaeu/epg_manager.py @@ -32,12 +32,12 @@ from datetime import datetime, timedelta, timezone from zoneinfo import ZoneInfo from typing import Any, Dict, List, Optional, Set, Tuple -from .utils import build_guest_headers from ...base.utils.logger import logger from ...base.models.epg_models import EPGEntry, EPGProgramDetails, EPGFlags from .constants import ( DEFAULT_REQUEST_TIMEOUT, SUPPORTED_COUNTRIES, + build_guest_headers, get_app_key, get_bifrost_url, get_language, diff --git a/lib/streaming_providers/providers/magentaeu/provider.py b/lib/streaming_providers/providers/magentaeu/provider.py index aaf6f36..47d54d1 100644 --- a/lib/streaming_providers/providers/magentaeu/provider.py +++ b/lib/streaming_providers/providers/magentaeu/provider.py @@ -171,7 +171,7 @@ class MagentaEUProvider(StreamingProvider): bifrost_url=get_bifrost_url(self.country) ) - from .utils import build_guest_headers + from .constants import build_guest_headers headers = build_guest_headers( self.country, device_id, session_id, flow="START_UP" diff --git a/lib/streaming_providers/providers/magentaeu/utils.py b/lib/streaming_providers/providers/magentaeu/utils.py index 4cd2748..c96aae9 100644 --- a/lib/streaming_providers/providers/magentaeu/utils.py +++ b/lib/streaming_providers/providers/magentaeu/utils.py @@ -1,40 +1,13 @@ # streaming_providers/providers/magentaeu/utils.py -import hashlib -import time -import uuid -from typing import Dict, Optional -from .constants import get_guest_headers +from .constants import build_guest_headers, _generate_txn_id - -def _generate_txn_id( - tracking_id: str, session_id: str, device_id: str, call_time: str -) -> str: - """SHA-256(trackingId + sessionId + deviceId + callTime)[:32]""" - raw = tracking_id + session_id + device_id + call_time - return hashlib.sha256(raw.encode()).hexdigest()[:32] - - -def build_guest_headers( - country: str, - device_id: str, - session_id: str, - flow: str, - step: Optional[str] = None, - tracking_id: Optional[str] = None, -) -> Dict[str, str]: - if tracking_id is None: - tracking_id = str(uuid.uuid4()) - - # Snapshot call_time once — x-txn-id is derived from it, so both - # headers must use the same value. - call_time = str(int(time.time() * 1000)) - - headers = get_guest_headers(country, device_id, session_id) - headers["x-call-time"] = call_time - headers["x-tv-flow"] = flow - headers["x-request-tracking-id"] = tracking_id - headers["x-txn-id"] = _generate_txn_id(tracking_id, session_id, device_id, call_time) - if step is not None: - headers["x-tv-step"] = step - return headers \ No newline at end of file +# Re-export for backward compatibility. These are the only two names that +# were previously defined here (verified against the prior utils.py before +# this change) -- both now live in constants.py alongside the rest of the +# header-generation logic (build_auth_headers, build_headers_with_txn), +# giving a single source of truth for all request headers. +__all__ = [ + "build_guest_headers", + "_generate_txn_id", +] \ No newline at end of file