magentaeu: add transaction id to auth.py

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
 

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
+    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,
+    )

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,

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"

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
+# 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",
+]