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

magenta2: Improve bootstrap

Browse Source
This commit is contained in:
Nirvana
2026-09-29 12:08:46 +02:00
parent 761e03fee8
commit 2c1a732344
12 changed files with 614 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/providers/lib_theplatform.py
+12
View File
@@ -456,6 +456,8 @@ def build_widevine_drm_config(
licence_url: str,
user_agent: str,
origin: Optional[str] = None,
session_id: Optional[str] = None,
call_id: Optional[str] = None,
) -> DRMConfig:
"""
Build a Widevine DRMConfig for theplatform licence acquisition.
@@ -465,6 +467,14 @@ def build_widevine_drm_config(
user_agent: Platform user-agent string for the licence request.
origin: Optional Origin header value (used by magentaeu to set
the country base URL; omit for magenta2).
session_id: Optional — if provided together with call_id, adds a
CID header formatted as "{session_id}::{call_id}",
matching the SMIL DRM path and confirmed against a
real ATV widevine capture (which shows CID but no
separate session-id header). session_id alone (without
call_id) adds nothing — additive; callers that don't
pass either keep prior behaviour unchanged.
call_id: Optional — see session_id above.
Returns:
DRMConfig ready for use by the base streaming provider.
@@ -476,6 +486,8 @@ def build_widevine_drm_config(
if origin:
headers["Origin"] = origin
headers["Referer"] = f"{origin}/"
if session_id and call_id:
headers["CID"] = f"{session_id}::{call_id}"
return DRMConfig(
system=DRMSystem.WIDEVINE,
lib/streaming_providers/providers/magenta2/auth.py
+65 -24
View File
@@ -23,15 +23,15 @@ from ...base.auth.base_oauth2_auth import OIDCConfiguration
from ...base.auth.credentials import ClientCredentials
from ...base.utils.logger import logger
from .constants import (
APPVERSION2,
DEFAULT_COUNTRY,
DEFAULT_PLATFORM,
IDM,
MAGENTA2_CLIENT_IDS,
MAGENTA2_LEGACY_CLIENT_IDS,
MAGENTA2_PLATFORMS,
SSO_USER_AGENT,
SUPPORTED_COUNTRIES,
TAA_REQUEST_TEMPLATE,
render_user_agent,
)
# Import Magenta2-specific components
@@ -58,10 +58,19 @@ class Magenta2Credentials(ClientCredentials):
if not hasattr(self, "client_secret") or self.client_secret is None:
self.client_secret = "" # Empty string for public client
# Set client_id from constant if not provided
# client_id is no longer defaulted here. The caller (provider.py)
# supplies it explicitly — initially a placeholder from
# MAGENTA2_LEGACY_CLIENT_IDS, then the real, server-provided
# sam3ClientId once bootstrap discovery completes (see
# Magenta2Authenticator._configure_authenticator_from_discovery /
# provider.py's post-discovery update). Not defaulting here prevents
# silently re-introducing a stale/legacy client_id if a caller
# forgets to pass one.
if not self.client_id:
self.client_id = MAGENTA2_CLIENT_IDS.get(
self.platform, MAGENTA2_CLIENT_IDS[DEFAULT_PLATFORM]
raise ValueError(
"Magenta2Credentials requires an explicit client_id "
"(pass a MAGENTA2_LEGACY_CLIENT_IDS placeholder pre-discovery, "
"or the bootstrap-provided sam3ClientId post-discovery)."
)
# Generate device ID if not provided
@@ -87,21 +96,48 @@ class Magenta2Credentials(ClientCredentials):
self.platform, MAGENTA2_PLATFORMS[DEFAULT_PLATFORM]
)
# Use provided models or fallback to platform defaults
# Single source of truth for the version: MAGENTA2_PLATFORMS[platform]
# ["version"] — the same value used to render the User-Agent. A real
# TAA capture confirms appVersion in both the JSON body and the
# keyValue string is identical to the version embedded in the UA
# (e.g. "3.134.4462" in both places) — there is no separate,
# independently-tracked TAA version.
version = platform_config["version"]
# device_name is unchanged as a key (still exists in
# MAGENTA2_PLATFORMS), only its value changed — from "Android TV" to
# the more specific "SHIELD Android TV" for the ATV platforms.
resolved_device_model = device_model or platform_config["device_name"]
resolved_client_model = client_model or f"ftv-{self.platform}"
# Do NOT synthesize a client_model when the caller doesn't pass one.
# A real TAA onboarding-login capture shows neither a
# "ClientModelParams(...)" segment in keyValue nor a "client" key in
# the JSON body when client_model is omitted — a previous version of
# this method always synthesized f"ftv-{platform}" here, which meant
# ClientModelParams/"client" were ALWAYS sent even when the real
# client never sends them for this call. Only include them when the
# caller explicitly supplies a client_model.
resolved_client_model = client_model
# "os" is "API level {N}", not "Android {N}" — confirmed from a real
# TAA capture ("os":"API level 30" for android-tv/atv-launcher,
# api_level="30"). A previous version of this method reconstructed
# "Android 11" from android_version, which was an unconfirmed guess
# and is now known to be wrong.
api_level = platform_config.get("api_level")
os_string = f"API level {api_level}" if api_level else resolved_device_model
# Build keyValue string with client model if available
key_value_parts = [IDM, APPVERSION2]
key_value_parts = [IDM, version]
# Add client model if available
# Add client model only if explicitly provided (see note above)
if resolved_client_model:
key_value_parts.append(f"ClientModelParams(id={resolved_client_model})")
key_value_parts.extend(
[
f"TokenChannelParams(id=Tv)",
f"TokenDeviceParams(id={self.device_id}, model={resolved_device_model}, os={platform_config['firmware']})",
f"TokenDeviceParams(id={self.device_id}, model={resolved_device_model}, os={os_string})",
"DE",
"telekom",
]
@@ -109,21 +145,24 @@ class Magenta2Credentials(ClientCredentials):
key_value = "/".join(key_value_parts)
# Start with template and populate fields
# Start with template and populate fields — appVersion is set here
# from the single version source, not baked into the template (the
# template can't hold a platform-specific value).
payload = TAA_REQUEST_TEMPLATE.copy()
payload.update(
{
"keyValue": key_value,
"accessToken": access_token,
"appVersion": version,
"device": {
"id": self.device_id,
"model": resolved_device_model,
"os": platform_config["firmware"],
"os": os_string,
},
}
)
# Add client model if available
# Add client model only if explicitly provided (see note above)
if resolved_client_model:
payload["client"] = {"model": resolved_client_model}
@@ -241,7 +280,7 @@ class Magenta2AuthConfig:
self.platform_config = MAGENTA2_PLATFORMS.get(
platform, MAGENTA2_PLATFORMS[DEFAULT_PLATFORM]
)
self.user_agent = self.platform_config["user_agent"]
self.user_agent = render_user_agent(self.platform, subscriber_suffix=False)
self.timeout = 30
self.endpoints = endpoints or {}
self.client_model = client_model
@@ -382,9 +421,13 @@ class Magenta2Authenticator(BaseAuthenticator):
self._device_model,
)
# Extract and cache client_id
self._client_id = self._sam3_client_id or MAGENTA2_CLIENT_IDS.get(
self.platform, MAGENTA2_CLIENT_IDS[DEFAULT_PLATFORM]
# Extract and cache client_id. Pre-discovery placeholder only —
# _configure_authenticator_from_discovery() (called by provider.py
# right after discover_provider_config() succeeds) overwrites this
# with the real, server-provided sam3ClientId before any actual
# token request is made.
self._client_id = self._sam3_client_id or MAGENTA2_LEGACY_CLIENT_IDS.get(
self.platform, MAGENTA2_LEGACY_CLIENT_IDS[DEFAULT_PLATFORM]
)
# Initialize credentials if not provided
@@ -581,13 +624,11 @@ class Magenta2Authenticator(BaseAuthenticator):
logger.debug(f"Token classified as USER_AUTHENTICATED (found {key} in JWT)")
return TokenAuthLevel.USER_AUTHENTICATED
# Check for client credentials patterns
client_id = claims.get("client_id", claims.get("clientId", ""))
if client_id in MAGENTA2_CLIENT_IDS.values():
logger.debug("Token classified as CLIENT_CREDENTIALS (known client ID)")
return TokenAuthLevel.CLIENT_CREDENTIALS
# Default to client credentials for TAA flow
# Everything else falls through to client-credentials classification
# for the TAA flow. (A prior version checked client_id against the
# legacy MAGENTA2_CLIENT_IDS table here, but both branches of that
# check returned CLIENT_CREDENTIALS regardless — it was a no-op,
# so it's been removed along with the dependency on that table.)
logger.debug("Token classified as CLIENT_CREDENTIALS (default for TAA)")
return TokenAuthLevel.CLIENT_CREDENTIALS
lib/streaming_providers/providers/magenta2/auth_bridge.py
+7 -3
View File
@@ -57,6 +57,8 @@ class AuthBridge:
provider_config: Any,
session_id: str,
serial_number: str,
user_agent_plain: str,
user_agent_subscriber: str,
generate_call_id, # callable() -> str
) -> None:
self._authenticator = authenticator
@@ -67,6 +69,8 @@ class AuthBridge:
self._provider_config = provider_config
self._session_id = session_id
self._serial_number = serial_number
self._ua_plain = user_agent_plain
self._ua_subscriber = user_agent_subscriber
self._generate_call_id = generate_call_id
self._persona_cache: Optional[PersonaResult] = None
@@ -187,7 +191,7 @@ class AuthBridge:
"x-dt-call-id": self._generate_call_id(),
"origin": "https://www.magenta.tv",
"referer": "https://www.magenta.tv/",
"user-agent": self._platform_config["user_agent"],
"user-agent": self._ua_subscriber,
"accept": "*/*",
"accept-encoding": "gzip, deflate, br, zstd",
"accept-language": "de-DE,de;q=0.9",
@@ -200,7 +204,7 @@ class AuthBridge:
"x-stbserialnumber": self._serial_number,
"dt-session-id": self._session_id,
"dt-call-id": self._generate_call_id(),
"user-agent": self._platform_config["user_agent"],
"user-agent": self._ua_subscriber,
"accept-encoding": "gzip",
}
@@ -209,7 +213,7 @@ class AuthBridge:
persona_token = self.ensure_authenticated()
return {
"Authorization": f"Basic {persona_token}",
"User-Agent": self._platform_config["user_agent"],
"User-Agent": self._ua_subscriber,
"Accept-Encoding": "gzip",
"CID": f"{self._session_id}::{self._generate_call_id()}",
}
lib/streaming_providers/providers/magenta2/channel_manager.py
+58 -8
View File
@@ -21,7 +21,6 @@ from .constants import (
DRM_SYSTEM_WIDEVINE,
ERROR_CODES,
MODE_LIVE,
QUALITY_RANK,
)
from .endpoint_manager import EndpointManager
from .config_models import ProviderConfig
@@ -58,6 +57,8 @@ class ChannelManager:
provider_config: Optional[ProviderConfig],
auth_callback: Callable[[], str],
build_scaled_image_url_callback: Callable[[str], Optional[str]],
user_agent_plain: str,
user_agent_subscriber: str,
catchup_window: int = 4,
):
self._http = http_manager
@@ -70,8 +71,19 @@ class ChannelManager:
self._provider_config = provider_config
self._ensure_authenticated = auth_callback
self._build_scaled_image_url = build_scaled_image_url_callback
self._ua_plain = user_agent_plain
self._ua_subscriber = user_agent_subscriber
self.catchup_window = catchup_window
# Special-rights rules (e.g. "uhd" -> liveTvOption False) from the
# manifest, used by _is_station_playable() to filter stations a
# non-managed device/account isn't allowed to play live.
self._special_rights = (
provider_config.manifest.special_rights_profiles
if provider_config and provider_config.manifest
else {}
)
# Populated on first get_channels() call
self._cached_channels: Optional[List[StreamingChannel]] = None
@@ -106,7 +118,7 @@ class ChannelManager:
try:
import uuid
cid = f"{self._session_id}::{str(uuid.uuid4())}"
user_agent = self._platform_config["user_agent"]
user_agent = self._ua_subscriber
# ── Step 1: distribution rights ──────────────────────────────────
rights_url = (
@@ -170,6 +182,14 @@ class ChannelManager:
try:
station_id = self._extract_station_id(tp_ch.station_id)
meta = self.station_metadata.get(station_id, {})
if not self._is_station_playable(meta.get("special_rights_profile")):
logger.debug(
f"Skipping {station_id} — special-rights blocked "
f"(profile={meta.get('special_rights_profile')})"
)
continue
name = meta.get("title") or tp_ch.station_id
logo_url = meta.get("logo_url")
quality = meta.get("quality")
@@ -362,7 +382,7 @@ class ChannelManager:
headers = {
"Authorization": f"Bearer {entitlement_token}",
"User-Agent": self._platform_config["user_agent"],
"User-Agent": self._ua_subscriber,
"Accept": "application/json",
}
@@ -567,17 +587,20 @@ class ChannelManager:
break
channel_number = entry.get("dt$displayChannelNumber")
special_rights_profile = (
station_info.get("dt$clientData", {}) or {}
).get("specialRightsProfile")
existing = metadata.get(station_id)
if not existing or QUALITY_RANK.get(quality, 1) > QUALITY_RANK.get(
existing["quality"], 1
):
if station_id in metadata:
logger.debug(f"Duplicate station entry: {station_id}")
else:
metadata[station_id] = {
"title": title,
"logo_url": logo_url,
"quality": quality,
"channel_number": channel_number,
"playback_id": playback_id,
"special_rights_profile": special_rights_profile,
}
except Exception as exc:
logger.debug(f"_fetch_station_metadata: skipping entry: {exc}")
@@ -591,9 +614,36 @@ class ChannelManager:
return metadata
def _is_station_playable(self, profile: Optional[str]) -> bool:
"""
Return True if the manifest allows live playback of a station
carrying this specialRightsProfile tag (e.g. "uhd").
`profile` comes from station_metadata[station_id]["special_rights_profile"],
which _fetch_station_metadata reads from the station feed's
dt$clientData.specialRightsProfile. It is NOT read from the
entitled-channels feed's tp_ch.extra — parse_entitled_channels_feed
only carries distributionRightIds into `extra`, never dt$clientData,
so reading it from there would always fail open (nothing filtered).
A station is blocked only if:
1. It carries a specialRightsProfile tag, AND
2. The manifest has a rule for that tag, AND
3. The rule says liveTvOption == False.
No rule for a given tag means unrestricted (e.g. this is the case
for UHD/Sky/DAZN on the MagentaTV One manifest).
"""
if not profile:
return True
rule = self._special_rights.get(profile)
if rule is None:
return True
return rule.live_tv_option
def _get_api_headers(self, require_auth: bool = False) -> Dict[str, str]:
headers = {
"User-Agent": self._platform_config["user_agent"],
"User-Agent": self._ua_subscriber if require_auth else self._ua_plain,
"Accept": "application/json",
"Content-Type": "application/json",
}
lib/streaming_providers/providers/magenta2/concurrency.py
+44 -15
View File
@@ -6,17 +6,33 @@ from ...base.utils.logger import logger
def extract_and_release_lock(
smil_content: str, http_manager: HTTPManager, client_id: str, user_agent: str
smil_content: str,
http_manager: HTTPManager,
device_id: str,
session_id: str,
call_id_callback,
user_agent: str,
) -> bool:
"""
Extract concurrency lock from SMIL and immediately release it
Returns True if lock was found and released, False otherwise
Extract concurrency lock from SMIL and immediately release it.
Returns True if lock was found and released, False otherwise.
We release the lock immediately rather than holding it during playback,
so we never call the /update endpoint the real client calls every
~30s (per updateLockInterval in the SMIL response) to keep a lock alive.
Implementing /update would require the playback manager to own the lock
lifecycle — out of scope here; noted for future work.
Args:
smil_content: SMIL XML content
http_manager: HTTP manager for making requests
client_id: The client ID used in SMIL request (will be formatted as player_{client_id})
user_agent: Platform-specific user agent
smil_content: SMIL XML content.
http_manager: HTTP manager for making requests.
device_id: Persisted device UUID (same value used in the SMIL
request's clientId param). Sent as player_{device_id} — matching
format confirmed from a real ATV One capture of both the SMIL
request and this unlock request.
session_id: Session UUID, used in the CID header.
call_id_callback: Callable () -> str returning a fresh UUID per request.
user_agent: Platform-specific (subscriber-suffixed) user agent.
"""
try:
import xml.etree.ElementTree as ET
@@ -49,14 +65,17 @@ def extract_and_release_lock(
logger.debug("SMIL doesn't contain complete concurrency lock")
return False
# Build release URL with the same client_id formatted as player_{client_id}
# player_{device_id} — same format the real client uses for both
# the SMIL clientId param and this unlock call's _clientId param.
client_id = f"player_{device_id}"
cid = f"{session_id}::{call_id_callback()}"
base_url = lock_params["concurrencyServiceUrl"].rstrip("/") + "/web/Concurrency/unlock"
formatted_client_id = f"player_{client_id}"
params = {
"schema": "1.0",
"form": "json",
"_clientId": formatted_client_id, # Use player_{smil_client_id}
"_clientId": client_id,
"_id": lock_params["lockId"],
"_sequenceToken": urllib.parse.quote(lock_params["lockSequenceToken"]),
"_encryptedLock": urllib.parse.quote(lock_params["lock"]),
@@ -66,15 +85,25 @@ def extract_and_release_lock(
release_url = f"{base_url}?{param_string}"
logger.debug(
f"Releasing concurrency lock: {lock_params['lockId']} with client: {formatted_client_id}"
f"Releasing concurrency lock: {lock_params['lockId']} with client: {client_id}"
)
headers = {
"User-Agent": user_agent, # Use platform-specific user agent
"User-Agent": user_agent,
"Accept": "application/json",
"CID": cid,
# Cookie names embed the client ID — confirmed from a real
# unlock-request capture. Not optional; the server appears to
# read the lock/sequence data from these cookies as well as
# (or instead of) the query params.
"Cookie": (
f"LockId_{client_id}="
f"id={lock_params['lockId']}"
f"&sequenceToken={lock_params['lockSequenceToken']}; "
f"LockEncr_{client_id}={lock_params['lock']}"
),
}
# Release the lock immediately
# Release the lock immediately
response = http_manager.get(
release_url, operation="concurrency_unlock", headers=headers, timeout=10
@@ -91,7 +120,7 @@ def extract_and_release_lock(
response_data = response.json()
if "unlockResponse" in response_data:
logger.info(
f"✓ Concurrency lock released successfully with client: {formatted_client_id}"
f"✓ Concurrency lock released successfully with client: {client_id}"
)
return True
else:
@@ -110,4 +139,4 @@ def extract_and_release_lock(
except Exception as e:
logger.warning(f"Error releasing concurrency lock: {e}")
return False
return False
lib/streaming_providers/providers/magenta2/config_models.py
+73 -6
View File
@@ -11,7 +11,7 @@ class BootstrapConfig:
client_model: str
device_model: str
subscriber_type: str = "FTV_OTT_DT" # resolved from platform via SUBSCRIBER_TYPES
subscriber_type: str = "FTV_OTT_DT" # resolved from MAGENTA2_PLATFORMS[platform] in from_api_response()
sam3_client_id: Optional[str] = None
taa_url: Optional[str] = None
device_tokens_url: Optional[str] = None
@@ -30,16 +30,35 @@ class BootstrapConfig:
@classmethod
def from_api_response(cls, bootstrap_data: Dict[str, Any], platform: str) -> "BootstrapConfig":
"""Create BootstrapConfig from API response"""
from .constants import SUBSCRIBER_TYPES
"""
Create BootstrapConfig from API response.
Raises:
ValueError: if clientModel, deviceModel or sam3ClientId is missing
from baseSettings. This is intentional — a bootstrap response
missing these fields means the server didn't recognize us as
a real client, and silently falling back (e.g. to the legacy
MAGENTA2_LEGACY_CLIENT_IDS table) would mask that. Do not add
a fallback here; let discovery fail loudly instead.
"""
from .constants import MAGENTA2_PLATFORMS
base_settings = bootstrap_data.get("baseSettings", {})
dcm_settings = bootstrap_data.get("dcm", {})
required = ("clientModel", "deviceModel", "sam3ClientId")
missing = [k for k in required if not base_settings.get(k)]
if missing:
raise ValueError(
f"Bootstrap response missing required fields: {missing}"
)
platform_cfg = MAGENTA2_PLATFORMS.get(platform, {})
return cls(
client_model=base_settings.get("clientModel", f"ftv-{platform}"),
device_model=base_settings.get("deviceModel", f"{platform.upper()}_FTV"),
subscriber_type=SUBSCRIBER_TYPES.get(platform, "FTV_OTT_DT"),
client_model=base_settings["clientModel"],
device_model=base_settings["deviceModel"],
subscriber_type=platform_cfg.get("subscriber_type", "FTV_OTT_DT"),
sam3_client_id=base_settings.get("sam3ClientId"),
taa_url=base_settings.get("taaUrl"),
device_tokens_url=base_settings.get("deviceTokensUrl"),
@@ -263,6 +282,52 @@ class TvHubConfig:
return self.base_urls.get("UnstructuredGrid")
@dataclass
class SpecialRightsRule:
"""
One entry from manifest.mpx.specialRightsProfiles.clientData — governs
whether a station tagged with this profile name (e.g. "uhd") may be
played live, timeshifted, caught up, or recorded on this device/account.
"""
name: str
live_tv_option: bool
timeshift_option: bool
catchup_option: bool
npvr_option: bool
manage_option: bool
def _parse_special_rights_profiles(manifest_data: Dict[str, Any]) -> Dict[str, "SpecialRightsRule"]:
"""
Parse manifest.mpx.specialRightsProfiles.clientData into a lookup by
profile name (e.g. "uhd"). A station with no matching entry here is
unrestricted — see channel_manager.ChannelManager._is_station_playable.
"""
rules: Dict[str, SpecialRightsRule] = {}
entries = (
manifest_data.get("mpx", {})
.get("specialRightsProfiles", {})
.get("clientData", [])
or []
)
for entry in entries:
name = entry.get("name")
if not name:
continue
live = entry.get("liveTv", {}) or {}
rec = entry.get("recording", {}) or {}
rules[name] = SpecialRightsRule(
name=name,
live_tv_option=bool(live.get("liveTvOption", False)),
timeshift_option=bool(live.get("timeshiftOption", False)),
catchup_option=bool(live.get("catchupOption", False)),
npvr_option=bool(rec.get("npvrOption", False)),
manage_option=bool(rec.get("manageOption", False)),
)
return rules
@dataclass
class ManifestConfig:
"""Complete configuration from manifest discovery"""
@@ -273,6 +338,7 @@ class ManifestConfig:
image_config: ImageConfig
youbora_config: Dict[str, Any] = field(default_factory=dict)
npvr_config: Dict[str, Any] = field(default_factory=dict)
special_rights_profiles: Dict[str, SpecialRightsRule] = field(default_factory=dict)
raw_data: Dict[str, Any] = field(default_factory=dict)
@classmethod
@@ -285,6 +351,7 @@ class ManifestConfig:
image_config=ImageConfig.from_manifest_data(manifest_data),
youbora_config=manifest_data.get("youbora", {}),
npvr_config=manifest_data.get("npvr", {}),
special_rights_profiles=_parse_special_rights_profiles(manifest_data),
raw_data=manifest_data,
)
lib/streaming_providers/providers/magenta2/constants.py
+142 -102
View File
@@ -1,4 +1,6 @@
# streaming_providers/providers/magenta2/constants.py
from typing import Optional
# ============================================================================
# Magenta2 Configuration
# ============================================================================
@@ -12,107 +14,129 @@ DEFAULT_COUNTRY = "de"
MAGENTA2_LOGO = "https://upload.wikimedia.org/wikipedia/commons/thumb/e/e4/Magenta_TV_Logo_2024.svg/2339px-Magenta_TV_Logo_2024.svg.png"
# Platform configuration
#
# Each entry describes only what's needed to build the bootstrap/manifest
# request and the User-Agent. Everything else (client_model, device_model,
# sam3_client_id, all endpoint URLs) comes from the server's bootstrap/
# manifest response — see config_models.BootstrapConfig / ManifestConfig.
#
# Versions below are confirmed from real-device captures/logs:
# - android-tv / atv-launcher: 3.180.7748 (Sebastian's current AndroidTV,
# shape confirmed against real AndroidTV request logs). SMIL, DRM and
# concurrency request *shapes* were only ever captured on atv-launcher
# (v3.136.4682) — carried over to android-tv on the assumption the
# selector/concurrency service doesn't vary by config_group. Treat that
# part as unverified for android-tv specifically until confirmed by a
# live-play capture.
# - web: 2.128.5 (MacBook web client capture).
DEFAULT_PLATFORM = "android-tv"
MAGENTA2_PLATFORMS = {
"web": {
"device_name": "Web Browser",
"firmware": "Chrome 120",
"user_agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"terminal_type": "WEB",
# TAA-specific device identification
"taa_device_model": "Web Browser",
"taa_os": "Chrome 120",
},
"android-tv": {
"device_name": "Android TV",
"firmware": "Android 11",
"user_agent": "Dalvik/2.1.0 (Linux; U; Android 11; SHIELD Android TV Build/RQ1A.210105.003) ((2.00T_ATV::3.134.4462::mdarcy::))",
"terminal_type": "ATV_ANDROIDTV",
# TAA-specific device identification (API level format required)
"taa_device_model": "SHIELD Android TV",
"taa_os": "API level 30",
"config_group": "atv-androidtv",
"subscriber_type": "FTV_OTT_DT",
"application_model": "DT:ATV-AndroidTV",
"api_level": "30",
"android_version": "11",
"device_name": "SHIELD Android TV",
"build_id": "RQ1A.210105.003",
"build_flavor": "mdarcy",
"version": "3.180.7748",
"ua_template_plain": (
"Dalvik/2.1.0 (Linux; U; Android {android_version}; "
"{device_name} Build/{build_id}) "
"((2.00T_ATV::{version}::{build_flavor}::))"
),
"ua_template_subscriber": (
"Dalvik/2.1.0 (Linux; U; Android {android_version}; "
"{device_name} Build/{build_id}) "
"((2.00T_ATV::{version}::{build_flavor}::{subscriber_type}))"
),
},
"atv-launcher": {
"device_name": "MagentaTV Stick",
"firmware": "Android 11",
"user_agent": "Mozilla/5.0 (Linux; Android 11; AFTS Build/PPR1.180610.011) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36",
"terminal_type": "ATV_LAUNCHER",
# TAA-specific device identification
"taa_device_model": "MagentaTV Stick",
"taa_os": "API level 30",
"config_group": "atv-launcher",
"subscriber_type": "FTV_OTT_DT",
"application_model": "DT:ATV-Launcher",
"api_level": "30",
"android_version": "11",
"device_name": "SHIELD Android TV",
"build_id": "RQ1A.210105.003",
"build_flavor": "mdarcy",
"version": "3.180.7748",
"ua_template_plain": (
"Dalvik/2.1.0 (Linux; U; Android {android_version}; "
"{device_name} Build/{build_id}) "
"((2.00T_ATV::{version}::{build_flavor}::))"
),
"ua_template_subscriber": (
"Dalvik/2.1.0 (Linux; U; Android {android_version}; "
"{device_name} Build/{build_id}) "
"((2.00T_ATV::{version}::{build_flavor}::{subscriber_type}))"
),
},
"android-mobile": {
"device_name": "Android Mobile",
"firmware": "Android 13",
"user_agent": "Mozilla/5.0 (Linux; Android 13; SM-G991B) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Mobile Safari/537.36",
"terminal_type": "ANDROID_MOBILE",
# TAA-specific device identification
"taa_device_model": "Android Mobile",
"taa_os": "API level 33",
},
"ios": {
"device_name": "iPhone",
"firmware": "iOS 15",
"user_agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 15_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/15.0 Mobile/15E148 Safari/604.1",
"terminal_type": "IOS",
# TAA-specific device identification
"taa_device_model": "iPhone",
"taa_os": "iOS 15.0",
"web": {
"config_group": "web-mtv",
"subscriber_type": "FTV_OTT_DT",
"application_model": "DT:WEB",
"api_level": "0",
"android_version": "", # unused for web
"device_name": "",
"build_id": "",
"build_flavor": "",
"version": "2.128.5",
"ua_template_plain": (
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
),
"ua_template_subscriber": None, # web UA never has a suffix
},
}
# ============================================================================
# Manifest Request Configuration
# ============================================================================
# App identification for manifest requests
MAGENTA2_APP_NAME = "MagentaTV"
MAGENTA2_APP_VERSION = "104180"
MAGENTA2_RUNTIME_VERSION = "1"
def render_user_agent(
platform: str,
subscriber_suffix: bool = False,
version_override: Optional[str] = None,
) -> str:
"""
Render the User-Agent string for a platform.
# Model names for manifest requests (different from device models!)
MANIFEST_MODEL_MAPPINGS = {
"web": "DT:WEB",
"android-tv": "DT:ATV-AndroidTV",
"atv-launcher": "DT:ATV-Launcher",
"android-mobile": "DT:Android-Mobile",
"ios": "DT:IOS",
}
Args:
platform: key into MAGENTA2_PLATFORMS.
subscriber_suffix: if True, use the template with the subscriber_type
suffix (SMIL and DRM requests). If False, use the plain template
(bootstrap, manifest, DCM requests).
version_override: optional; replaces the platform's configured version.
# Firmware strings for manifest requests
MANIFEST_FIRMWARE_MAPPINGS = {
"web": "Chrome 120",
"android-tv": "API level 30",
"atv-launcher": "API level 30",
"android-mobile": "API level 33",
"ios": "iOS 15.0",
}
Raises:
ValueError: if `platform` is not a known key in MAGENTA2_PLATFORMS.
"""
cfg = MAGENTA2_PLATFORMS.get(platform)
if cfg is None:
raise ValueError(f"Unknown platform: {platform}")
# Also update the fallback mappings:
CLIENT_MODEL_MAPPINGS = {
"web": "ftv-web",
"android-tv": "ftv-androidtv",
"atv-launcher": "ftv-androidtv",
"android-mobile": "ftv-android",
"ios": "ftv-ios",
}
template = (
cfg["ua_template_subscriber"] if subscriber_suffix
else cfg["ua_template_plain"]
)
if template is None:
template = cfg["ua_template_plain"]
DEVICE_MODEL_MAPPINGS = {
"web": "WebBrowser_FTV",
"android-tv": "AndroidTV_FTV",
"atv-launcher": "AndroidTV_FTV",
"android-mobile": "AndroidMobile_FTV",
"ios": "iOS_FTV",
}
version = version_override or cfg["version"]
# And update subscriber types if needed:
SUBSCRIBER_TYPES = {
"web": "WEB_OTT_DT",
"android-tv": "FTV_OTT_DT",
"atv-launcher": "FTV_OTT_DT", # Same as android-tv
"android-mobile": "MOB_OTT_DT", # Different for mobile
"ios": "IOS_OTT_DT",
}
if "{" not in template:
return template
return template.format(
android_version=cfg["android_version"],
device_name=cfg["device_name"],
build_id=cfg["build_id"],
build_flavor=cfg["build_flavor"],
version=version,
subscriber_type=cfg["subscriber_type"],
)
# ============================================================================
# API Configuration - MINIMAL HARDCODING
@@ -120,8 +144,8 @@ SUBSCRIBER_TYPES = {
# Only bootstrap endpoint is hardcoded - everything else discovered dynamically
MAGENTA2_BASE_URL = "https://prod.dcm.telekom-dienste.de/v1"
MAGENTA2_BOOTSTRAP_URL = MAGENTA2_BASE_URL + "/settings/{terminal_type}/bootstrap"
MAGENTA2_MANIFEST_URL = MAGENTA2_BASE_URL + "/settings/{terminal_type}/manifest"
MAGENTA2_BOOTSTRAP_URL = MAGENTA2_BASE_URL + "/settings/{config_group}/bootstrap"
MAGENTA2_MANIFEST_URL = MAGENTA2_BASE_URL + "/settings/{config_group}/manifest"
# Fallback endpoints if discovery fails
MAGENTA2_FALLBACK_ENDPOINTS = {
@@ -141,7 +165,10 @@ MAGENTA2_FALLBACK_ACCOUNT_URI = "http://access.auth.theplatform.com/data/Account
# Application identifiers
IDM = "TDGIDM"
APPVERSION2 = "3.134.4462"
# NOTE: no separate app-version constant. auth.py::to_taa_payload reads
# MAGENTA2_PLATFORMS[platform]["version"] directly, confirmed by a real TAA
# capture to be identical to the version embedded in the User-Agent — a
# previously separate APPVERSION2 constant here could drift from that value.
SSO_URL = "https://ssom.magentatv.de/login"
# SSO User Agent
@@ -151,16 +178,19 @@ SSO_USER_AGENT = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, lik
# OAuth2 Configuration
# ============================================================================
# Client IDs for different platforms
MAGENTA2_CLIENT_IDS = {
# Legacy UUIDv4 client IDs.
#
# Used ONLY as a last-resort fallback if the bootstrap response fails to
# return a sam3ClientId — which in practice never happens, because a missing
# sam3ClientId already makes BootstrapConfig.from_api_response raise (see
# config_models.py), so discovery itself fails before this table would be
# consulted. Modern clients read sam3ClientId from the bootstrap response
# (baseSettings.sam3ClientId), not from this table.
MAGENTA2_LEGACY_CLIENT_IDS = {
"web": "709115c2-f87e-4bad-9b94-28ac08d72cd9",
"android-tv": "05f5f3df-1130-4707-a761-c04d0c50b7f2",
"ios": "21218403-52ec-4a65-abf4-f36a0eadd631",
}
# Client ID used for SMIL/selector manifest requests
SMIL_CLIENT_ID = "a8198f31-b406-4177-8dee-f6216c356c75"
# OAuth2 scopes
MAGENTA2_OAUTH_SCOPE = "openid profile offline_access tvhubs"
@@ -256,10 +286,13 @@ ERROR_CODES = {
# TAA Configuration
# ============================================================================
# TAA request template
# TAA request template.
# NOTE: no "appVersion" key here — it must vary by platform (each platform
# has its own version in MAGENTA2_PLATFORMS), so auth.py::to_taa_payload
# sets it dynamically from platform_config["version"] instead of baking in
# a single fixed value that couldn't be correct for every platform.
TAA_REQUEST_TEMPLATE = {
"accessTokenSource": IDM,
"appVersion": APPVERSION2,
"channel": {"id": "Tv"},
"natco": "DE",
"type": "telekom",
@@ -287,8 +320,15 @@ QUALITY_FALLBACK: dict = {
"SD": ["SD"],
}
# Integer rank per quality label — used for fast channel deduplication when
# multiple entries share the same display number (keep the highest-ranked one).
# Integer rank per quality label.
#
# NOTE: no longer used for live-channel dedup in channel_manager.py —
# SD/HD/UHD variants of a station have distinct station_ids in the entitled-
# channels feed, so a rank-based "keep the highest quality" merge in
# _fetch_station_metadata never actually fired on real data and has been
# removed (first entry wins for a genuinely duplicate station_id, with a
# debug log). Kept here because it may still be read by VOD quality
# selection alongside QUALITY_FALLBACK above — grep before deleting.
QUALITY_RANK: dict = {
"SD": 1,
"HD": 2,
@@ -302,9 +342,9 @@ QUALITY_RANK: dict = {
# ============================================================================
# tvhubs base URL template — {client_model} is resolved at runtime from
# CLIENT_MODEL_MAPPINGS. This is a fallback; the live value should come
# from the manifest's tv_hubs.base_urls["ftv"] (or equivalent) via
# ProviderConfig.get_resolved_tvhub_url().
# BootstrapConfig.client_model (server-provided). This URL is only a
# fallback; the live value should come from the manifest's
# tv_hubs.base_urls["ftv"] (or equivalent) via ProviderConfig.get_resolved_tvhub_url().
TVHUBS_BASE_URL = "https://tvhubs.t-online.de/v3/{client_model}"
# Flex IDs for the VOD catalogue.
lib/streaming_providers/providers/magenta2/discovery.py
+57 -40
View File
@@ -11,8 +11,9 @@ from .constants import (
DEFAULT_REQUEST_TIMEOUT,
MAGENTA2_BOOTSTRAP_URL,
MAGENTA2_MANIFEST_URL,
MAGENTA2_PLATFORMS,
OPENID_CONFIG_CACHE_DURATION,
SUBSCRIBER_TYPES,
render_user_agent,
)
@@ -24,19 +25,26 @@ class DiscoveryService:
def __init__(
self,
platform: str,
terminal_type: str,
device_id: str,
session_id: str,
http_manager: HTTPManager,
proxy_config: Optional[ProxyConfig] = None,
user_type: str = "normal",
):
cfg = MAGENTA2_PLATFORMS[platform]
self.platform = platform
self.terminal_type = terminal_type
self._config_group = cfg["config_group"]
self._subscriber_type = cfg["subscriber_type"]
self._application_model = cfg["application_model"]
self._api_level = cfg["api_level"]
self._version = cfg["version"]
self._ua_plain = render_user_agent(platform, subscriber_suffix=False)
self._ua_subscriber = render_user_agent(platform, subscriber_suffix=True)
self.device_id = device_id
self.session_id = session_id
self.http_manager = http_manager
self.proxy_config = proxy_config
self.subscriber_type = SUBSCRIBER_TYPES.get(platform, "FTV_OTT_DT")
self._user_type = user_type
# Cache storage
self._bootstrap_config: Optional[BootstrapConfig] = None
@@ -118,16 +126,26 @@ class DiscoveryService:
try:
logger.info("Discovering bootstrap configuration")
terminal_type = self.terminal_type.lower().replace("_", "-")
url = MAGENTA2_BOOTSTRAP_URL.format(terminal_type=terminal_type)
url = MAGENTA2_BOOTSTRAP_URL.format(config_group=self._config_group)
# Superset of the modern web-client shape (deviceId, portal,
# subscriberType, $redirect, sid — from MacBook web capture) and
# the legacy/ATV shape (applicationModel, version — confirmed
# against real AndroidTV request logs, v3.180.7748). deviceModel
# is intentionally NOT sent on bootstrap: it's absent from every
# modern-shape capture we have (web and ATV alike) and only
# appears on the legacy ATV bootstrap shape.
params = {
"deviceid": self.device_id,
"sid": self.session_id,
"deviceId": self.device_id,
"portal": "release",
"subscriberType": self._subscriber_type,
"$redirect": "false",
"sid": self.session_id,
"applicationModel": self._application_model,
"version": self._version,
}
headers = self._get_dcm_headers()
headers = self._get_dcm_headers(subscriber_suffix=False)
response = self.http_manager.get(
url,
@@ -183,42 +201,45 @@ class DiscoveryService:
try:
logger.info("Discovering manifest configuration")
# Manifest discovery only runs after a successful bootstrap — it
# needs deviceModel from the bootstrap response.
if not self._bootstrap_config:
logger.error("Cannot discover manifest: no bootstrap config available")
return None
# Priority 1: Use dcm.manifestBaseUrl from bootstrap
if self._bootstrap_config and self._bootstrap_config.manifest_base_url:
terminal_type = self.terminal_type.lower().replace("_", "-")
if self._bootstrap_config.manifest_base_url:
manifest_url = self._bootstrap_config.manifest_base_url.replace(
"{configGroupId}", terminal_type
"{configGroupId}", self._config_group
)
logger.debug(f"Using bootstrap manifestBaseUrl: {manifest_url}")
# Priority 2: Fallback to hardcoded manifest URL
else:
terminal_type = self.terminal_type.lower().replace("_", "-")
manifest_url = MAGENTA2_MANIFEST_URL.format(terminal_type=terminal_type)
manifest_url = MAGENTA2_MANIFEST_URL.format(config_group=self._config_group)
logger.debug(f"Using fallback manifest URL: {manifest_url}")
# Build correct manifest parameters
from .constants import (
MAGENTA2_APP_NAME,
MAGENTA2_APP_VERSION,
MAGENTA2_RUNTIME_VERSION,
MANIFEST_FIRMWARE_MAPPINGS,
MANIFEST_MODEL_MAPPINGS,
)
# Superset of the modern web-client shape (deviceId, deviceModel,
# portal, subscriberType, $redirect, sid — from MacBook web
# capture) and the legacy/ATV shape (applicationModel, version,
# apiLevel, userType — confirmed against real AndroidTV request
# logs, v3.180.7748).
params = {
"model": MANIFEST_MODEL_MAPPINGS.get(self.platform, "DT:ATV-AndroidTV"),
"deviceId": self.device_id,
"appname": MAGENTA2_APP_NAME,
"appVersion": MAGENTA2_APP_VERSION,
"firmware": MANIFEST_FIRMWARE_MAPPINGS.get(self.platform, "API level 30"),
"runtimeVersion": MAGENTA2_RUNTIME_VERSION,
"duid": self.device_id, # Same as deviceId
"portal": "release",
"subscriberType": self._subscriber_type,
"$redirect": "false",
"sid": self.session_id,
"deviceModel": self._bootstrap_config.device_model,
"applicationModel": self._application_model,
"version": self._version,
"apiLevel": self._api_level,
"userType": self._user_type,
}
logger.debug(f"Manifest request params: {params}")
headers = self._get_dcm_headers()
headers = self._get_dcm_headers(subscriber_suffix=False)
response = self.http_manager.get(
manifest_url,
@@ -315,18 +336,14 @@ class DiscoveryService:
self._last_openid = None
return None
def _get_dcm_headers(self) -> Dict[str, str]:
"""Get headers for DCM requests"""
from .constants import DEFAULT_PLATFORM, MAGENTA2_PLATFORMS
platform_config = MAGENTA2_PLATFORMS.get(
self.platform, MAGENTA2_PLATFORMS[DEFAULT_PLATFORM]
)
def _get_dcm_headers(self, subscriber_suffix: bool = False) -> Dict[str, str]:
"""Get headers for DCM requests (bootstrap/manifest)."""
ua = self._ua_subscriber if subscriber_suffix else self._ua_plain
return {
"User-Agent": platform_config["user_agent"],
"User-Agent": ua,
"Content-Type": "application/json",
"Accept": "application/json",
"Accept-Encoding": "gzip",
"x-dt-session-id": self.session_id,
"x-dt-call-id": self._generate_call_id(),
}
@@ -395,4 +412,4 @@ class DiscoveryService:
self._last_bootstrap = None
self._last_manifest = None
self._last_openid = None
logger.info("Discovery cache cleared")
logger.info("Discovery cache cleared")
lib/streaming_providers/providers/magenta2/playback_manager.py
+27 -2
View File
@@ -74,7 +74,21 @@ class PlaybackManager:
provider_config:
ProviderConfig after discovery.
platform_config:
Platform-specific dict from MAGENTA2_PLATFORMS (user_agent, etc.).
Platform-specific dict from MAGENTA2_PLATFORMS. Kept for any fields
not covered by user_agent_subscriber (subscriber_type, etc.) — no
longer used for platform_config["user_agent"], which doesn't exist
anymore; use user_agent_subscriber instead.
user_agent_subscriber:
Rendered UA (with subscriber_type suffix) — used for the live-DRM
fast path in get_drm(), matching what SmilManager uses for its own
(SMIL/VOD) DRM path.
session_id / call_id_callback:
Needed so the live-DRM fast path can send a CID header
("{session_id}::{call_id}"), matching the SMIL DRM path (see
smil_manager.get_drm) and confirmed against a real ATV widevine
capture, which shows only CID — no separate session-id header.
call_id_callback is optional — if omitted, CID is left off entirely
(better than sending a malformed header).
auth_callback:
Callable[[], str] — returns a valid persona token (Basic-auth value).
recording_url_cache:
@@ -91,6 +105,9 @@ class PlaybackManager:
platform_config: Dict,
auth_callback: Callable[[], str],
recording_url_cache: Dict[str, str],
user_agent_subscriber: Optional[str] = None,
session_id: Optional[str] = None,
call_id_callback: Optional[Callable[[], str]] = None,
):
self._channel_manager = channel_manager
self._smil_manager = smil_manager
@@ -99,6 +116,9 @@ class PlaybackManager:
self._platform_config = platform_config
self._ensure_authenticated = auth_callback
self._recording_url_cache = recording_url_cache
self._ua_subscriber = user_agent_subscriber
self._session_id = session_id
self._call_id = call_id_callback
# ------------------------------------------------------------------ #
# Public API #
@@ -340,10 +360,15 @@ class PlaybackManager:
persona_jwt=raw_jwt,
account_uri=account_uri,
)
call_id = self._call_id() if self._call_id else None
return [
build_widevine_drm_config(
licence_url=licence_url,
user_agent=self._platform_config["user_agent"],
# Subscriber-suffixed UA — matches what SmilManager.get_drm
# sends for the SMIL/VOD DRM path.
user_agent=self._ua_subscriber,
session_id=self._session_id,
call_id=call_id,
)
]
except Exception as exc:
lib/streaming_providers/providers/magenta2/provider.py
+77 -17
View File
@@ -15,7 +15,7 @@ that delegates to the three domain managers:
"""
import uuid
from datetime import datetime
from typing import Any, ClassVar, Dict, List, Optional, Tuple, cast
from typing import Any, ClassVar, Dict, List, Optional, Tuple, cast, Union
from urllib.parse import quote
from ...base.models import DRMConfig, StreamingChannel, Event
@@ -41,10 +41,11 @@ from .constants import (
DEFAULT_MAX_RETRIES,
DEFAULT_PLATFORM,
DEFAULT_REQUEST_TIMEOUT,
MAGENTA2_CLIENT_IDS,
MAGENTA2_LEGACY_CLIENT_IDS,
MAGENTA2_LOGO,
MAGENTA2_PLATFORMS,
SUPPORTED_COUNTRIES,
render_user_agent,
)
from .discovery import DiscoveryService
from .endpoint_manager import EndpointManager
@@ -78,15 +79,33 @@ class Magenta2Provider(StreamingProvider):
)
self.platform = platform
self.platform_config = MAGENTA2_PLATFORMS.get(
platform, MAGENTA2_PLATFORMS[DEFAULT_PLATFORM]
)
self.terminal_type = self.platform_config["terminal_type"]
if platform not in MAGENTA2_PLATFORMS:
raise ValueError(
f"Unknown platform '{platform}'. Supported: {list(MAGENTA2_PLATFORMS.keys())}"
)
self.platform_config = MAGENTA2_PLATFORMS[platform]
self.user_agent_plain = render_user_agent(platform, subscriber_suffix=False)
self.user_agent_subscriber = render_user_agent(platform, subscriber_suffix=True)
# Stable UUIDs for the lifetime of this provider instance.
self.session_id = self._generate_uuid()
self.device_id = self._generate_uuid()
self.serial_number = self._generate_uuid()
# session_id is fresh per process launch (matches the real client).
self.session_id = str(uuid.uuid1())
# device_id: read from the SAME persisted source TokenFlowManager
# already uses (SessionManager.get_device_id), which itself
# generates-and-persists on first call. Do NOT generate a second,
# independent device_id here — doing so would make discovery/SMIL
# send one device_id while the TAA/yo_digital token flow uses a
# different one, and the server would see two "devices" for one
# installation. get_device_id() currently persists a uuid4 — keep
# that format; do not switch to uuid1 for this value.
self.device_id = self.settings_manager.session_manager.get_device_id(
self.provider_name, self.country
)
# serial_number has no existing persisted home in SessionManager, so
# it gets its own small persisted key here (mirrors get_device_id's
# load-or-generate pattern without changing the shared SessionManager).
self.serial_number = self._get_or_create_serial_number()
# ── Proxy ────────────────────────────────────────────────────────────
self.proxy_config = (
@@ -103,7 +122,7 @@ class Magenta2Provider(StreamingProvider):
self.http_manager = HTTPManagerFactory.create_for_provider(
provider_name="magenta2",
proxy_config=self.proxy_config,
user_agent=self.platform_config["user_agent"],
user_agent=self.user_agent_plain,
timeout=DEFAULT_REQUEST_TIMEOUT,
max_retries=DEFAULT_MAX_RETRIES,
)
@@ -111,7 +130,6 @@ class Magenta2Provider(StreamingProvider):
# ── Discovery service ────────────────────────────────────────────────
self.discovery_service = DiscoveryService(
platform=platform,
terminal_type=self.terminal_type,
device_id=self.device_id,
session_id=self.session_id,
http_manager=self.http_manager,
@@ -127,12 +145,19 @@ class Magenta2Provider(StreamingProvider):
self.provider_config = cast(ProviderConfig, cast(object, None))
# ── Authenticator (minimal config; updated after discovery) ──────────
fallback_client_id = MAGENTA2_CLIENT_IDS.get(
platform, MAGENTA2_CLIENT_IDS[DEFAULT_PLATFORM]
# This placeholder client_id is overwritten by the real,
# server-provided sam3ClientId in _configure_authenticator_from_discovery()
# below, which always runs before __init__ returns (or __init__ raises
# and construction never completes — see _perform_configuration_discovery).
# It is unreachable in normal operation; it only matters for the
# handful of authenticator calls made before discovery finishes, none
# of which occur in this constructor.
fallback_client_id = MAGENTA2_LEGACY_CLIENT_IDS.get(
platform, MAGENTA2_LEGACY_CLIENT_IDS[DEFAULT_PLATFORM]
)
if username and password:
credentials: Magenta2Credentials | Magenta2UserCredentials = (
credentials: Union[Magenta2Credentials, Magenta2UserCredentials] = (
Magenta2UserCredentials(
client_id=fallback_client_id,
platform=platform,
@@ -221,6 +246,9 @@ class Magenta2Provider(StreamingProvider):
http_manager=self.http_manager,
provider_name=self.provider_name,
session_id=self.session_id,
device_id=self.device_id,
user_agent_plain=self.user_agent_plain,
user_agent_subscriber=self.user_agent_subscriber,
call_id_callback=self._generate_call_id,
auth_callback=self._ensure_authenticated,
platform_config=self.platform_config,
@@ -241,6 +269,8 @@ class Magenta2Provider(StreamingProvider):
provider_config=self.provider_config,
auth_callback=self._ensure_authenticated,
build_scaled_image_url_callback=self._build_scaled_image_url,
user_agent_plain=self.user_agent_plain,
user_agent_subscriber=self.user_agent_subscriber,
catchup_window=self.catchup_window,
)
logger.info("✓ ChannelManager initialized")
@@ -253,6 +283,9 @@ class Magenta2Provider(StreamingProvider):
platform_config=self.platform_config,
auth_callback=self._ensure_authenticated,
recording_url_cache=self._recording_url_cache,
user_agent_subscriber=self.user_agent_subscriber,
session_id=self.session_id,
call_id_callback=self._generate_call_id,
)
logger.info("✓ PlaybackManager initialized")
@@ -279,6 +312,8 @@ class Magenta2Provider(StreamingProvider):
provider_config=self.provider_config,
session_id=self.session_id,
serial_number=self.serial_number,
user_agent_plain=self.user_agent_plain,
user_agent_subscriber=self.user_agent_subscriber,
generate_call_id=self._generate_call_id,
)
@@ -295,6 +330,31 @@ class Magenta2Provider(StreamingProvider):
def _generate_call_id(self) -> str:
return self._generate_uuid()
def _get_or_create_serial_number(self) -> str:
"""
Return a stable serial number for this installation, persisted
across runs.
SessionManager has a dedicated get_device_id() that TokenFlowManager
also relies on, but no equivalent for serial_number, so this mirrors
the same load-or-generate pattern locally (same session_data blob,
different key) rather than adding a new public method to the shared
SessionManager.
"""
session_manager = self.settings_manager.session_manager
session_data = session_manager.load_session(self.provider_name, self.country) or {}
serial_number = session_data.get("serial_number")
if not serial_number:
serial_number = str(uuid.uuid4())
session_data["serial_number"] = serial_number
session_manager.save_session(self.provider_name, session_data, self.country)
logger.info(f"Generated new serial number: {serial_number}")
else:
logger.debug(f"Using existing serial number: {serial_number}")
return serial_number
def _load_proxy_from_manager(self, config_dir: Optional[str]) -> Optional[ProxyConfig]:
try:
proxy_manager = ProxyConfigManager(config_dir)
@@ -630,7 +690,7 @@ class Magenta2Provider(StreamingProvider):
def _get_dcm_headers(self) -> Dict[str, str]:
return {
"User-Agent": self.platform_config["user_agent"],
"User-Agent": self.user_agent_plain,
"Content-Type": "application/json",
"Accept": "application/json",
"x-dt-session-id": self.session_id,
@@ -639,7 +699,7 @@ class Magenta2Provider(StreamingProvider):
def _get_api_headers(self, require_auth: bool = False) -> Dict[str, str]:
headers: Dict[str, str] = {
"User-Agent": self.platform_config["user_agent"],
"User-Agent": self.user_agent_subscriber if require_auth else self.user_agent_plain,
"Accept": "application/json",
"Content-Type": "application/json",
}
lib/streaming_providers/providers/magenta2/smil_manager.py
+44 -10
View File
@@ -50,7 +50,6 @@ from .constants import (
DEFAULT_REQUEST_TIMEOUT,
MAGENTA2_FALLBACK_ACCOUNT_URI,
SMIL_CACHE_DURATION,
SMIL_CLIENT_ID,
VOD_PREFIX_EPISODE,
VOD_PREFIX_MOVIE_MV,
VOD_PREFIX_MOVIE_SH,
@@ -65,11 +64,27 @@ class SmilManager:
http_manager: HTTPManager instance from the parent provider.
provider_name: Provider identifier string (e.g. ``"magenta2"``).
session_id: Stable session UUID used in ``cid`` correlation header.
device_id: Persisted device UUID (from
``session_manager.get_device_id``). Used as
``player_{device_id}`` in the SMIL ``clientId``
query param and passed through to the
concurrency-unlock call, which must use the same
value the SMIL request used.
user_agent_plain: Rendered UA without the subscriber_type suffix.
Currently unused directly by this class (SMIL and
DRM requests are always subscriber-suffixed per
capture) but kept for parity with DiscoveryService
and in case a future endpoint needs it.
user_agent_subscriber: Rendered UA with the subscriber_type suffix —
used for SMIL, DRM and concurrency-unlock requests,
matching the real client's captured headers.
call_id_callback: Callable ``() -> str`` returning a fresh UUID per request.
auth_callback: Callable ``() -> str`` returning the current
Base64-encoded persona token.
platform_config: Platform dict from ``MAGENTA2_PLATFORMS`` — supplies
User-Agent and DRM request headers.
platform_config: Platform dict from ``MAGENTA2_PLATFORMS``. Kept for
any remaining legacy fields; no longer used for
platform_config["user_agent"], which doesn't exist
anymore — use user_agent_subscriber instead.
endpoint_manager: EndpointManager for selector / widevine endpoint lookup.
provider_config: ProviderConfig — account PID and account URI.
vod_manager: Optional VodManager — required only for GN id resolution.
@@ -81,6 +96,9 @@ class SmilManager:
http_manager,
provider_name: str,
session_id: str,
device_id: str,
user_agent_plain: str,
user_agent_subscriber: str,
call_id_callback,
auth_callback,
platform_config: Dict,
@@ -92,6 +110,9 @@ class SmilManager:
self._http = http_manager
self._provider = provider_name
self._session_id = session_id
self._device_id = device_id
self._ua_plain = user_agent_plain
self._ua_subscriber = user_agent_subscriber
self._call_id = call_id_callback
self._auth = auth_callback
self._platform_config = platform_config
@@ -217,6 +238,9 @@ class SmilManager:
f"account={quote(account_uri, safe='')}"
)
call_id = self._call_id()
cid = f"{self._session_id}::{call_id}"
drm_config = DRMConfig(
system=DRMSystem.WIDEVINE,
priority=1,
@@ -226,8 +250,9 @@ class SmilManager:
server_certificate=None,
req_headers=json.dumps(
{
"User-Agent": self._platform_config["user_agent"],
"User-Agent": self._ua_subscriber,
"Content-Type": "application/octet-stream",
"CID": cid,
}
),
use_http_get_request=False,
@@ -463,8 +488,15 @@ class SmilManager:
logger.error(f"{self._provider}: No persona token for SMIL request")
return None
cid = f"{self._session_id}::{self._call_id()}"
smil_params = f"?format=SMIL&formats=MPEG-DASH&tracking=true&cid={cid}"
call_id = self._call_id()
cid = f"{self._session_id}::{call_id}"
smil_params = (
f"?format=smil"
f"&formats=MPEG-DASH"
f"&tracking=true"
f"&clientId=player_{self._device_id}"
f"&cid={cid}"
)
if smil_base_url:
smil_url = f"{smil_base_url.split('?')[0]}{smil_params}"
@@ -487,8 +519,8 @@ class SmilManager:
headers = {
"Authorization": f"Basic {persona_token}",
"User-Agent": self._platform_config["user_agent"],
"Accept": "application/smil+xml, application/xml;q=0.9, */*;q=0.8",
"User-Agent": self._ua_subscriber,
"CID": cid,
}
logger.debug(f"{self._provider}: SMIL URL: {smil_url}")
@@ -528,8 +560,10 @@ class SmilManager:
extract_and_release_lock(
smil_content,
self._http,
client_id=SMIL_CLIENT_ID,
user_agent=self._platform_config["user_agent"],
device_id=self._device_id,
session_id=self._session_id,
call_id_callback=self._call_id,
user_agent=self._ua_subscriber,
)
return smil_content
lib/streaming_providers/providers/magenta2/vod_manager.py
+8 -4
View File
@@ -72,7 +72,6 @@ from ...base.models.quality import Quality
from .constants import (
QUALITY_FALLBACK,
SUBSCRIBER_TYPES,
TVHUBS_BASE_URL,
VOD_DEFAULT_PAGE_SIZE,
VOD_FLEX_ID_DETAILS,
@@ -222,9 +221,14 @@ class VodManager:
# request via _playback_params(). Not available from bootstrap because it
# is portal-specific metadata returned by the server, not a device identity.
self._partner_map_id: Optional[str] = None
# subscriber_type is not in bootstrap; derive from platform once at init.
_platform = getattr(bootstrap, "platform", "")
self._subscriber_type: str = SUBSCRIBER_TYPES.get(_platform, "FTV_OTT_DT")
# subscriber_type comes straight from BootstrapConfig.subscriber_type,
# which config_models.BootstrapConfig.from_api_response populates from
# MAGENTA2_PLATFORMS[platform]["subscriber_type"] at discovery time.
# (Previously this tried bootstrap.platform, which BootstrapConfig
# never actually had -- getattr silently returned "" and the
# SUBSCRIBER_TYPES lookup always fell through to its "FTV_OTT_DT"
# default regardless of platform. This reads the real value now.)
self._subscriber_type: str = getattr(bootstrap, "subscriber_type", None) or "FTV_OTT_DT"
# Node registry: opaque content_id → (fetch_url, extra_params)
# Populated when lanes/series/seasons are discovered so that