feat(drm): license a track only when its own KID has no key

A vault key for the track's KID now stands in for the licence on the client and the server (PlayReady included). decrypt_verified licenses any other header KID the file uses.
This commit is contained in:
imSp4rky
2026-11-12 18:03:16 -07:00
parent 16fbf5d17a
commit 7f91a1d36b
15 changed files with 343 additions and 66 deletions
+21 -3
View File
@@ -72,7 +72,7 @@ def widevine(*kids: UUID) -> Widevine:
return Widevine(pssh=PSSH.new(system_id=PSSH.SystemId.Widevine, key_ids=list(kids)))
def prepare(cmd: dl, drm: Widevine, track_kid: UUID, table: Table, **kwargs: Any) -> None:
def prepare(cmd: dl, drm: Widevine, track_kid: Optional[UUID], table: Table, **kwargs: Any) -> None:
cmd.prepare_drm(
drm,
track=SimpleNamespace(drm=[drm]),
@@ -220,7 +220,25 @@ def test_a_refused_licence_without_the_track_key_reports_the_licence_error(
assert "Licence request failed" in str(tree.children[-1].label)
def test_a_refused_licence_keeps_the_known_track_key(
def test_a_held_track_key_needs_no_licence(cmd: dl, licences: list[str], monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(dl, "LICENSE_KEY_CACHE", {KID_A: KEYS[KID_A]})
drm = widevine(KID_A, KID_B)
prepare(cmd, drm, KID_A, new_table())
assert licences == []
assert drm.content_keys == {KID_A: KEYS[KID_A]}
def test_force_licenses_a_held_track_key_again(cmd: dl, licences: list[str], monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(dl, "LICENSE_KEY_CACHE", {KID_A: KEYS[KID_A]})
prepare(cmd, widevine(KID_A, KID_B), KID_A, new_table(), force=True)
assert len(licences) == 1
def test_a_refused_licence_keeps_the_known_keys_when_the_track_kid_is_unknown(
cmd: dl, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
monkeypatch.setattr(Widevine, "get_content_keys", refuse_licence)
@@ -228,7 +246,7 @@ def test_a_refused_licence_keeps_the_known_track_key(
monkeypatch.setattr(dl, "DRM_LOCKS", {})
drm = widevine(KID_A, KID_B)
prepare(cmd, drm, KID_A, new_table())
prepare(cmd, drm, None, new_table())
assert drm.content_keys == {KID_A: KEYS[KID_A]}
assert [r.getMessage() for r in caplog.records if r.levelname == "WARNING"] == [
+12 -1
View File
@@ -131,7 +131,18 @@ def test_segment_decrypt_needs_a_key_for_a_kidless_track(
drm.content_keys[key_kid] = KEY
assert can_use(drm, "mp4decrypt", movie(two_sinf(A, piff_first=True))) is usable
assert can_use(drm, "mp4decrypt", movie(encv(A))) is True
assert can_use(drm, "mp4decrypt", movie(encv(A))) is usable
def test_segment_decrypt_waits_for_a_key_to_every_kid_the_drm_names(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(config, "decrypt_segments", True, raising=False)
monkeypatch.setattr(binaries, "Mp4decrypt", tmp_path / "mp4decrypt")
drm = _widevine(A, B)
drm.content_keys[A] = KEY
assert can_use(drm, "mp4decrypt", movie(encv(A))) is False
class StubDRM:
+26
View File
@@ -384,6 +384,32 @@ def test_stale_backup_from_a_killed_run_is_not_restored(tmp_path: Path, monkeypa
assert path.read_bytes() == b"cipher"
@pytest.mark.parametrize(
("windows", "licensed"),
[({KID: [(0.0, 2.0)], SIBLING: [(2.0, 4.0)]}, [SIBLING]), ({KID: [(0.0, 2.0)]}, []), (None, [SIBLING])],
ids=["file-uses-it", "file-does-not", "unmapped"],
)
def test_an_unkeyed_kid_the_file_uses_is_licensed_before_the_decrypt(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, windows: Optional[dict], licensed: list[UUID]
) -> None:
cmd = make_cmd(monkeypatch, Vaults("SVC"))
drm = FakeDRM({KID: GOOD})
drm.kids = [KID, SIBLING]
monkeypatch.setattr(
dl_module.verify,
"kid_windows",
lambda path: None if windows is None else dl_module.verify.KidMap(windows, True),
)
path = tmp_path / "video.mp4"
path.write_bytes(b"cipher")
licences: list[UUID] = []
cmd.decrypt_verified(drm, path, lambda drm, track_kid: licences.append(track_kid), KID)
assert licences == licensed
assert drm.decrypted_with == [{KID: GOOD}]
def test_drm_without_content_keys_still_decrypts(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""HLS AES-128 ClearKey has no KID:KEY map; the hook must pass it straight through."""
cmd = make_cmd(monkeypatch, Vaults("SVC"))