From 6b04d7b1379443f37347f8c3370078d2c7007443 Mon Sep 17 00:00:00 2001 From: imSp4rky Date: Tue, 13 Oct 2026 17:32:29 -0600 Subject: [PATCH] fix(vaults): lengthen the whole-file key check and steady its HEVC test Wrong-key HEVC noise can decode clean for seconds in small frames, so the tail check now decodes 8 s. The test clips are 640x360 with a fixed IV. --- docs/guide/vaults.md | 4 +-- tests/core/test_ffmpeg_decodes.py | 40 +++++++++++++++++++----------- unshackle/core/drm/verify.py | 4 +-- unshackle/core/utils/subprocess.py | 14 +++++++---- 4 files changed, 38 insertions(+), 24 deletions(-) diff --git a/docs/guide/vaults.md b/docs/guide/vaults.md index e76d5770..4a8954ed 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. A KID that is not in this map, 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. 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. A KID that is not in this map, 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: @@ -46,7 +46,7 @@ The download stops with an error instead of writing noise when no ciphertext is Per-segment decryption gives up the retry. With `decrypt_segments` on, and always with MonaLisa, the plaintext replaces each segment as it arrives, so no ciphertext is left. A failed decode then flags the pair and stops the download. The next run skips the flagged pair. -unshackle copies a content key from one vault to the others only after this check passes, so a wrong content key does not spread. Without FFmpeg, a content key from a vault is not copied at all. To see or clear the flags, open the SQLite file and read or delete rows in `bad_keys`. +unshackle copies a content key from one vault to the others only after this check passes, so a content key that fails the check does not spread. Without FFmpeg, a content key from a vault is not copied at all. To see or clear the flags, open the SQLite file and read or delete rows in `bad_keys`. !!! tip "A good two-vault setup" A common arrangement is a fast local **SQLite** vault plus a shared remote vault (**MySQL**, **HTTP**, or **API**). The local vault answers instantly and works offline; the remote vault lets you share keys across machines or with a group. diff --git a/tests/core/test_ffmpeg_decodes.py b/tests/core/test_ffmpeg_decodes.py index f39959f8..404dbb2e 100644 --- a/tests/core/test_ffmpeg_decodes.py +++ b/tests/core/test_ffmpeg_decodes.py @@ -22,8 +22,10 @@ pytestmark = pytest.mark.skipif(binaries.FFMPEG is None, reason="requires ffmpeg KID = "00000000098ac6820000000000000000" KEY = "13c795e9383efc0387c0bad6a10df73f" WRONG_KEY = "00000000000000000000000000000000" +IV = "0123456789abcdef" # the packager's random IV would change the wrong-key noise on each run # 24 fps with a keyframe every 48 frames: one 2 s GOP for each 2 s fragment. -GOP = ["-x264-params", "keyint=48:min-keyint=48:scenecut=0"] +GOP_PARAMS = "keyint=48:min-keyint=48:scenecut=0" +GOP = ["-x264-params", GOP_PARAMS] def ffmpeg(out: Path, source: str, *codec: str) -> Path: @@ -34,10 +36,10 @@ def ffmpeg(out: Path, source: str, *codec: str) -> Path: return out -def video(out: Path, *x264: str, codec: str = "libx264") -> Path: +def video(out: Path, *x264: str, codec: str = "libx264", size: str = "320x180") -> Path: return ffmpeg( out, - "testsrc2=size=320x180:rate=24:duration=12", + f"testsrc2=size={size}:rate=24:duration=12", "-c:v", codec, "-preset", @@ -63,7 +65,7 @@ def test_mpegts_video_passes_fallback(tmp_path: Path) -> None: def test_open_gop_passes_windows(tmp_path: Path) -> None: # The B-frames before each open-GOP keyframe refer to the GOP before the window. - og = video(tmp_path / "og.mp4", "-x264-params", "open-gop=1:keyint=48:min-keyint=48:scenecut=0:bframes=3") + og = video(tmp_path / "og.mp4", "-x264-params", f"open-gop=1:{GOP_PARAMS}:bframes=3") assert all(ffmpeg_decodes(og, start=start, seconds=2, video=True) for start in (2.0, 4.0, 6.0, 8.0)) @@ -71,26 +73,31 @@ def test_open_gop_hevc_passes_windows(tmp_path: Path) -> None: og = video( tmp_path / "og.mp4", "-x265-params", - "open-gop=1:keyint=48:min-keyint=48:scenecut=0:log-level=none", + f"open-gop=1:{GOP_PARAMS}:log-level=none", codec="libx265", ) assert all(ffmpeg_decodes(og, start=start, seconds=2, video=True) for start in (2.0, 4.0, 6.0, 8.0)) -# An HEVC keyframe decrypted with a wrong key can decode without an error, so the HEVC case -# proves that the check decodes the frames after it too. -@pytest.fixture(scope="module", params=["h264", "hevc"]) +# An HEVC keyframe decrypted with a wrong key can decode without an error, so the HEVC cases +# prove that the check decodes the frames after it too. HEVC noise trips fewer decoder checks +# in a small frame, and a 320x180 window often decodes clean, so the HEVC clips are 640x360. +# Each param is the x265 params to add, or None for H.264. +@pytest.fixture( + scope="module", + params=[None, "", ":open-gop=1"], + ids=["h264", "hevc", "hevc-open-gop"], +) def encrypted(request: pytest.FixtureRequest, tmp_path_factory: pytest.TempPathFactory) -> tuple[KidMap, Path, Path]: """The KID map of a clip encrypted after a clear lead, with its right-key and wrong-key decrypts.""" if not binaries.ShakaPackager or not binaries.Mp4decrypt: pytest.skip("requires shaka packager and mp4decrypt") tmp = tmp_path_factory.mktemp("wrong_key") - if request.param == "hevc": - src = video( - tmp / "src.mp4", "-x265-params", "keyint=48:min-keyint=48:scenecut=0:log-level=none", codec="libx265" - ) - else: + if request.param is None: src = video(tmp / "src.mp4", *GOP) + else: + x265 = f"{GOP_PARAMS}:log-level=none{request.param}" + src = video(tmp / "src.mp4", "-x265-params", x265, codec="libx265", size="640x360") enc = tmp / "enc.mp4" subprocess.run( [ @@ -99,6 +106,8 @@ def encrypted(request: pytest.FixtureRequest, tmp_path_factory: pytest.TempPathF "--enable_raw_key_encryption", "--keys", f"label=:key_id={KID}:key={KEY}", + "--iv", + IV, "--clear_lead", "4", "--fragment_duration", @@ -126,9 +135,10 @@ def test_right_key_passes_every_window(encrypted: tuple[KidMap, Path, Path]) -> assert all(windows_pass(kid_map, right)) -def test_wrong_key_fails_every_window(encrypted: tuple[KidMap, Path, Path]) -> None: +def test_wrong_key_fails_the_kid(encrypted: tuple[KidMap, Path, Path]) -> None: + # One failed window flags the KID, so one clean window of noise does not hide a wrong key. kid_map, _, wrong = encrypted - assert not any(windows_pass(kid_map, wrong)) + assert not all(windows_pass(kid_map, wrong)) def test_wrong_key_passes_the_clear_lead(encrypted: tuple[KidMap, Path, Path]) -> None: diff --git a/unshackle/core/drm/verify.py b/unshackle/core/drm/verify.py index b24e885b..38d14651 100644 --- a/unshackle/core/drm/verify.py +++ b/unshackle/core/drm/verify.py @@ -240,8 +240,8 @@ def pick_windows(spans: list[tuple[float, float]], longest: float = 4.0) -> list A window starts at a fragment and runs on through the next fragments of the same KID, up to ``longest`` seconds, but never into a fragment of another KID or a clear one. A short window can decode noise without an error: the decoder often needs several frames to find it. - A wrong key fails on the first inter frame in every measured case, so ``longest`` buys - decoder frames, not certainty; every second of it is decoded in full for a right key. + ``longest`` buys decoder frames, not certainty; every second of it is decoded in full for + a right key. The last window ends where the KID's last run of fragments ends. """ spans = sorted(span for span in spans if span[1] > span[0]) # a duplicated tfdt gives an empty span diff --git a/unshackle/core/utils/subprocess.py b/unshackle/core/utils/subprocess.py index e02bfebb..6a224a6e 100644 --- a/unshackle/core/utils/subprocess.py +++ b/unshackle/core/utils/subprocess.py @@ -56,6 +56,7 @@ FFMPEG_MISSING_REFERENCE = ( "could not find ref with poc", ) FFMPEG_CHECK_TIMEOUT = 120 +FFMPEG_CHECK_TAIL = 8 # seconds def ffmpeg_decodes(path: Path, start: Optional[float] = None, seconds: float = 3, video: Optional[bool] = None) -> bool: @@ -66,13 +67,15 @@ def ffmpeg_decodes(path: Path, start: Optional[float] = None, seconds: float = 3 With ``start``, the check decodes ``seconds`` from that time. An FFmpeg copy cuts the window first, so no sample after the window gets to the decoder. Without ``start``, the - check decodes the first seconds and the last 4 seconds, because a title can open with + check decodes the first seconds and the last seconds, because a title can open with a clear or separately keyed lead. Every frame is decoded, because an HEVC keyframe decrypted with a wrong key can decode - without an error. When a window starts inside a GOP and the only error is a missing - reference frame, the check decodes the keyframes of that window again and uses that - verdict. ``video=False`` skips that second pass. + without an error. A pass is not proof: HEVC noise can decode clean for seconds, most + often in small frames, so a longer window finds more wrong keys. When a window starts + inside a GOP and the only error is a missing reference frame, the check decodes the + keyframes of that window again and uses that verdict. ``video=False`` skips that second + pass. A file this FFmpeg build cannot judge counts as a pass, never as a wrong key, and so does a check that does not finish in time. @@ -130,9 +133,10 @@ def ffmpeg_decodes(path: Path, start: Optional[float] = None, seconds: float = 3 return decode(window, keyframes=True) return False + tail = ["-sseof", f"-{FFMPEG_CHECK_TAIL}", "-t", str(FFMPEG_CHECK_TAIL)] if start is not None: return decode(["-ss", f"{start:.6f}", "-t", f"{max(seconds - 0.5, seconds / 2):.6f}"], keyframes=False) - return decode(["-t", str(seconds)], keyframes=False) and decode(["-sseof", "-4", "-t", "4"], keyframes=False) + return decode(["-t", str(seconds)], keyframes=False) and decode(tail, keyframes=False) def ffprobe(uri: Union[bytes, Path]) -> dict: