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

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

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

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

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
+        not hold, so a clear file judges and copies no KID. A KID that is not in an incomplete map,
-        start and end check of the whole file, and a failure there flags each of those KIDs. On
+        and every KID of any other file, gets a start and end check of the whole file, and a
-        failure the pair is flagged in the local vaults, the ciphertext is restored, and the
+        failure there flags each of those KIDs. On failure the pair is flagged in the local
-        licence runs again; ``Vaults`` tells the vault that served it. Every vault skips a
+        vaults, the ciphertext is restored, and the licence runs again; ``Vaults`` tells the
-        flagged pair, so each pass burns at most one vault; when the vaults run out the CDM
+        vault that served it. Every vault skips a flagged pair, so each pass burns at most one
-        answers, and a CDM key is not checked. A CDM that returns the flagged key clears the
+        vault; when the vaults run out the CDM answers, and a CDM key is not checked. A CDM
-        flag and stores the content key again, because then the decode check was wrong.
+        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
 

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]]:
+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."""
+    """Parse an ``stsd`` box. Returns the default KID of each sample entry (None for a clear one), and
-    return [entry_kid(data, kind, entry, entry_end) for kind, entry, entry_end in children(data, body + 8, end)]
+    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
+    they are not known. Returns an empty map for a clear MP4: no track with fragments (or,
-    can place in time, so the caller must use another check.
+    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))

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.