From 9f054269dbbbe2862dc07236fc234e979486bb00 Mon Sep 17 00:00:00 2001 From: imSp4rky Date: Fri, 23 Oct 2026 16:59:40 -0600 Subject: [PATCH] fix(vaults): copy no vault key from a clear track kid_windows returns an empty map for an MP4 with no encrypted sample entry or protecting senc, so the key check judges, flags and copies nothing for it. --- docs/guide/vaults.md | 2 +- tests/core/test_kid_windows.py | 65 +++++++++++++++++++++++++++++-- tests/core/test_vault_bad_keys.py | 26 +++++++++++++ unshackle/commands/dl.py | 17 ++++---- unshackle/core/drm/verify.py | 48 ++++++++++++++++++----- unshackle/core/tracks/track.py | 16 +------- 6 files changed, 136 insertions(+), 38 deletions(-) diff --git a/docs/guide/vaults.md b/docs/guide/vaults.md index ffa8cf27..331acc38 100644 --- a/docs/guide/vaults.md +++ b/docs/guide/vaults.md @@ -34,7 +34,7 @@ unshackle pushes keys from every source, not only a local CDM. A run against a r ### Wrong keys from a poisoned vault -A public vault can return the right KID with a wrong content key. The decrypters accept a wrong content key and write noise. unshackle therefore trusts a content key from a vault only after FFmpeg decodes the decrypted track with strict error detection. For a fragmented MP4, unshackle reads from the encrypted file which fragments each KID encrypts, and FFmpeg decodes up to three windows of each KID: from its first, middle and last fragment on through the next fragments of that KID, for up to 4 seconds. A window never runs into a fragment of another KID or a clear fragment. A clear lead or a lead under another KID therefore cannot hide a wrong content key, and only the KID that fails is flagged. When every KID in the map has a content key, the map is complete, and a KID that is not in it does not encrypt this file. This file does not judge that KID; the file of its own track judges it. A KID that is not in an incomplete map, for example one with an all-zero KID, and every KID of any other file, gets one check of the whole file. FFmpeg decodes the first seconds and the last seconds. A failure there flags each of those KIDs. When a mapped KID fails, the KIDs that are not in the map get no verdict from that pass; the retry with the replaced key judges them. Content keys from a licence skip this check. A file this FFmpeg build cannot judge counts as a pass, never as a wrong content key: a codec it has no decoder for, or a header it cannot read. A pass is not proof. A wrong content key for HEVC video can decode without an error for some seconds, most often at a low resolution, and that content key passes the check. Without FFmpeg there is no check at all. +A public vault can return the right KID with a wrong content key. The decrypters accept a wrong content key and write noise. unshackle therefore trusts a content key from a vault only after FFmpeg decodes the decrypted track with strict error detection. For a fragmented MP4, unshackle reads from the encrypted file which fragments each KID encrypts, and FFmpeg decodes up to three windows of each KID: from its first, middle and last fragment on through the next fragments of that KID, for up to 4 seconds. A window never runs into a fragment of another KID or a clear fragment. A clear lead or a lead under another KID therefore cannot hide a wrong content key, and only the KID that fails is flagged. When every KID in the map has a content key, the map is complete, and a KID that is not in it does not encrypt this file. This file does not judge that KID; the file of its own track judges it. A file with no encrypted sample entry and no encrypted fragment, for example a clear audio track, judges no KID, so it neither flags nor copies a content key. A KID that is not in an incomplete map, for example one with an all-zero KID, and every KID of any other file, gets one check of the whole file. FFmpeg decodes the first seconds and the last seconds. A failure there flags each of those KIDs. When a mapped KID fails, the KIDs that are not in the map get no verdict from that pass; the retry with the replaced key judges them. Content keys from a licence skip this check. A file this FFmpeg build cannot judge counts as a pass, never as a wrong content key: a codec it has no decoder for, or a header it cannot read. A pass is not proof. A wrong content key for HEVC video can decode without an error for some seconds, most often at a low resolution, and that content key passes the check. Without FFmpeg there is no check at all. When the decode fails, unshackle: diff --git a/tests/core/test_kid_windows.py b/tests/core/test_kid_windows.py index 6509f3ba..25218c48 100644 --- a/tests/core/test_kid_windows.py +++ b/tests/core/test_kid_windows.py @@ -3,6 +3,8 @@ from pathlib import Path from typing import Optional from uuid import UUID +import pytest + from unshackle.core.drm.verify import PIFF_TENC, PIFF_TFXD, KidMap, kid_windows, pick_windows A, B, C = UUID(int=0xA), UUID(int=0xB), UUID(int=0xC) @@ -44,15 +46,28 @@ def sbgp(*runs: tuple[int, int]) -> bytes: return full(b"sbgp", b"seig" + struct.pack(">I", len(runs)) + b"".join(struct.pack(">II", *run) for run in runs)) -def movie(*sample_entries: bytes, handler: bytes = b"vide", stbl: bytes = b"") -> bytes: +def trak(*sample_entries: bytes, track: int = 1, handler: bytes = b"vide", stbl: bytes = b"") -> bytes: stsd = full(b"stsd", struct.pack(">I", len(sample_entries)) + b"".join(sample_entries)) hdlr = full(b"hdlr", bytes(4) + handler + bytes(12)) - trak = box( + return box( b"trak", - full(b"tkhd", bytes(8) + struct.pack(">I", 1)), + full(b"tkhd", bytes(8) + struct.pack(">I", track)), box(b"mdia", full(b"mdhd", bytes(8) + struct.pack(">I", 1000)), hdlr, box(b"minf", box(b"stbl", stsd, stbl))), ) - return box(b"ftyp", b"iso6") + box(b"moov", trak, box(b"mvex", full(b"trex", struct.pack(">III", 1, 1, 0)))) + + +def movie(*sample_entries: bytes, handler: bytes = b"vide", stbl: bytes = b"", others: bytes = b"") -> bytes: + moov = box( + b"moov", + trak(*sample_entries, handler=handler, stbl=stbl), + others, + box(b"mvex", full(b"trex", struct.pack(">III", 1, 1, 0))), + ) + return box(b"ftyp", b"iso6") + moov + + +def clear_audio(others: bytes = b"") -> bytes: + return movie(box(b"mp4a", bytes(28)), handler=b"soun", others=others) def moof( @@ -178,3 +193,45 @@ def test_not_a_fragmented_mp4(tmp_path: Path) -> None: path = tmp_path / "video.ts" path.write_bytes(b"\x47" * 188 * 4) assert kid_windows(path) is None + + +def test_clear_mp4_maps_no_kid(tmp_path: Path) -> None: + path = tmp_path / "clear.m4a" + path.write_bytes(clear_audio() + moof(1, 0)) + assert kid_windows(path) == KidMap({}, False) + + +def test_encrypted_entry_without_a_tenc_is_not_clear(tmp_path: Path) -> None: + path = tmp_path / "enc.mp4" + path.write_bytes(movie(box(b"encv", bytes(78), box(b"sinf", box(b"frma", b"avc1")))) + moof(1, 0)) + assert kid_windows(path) is None + + +@pytest.mark.parametrize("kind", [b"enct", b"encs", b"drmi"]) +def test_other_encrypted_entry_kinds_are_not_clear(tmp_path: Path, kind: bytes) -> None: + path = tmp_path / "enc.mp4" + path.write_bytes(movie(box(kind, bytes(16)), handler=b"text") + moof(1, 0)) + assert kid_windows(path) is None + + +def test_a_clear_entry_with_a_protecting_senc_is_not_clear(tmp_path: Path) -> None: + path = tmp_path / "enc.m4a" + senc = full(b"senc", struct.pack(">I", 1) + bytes(8)) + path.write_bytes(clear_audio() + moof(1, 0, senc)) + assert kid_windows(path) is None + path.write_bytes(clear_audio() + moof(1, 0, full(b"senc", bytes(4)))) + assert kid_windows(path) == KidMap({}, False) + + +def test_only_the_tracks_with_fragments_decide_clear(tmp_path: Path) -> None: + path = tmp_path / "clear.m4a" + path.write_bytes(clear_audio(others=trak(encv(A), track=2)) + moof(1, 0)) + assert kid_windows(path) == KidMap({}, False) + path.write_bytes(clear_audio(others=trak(encv(A), track=2))) + assert kid_windows(path) is None + + +def test_a_clear_lead_beside_an_encrypted_entry_is_not_clear(tmp_path: Path) -> None: + path = tmp_path / "lead.mp4" + path.write_bytes(movie(box(b"avc1", bytes(78)), encv(A)) + moof(1, 0)) + assert kid_windows(path) is None diff --git a/tests/core/test_vault_bad_keys.py b/tests/core/test_vault_bad_keys.py index 5a5b3530..573eefb7 100644 --- a/tests/core/test_vault_bad_keys.py +++ b/tests/core/test_vault_bad_keys.py @@ -639,6 +639,32 @@ def test_only_checked_kids_are_copied_to_other_vaults(tmp_path: Path, monkeypatc assert local.get_key(KID, "SVC") is None +def test_a_clear_file_judges_and_copies_no_vault_key(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + local = SQLite("local", tmp_path / "local.db") + remote = Remote("remote", tmp_path / "remote.db") + remote.add_key("SVC", KID, BAD) + vaults = Vaults("SVC") + vaults.vaults = [local, remote] + cmd = make_cmd(monkeypatch, vaults) + drm = FakeDRM({KID: vaults.get_key(KID)[0]}) + checks: list = [] + + def decodes(path: Path, start: object = None, seconds: float = 3, video: object = None) -> bool: + checks.append(start) + return True + + monkeypatch.setattr(dl_module.verify, "kid_windows", lambda path: dl_module.verify.KidMap({}, False)) + monkeypatch.setattr(dl_module, "ffmpeg_decodes", decodes) + path = tmp_path / "audio.mp4" + path.write_bytes(b"clear") + + cmd.decrypt_verified(drm, path, None, KID) + cmd.wait_vault_writes() + assert checks == [] + assert local.get_key(KID, "SVC") is None + assert not local.is_bad_key(KID, BAD) + + def test_flagged_pair_warns_once_per_vault(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: local = SQLite("local", tmp_path / "local.db") poisoned = Remote("poisoned", tmp_path / "poisoned.db") diff --git a/unshackle/commands/dl.py b/unshackle/commands/dl.py index bbee91c3..7ee5441b 100644 --- a/unshackle/commands/dl.py +++ b/unshackle/commands/dl.py @@ -4508,13 +4508,14 @@ class dl: neither happens. For a fragmented MP4, ``verify.kid_windows`` reads from the ciphertext which fragments each KID encrypts, and FFmpeg decodes windows inside each KID's fragments, so only a KID that fails is flagged. A complete map does not judge a KID it does - not hold. A KID that is not in an incomplete map, and every KID of any other file, gets a - start and end check of the whole file, and a failure there flags each of those KIDs. On - failure the pair is flagged in the local vaults, the ciphertext is restored, and the - licence runs again; ``Vaults`` tells the vault that served it. Every vault skips a - flagged pair, so each pass burns at most one vault; when the vaults run out the CDM - answers, and a CDM key is not checked. A CDM that returns the flagged key clears the - flag and stores the content key again, because then the decode check was wrong. + not hold, so a clear file judges and copies no KID. A KID that is not in an incomplete map, + and every KID of any other file, gets a start and end check of the whole file, and a + failure there flags each of those KIDs. On failure the pair is flagged in the local + vaults, the ciphertext is restored, and the licence runs again; ``Vaults`` tells the + vault that served it. Every vault skips a flagged pair, so each pass burns at most one + vault; when the vaults run out the CDM answers, and a CDM key is not checked. A CDM + that returns the flagged key clears the flag and stores the content key again, because + then the decode check was wrong. With ``decrypt=False`` the downloader already decrypted the segments in place, so a failure can only flag and raise; the next run skips the flagged pair. @@ -4578,7 +4579,7 @@ class dl: def judged_kids() -> dict[UUID, Vault]: """The vault KIDs this file can judge. A map whose KIDs all have keys skips the KIDs it lacks.""" kids = vault_kids() - if windows and all(kid in keys for kid in windows): + if kid_map is not None and all(kid in keys for kid in windows): return {kid: vault for kid, vault in kids.items() if kid in windows} return kids diff --git a/unshackle/core/drm/verify.py b/unshackle/core/drm/verify.py index 38d14651..f39b6db2 100644 --- a/unshackle/core/drm/verify.py +++ b/unshackle/core/drm/verify.py @@ -9,7 +9,9 @@ DECRYPT_HOOK: Optional[Callable[..., None]] = None CONTAINERS = {b"moov", b"trak", b"mdia", b"minf", b"stbl", b"mvex", b"moof", b"traf", b"sinf", b"schi"} PIFF_TENC = bytes.fromhex("8974dbce7be74c5184f97148f9882554") PIFF_TFXD = bytes.fromhex("6d1d9b0542d544e680e2141daff757b2") +PIFF_SENC = bytes.fromhex("a2394f525a9b4f14a2446c427c648df4") TENC = {b"tenc", b"uuid" + PIFF_TENC} +SENC = {b"senc", b"uuid" + PIFF_SENC} # the fixed sample entry fields before the child boxes: visual; sound v0 (and ISO v1), QuickTime sound v1 and v2 ENTRY_FIELDS = {b"encv": (78,), b"enca": (28, 44, 64)} @@ -123,9 +125,26 @@ def entry_kid(data: bytes, kind: bytes, entry: int, end: int) -> Optional[UUID]: return None -def sample_entries(data: bytes, body: int, end: int) -> list[Optional[UUID]]: - """Parse an ``stsd`` box. Returns the default KID of each sample entry, None for a clear one.""" - return [entry_kid(data, kind, entry, entry_end) for kind, entry, entry_end in children(data, body + 8, end)] +def sample_entries(data: bytes, body: int, end: int) -> tuple[list[Optional[UUID]], bool]: + """Parse an ``stsd`` box. Returns the default KID of each sample entry (None for a clear one), and + whether any entry is encrypted.""" + found = list(children(data, body + 8, end)) + kids = [entry_kid(data, kind, entry, entry_end) for kind, entry, entry_end in found] + return kids, any(kind[:3] in (b"enc", b"drm") for kind, _, _ in found) + + +def senc_protects_samples(buf: bytes, body: int, end: int) -> bool: + """True if a senc/PIFF-uuid box describes protected samples. + + An empty senc (sample_count 0) sits on a genuinely clear fragment, so treating it as + a skipped one would raise on a healthy file. Field order follows + read_per_sample_iv_size. + """ + flags = int.from_bytes(buf[body + 1 : body + 4], "big") + pos = body + 4 + (20 if flags & 0x1 else 0) # PIFF override: AlgorithmID(3)+IV_size(1)+KID(16) + if pos + 4 > end: + return False + return int.from_bytes(buf[pos : pos + 4], "big") > 0 def read_top_boxes(f: BinaryIO) -> Iterator[tuple[bytes, Optional[bytes]]]: @@ -164,8 +183,10 @@ def kid_windows(path: Path) -> Optional[KidMap]: the result holds up to three (start, end) windows from ``pick_windows``, relative to the first fragment of the track. A fragment ends where the next fragment of the track starts. The last fragment ends after its sample durations, or after 3 seconds when - they are not known. Returns None when the file has no encrypted fragment this reader - can place in time, so the caller must use another check. + they are not known. Returns an empty map for a clear MP4: no track with fragments (or, + without fragments, no track) has an encrypted sample entry or a senc. Returns None when + the file has no encrypted fragment this reader can place in time, so the caller must + use another check; that includes an encrypted sample entry whose fragments are all clear. """ timescales: dict[int, int] = {} handlers: dict[int, bytes] = {} @@ -174,6 +195,7 @@ def kid_windows(path: Path) -> Optional[KidMap]: moov_groups: dict[int, list[Seig]] = {} fragments: list[Fragment] = [] clock: dict[int, int] = {} + protected: set[int] = set() try: with path.open("rb") as f: for kind, box in read_top_boxes(f): @@ -197,13 +219,15 @@ def kid_windows(path: Path) -> Optional[KidMap]: # the mdia handler comes first; a QuickTime minf can hold a data handler after it handlers.setdefault(track, box[body + 8 : body + 12]) elif kind_ == b"stsd": - entries[track] = sample_entries(box, body, box_end) + entries[track], encrypted = sample_entries(box, body, box_end) + if encrypted: + protected.add(track) elif kind_ == b"sgpd": groups = seig_entries(box, body) if groups is not None: moov_groups[track] = groups elif kind == b"moof" and box: - fragments.extend(read_fragment(box, entries, defaults, moov_groups, clock)) + fragments.extend(read_fragment(box, entries, defaults, moov_groups, clock, protected)) except (OSError, struct.error, IndexError, ValueError, MemoryError, OverflowError, RecursionError): return None @@ -230,7 +254,7 @@ def kid_windows(path: Path) -> Optional[KidMap]: for kid in kids: spans.setdefault(kid, []).append((start, end)) if not spans: - return None + return KidMap({}, False) if entries and protected.isdisjoint(tracks or entries) else None return KidMap({kid: pick_windows(fragment_spans) for kid, fragment_spans in spans.items()}, video) @@ -284,11 +308,13 @@ def read_fragment( defaults: dict[int, tuple[int, int]], moov_groups: dict[int, list[Seig]], clock: dict[int, int], + protected: set[int], ) -> Iterator[Fragment]: """Yield (track, decode time, duration, KIDs that encrypt samples) for each track fragment in a moof. A fragment without a tfdt or tfxd starts where the track's previous fragment ended, by - the running sum of sample durations in ``clock``. + the running sum of sample durations in ``clock``. A track with a senc that protects + samples goes into ``protected``. """ for kind, traf, traf_end in children(moof): if kind != b"traf": @@ -297,7 +323,7 @@ def read_fragment( groups: list[Seig] = [] runs: Optional[list[tuple[int, int]]] = None truns: list[int] = [] - for kind_, body, _ in children(moof, traf, traf_end): + for kind_, body, box_end in children(moof, traf, traf_end): if kind_ == b"tfhd": flags = int.from_bytes(moof[body + 1 : body + 4], "big") track = struct.unpack_from(">I", moof, body + 4)[0] @@ -317,6 +343,8 @@ def read_fragment( groups = seig_entries(moof, body) or groups elif kind_ == b"sbgp": runs = sbgp_runs(moof, body) or runs + elif kind_ in SENC and senc_protects_samples(moof, body, box_end): + protected.add(track) if track not in entries: continue default_entry, trex_duration = defaults.get(track, (1, 0)) diff --git a/unshackle/core/tracks/track.py b/unshackle/core/tracks/track.py index 9fe233de..cbeb4c0e 100644 --- a/unshackle/core/tracks/track.py +++ b/unshackle/core/tracks/track.py @@ -30,7 +30,7 @@ from unshackle.core.config import config from unshackle.core.constants import DOWNLOAD_ALL_DRM, DOWNLOAD_CANCELLED, DOWNLOAD_LICENCE_ONLY, DownloadCancelled from unshackle.core.downloaders import requests from unshackle.core.drm import DRM_T, ClearKeyCENC, PlayReady, Widevine -from unshackle.core.drm.verify import decrypt_track +from unshackle.core.drm.verify import decrypt_track, senc_protects_samples from unshackle.core.events import events from unshackle.core.session import MAX_BACKOFF, RnetSession, utf8_text from unshackle.core.temp import restore_segments, stash_segments @@ -268,20 +268,6 @@ def has_encrypted_sample_entry(path: Path) -> bool: return False -def senc_protects_samples(buf: bytes, body: int, end: int) -> bool: - """True if a senc/PIFF-uuid box describes protected samples. - - An empty senc (sample_count 0) sits on a genuinely clear fragment, so treating it as - a skipped one would raise on a healthy file. Field order follows - read_per_sample_iv_size. - """ - flags = int.from_bytes(buf[body + 1 : body + 4], "big") - pos = body + 4 + (20 if flags & 0x1 else 0) # PIFF override: AlgorithmID(3)+IV_size(1)+KID(16) - if pos + 4 > end: - return False - return int.from_bytes(buf[pos : pos + 4], "big") > 0 - - def moof_still_encrypted(moof: bytes) -> bool: # senc only: a decrypter detaches just the atom it consumed, so a PIFF uuid can # outlive a good decrypt. ISM arrives as senc via piff_senc_to_cenc.