"""Stream manifest parsing: bento (JSON) and DASH (XML).
Manifests are produced by Tidal's CDN and embedded (base64) in API
responses, but they are still *remote-supplied input*: XML is parsed
with ``defusedxml`` (entity-expansion bombs) and every parser enforces
hard bounds (bytes, nodes, segments, expanded URL size). A manifest
that asks for DRM (``ContentProtection``) is rejected outright —
returning those URLs would save undecryptable bytes as a "successful"
download.
Both parsers return plain dicts::
{"codec": "FLAC", "urls": [...], "bandwidth": 900000,
"sample_rate": 192000, "bit_depth": 24, "channels": 2,
"representation_id": "FLAC_HIRES,192000,24"}
``urls`` is ``[initialization, segment1, segment2, ...]`` for DASH and
``[single_file_url]`` for bento.
"""
from __future__ import annotations
import base64
import json
import logging
import math
import re
from typing import Any
from urllib.parse import urljoin
import defusedxml.ElementTree as ET
from .errors import ManifestError
logger = logging.getLogger("tidalpy.manifests")
MAX_MANIFEST_BYTES = 2 * 1024 * 1024
MAX_MANIFEST_NODES = 200_000
MAX_SEGMENTS = 100_000
MAX_EXPANDED_URL_BYTES = 16 * 1024 * 1024
MAX_TEMPLATE_WIDTH = 20
def _text(content: bytes | str) -> str:
if isinstance(content, bytes):
if len(content) > MAX_MANIFEST_BYTES:
raise ManifestError(
f"manifest too large ({len(content)} bytes > {MAX_MANIFEST_BYTES})"
)
text = content.decode("utf-8-sig", errors="strict")
else:
if len(content.encode("utf-8")) > MAX_MANIFEST_BYTES:
raise ManifestError(f"manifest too large (>{MAX_MANIFEST_BYTES} bytes)")
text = content
return text
[docs]
def decode_manifest(manifest_b64: str | None, mime_type: str | None = "") -> str:
"""base64-decode an API manifest payload with size bounds."""
if not manifest_b64:
raise ManifestError("stream manifest is empty")
try:
raw = base64.b64decode(manifest_b64, validate=True)
except (ValueError, TypeError) as exc:
raise ManifestError(f"stream manifest is not valid base64: {exc}") from exc
if not raw:
raise ManifestError("stream manifest decodes to empty bytes")
return _text(raw)
[docs]
def decode_data_uri(uri: str | None) -> str:
"""Decode a ``data:application/dash+xml;base64,...`` URI (OpenAPI v2)."""
if not isinstance(uri, str) or "," not in uri:
raise ManifestError("stream manifest data URI is empty")
header, payload = uri.split(",", 1)
if ";base64" not in header.lower():
raise ManifestError(f"unsupported manifest data URI encoding: {header!r}")
if not payload:
raise ManifestError("stream manifest data URI has no payload")
try:
raw = base64.b64decode(payload, validate=True)
except ValueError as exc:
raise ManifestError("stream manifest data URI is not valid base64") from exc
if len(raw) > MAX_MANIFEST_BYTES:
raise ManifestError(f"stream manifest too large ({len(raw)} bytes)")
try:
return raw.decode("utf-8")
except UnicodeDecodeError as exc:
raise ManifestError("stream manifest is not valid UTF-8") from exc
# ---------------------------------------------------------------------------
# bento
# ---------------------------------------------------------------------------
[docs]
def parse_bento(manifest_b64: str) -> dict[str, Any]:
"""Parse a ``application/vnd.tidal.bts*`` (bento) manifest.
Result includes ``codec`` (normalised), ``urls``, and
``encryption_key`` (the raw ``keyId`` security token or ``None``).
"""
text = decode_manifest(manifest_b64, "bento")
try:
data = json.loads(text)
except ValueError as exc:
raise ManifestError(f"bento manifest is not valid JSON: {exc}") from exc
if not isinstance(data, dict):
raise ManifestError("bento manifest is not a JSON object")
urls = data.get("urls") or []
if not isinstance(urls, list) or not urls:
raise ManifestError("bento manifest has no URLs")
urls = [str(u) for u in urls if u]
if not urls:
raise ManifestError("bento manifest URLs are empty")
codecs = str(data.get("codecs") or "")
encryption_key = data.get("keyId") or None
if encryption_key is not None:
encryption_key = str(encryption_key)
return {
"codec": codecs,
"urls": urls,
"encryption_key": encryption_key,
"bandwidth": _int(data.get("bandwidth")),
"sample_rate": _int(data.get("samplingRate")),
"bit_depth": _int(data.get("bitDepth")),
"channels": _int(data.get("channelCount") or data.get("channels")),
"representation_id": None,
}
# ---------------------------------------------------------------------------
# DASH
# ---------------------------------------------------------------------------
def _seconds(value: str | None) -> float | None:
match = re.fullmatch(
r"PT(?:(\d+(?:\.\d+)?)H)?(?:(\d+(?:\.\d+)?)M)?(?:(\d+(?:\.\d+)?)S)?",
value or "",
)
if not match:
return None
return sum(
float(part or 0) * factor for part, factor in zip(match.groups(), (3600, 60, 1))
)
def _xml_root(content: str) -> ET.Element:
try:
root = ET.fromstring(content)
except Exception as exc: # defusedxml raises DefusedXmlException
raise ManifestError(f"invalid DASH manifest XML: {exc}") from exc
node_count = sum(1 for _ in root.iter())
if node_count > MAX_MANIFEST_NODES:
raise ManifestError(f"DASH manifest has too many XML nodes ({node_count})")
return root
def _bit_depth_from_id(representation_id: str) -> str:
"""Tidal encodes bit depth in representation ids like ``FLAC,192000,24``."""
match = re.fullmatch(
r"FLAC(?:_[A-Z0-9]+)*,\d+,(\d{1,2})", representation_id, re.IGNORECASE
)
return match.group(1) if match else ""
def _expand_template(pattern: str, values: dict[str, str]) -> str:
def replace(match: re.Match) -> str:
name, width_spec, width = match.group(1), match.group(2), match.group(3)
digits = int(width) if width else 0
if digits > MAX_TEMPLATE_WIDTH:
raise ManifestError(f"DASH template width too large: %{width_spec}")
value = str(values.get(name, ""))
return value.zfill(digits) if digits else value
expanded = re.sub(
r"\$(RepresentationID|Bandwidth|Number|Time)(%0(\d+)d)?\$",
replace,
pattern.replace("$$", "\x00"),
)
expanded = expanded.replace("\x00", "$")
if not expanded:
raise ManifestError("DASH segment template expanded to an empty URL")
return expanded
[docs]
def parse_dash(manifest: str | bytes) -> list[dict[str, Any]]:
"""Parse a DASH MPD into audio representations (see module docstring)."""
root = _xml_root(_text(manifest))
# Strip namespaces for simpler traversal (safe: bounds already applied).
for element in root.iter():
element.tag = element.tag.rsplit("}", 1)[-1]
if element.tag == "ContentProtection":
raise ManifestError("DRM-protected DASH stream is not supported")
periods = root.findall("Period")
if len(periods) != 1:
raise ManifestError(f"DASH must have exactly one period, found {len(periods)}")
period = periods[0]
period_duration = _seconds(period.get("duration"))
if period_duration is None:
period_duration = _seconds(root.get("mediaPresentationDuration"))
def base_prefix(parent: str, node: ET.Element | None) -> str:
if node is None:
return parent
child = node.find("BaseURL")
if child is not None and child.text:
text: str = child.text.strip() # defusedxml is untyped
return urljoin(parent, text)
return parent
period_prefix = base_prefix(base_prefix("", root), period)
representations: list[dict[str, Any]] = []
expanded_count = 0
for adaptation in period.findall("AdaptationSet"):
media_type = adaptation.get("mimeType") or ""
for rep in adaptation.findall("Representation"):
media_type = rep.get("mimeType") or media_type
if adaptation.get("contentType") != "audio" and not media_type.startswith(
"audio/"
):
continue
# SegmentTemplate inherits: Period -> AdaptationSet -> Representation
template_attrs: dict[str, str] = {}
timeline: ET.Element | None = None
for node in (period, adaptation, rep):
template = node.find("SegmentTemplate")
if template is None:
continue
template_attrs.update(template.attrib)
found = template.find("SegmentTimeline")
if found is not None:
timeline = found
if not template_attrs.get("initialization") or not template_attrs.get(
"media"
):
raise ManifestError("DASH SegmentTemplate lacks initialization/media")
prefix = base_prefix(base_prefix(period_prefix, adaptation), rep)
start_number = int(template_attrs.get("startNumber") or 1)
timescale = int(template_attrs.get("timescale") or 1)
if timescale <= 0:
raise ManifestError("DASH timescale must be positive")
timestamps: list[int] = []
if timeline is not None:
current = 0
entries = timeline.findall("S")
for index, s in enumerate(entries):
if s.get("t") is not None:
current = int(s.get("t"))
step = int(s.get("d") or 0)
if step <= 0:
raise ManifestError("DASH segment duration must be positive")
repeat = int(s.get("r") or 0)
if repeat < 0:
# Open-ended repeat (``r="-1"`` — the DASH live
# idiom): bound it by the following entry's
# start time, falling back to the presentation
# duration. An open repeat with no known end is
# live-only and cannot be a VOD file.
following = (
entries[index + 1] if index + 1 < len(entries) else None
)
end: float | None = None
if following is not None and following.get("t"):
end = float(following.get("t"))
elif period_duration is not None:
end = period_duration * timescale + int(
template_attrs.get("presentationTimeOffset") or 0
)
if end is None:
raise ManifestError(
"unbounded DASH timeline is not supported"
)
repeat = math.ceil((end - current) / step) - 1
count = repeat + 1
if count < 0 or len(timestamps) + count > MAX_SEGMENTS:
raise ManifestError(
"invalid or excessively large DASH timeline"
)
timestamps.extend(current + i * step for i in range(count))
current += step * count
elif (
period_duration is not None
and int(template_attrs.get("duration") or 0) > 0
):
step = int(template_attrs["duration"])
count = math.ceil(period_duration * timescale / step)
if count > MAX_SEGMENTS:
raise ManifestError("DASH timeline exceeds the segment bound")
timestamps = [i * step for i in range(int(count))]
else:
raise ManifestError("DASH manifest has no finite segment timeline")
if not timestamps or len(timestamps) > MAX_SEGMENTS:
raise ManifestError(
"DASH timeline is empty or exceeds the segment bound"
)
values = {
"RepresentationID": rep.get("id", ""),
"Bandwidth": rep.get("bandwidth", ""),
}
urls: list[str] = []
expanded_bytes = 0
init_url = _expand_template(template_attrs["initialization"], values)
urls.append(urljoin(prefix, init_url))
expanded_bytes += len(init_url.encode("utf-8"))
for index, timestamp in enumerate(timestamps):
values.update(
{"Number": str(start_number + index), "Time": str(timestamp)}
)
segment = _expand_template(template_attrs["media"], values)
location = urljoin(prefix, segment)
expanded_bytes += len(location.encode("utf-8"))
if expanded_bytes > MAX_EXPANDED_URL_BYTES:
raise ManifestError("expanded DASH manifest is too large")
urls.append(location)
expanded_count += len(urls)
if expanded_count > MAX_SEGMENTS + 1:
raise ManifestError("DASH manifest exceeds the total segment bound")
bit_depth = (
rep.get("audioBitDepth") or adaptation.get("audioBitDepth") or ""
)
if not bit_depth:
bit_depth = _bit_depth_from_id(rep.get("id", ""))
sample_rate = (
rep.get("audioSamplingRate")
or adaptation.get("audioSamplingRate")
or ""
)
channels = ""
for node in (rep, adaptation):
config = node.find("AudioChannelConfiguration")
if config is not None:
channels = config.get("value", "")
break
representations.append(
{
"urls": urls,
"codec": rep.get("codecs") or adaptation.get("codecs") or "",
"mimeType": media_type,
"bandwidth": _int(rep.get("bandwidth")),
"sample_rate": _int(sample_rate),
"bit_depth": _int(bit_depth),
"channels": _int(channels),
"representation_id": rep.get("id", ""),
}
)
if not representations:
raise ManifestError("DASH manifest contains no audio representations")
return representations
def _int(value: Any) -> int | None:
try:
result = int(value)
return result if result > 0 else None
except (TypeError, ValueError):
return None