From c06b00ec6386a6fda01c826dc82a61500c69ecbd Mon Sep 17 00:00:00 2001 From: imSp4rky Date: Wed, 23 Sep 2026 16:08:58 -0600 Subject: [PATCH] fix(import): read an HLS export that names no title language or master key Probe each media playlist for its KIDs and refuse a manifest unshackle cannot read. --- docs/guide/downloading.md | 4 + tests/core/test_import_foreign_hls.py | 144 ++++++++++++++++++++++++++ unshackle/core/import_service.py | 58 ++++++++++- 3 files changed, 204 insertions(+), 2 deletions(-) create mode 100644 tests/core/test_import_foreign_hls.py diff --git a/docs/guide/downloading.md b/docs/guide/downloading.md index 654555be..0c85c4c6 100644 --- a/docs/guide/downloading.md +++ b/docs/guide/downloading.md @@ -843,6 +843,10 @@ The [mediaexport](https://github.com/unshackle-dl/mediaexport) package specifies service. If a service never sets it, importing that export fails with a message naming the service. Neither end guesses a language for you, so the fix belongs in the service. + An HLS master playlist gives no language for a video track. When the export gives no + title language, the video track takes the language of the default audio track. When + the master playlist gives no audio language either, the import stops with a message. + ## Metadata and tagging unshackle looks up metadata automatically, but you can override the identifiers used for diff --git a/tests/core/test_import_foreign_hls.py b/tests/core/test_import_foreign_hls.py new file mode 100644 index 00000000..d07e2156 --- /dev/null +++ b/tests/core/test_import_foreign_hls.py @@ -0,0 +1,144 @@ +"""An HLS export from another tool: only a master playlist URL, title keys, and no track rows. + +The importer parses the master playlist for the tracks. Such a file can have no title +language, and its master playlist can name no key, so each track's own media playlist is +the only place that names its KIDs. +""" + +from __future__ import annotations + +import base64 +import json +import struct +from pathlib import Path +from types import SimpleNamespace +from typing import Any, Optional, cast +from unittest.mock import patch +from uuid import UUID + +import click +import pytest + +from unshackle.core.import_service import ImportService +from unshackle.core.manifests import HLS +from unshackle.core.tracks import Video + +MASTER_URL = "https://example.invalid/hls/playlist.m3u8?t=1" +VIDEO_KID = UUID("00000000-499d-c8c7-6332-202020202020") +AUDIO_KID = UUID("00000000-499d-c8c7-6336-202020202020") +OTHER_KID = UUID("11111111-2222-3333-4444-555555555555") +KEYS = {VIDEO_KID: "0000000000000000000000000000000a", AUDIO_KID: "0000000000000000000000000000000b"} +KEYS[OTHER_KID] = "0000000000000000000000000000000c" + + +def playready_object(kid: UUID) -> str: + """A base64 PlayReady object whose one WRM header names ``kid``.""" + value = base64.b64encode(kid.bytes_le).decode() + xml = ( + '' + f'' + ).encode("utf-16-le") + record = struct.pack(" str: + key = "" + if kid: + key = ( + '#EXT-X-KEY:METHOD=SAMPLE-AES,KEYFORMAT="com.microsoft.playready",KEYFORMATVERSIONS="1",' + f'URI="data:text/plain;charset=UTF-16;base64,{playready_object(kid)}"\n' + ) + return ( + "#EXTM3U\n#EXT-X-TARGETDURATION:6\n#EXT-X-PLAYLIST-TYPE:VOD\n" + f'#EXT-X-MAP:URI="init.mp4"\n{key}#EXTINF:6.0,\nseg0.m4s\n#EXT-X-ENDLIST\n' + ) + + +def master(audio_language: str = "en") -> str: + lang = f'LANGUAGE="{audio_language}",' if audio_language else "" + return ( + "#EXTM3U\n" + f'#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID="aud",{lang}NAME="Main",DEFAULT=YES,URI="audio.m3u8"\n' + '#EXT-X-MEDIA:TYPE=SUBTITLES,GROUP-ID="sub",LANGUAGE="en",NAME="English",URI="sub.m3u8"\n' + '#EXT-X-STREAM-INF:BANDWIDTH=6000000,RESOLUTION=1920x1080,CODECS="avc1.640028,ec-3",' + 'AUDIO="aud",SUBTITLES="sub"\nvideo.m3u8\n' + ) + + +def write_export(tmp_path: Path, language: str = "", manifest_url: str = MASTER_URL) -> Path: + """An export in the other tool's own format, as that tool writes it.""" + title: dict[str, Any] = {"id": "ep-1", "kind": "episode", "name": "Example Show", "season": 1, "episode": 1} + if language: + title["language"] = language + doc = { + "kind": "unidl-export", + "version": 1, + "service": "example", + "titles": [ + { + "save_name": "Example.Show.S01E01", + "title": title, + "keys": [f"{kid.hex}:{key}" for kid, key in KEYS.items()], + "manifest_url": manifest_url, + "headers": {"User-Agent": "Example/1.0"}, + "drm": {"system": "playready", "wrm_header": ""}, + } + ], + } + path = tmp_path / "export.json" + path.write_text(json.dumps(doc), encoding="utf8") + return path + + +def import_title(export: Path, master_text: str, playlists: dict[str, Optional[UUID]]) -> tuple[ImportService, Any]: + ctx = cast(click.Context, SimpleNamespace(parent=None, params={})) + svc = ImportService(ctx, "EXAMPLE", "ep-1", str(export)) + + def get(url: str, **_: Any) -> SimpleNamespace: + name = url.split("?")[0].rsplit("/", 1)[-1] + return SimpleNamespace(text=media_playlist(playlists.get(name)), raise_for_status=lambda: None) + + setattr(svc.session, "get", get) + episode = next(iter(svc.get_titles())) + with patch.object(HLS, "from_url", side_effect=lambda url, session=None, **kw: HLS.from_text(master_text, url)): + episode.tracks = svc.get_tracks(episode) + return svc, episode + + +def test_no_title_language_takes_the_default_audio_language(tmp_path: Path) -> None: + _, episode = import_title(write_export(tmp_path), master(), {}) + + assert [str(t.language) for t in episode.tracks.videos] == ["en"] + + +def test_no_language_anywhere_names_the_missing_language(tmp_path: Path) -> None: + with pytest.raises(click.ClickException, match="No language"): + import_title(write_export(tmp_path), master(audio_language=""), {}) + + +def test_each_hls_track_gets_the_keys_its_media_playlist_names(tmp_path: Path) -> None: + """The master playlist names no key, so only the media playlist tells the tracks apart.""" + svc, episode = import_title( + write_export(tmp_path, "en"), master(), {"video.m3u8": VIDEO_KID, "audio.m3u8": AUDIO_KID} + ) + svc.resolve_server_keys(episode) + + video, audio = episode.tracks.videos[0], episode.tracks.audio[0] + assert video.drm and video.drm[0].content_keys == {VIDEO_KID: KEYS[VIDEO_KID]} + assert audio.drm and audio.drm[0].content_keys == {AUDIO_KID: KEYS[AUDIO_KID]} + assert all(not t.drm for t in episode.tracks.subtitles) + + +def test_an_hls_track_whose_playlist_names_no_key_stays_clear(tmp_path: Path) -> None: + svc, episode = import_title(write_export(tmp_path, "en"), master(), {"video.m3u8": VIDEO_KID}) + svc.resolve_server_keys(episode) + + assert not episode.tracks.audio[0].drm + assert isinstance(episode.tracks.videos[0], Video) and episode.tracks.videos[0].drm + + +def test_a_manifest_that_unshackle_cannot_read_stops_the_import(tmp_path: Path) -> None: + """A manifest held only in the other tool's own form must not import as a title with no tracks.""" + with pytest.raises(click.ClickException, match="cannot read"): + import_title(write_export(tmp_path, "en", manifest_url="x-unidl:json_manifest"), master(), {}) diff --git a/unshackle/core/import_service.py b/unshackle/core/import_service.py index 5f8537c3..efba9b18 100644 --- a/unshackle/core/import_service.py +++ b/unshackle/core/import_service.py @@ -8,6 +8,7 @@ from typing import Any, Callable, Optional, TypeVar, Union from uuid import UUID import click +import m3u8 import mediaexport import requests from langcodes import tag_is_valid @@ -42,6 +43,43 @@ def fallback_language(language: Any) -> Optional[Any]: return language +def hls_audio_language(manifest: HLS) -> Optional[str]: + """The language of the default audio rendition, else of the first audio rendition, else None. + + An HLS master playlist gives no language for a video variant, so the parser needs a + fallback. An export from another tool can give no title language. + """ + audio = [m for m in manifest.manifest.media if m.type == "AUDIO" and fallback_language(m.language)] + audio.sort(key=lambda m: m.default != "YES") + return str(audio[0].language) if audio else None + + +def hls_playlist_kids(track: Any, session: Optional[requests.Session] = None) -> Optional[set[UUID]]: + """Return the KIDs that the Widevine and PlayReady keys of the track's media playlist name. + + Return None when the playlist has no key or only AES-128 keys: the download gets an AES-128 + key from its URI, so such a track takes no exported content key. An empty set means that + the track is encrypted but names no KID that unshackle can read. + """ + try: + text = (session or requests.Session()).get(str(track.url)).text + keys = [k for k in m3u8.loads(text, uri=str(track.url)).keys if k is not None and k.method != "NONE"] + except Exception as e: + log.warning(f"Cannot read the media playlist of {track.id} for its KIDs: {e!r}") + return None + kids: set[UUID] = set() + for key in keys: + if key.method == "AES-128": + continue + try: + kids.update(real_kids(HLS.get_drm(key))) + except Exception as e: + log.debug(f"Skipping HLS key {key.keyformat or key.method} of {track.id}: {e!r}") + if kids or any(k.method != "AES-128" for k in keys): + return kids + return None + + def resolve_import_manifest_data( tracks: Tracks, manifest_type: Optional[str], @@ -316,8 +354,16 @@ class ImportService: f"Failed to fetch the {manifest_type} manifest for '{title}'. " f"The manifest URL may have expired since export. ({e})" ) + language = fallback_language(title.language) + if language is None and isinstance(manifest, HLS): + language = hls_audio_language(manifest) + if language is None: + raise click.ClickException( + f"No language for '{title}': the export gives no title language and the HLS " + "manifest gives no audio language." + ) try: - parsed = manifest.to_tracks(language=fallback_language(title.language)) + parsed = manifest.to_tracks(language=language) except ValueError as e: if "Language information could not be derived" in str(e): raise click.ClickException( @@ -365,6 +411,11 @@ class ImportService: track.drm = drm tracks.add(track, warn_only=True) + if manifest_url and parser is None and not [*tracks.videos, *tracks.audio, *tracks.subtitles]: + raise click.ClickException( + f"Cannot import '{title}': unshackle cannot read its manifest {str(manifest_url)[:80]!r}." + ) + resolve_import_manifest_data( tracks, manifest_type, @@ -608,7 +659,8 @@ class ImportService: reads only ``track.drm``. A rung that misses content key injection here downloads encrypted and muxes without error. A DASH manifest with no ContentProtection can still describe an encrypted track, with the PSSH only in the init segment, so that case - probes the init segment with `session`. + probes the init segment with `session`. An HLS track with no DRM gets its KIDs from the + EXT-X-KEY tags of its media playlist, because a master playlist can name no KID. """ if track.drm: return {kid for drm in track.drm for kid in real_kids(drm)} @@ -642,6 +694,8 @@ class ImportService: manifest = ism.get("manifest") if manifest is not None and manifest.findall(".//ProtectionHeader"): return set() + if track.descriptor == Track.Descriptor.HLS and track.url: + return hls_playlist_kids(track, session) return None def exported_drm_system(self) -> str: