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).

tests/core/test_cdm_detect.py

@@ -0,0 +1,201 @@
+"""CDM detection and load-time DRM stamping for local and remote CDMs.
+
+Covers the predicates in ``unshackle.core.cdm.detect`` (playready/widevine, local/remote,
+and the remote combinations) and the loader stamping that sets a remote CDM's DRM type at
+load time instead of guessing it on each call. Services such as AMZN route to a device
+profile off that type, so a wrong classification here is what produces a Downgrade.Hd
+license denial.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from unshackle.core.cdm import loader as loader_mod
+from unshackle.core.cdm.detect import (cdm_location, is_local_cdm, is_playready_cdm, is_remote_cdm,
+                                       is_remote_playready_cdm, is_remote_widevine_cdm, is_widevine_cdm)
+from unshackle.core.cdm.loader import _stamp_remote
+
+pytestmark = pytest.mark.unit
+
+REPO = Path(__file__).resolve().parents[2]
+PRD_DIR = REPO / "unshackle" / "PRDs"
+
+
+def test_stamped_remote_playready():
+    cdm = SimpleNamespace(drm="playready", is_remote_cdm=True)
+    assert is_playready_cdm(cdm)
+    assert not is_widevine_cdm(cdm)
+    assert is_remote_cdm(cdm)
+    assert is_remote_playready_cdm(cdm)
+    assert not is_remote_widevine_cdm(cdm)
+    assert cdm_location(cdm) == "remote"
+
+
+def test_stamped_remote_widevine():
+    cdm = SimpleNamespace(drm="widevine", is_remote_cdm=True)
+    assert is_widevine_cdm(cdm)
+    assert not is_playready_cdm(cdm)
+    assert is_remote_widevine_cdm(cdm)
+    assert not is_remote_playready_cdm(cdm)
+    assert cdm_location(cdm) == "remote"
+
+
+def test_drm_stamp_beats_conflicting_is_playready_attr():
+    # The stamped drm takes priority over the wrapper's own is_playready guess.
+    cdm = SimpleNamespace(drm="playready", is_playready=False, is_remote_cdm=True)
+    assert is_playready_cdm(cdm)
+    assert not is_widevine_cdm(cdm)
+
+
+def test_drm_stamp_case_insensitive():
+    cdm = SimpleNamespace(drm="PlayReady", is_remote_cdm=True)
+    assert is_playready_cdm(cdm)
+
+
+def test_none_classifies_as_nothing():
+    for fn in (
+        is_playready_cdm,
+        is_widevine_cdm,
+        is_remote_cdm,
+        is_local_cdm,
+        is_remote_playready_cdm,
+        is_remote_widevine_cdm,
+    ):
+        assert fn(None) is False
+    assert cdm_location(None) == "unknown"
+
+
+@pytest.mark.parametrize("device_name", ["SL2", "SL3", "SL2000", "SL3000", "sl3000"])
+def test_decryptlabs_sl_names_are_playready(device_name):
+    from unshackle.core.cdm.decrypt_labs_remote_cdm import DecryptLabsRemoteCDM
+
+    cdm = DecryptLabsRemoteCDM(secret="x", device_name=device_name)
+    assert cdm.is_playready
+    assert is_playready_cdm(cdm)
+    assert is_remote_cdm(cdm)
+    assert is_remote_playready_cdm(cdm)
+    assert not is_local_cdm(cdm)
+
+
+@pytest.mark.parametrize("device_name", ["ChromeCDM", "L1", "L2"])
+def test_decryptlabs_widevine_names(device_name):
+    from unshackle.core.cdm.decrypt_labs_remote_cdm import DecryptLabsRemoteCDM
+
+    cdm = DecryptLabsRemoteCDM(secret="x", device_name=device_name)
+    assert not cdm.is_playready
+    assert is_widevine_cdm(cdm)
+    assert is_remote_widevine_cdm(cdm)
+
+
+def test_decryptlabs_device_type_forces_playready():
+    from unshackle.core.cdm.decrypt_labs_remote_cdm import DecryptLabsRemoteCDM
+
+    cdm = DecryptLabsRemoteCDM(secret="x", device_name="ChromeCDM", device_type="PLAYREADY")
+    assert cdm.is_playready
+
+
+@pytest.mark.parametrize(
+    ("device", "expect_pr"),
+    [
+        ({"name": "SL3000"}, True),
+        ({"name": "SL2"}, True),
+        ({"name": "ChromeCDM"}, False),
+        ({"type": "PLAYREADY", "name": "ChromeCDM"}, True),
+    ],
+)
+def test_custom_remote_cdm_classification(device, expect_pr):
+    from unshackle.core.cdm.custom_remote_cdm import CustomRemoteCDM
+
+    cdm = CustomRemoteCDM(host="http://example.invalid", device=device)
+    assert cdm.is_playready is expect_pr
+    assert is_playready_cdm(cdm) is expect_pr
+    assert is_remote_cdm(cdm)
+
+
+def test_stamp_remote_sets_attrs_and_returns_same_object():
+    obj = SimpleNamespace()
+    out = _stamp_remote(obj, "playready")
+    assert out is obj
+    assert obj.drm == "playready"
+    assert obj.is_remote_cdm is True
+
+
+def test_stamp_remote_is_best_effort_on_unsettable_object():
+    class Slotted:
+        __slots__ = ()
+
+    obj = Slotted()
+    out = _stamp_remote(obj, "widevine")  # must not raise
+    assert out is obj
+    assert not hasattr(obj, "drm")
+
+
+def test_load_remote_decryptlabs_playready_is_stamped():
+    cdm_api = {"name": "dl-pr", "type": "decrypt_labs", "secret": "x", "device_name": "SL3000"}
+    cdm = loader_mod._load_remote_cdm(dict(cdm_api), "dl-pr", "AMZN", None)
+    assert cdm.drm == "playready"
+    assert cdm.is_remote_cdm is True
+    assert is_remote_playready_cdm(cdm)
+
+
+def test_load_remote_decryptlabs_widevine_is_stamped():
+    cdm_api = {"name": "dl-wv", "type": "decrypt_labs", "secret": "x", "device_name": "ChromeCDM"}
+    cdm = loader_mod._load_remote_cdm(dict(cdm_api), "dl-wv", "AMZN", None)
+    assert cdm.drm == "widevine"
+    assert is_remote_widevine_cdm(cdm)
+
+
+def test_load_remote_native_playready_is_stamped(monkeypatch):
+    import pyplayready.remote.remotecdm as prmod
+
+    class FakePlayReadyRemote:
+        def __init__(self, **kwargs):
+            self.kwargs = kwargs
+
+    monkeypatch.setattr(prmod, "RemoteCdm", FakePlayReadyRemote)
+    cdm_api = {"name": "pr", "Device Type": "PLAYREADY", "host": "h", "secret": "s", "device_name": "d"}
+    cdm = loader_mod._load_remote_cdm(dict(cdm_api), "pr", "AMZN", None)
+    assert isinstance(cdm, FakePlayReadyRemote)
+    assert cdm.drm == "playready"
+    assert cdm.is_remote_cdm is True
+    assert is_remote_playready_cdm(cdm)
+
+
+def test_load_remote_native_widevine_is_stamped(monkeypatch):
+    import pywidevine.remotecdm as wvmod
+
+    class FakeWidevineRemote:
+        def __init__(self, **kwargs):
+            self.kwargs = kwargs
+
+    monkeypatch.setattr(wvmod, "RemoteCdm", FakeWidevineRemote)
+    cdm_api = {"name": "wv", "Device Type": "ANDROID", "host": "h", "secret": "s", "device_name": "d"}
+    cdm = loader_mod._load_remote_cdm(dict(cdm_api), "wv", "AMZN", None)
+    assert isinstance(cdm, FakeWidevineRemote)
+    assert cdm.drm == "widevine"
+    assert is_remote_widevine_cdm(cdm)
+
+
+@pytest.fixture
+def local_playready_cdm():
+    prds = sorted(PRD_DIR.glob("*.prd"))
+    if not prds:
+        pytest.skip(f"no local .prd present in {PRD_DIR}")
+    from pyplayready.cdm import Cdm
+    from pyplayready.device import Device
+
+    return Cdm.from_device(Device.load(str(prds[0])))
+
+
+def test_local_playready_is_playready_and_local(local_playready_cdm):
+    cdm = local_playready_cdm
+    assert is_playready_cdm(cdm)
+    assert not is_widevine_cdm(cdm)
+    assert is_local_cdm(cdm)
+    assert not is_remote_cdm(cdm)
+    assert not is_remote_playready_cdm(cdm)
+    assert cdm_location(cdm) == "local"

tests/core/test_ism_init.py

@@ -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))

tests/core/test_ism_parse.py

@@ -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()

unshackle/core/cdm/custom_remote_cdm.py

@@ -152,7 +152,9 @@ class CustomRemoteCDM:
         self.security_level = device.get("security_level", 3)
 
         # Determine if this is a PlayReady CDM
-        self._is_playready = self.device_type_str.upper() == "PLAYREADY" or self.device_name in ["SL2", "SL3"]
+        self._is_playready = self.device_type_str.upper() == "PLAYREADY" or (
+            bool(self.device_name) and self.device_name.upper().startswith("SL")
+        )
 
         # Get device type enum for compatibility
         if self.device_type_str:

unshackle/core/cdm/decrypt_labs_remote_cdm.py

@@ -134,7 +134,9 @@ class DecryptLabsRemoteCDM:
         if device_type:
             self.device_type = self._get_device_type_enum(device_type)
 
-        self._is_playready = (device_type and device_type.upper() == "PLAYREADY") or (device_name in ["SL2", "SL3"])
+        self._is_playready = (device_type and device_type.upper() == "PLAYREADY") or (
+            bool(device_name) and device_name.upper().startswith("SL")
+        )
 
         if self._is_playready:
             self.system_id = system_id or 0

unshackle/core/cdm/detect.py

@@ -112,6 +112,10 @@ def is_playready_cdm(cdm: Any) -> bool:
     if cdm is None:
         return False
 
+    drm = getattr(cdm, "drm", None)
+    if drm:
+        return str(drm).lower() == "playready"
+
     if hasattr(cdm, "is_playready"):
         try:
             return bool(getattr(cdm, "is_playready"))
@@ -144,6 +148,18 @@ def is_playready_cdm(cdm: Any) -> bool:
     return "pyplayready" in mod
 
 
+def is_remote_playready_cdm(cdm: Any) -> bool:
+    """Return True if the CDM is a remote PlayReady CDM (as opposed to a local .prd)."""
+
+    return is_playready_cdm(cdm) and is_remote_cdm(cdm)
+
+
+def is_remote_widevine_cdm(cdm: Any) -> bool:
+    """Return True if the CDM is a remote Widevine CDM (as opposed to a local .wvd)."""
+
+    return is_widevine_cdm(cdm) and is_remote_cdm(cdm)
+
+
 def is_widevine_cdm(cdm: Any) -> bool:
     """
     Return True if the given CDM should be treated as Widevine.
@@ -155,6 +171,10 @@ def is_widevine_cdm(cdm: Any) -> bool:
     if cdm is None:
         return False
 
+    drm = getattr(cdm, "drm", None)
+    if drm:
+        return str(drm).lower() == "widevine"
+
     if hasattr(cdm, "is_playready"):
         try:
             return not bool(getattr(cdm, "is_playready"))

unshackle/core/cdm/loader.py

@@ -13,6 +13,22 @@ from typing import Any, Optional
 log = logging.getLogger("cdm")
 
 
+def _stamp_remote(cdm: Any, drm: str) -> Any:
+    """Record a remote CDM's DRM type and mark it as remote.
+
+    The loader knows the DRM type from config, so it sets it here as the value
+    `detect.py` reads before it falls back to class or module checks. Setting the
+    attributes is best effort: a third-party CDM object that rejects them still
+    works, since detect.py keeps its existing checks.
+    """
+    try:
+        cdm.drm = drm
+        cdm.is_remote_cdm = True
+    except (AttributeError, TypeError):
+        pass
+    return cdm
+
+
 def load_cdm(
     cdm_name: str,
     *,
@@ -63,35 +79,43 @@ def _load_remote_cdm(
                     f"No secret provided for DecryptLabs CDM '{cdm_name}' and no global decrypt_labs_api_key configured"
                 )
 
-        return DecryptLabsRemoteCDM(service_name=service_name, vaults=vaults, **cdm_api)
+        cdm: Any = DecryptLabsRemoteCDM(service_name=service_name, vaults=vaults, **cdm_api)
+        return _stamp_remote(cdm, "playready" if cdm.is_playready else "widevine")
 
     if cdm_type == "custom_api":
         from unshackle.core.cdm.custom_remote_cdm import CustomRemoteCDM
 
         del cdm_api["name"]
         del cdm_api["type"]
-        return CustomRemoteCDM(service_name=service_name, vaults=vaults, **cdm_api)
+        cdm = CustomRemoteCDM(service_name=service_name, vaults=vaults, **cdm_api)
+        return _stamp_remote(cdm, "playready" if cdm.is_playready else "widevine")
 
     device_type = cdm_api.get("Device Type", cdm_api.get("device_type", ""))
     if str(device_type).upper() == "PLAYREADY":
         from pyplayready.remote.remotecdm import RemoteCdm as PlayReadyRemoteCdm
 
-        return PlayReadyRemoteCdm(
-            security_level=cdm_api.get("Security Level", cdm_api.get("security_level", 3000)),
-            host=cdm_api.get("Host", cdm_api.get("host")),
-            secret=cdm_api.get("Secret", cdm_api.get("secret")),
-            device_name=cdm_api.get("Device Name", cdm_api.get("device_name")),
+        return _stamp_remote(
+            PlayReadyRemoteCdm(
+                security_level=cdm_api.get("Security Level", cdm_api.get("security_level", 3000)),
+                host=cdm_api.get("Host", cdm_api.get("host")),
+                secret=cdm_api.get("Secret", cdm_api.get("secret")),
+                device_name=cdm_api.get("Device Name", cdm_api.get("device_name")),
+            ),
+            "playready",
         )
 
     from pywidevine.remotecdm import RemoteCdm
 
-    return RemoteCdm(
-        device_type=cdm_api.get("Device Type", cdm_api.get("device_type", "")),
-        system_id=cdm_api.get("System ID", cdm_api.get("system_id", "")),
-        security_level=cdm_api.get("Security Level", cdm_api.get("security_level", 3000)),
-        host=cdm_api.get("Host", cdm_api.get("host")),
-        secret=cdm_api.get("Secret", cdm_api.get("secret")),
-        device_name=cdm_api.get("Device Name", cdm_api.get("device_name")),
+    return _stamp_remote(
+        RemoteCdm(
+            device_type=cdm_api.get("Device Type", cdm_api.get("device_type", "")),
+            system_id=cdm_api.get("System ID", cdm_api.get("system_id", "")),
+            security_level=cdm_api.get("Security Level", cdm_api.get("security_level", 3000)),
+            host=cdm_api.get("Host", cdm_api.get("host")),
+            secret=cdm_api.get("Secret", cdm_api.get("secret")),
+            device_name=cdm_api.get("Device Name", cdm_api.get("device_name")),
+        ),
+        "widevine",
     )
 
 

unshackle/core/manifests/ism.py

@@ -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(

unshackle/core/manifests/ism_init.py

@@ -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]]: