mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-07-15 12:27:24 +00:00
feat(ism): harden Smooth parsing and derive fps/codec from spec fields
Regex-based UrlPattern substitution (handles {Bitrate}/{start_time} spellings), AudioTag/FourCC codec fallback, Width/Height fallbacks, integer fragment times, live-manifest and missing-language guards. Extract fps from the HEVC SPS VUI (HEVC-only; AVC timing untrusted).
This commit is contained in:
imSp4rky
@@ -14,9 +14,10 @@ import pytest
|
||||
|
||||
from unshackle.core.manifests.ism_init import (NAL_START_CODE, PIFF_SENC_UUID, box, build_avcc, build_dec3,
|
||||
build_hvcc, build_init_segment, full_box,
|
||||
parse_codec_private_data_colour, parse_hevc_sps_format,
|
||||
read_per_sample_iv_size, read_track_id, remove_emulation_prevention,
|
||||
split_nal_units, synthesize_aac_codec_private_data)
|
||||
parse_codec_private_data_colour, parse_codec_private_data_vui,
|
||||
parse_hevc_sps_format, read_per_sample_iv_size, read_track_id,
|
||||
remove_emulation_prevention, split_nal_units,
|
||||
synthesize_aac_codec_private_data)
|
||||
|
||||
# Real CodecPrivateData taken from a Smooth Streaming manifest.
|
||||
VIDEO_HEVC_CPD = (
|
||||
@@ -472,6 +473,15 @@ def test_parse_colour_absent_or_unknown_returns_none():
|
||||
assert parse_codec_private_data_colour("HVC1", b"\x00\x00\x00\x01\x42") is None
|
||||
|
||||
|
||||
def test_parse_vui_fps():
|
||||
assert parse_codec_private_data_vui("HVC1", bytes.fromhex(VIDEO_HEVC_CPD))[1] == pytest.approx(24000 / 1001)
|
||||
assert parse_codec_private_data_vui("HVC1", bytes.fromhex(VIDEO_HEVC_PQ_CPD))[1] == 25.0
|
||||
# VUI timing can be present even when colour info is absent.
|
||||
assert parse_codec_private_data_vui("HVC1", bytes.fromhex(VIDEO_HEVC10_CPD)) == (None, 25.0)
|
||||
# AVC is deliberately untrusted for fps (field-based VUI timing).
|
||||
assert parse_codec_private_data_vui("H264", bytes.fromhex(VIDEO_AVC_CPD)) == (None, None)
|
||||
|
||||
|
||||
def test_hvcc_profile_tier_level_is_nonzero():
|
||||
# De-emulated PTL must yield real profile/level, not the off-by-one garbage.
|
||||
hvcc = build_hvcc(bytes.fromhex(VIDEO_HEVC_CPD))
|
||||
|
||||
43
tests/core/test_ism_parse.py
Normal file
43
tests/core/test_ism_parse.py
Normal file
@@ -0,0 +1,43 @@
|
||||
import pytest
|
||||
|
||||
from unshackle.core.manifests.ism import ISM
|
||||
from unshackle.core.tracks import Audio
|
||||
|
||||
MANIFEST = """<?xml version="1.0"?>
|
||||
<SmoothStreamingMedia MajorVersion="2" MinorVersion="0" Duration="60000000" TimeScale="10000000">
|
||||
<StreamIndex Type="audio" Name="audio" Language="en"
|
||||
Url="QualityLevels({Bitrate})/Fragments(audio={start_time})">
|
||||
<QualityLevel Index="0" Bitrate="128000" AudioTag="255" SamplingRate="48000"
|
||||
Channels="2" BitsPerSample="16" CodecPrivateData="1190"/>
|
||||
<c t="0" d="20000000" r="2"/>
|
||||
<c/>
|
||||
</StreamIndex>
|
||||
</SmoothStreamingMedia>
|
||||
"""
|
||||
|
||||
|
||||
def test_placeholder_spellings_repeat_and_audiotag_fallback() -> None:
|
||||
tracks = ISM.from_text(MANIFEST, "https://cdn.example/x.ism/manifest").to_tracks()
|
||||
(audio,) = tracks.audio
|
||||
segments = audio.data["ism"]["segments"]
|
||||
assert segments == [
|
||||
"https://cdn.example/x.ism/QualityLevels(128000)/Fragments(audio=0)",
|
||||
"https://cdn.example/x.ism/QualityLevels(128000)/Fragments(audio=20000000)",
|
||||
"https://cdn.example/x.ism/QualityLevels(128000)/Fragments(audio=40000000)",
|
||||
]
|
||||
assert audio.codec == Audio.Codec.AAC # AudioTag 255 means AACL
|
||||
|
||||
|
||||
def test_missing_language_without_fallback_raises() -> None:
|
||||
# MS-SSTR Language is optional; video StreamIndexes commonly omit it.
|
||||
no_lang = MANIFEST.replace(' Language="en"', "")
|
||||
with pytest.raises(ValueError, match="fallback language"):
|
||||
ISM.from_text(no_lang, "https://cdn.example/x.ism/manifest").to_tracks()
|
||||
tracks = ISM.from_text(no_lang, "https://cdn.example/x.ism/manifest").to_tracks(language="en")
|
||||
assert str(tracks.audio[0].language) == "en"
|
||||
|
||||
|
||||
def test_live_manifest_rejected() -> None:
|
||||
live = MANIFEST.replace('Duration="60000000"', 'Duration="60000000" IsLive="TRUE"', 1)
|
||||
with pytest.raises(ValueError, match="Live"):
|
||||
ISM.from_text(live, "https://cdn.example/x.ism/manifest").to_tracks()
|
||||
@@ -3,6 +3,7 @@ from __future__ import annotations
|
||||
import base64
|
||||
import hashlib
|
||||
import html
|
||||
import re
|
||||
import shutil
|
||||
import struct
|
||||
import urllib.parse
|
||||
@@ -18,7 +19,7 @@ from requests import Session
|
||||
from unshackle.core.constants import DOWNLOAD_CANCELLED, DOWNLOAD_LICENCE_ONLY, AnyTrack
|
||||
from unshackle.core.drm import DRM_T, PlayReady, Widevine
|
||||
from unshackle.core.events import events
|
||||
from unshackle.core.manifests.ism_init import (build_init_segment, parse_codec_private_data_colour,
|
||||
from unshackle.core.manifests.ism_init import (build_init_segment, parse_codec_private_data_vui,
|
||||
read_per_sample_iv_size, read_track_id)
|
||||
from unshackle.core.session import RnetSession
|
||||
from unshackle.core.tracks import Audio, DownloadContext, Subtitle, Track, Tracks, Video
|
||||
@@ -26,6 +27,12 @@ from unshackle.core.utilities import log_event, try_ensure_utf8
|
||||
from unshackle.core.utils.redact import safe_display_url
|
||||
from unshackle.core.utils.xml import load_xml
|
||||
|
||||
# MS-SSTR: FourCC may be absent; AudioTag carries the WAVE format tag instead.
|
||||
AUDIO_TAG_FOURCC = {"255": "AACL", "65534": "EC-3"}
|
||||
|
||||
# Smooth FourCCs that Codec.from_mime (RFC 6381 names) doesn't know directly.
|
||||
FOURCC_MIME = {"H264": "avc1", "H265": "hvc1", "HEVC": "hvc1", "AACL": "mp4a", "AACH": "mp4a", "AACP": "mp4a"}
|
||||
|
||||
|
||||
class ISM:
|
||||
def __init__(self, manifest: Element, url: str) -> None:
|
||||
@@ -88,20 +95,25 @@ class ISM:
|
||||
return drm
|
||||
|
||||
@staticmethod
|
||||
def get_video_range(fourcc: str, codec_private_data: str) -> Video.Range:
|
||||
"""Derive colour range from the SPS VUI in CodecPrivateData — Smooth
|
||||
manifests carry no range attributes. Soft-fails to SDR."""
|
||||
def get_video_range_and_fps(fourcc: str, codec_private_data: str) -> tuple[Video.Range, Optional[float]]:
|
||||
"""Derive colour range and fps from the SPS VUI in CodecPrivateData,
|
||||
since Smooth manifests carry neither as attributes. Range soft-fails to
|
||||
SDR; fps is None for non-HEVC codecs and VUIs without timing info."""
|
||||
fourcc = (fourcc or "").upper()
|
||||
if fourcc in ("DVHE", "DVH1"):
|
||||
return Video.Range.DV
|
||||
try:
|
||||
cpd = bytes.fromhex(codec_private_data or "")
|
||||
except ValueError:
|
||||
return Video.Range.SDR
|
||||
cicp = parse_codec_private_data_colour(fourcc, cpd)
|
||||
cpd = b""
|
||||
cicp, fps = parse_codec_private_data_vui(fourcc, cpd)
|
||||
if fourcc in ("DVHE", "DVH1"):
|
||||
return Video.Range.DV, fps
|
||||
if not cicp:
|
||||
return Video.Range.SDR
|
||||
return Video.Range.from_cicp(*cicp)
|
||||
return Video.Range.SDR, fps
|
||||
return Video.Range.from_cicp(*cicp), fps
|
||||
|
||||
@staticmethod
|
||||
def get_video_range(fourcc: str, codec_private_data: str) -> Video.Range:
|
||||
return ISM.get_video_range_and_fps(fourcc, codec_private_data)[0]
|
||||
|
||||
@staticmethod
|
||||
def _init_segment(
|
||||
@@ -121,7 +133,7 @@ class ISM:
|
||||
# CodecPrivateData may legitimately be empty (AAC config is synthesized,
|
||||
# EC-3 decoders sync from the frames); the builder handles each case.
|
||||
cpd = quality_level.get("CodecPrivateData") or ""
|
||||
fourcc = quality_level.get("FourCC") or ""
|
||||
fourcc = quality_level.get("FourCC") or AUDIO_TAG_FOURCC.get(quality_level.get("AudioTag") or "") or ""
|
||||
|
||||
root_timescale = manifest.get("TimeScale") if manifest is not None else None
|
||||
timescale = int(stream_index.get("TimeScale") or root_timescale or 10000000)
|
||||
@@ -202,6 +214,8 @@ class ISM:
|
||||
return None
|
||||
|
||||
def to_tracks(self, language: Optional[Union[str, Language]] = None) -> Tracks:
|
||||
if (self.manifest.get("IsLive") or "").upper() == "TRUE":
|
||||
raise ValueError("Live Smooth Streaming manifests are not supported")
|
||||
tracks = Tracks()
|
||||
base_url = self.url
|
||||
duration = int(self.manifest.get("Duration") or 0)
|
||||
@@ -212,17 +226,30 @@ class ISM:
|
||||
if not content_type:
|
||||
raise ValueError("No content type value could be found")
|
||||
for ql in stream_index.findall("QualityLevel"):
|
||||
codec = ql.get("FourCC")
|
||||
codec = ql.get("FourCC") or AUDIO_TAG_FOURCC.get(ql.get("AudioTag") or "")
|
||||
if codec == "TTML":
|
||||
codec = "STPP"
|
||||
track_lang = None
|
||||
lang = (stream_index.get("Language") or "").strip()
|
||||
if lang and tag_is_valid(lang) and not lang.startswith("und"):
|
||||
track_lang = Language.get(lang)
|
||||
if not track_lang and not language:
|
||||
# Language is optional in MS-SSTR; video streams commonly omit it.
|
||||
raise ValueError(
|
||||
"Language information could not be derived from the manifest and no fallback "
|
||||
"language was provided when calling ISM.to_tracks()."
|
||||
)
|
||||
|
||||
track_urls: list[str] = []
|
||||
fragment_time = 0
|
||||
fragments = stream_index.findall("c")
|
||||
# MS-SSTR UrlPattern; regex over str.format so {Bitrate}/{start_time}
|
||||
# spellings work and unknown placeholders like {CustomAttributes}
|
||||
# or stray braces in query strings don't raise.
|
||||
url_template = urllib.parse.urljoin(
|
||||
base_url,
|
||||
re.sub(r"\{[Bb]itrate\}", str(ql.get("Bitrate") or 0), stream_index.get("Url") or ""),
|
||||
)
|
||||
# Some manifests omit the first fragment in the <c> list but
|
||||
# still expect a request for start time 0 which contains the
|
||||
# initialization segment. If the first declared fragment is not
|
||||
@@ -230,17 +257,7 @@ class ISM:
|
||||
if fragments:
|
||||
first_time = int(fragments[0].get("t") or 0)
|
||||
if first_time != 0:
|
||||
track_urls.append(
|
||||
urllib.parse.urljoin(
|
||||
base_url,
|
||||
stream_index.get("Url").format_map(
|
||||
{
|
||||
"bitrate": ql.get("Bitrate"),
|
||||
"start time": "0",
|
||||
}
|
||||
),
|
||||
)
|
||||
)
|
||||
track_urls.append(re.sub(r"\{start[ _]time\}", "0", url_template))
|
||||
|
||||
for idx, frag in enumerate(fragments):
|
||||
fragment_time = int(frag.get("t", fragment_time))
|
||||
@@ -251,19 +268,11 @@ class ISM:
|
||||
next_time = int(fragments[idx + 1].get("t"))
|
||||
except (IndexError, AttributeError):
|
||||
next_time = duration
|
||||
duration_frag = (next_time - fragment_time) / repeat
|
||||
# floor division: float times would corrupt segment URLs;
|
||||
# any drift is reset by the next fragment's explicit t.
|
||||
duration_frag = (next_time - fragment_time) // repeat
|
||||
for _ in range(repeat):
|
||||
track_urls.append(
|
||||
urllib.parse.urljoin(
|
||||
base_url,
|
||||
stream_index.get("Url").format_map(
|
||||
{
|
||||
"bitrate": ql.get("Bitrate"),
|
||||
"start time": str(fragment_time),
|
||||
}
|
||||
),
|
||||
)
|
||||
)
|
||||
track_urls.append(re.sub(r"\{start[ _]time\}", str(fragment_time), url_template))
|
||||
fragment_time += duration_frag
|
||||
|
||||
track_id = hashlib.md5(
|
||||
@@ -288,20 +297,25 @@ class ISM:
|
||||
|
||||
if content_type == "video":
|
||||
try:
|
||||
vcodec = Video.Codec.from_mime(codec) if codec else None
|
||||
vcodec = Video.Codec.from_mime(FOURCC_MIME.get(codec.upper(), codec)) if codec else None
|
||||
except ValueError:
|
||||
vcodec = None
|
||||
range_, fps = self.get_video_range_and_fps(codec or "", ql.get("CodecPrivateData") or "")
|
||||
tracks.add(
|
||||
Video(
|
||||
id_=track_id,
|
||||
url=self.url,
|
||||
codec=vcodec,
|
||||
range_=self.get_video_range(codec or "", ql.get("CodecPrivateData") or ""),
|
||||
range_=range_,
|
||||
fps=fps,
|
||||
language=track_lang or language,
|
||||
is_original_lang=bool(language and track_lang and str(track_lang) == str(language)),
|
||||
bitrate=ql.get("Bitrate"),
|
||||
width=int(ql.get("MaxWidth") or 0) or int(stream_index.get("MaxWidth") or 0),
|
||||
height=int(ql.get("MaxHeight") or 0) or int(stream_index.get("MaxHeight") or 0),
|
||||
# Width/Height are non-spec but common when Max* are absent
|
||||
width=int(ql.get("MaxWidth") or ql.get("Width") or 0)
|
||||
or int(stream_index.get("MaxWidth") or stream_index.get("Width") or 0),
|
||||
height=int(ql.get("MaxHeight") or ql.get("Height") or 0)
|
||||
or int(stream_index.get("MaxHeight") or stream_index.get("Height") or 0),
|
||||
descriptor=Video.Descriptor.ISM,
|
||||
drm=drm,
|
||||
data=data,
|
||||
@@ -309,7 +323,7 @@ class ISM:
|
||||
)
|
||||
elif content_type == "audio":
|
||||
try:
|
||||
acodec = Audio.Codec.from_mime(codec) if codec else None
|
||||
acodec = Audio.Codec.from_mime(FOURCC_MIME.get(codec.upper(), codec)) if codec else None
|
||||
except ValueError:
|
||||
acodec = None
|
||||
tracks.add(
|
||||
|
||||
@@ -215,9 +215,9 @@ def skip_hevc_st_ref_pic_set(r: BitReader, idx: int, num_delta_pocs: list[int])
|
||||
return num_negative + num_positive
|
||||
|
||||
|
||||
def parse_hevc_sps_colour(sps_rbsp: bytes) -> Optional[tuple[int, int, int]]:
|
||||
"""VUI (colour_primaries, transfer_characteristics, matrix_coeffs) from a
|
||||
de-emulated HEVC SPS, or None when no colour description is present."""
|
||||
def parse_hevc_sps_vui(sps_rbsp: bytes) -> tuple[Optional[tuple[int, int, int]], Optional[float]]:
|
||||
"""((colour_primaries, transfer_characteristics, matrix_coeffs), fps) from a
|
||||
de-emulated HEVC SPS VUI; either element is None when absent."""
|
||||
r = BitReader(sps_rbsp)
|
||||
_, _, _, max_sub_layers_minus1 = read_hevc_sps_to_bit_depth(r)
|
||||
log2_max_poc_lsb_minus4 = r.read_ue()
|
||||
@@ -245,37 +245,62 @@ def parse_hevc_sps_colour(sps_rbsp: bytes) -> Optional[tuple[int, int, int]]:
|
||||
r.read_bits(1) # used_by_curr_pic_lt_sps_flag
|
||||
r.read_bits(2) # sps_temporal_mvp_enabled + strong_intra_smoothing_enabled
|
||||
if not r.read_bits(1): # vui_parameters_present_flag
|
||||
return None
|
||||
return None, None
|
||||
if r.read_bits(1): # aspect_ratio_info_present_flag
|
||||
if r.read_bits(8) == 255: # aspect_ratio_idc == EXTENDED_SAR
|
||||
r.read_bits(32) # sar_width + sar_height
|
||||
if r.read_bits(1): # overscan_info_present_flag
|
||||
r.read_bits(1) # overscan_appropriate_flag
|
||||
if not r.read_bits(1): # video_signal_type_present_flag
|
||||
return None
|
||||
r.read_bits(3) # video_format
|
||||
r.read_bits(1) # video_full_range_flag
|
||||
if not r.read_bits(1): # colour_description_present_flag
|
||||
return None
|
||||
return r.read_bits(8), r.read_bits(8), r.read_bits(8)
|
||||
colour: Optional[tuple[int, int, int]] = None
|
||||
if r.read_bits(1): # video_signal_type_present_flag
|
||||
r.read_bits(4) # video_format + video_full_range_flag
|
||||
if r.read_bits(1): # colour_description_present_flag
|
||||
colour = (r.read_bits(8), r.read_bits(8), r.read_bits(8))
|
||||
fps: Optional[float] = None
|
||||
try:
|
||||
if r.read_bits(1): # chroma_loc_info_present_flag
|
||||
r.read_ue() # chroma_sample_loc_type_top_field
|
||||
r.read_ue() # chroma_sample_loc_type_bottom_field
|
||||
r.read_bits(3) # neutral_chroma_indication + field_seq + frame_field_info flags
|
||||
if r.read_bits(1): # default_display_window_flag
|
||||
for _ in range(4):
|
||||
r.read_ue() # def_disp_win offsets
|
||||
if r.read_bits(1): # vui_timing_info_present_flag
|
||||
num_units_in_tick = r.read_bits(32)
|
||||
time_scale = r.read_bits(32)
|
||||
if num_units_in_tick:
|
||||
fps = time_scale / num_units_in_tick
|
||||
except (IndexError, ValueError):
|
||||
pass # truncated VUI: keep whatever colour info was already read
|
||||
return colour, fps
|
||||
|
||||
|
||||
HEVC_FOURCCS = frozenset(("HVC1", "HEV1", "HEVC", "H265", "DVHE", "DVH1"))
|
||||
|
||||
|
||||
def parse_codec_private_data_colour(fourcc: str, codec_private_data: bytes) -> Optional[tuple[int, int, int]]:
|
||||
"""SPS VUI colour triple from HEVC CodecPrivateData; None when the codec
|
||||
is unsupported, no colour description, or malformed data."""
|
||||
def parse_codec_private_data_vui(
|
||||
fourcc: str, codec_private_data: bytes
|
||||
) -> tuple[Optional[tuple[int, int, int]], Optional[float]]:
|
||||
"""(colour triple, fps) from the SPS VUI in HEVC CodecPrivateData; (None, None)
|
||||
when the codec is unsupported or the data is malformed. H.264 is excluded
|
||||
on purpose: its VUI timing is field-based and often misdeclared, so fps
|
||||
read from it can't be trusted."""
|
||||
if (fourcc or "").upper() not in HEVC_FOURCCS:
|
||||
return None
|
||||
return None, None
|
||||
try:
|
||||
nals = split_nal_units(codec_private_data)
|
||||
sps = next((n for n in nals if (n[0] >> 1) & 0x3F == 33), None)
|
||||
if sps is None:
|
||||
return None
|
||||
return parse_hevc_sps_colour(remove_emulation_prevention(sps))
|
||||
return None, None
|
||||
return parse_hevc_sps_vui(remove_emulation_prevention(sps))
|
||||
except (IndexError, ValueError):
|
||||
return None
|
||||
return None, None
|
||||
|
||||
|
||||
def parse_codec_private_data_colour(fourcc: str, codec_private_data: bytes) -> Optional[tuple[int, int, int]]:
|
||||
"""SPS VUI colour triple from HEVC CodecPrivateData; None when the codec
|
||||
is unsupported, no colour description, or malformed data."""
|
||||
return parse_codec_private_data_vui(fourcc, codec_private_data)[0]
|
||||
|
||||
|
||||
def iter_boxes(data: bytes, start: int, end: int) -> Iterator[tuple[bytes, Optional[bytes], int, int]]:
|
||||
|
||||
Reference in New Issue
Block a user