From bc501677919b4633c5c632034f4170fd991914ec Mon Sep 17 00:00:00 2001 From: imSp4rky Date: Wed, 23 Sep 2026 15:52:43 -0600 Subject: [PATCH] fix(import): give no title keys to a track that is clear A subtitle, or a track the export lists no KID for while it lists others, gets no DRM. --- tests/core/test_import_track_keys.py | 71 ++++++++++++++++++++++++++-- unshackle/core/import_service.py | 15 +++++- 2 files changed, 82 insertions(+), 4 deletions(-) diff --git a/tests/core/test_import_track_keys.py b/tests/core/test_import_track_keys.py index 65c870bb..f8c373c9 100644 --- a/tests/core/test_import_track_keys.py +++ b/tests/core/test_import_track_keys.py @@ -22,7 +22,7 @@ from unshackle.commands.dl import dl from unshackle.core.drm import Widevine from unshackle.core.drm.key_args import ZERO_KID from unshackle.core.import_service import ImportService -from unshackle.core.manifests import DASH +from unshackle.core.manifests import DASH, HLS from unshackle.core.titles import Movie from unshackle.core.tracks import Video @@ -179,7 +179,7 @@ def test_a_rebuilt_drm_holds_only_the_keys_for_the_kids_its_pssh_names(tmp_path: single: list[dict[str, Any]] = [widevine(VIDEO_KID, VIDEO_KEY).to_dict()] with patch.object(svc, "title_drm_dicts", return_value=single): - rebuilt = svc.rebuild_drm({"id": "u1"}, "movie-1") + rebuilt = svc.rebuild_drm({"id": "u1", "type": "Video"}, "movie-1") assert rebuilt and rebuilt[0].content_keys == {VIDEO_KID: VIDEO_KEY} @@ -189,9 +189,74 @@ def test_a_title_with_several_drm_entries_gives_no_track_a_guessed_own_key(tmp_p pool, and the zero-KID fallback stays empty instead of taking the first entry's key.""" svc = import_service(write_export(tmp_path, kids=False)) - rebuilt = svc.rebuild_drm({"id": "u1"}, "movie-1") + rebuilt = svc.rebuild_drm({"id": "u1", "type": "Video"}, "movie-1") assert rebuilt drm = rebuilt[0] assert drm.content_keys == POOL assert not [arg for arg in drm.mp4decrypt_key_args() if arg.startswith(ZERO_KID)] + + +HLS_URL = "https://example.invalid/master.m3u8" +HLS_MASTER = """#EXTM3U +#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID="aud",LANGUAGE="en",NAME="English",DEFAULT=YES,URI="audio_en.m3u8" +#EXT-X-MEDIA:TYPE=SUBTITLES,GROUP-ID="sub",LANGUAGE="es",NAME="Spanish",URI="sub_es.m3u8" +#EXT-X-STREAM-INF:BANDWIDTH=6000000,RESOLUTION=1920x1080,CODECS="avc1.640028,mp4a.40.2",AUDIO="aud",SUBTITLES="sub" +1080.m3u8 +""" +SUB_KID = UUID("0123456789abcdef0123456789abcdef") +SUB_KEY = "0000000000000000000000000000000c" + + +def hls_import( + tmp_path: Path, audio_drm: Widevine | None = None, subtitle_drm: Widevine | None = None, kids: bool = True +) -> dict[str, Any]: + """Export an HLS ladder with a licensed video track, import it, and return each track's DRM. + + HLS tracks come back from their exported dicts, and HLS.download_track decrypts every + track that holds a DRM. + """ + title = Movie(id_="movie-1", service=StubService, name="Example Movie", year=2024, language="en") + for track in HLS.from_text(HLS_MASTER, HLS_URL).to_tracks(language="en"): + title.tracks.add(track) + title.tracks.manifest_url = HLS_URL + runner = dl.__new__(dl) + runner.service = "EXAMPLE" + runner.log = logging.getLogger("download") + export = tmp_path / "export.json" + drm = {"video": widevine(VIDEO_KID, VIDEO_KEY), "en": audio_drm, "es": subtitle_drm} + for track in [*title.tracks.videos, *title.tracks.audio, *title.tracks.subtitles]: + runner.write_export(export, title, track, drm.get(label(track))) + if not kids: + doc = json.loads(export.read_text(encoding="utf8")) + for row in doc["titles"][0]["tracks"]: + row.pop("kids", None) + export.write_text(json.dumps(doc), encoding="utf8") + svc = import_service(export) + movie = next(iter(svc.get_titles())) + movie.tracks = svc.get_tracks(movie) + svc.resolve_server_keys(movie) + return {label(t): t.drm for t in movie.tracks} + + +def test_a_clear_subtitle_gets_no_drm_from_the_key_pool(tmp_path: Path) -> None: + """An export from before the per-track KID lists cannot tell a clear video or audio track + from an encrypted one, so those keep the pool. A subtitle is clear.""" + drm = hls_import(tmp_path, audio_drm=widevine(AUDIO_KID, AUDIO_KEY), kids=False) + + assert drm["es"] is None, "a clear subtitle that holds a DRM goes to a decrypter that empties it" + assert drm["video"] and drm["video"][0].content_keys == POOL, "a video that declares no KID keeps the pool" + + +def test_a_keyed_video_leaves_a_clear_audio_track_clear(tmp_path: Path) -> None: + drm = hls_import(tmp_path) + + assert drm["video"] and drm["video"][0].content_keys == {VIDEO_KID: VIDEO_KEY} + assert drm["en"] is None, "the export lists KIDs for its encrypted tracks, so an audio track with none is clear" + assert drm["es"] is None + + +def test_a_subtitle_that_lists_its_kid_keeps_its_key(tmp_path: Path) -> None: + drm = hls_import(tmp_path, subtitle_drm=widevine(SUB_KID, SUB_KEY)) + + assert drm["es"] and drm["es"][0].content_keys == {SUB_KID: SUB_KEY} diff --git a/unshackle/core/import_service.py b/unshackle/core/import_service.py index 21ec341c..5f8537c3 100644 --- a/unshackle/core/import_service.py +++ b/unshackle/core/import_service.py @@ -454,7 +454,7 @@ class ImportService: except (TypeError, ValueError, NotImplementedError) as e: self.log.warning(f"Skipping exported track {row.get('id')!r}: {e}") continue - track.drm = self.rebuild_drm({"id": track_id}, title_id) + track.drm = self.rebuild_drm({"id": track_id, "type": type(track).__name__}, title_id) out.append(track) return out @@ -489,6 +489,11 @@ class ImportService: row = next((r for r in (entry.tracks if entry else []) if str(r.get("id", "")) == track_id), None) return set(row["kids"]) if row and row.get("kids") else None + def lists_track_kids(self, title_id: str) -> bool: + """Return True when the export lists the KIDs of at least one track of the title.""" + entry = self.doc.get(title_id) + return any(row.get("kids") for row in (entry.tracks if entry else [])) + def exported_keys(self, title_id: str, track_id: str) -> dict[UUID, str]: """Return the exported content keys for the KIDs the row of the track lists, else all of the title.""" entry = self.doc.get(title_id) @@ -503,10 +508,18 @@ class ImportService: list, a title with more than one DRM entry does not say which entry is this track's, so the track gets a stub that names every exported KID instead of a guess. A rebuilt PSSH holds only the keys for the KIDs it names. + + A track that declares no content key, DRM or KID of its own is clear if it is a subtitle, + or if the export lists the KIDs of its other tracks. Only a video or audio track in an + export with no KID lists takes the title keys, because that export cannot tell. """ own_keys = track_dict.get("keys") own_drm = track_dict.get("drm") row_kids = None if own_keys else self.exported_kids(title_id, str(track_dict.get("id", ""))) + if not (own_keys or own_drm or row_kids) and ( + track_dict.get("type") not in ("Video", "Audio") or self.lists_track_kids(title_id) + ): + return None if own_keys: keys = own_keys else: