mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-14 13:05:09 +00:00
fix(vaults): verify vault content keys per KID from the fragment map
Map each KID to its fragments from the ciphertext moov/moof (tenc, seig, ISM duration clock) and decode up to three windows per KID that never cross into another KID, so a clear lead cannot hide a wrong key and only the failing KID is flagged. Decode every frame (wrong-key HEVC keyframes decode clean), re-judge mid-GOP starts on keyframes, time out every FFmpeg run, and refuse to store a flagged pair again in SQLite.
This commit is contained in:
imSp4rky
@@ -0,0 +1,143 @@
|
||||
"""``ffmpeg_decodes`` must pass good media and fail a wrong content key, judged by a real FFmpeg.
|
||||
|
||||
Each test builds its media from FFmpeg lavfi sources, so no media file is kept in the repo.
|
||||
The encrypted cases use Shaka Packager to encrypt with a clear lead and mp4decrypt to
|
||||
decrypt with a wrong key, which writes noise in place of the encrypted samples.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from uuid import UUID
|
||||
|
||||
import pytest
|
||||
|
||||
from unshackle.core import binaries
|
||||
from unshackle.core.drm.verify import KidMap, kid_windows
|
||||
from unshackle.core.utils.subprocess import ffmpeg_decodes
|
||||
|
||||
pytestmark = pytest.mark.skipif(binaries.FFMPEG is None, reason="requires ffmpeg")
|
||||
|
||||
KID = "00000000098ac6820000000000000000"
|
||||
KEY = "13c795e9383efc0387c0bad6a10df73f"
|
||||
WRONG_KEY = "00000000000000000000000000000000"
|
||||
# 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"]
|
||||
|
||||
|
||||
def ffmpeg(out: Path, source: str, *codec: str) -> Path:
|
||||
subprocess.run(
|
||||
[str(binaries.FFMPEG), "-nostdin", "-v", "error", "-y", "-f", "lavfi", "-i", source, *codec, str(out)],
|
||||
check=True,
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def video(out: Path, *x264: str, codec: str = "libx264") -> Path:
|
||||
return ffmpeg(
|
||||
out,
|
||||
"testsrc2=size=320x180:rate=24:duration=12",
|
||||
"-c:v",
|
||||
codec,
|
||||
"-preset",
|
||||
"veryfast",
|
||||
"-pix_fmt",
|
||||
"yuv420p",
|
||||
*x264,
|
||||
)
|
||||
|
||||
|
||||
def test_audio_only_passes_fallback(tmp_path: Path) -> None:
|
||||
# -skip_frame on a file with no video decoder makes FFmpeg refuse to open it.
|
||||
audio = ffmpeg(tmp_path / "a.m4a", "sine=f=440:duration=12", "-c:a", "aac")
|
||||
assert ffmpeg_decodes(audio)
|
||||
assert ffmpeg_decodes(audio, video=False)
|
||||
|
||||
|
||||
def test_mpegts_video_passes_fallback(tmp_path: Path) -> None:
|
||||
ts = video(tmp_path / "v.ts", *GOP)
|
||||
assert ffmpeg_decodes(ts)
|
||||
assert ffmpeg_decodes(ts, video=True)
|
||||
|
||||
|
||||
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")
|
||||
assert all(ffmpeg_decodes(og, start=start, seconds=2, video=True) for start in (2.0, 4.0, 6.0, 8.0))
|
||||
|
||||
|
||||
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",
|
||||
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"])
|
||||
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:
|
||||
src = video(tmp / "src.mp4", *GOP)
|
||||
enc = tmp / "enc.mp4"
|
||||
subprocess.run(
|
||||
[
|
||||
str(binaries.ShakaPackager),
|
||||
f"in={src},stream=video,output={enc}",
|
||||
"--enable_raw_key_encryption",
|
||||
"--keys",
|
||||
f"label=:key_id={KID}:key={KEY}",
|
||||
"--clear_lead",
|
||||
"4",
|
||||
"--fragment_duration",
|
||||
"2",
|
||||
"--segment_duration",
|
||||
"2",
|
||||
],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
)
|
||||
right, wrong = tmp / "right.mp4", tmp / "wrong.mp4"
|
||||
subprocess.run([str(binaries.Mp4decrypt), "--key", f"{KID}:{KEY}", str(enc), str(right)], check=True)
|
||||
subprocess.run([str(binaries.Mp4decrypt), "--key", f"{KID}:{WRONG_KEY}", str(enc), str(wrong)], check=True)
|
||||
kid_map = kid_windows(enc)
|
||||
assert kid_map and list(kid_map.windows) == [UUID(KID)]
|
||||
return kid_map, right, wrong
|
||||
|
||||
|
||||
def windows_pass(kid_map: KidMap, path: Path) -> list[bool]:
|
||||
return [ffmpeg_decodes(path, start=s, seconds=e - s, video=kid_map.video) for s, e in kid_map.windows[UUID(KID)]]
|
||||
|
||||
|
||||
def test_right_key_passes_every_window(encrypted: tuple[KidMap, Path, Path]) -> None:
|
||||
kid_map, right, _ = encrypted
|
||||
assert all(windows_pass(kid_map, right))
|
||||
|
||||
|
||||
def test_wrong_key_fails_every_window(encrypted: tuple[KidMap, Path, Path]) -> None:
|
||||
kid_map, _, wrong = encrypted
|
||||
assert not any(windows_pass(kid_map, wrong))
|
||||
|
||||
|
||||
def test_wrong_key_passes_the_clear_lead(encrypted: tuple[KidMap, Path, Path]) -> None:
|
||||
# The window ends where the encrypted fragment starts, and B-frames put that fragment's
|
||||
# keyframe first in decode order: the decoder must not get it.
|
||||
kid_map, _, wrong = encrypted
|
||||
lead = kid_map.windows[UUID(KID)][0][0]
|
||||
assert ffmpeg_decodes(wrong, start=0.0, seconds=lead, video=True)
|
||||
|
||||
|
||||
def test_wrong_key_fails_fallback(encrypted: tuple[KidMap, Path, Path]) -> None:
|
||||
assert not ffmpeg_decodes(encrypted[2])
|
||||
@@ -0,0 +1,160 @@
|
||||
import struct
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
from uuid import UUID
|
||||
|
||||
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)
|
||||
|
||||
|
||||
def box(kind: bytes, *parts: bytes) -> bytes:
|
||||
body = b"".join(parts)
|
||||
return struct.pack(">I4s", 8 + len(body), kind) + body
|
||||
|
||||
|
||||
def full(kind: bytes, payload: bytes, version: int = 0, flags: int = 0) -> bytes:
|
||||
return box(kind, bytes([version]) + flags.to_bytes(3, "big") + payload)
|
||||
|
||||
|
||||
def encv(kid: UUID, piff: bool = False) -> bytes:
|
||||
tenc_body = b"\x00\x00\x01\x08" + kid.bytes
|
||||
tenc = box(b"uuid", PIFF_TENC, bytes(4) + tenc_body) if piff else full(b"tenc", tenc_body)
|
||||
return box(b"encv", bytes(78), box(b"sinf", box(b"frma", b"avc1"), box(b"schi", tenc)))
|
||||
|
||||
|
||||
def enca(kid: UUID) -> bytes:
|
||||
return box(b"enca", bytes(28), box(b"sinf", box(b"schi", full(b"tenc", b"\x00\x00\x01\x08" + kid.bytes))))
|
||||
|
||||
|
||||
def seig(protected: bool, kid: UUID) -> bytes:
|
||||
return bytes([0, 0, int(protected), 8]) + kid.bytes
|
||||
|
||||
|
||||
def sgpd(*entries: bytes, version: int = 1) -> bytes:
|
||||
head = b"seig"
|
||||
if version >= 1:
|
||||
head += struct.pack(">I", 20)
|
||||
if version >= 2:
|
||||
head += struct.pack(">I", 0)
|
||||
return full(b"sgpd", head + struct.pack(">I", len(entries)) + b"".join(entries), version=version)
|
||||
|
||||
|
||||
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:
|
||||
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(
|
||||
b"trak",
|
||||
full(b"tkhd", bytes(8) + struct.pack(">I", 1)),
|
||||
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 moof(
|
||||
entry: int,
|
||||
time: int,
|
||||
*groups: bytes,
|
||||
samples: int = 48,
|
||||
duration: Optional[int] = None,
|
||||
tfxd: bool = False,
|
||||
) -> bytes:
|
||||
flags = 0x02 | (0x08 if duration is not None else 0)
|
||||
tfhd = struct.pack(">II", 1, entry) + (struct.pack(">I", duration) if duration is not None else b"")
|
||||
if tfxd:
|
||||
timing = box(b"uuid", PIFF_TFXD, bytes([1, 0, 0, 0]) + struct.pack(">QQ", time, 0))
|
||||
else:
|
||||
timing = full(b"tfdt", struct.pack(">Q", time), version=1)
|
||||
trun = full(b"trun", struct.pack(">I", samples))
|
||||
return box(b"moof", box(b"traf", full(b"tfhd", tfhd, flags=flags), timing, trun, *groups)) + box(b"mdat", bytes(64))
|
||||
|
||||
|
||||
def local(kid: UUID, protected: bool = True) -> tuple[bytes, bytes]:
|
||||
return sgpd(seig(protected, kid)), sbgp((48, 0x10001))
|
||||
|
||||
|
||||
def test_each_kid_maps_to_its_own_fragments(tmp_path: Path) -> None:
|
||||
path = tmp_path / "enc.mp4"
|
||||
path.write_bytes(
|
||||
movie(encv(A), encv(B))
|
||||
+ moof(1, 5000) # lead under A
|
||||
+ moof(2, 7000) # the rest under B
|
||||
+ moof(2, 9000)
|
||||
+ moof(1, 11000, *local(C)) # a seig group overrides the sample entry's KID
|
||||
+ moof(1, 13000, *local(A, protected=False)) # a clear seig group names no KID
|
||||
)
|
||||
assert kid_windows(path) == KidMap({A: [(0.0, 2.0)], B: [(2.0, 6.0), (4.0, 6.0)], C: [(6.0, 8.0)]}, True)
|
||||
|
||||
|
||||
def test_windows_run_through_a_kid_but_stop_at_another_or_after_ten_seconds() -> None:
|
||||
long_run = [(t * 2.0, t * 2.0 + 2) for t in range(20)]
|
||||
assert pick_windows(long_run) == [(0.0, 10.0), (20.0, 30.0), (30.0, 40.0)]
|
||||
# a gap is another KID or a clear stretch: no window crosses it, and the last one ends the last run
|
||||
assert pick_windows([(0, 2), (2, 4), (10, 12), (12, 14), (14, 16)]) == [(0, 4), (10, 16)]
|
||||
|
||||
|
||||
def test_first_middle_and_last_fragment(tmp_path: Path) -> None:
|
||||
path = tmp_path / "enc.mp4"
|
||||
path.write_bytes(movie(encv(A)) + b"".join(moof(1, t * 1000) for t in range(5)))
|
||||
assert kid_windows(path) == KidMap({A: [(0.0, 7.0), (2.0, 7.0)]}, True)
|
||||
|
||||
|
||||
def test_last_fragment_ends_after_its_sample_durations(tmp_path: Path) -> None:
|
||||
path = tmp_path / "enc.mp4"
|
||||
path.write_bytes(movie(encv(A)) + moof(1, 0, samples=10, duration=250))
|
||||
assert kid_windows(path) == KidMap({A: [(0.0, 2.5)]}, True)
|
||||
|
||||
|
||||
def test_audio_track_is_not_video(tmp_path: Path) -> None:
|
||||
path = tmp_path / "enc.m4a"
|
||||
path.write_bytes(movie(enca(A), handler=b"soun") + moof(1, 0))
|
||||
assert kid_windows(path) == KidMap({A: [(0.0, 3.0)]}, False)
|
||||
|
||||
|
||||
def test_moov_seig_table_survives_another_grouping(tmp_path: Path) -> None:
|
||||
roll = full(b"sgpd", b"roll" + struct.pack(">II", 2, 1) + b"\x00\x01", version=1)
|
||||
path = tmp_path / "enc.mp4"
|
||||
path.write_bytes(movie(encv(A), stbl=sgpd(seig(True, C)) + roll) + moof(1, 0, sbgp((48, 1))))
|
||||
assert kid_windows(path) == KidMap({C: [(0.0, 3.0)]}, True)
|
||||
|
||||
|
||||
def test_sgpd_version_2(tmp_path: Path) -> None:
|
||||
path = tmp_path / "enc.mp4"
|
||||
path.write_bytes(movie(encv(A)) + moof(1, 0, sgpd(seig(True, C), version=2), sbgp((48, 0x10001))))
|
||||
assert kid_windows(path) == KidMap({C: [(0.0, 3.0)]}, True)
|
||||
|
||||
|
||||
def test_sgpd_version_0_with_constant_iv(tmp_path: Path) -> None:
|
||||
constant_iv = bytes([0, 0, 1, 0]) + B.bytes + b"\x08" + bytes(8)
|
||||
groups = sgpd(constant_iv, seig(True, C), version=0)
|
||||
path = tmp_path / "enc.mp4"
|
||||
path.write_bytes(movie(encv(A)) + moof(1, 0, groups, sbgp((24, 0x10001), (24, 0x10002))))
|
||||
assert kid_windows(path) == KidMap({B: [(0.0, 3.0)], C: [(0.0, 3.0)]}, True)
|
||||
|
||||
|
||||
def test_samples_past_the_sbgp_runs_take_the_sample_entry_kid(tmp_path: Path) -> None:
|
||||
path = tmp_path / "enc.mp4"
|
||||
path.write_bytes(movie(encv(A)) + moof(1, 0, sgpd(seig(True, C)), sbgp((10, 0x10001))))
|
||||
assert kid_windows(path) == KidMap({A: [(0.0, 3.0)], C: [(0.0, 3.0)]}, True)
|
||||
|
||||
|
||||
def test_piff_tenc_and_tfxd(tmp_path: Path) -> None:
|
||||
path = tmp_path / "enc.ismv"
|
||||
path.write_bytes(movie(encv(A, piff=True)) + moof(1, 0, tfxd=True) + moof(1, 2000, tfxd=True))
|
||||
assert kid_windows(path) == KidMap({A: [(0.0, 5.0), (2.0, 5.0)]}, True)
|
||||
|
||||
|
||||
def test_oversized_box(tmp_path: Path) -> None:
|
||||
path = tmp_path / "enc.mp4"
|
||||
path.write_bytes(movie(encv(A)) + moof(1, 0) + struct.pack(">I4s", 0x7FFFFFFF, b"moof") + bytes(16))
|
||||
assert kid_windows(path) is None
|
||||
|
||||
|
||||
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
|
||||
@@ -77,7 +77,11 @@ class FakeDRM:
|
||||
|
||||
def make_cmd(monkeypatch: pytest.MonkeyPatch, vaults: Vaults) -> dl:
|
||||
monkeypatch.setattr(dl_module.binaries, "FFMPEG", "ffmpeg")
|
||||
monkeypatch.setattr(dl_module, "ffmpeg_decodes", lambda path: path.read_bytes() == b"plain:" + GOOD.encode())
|
||||
monkeypatch.setattr(
|
||||
dl_module,
|
||||
"ffmpeg_decodes",
|
||||
lambda path, start=None, seconds=3, video=None: path.read_bytes() == b"plain:" + GOOD.encode(),
|
||||
)
|
||||
cmd = dl.__new__(dl)
|
||||
cmd.log = SimpleNamespace(warning=print, info=print, debug=print)
|
||||
cmd.vaults = vaults
|
||||
@@ -248,7 +252,7 @@ def test_cdm_returning_the_flagged_key_clears_the_flag(tmp_path: Path, monkeypat
|
||||
vaults = Vaults("SVC")
|
||||
vaults.vaults = [local, poisoned]
|
||||
cmd = make_cmd(monkeypatch, vaults)
|
||||
monkeypatch.setattr(dl_module, "ffmpeg_decodes", lambda path: False)
|
||||
monkeypatch.setattr(dl_module, "ffmpeg_decodes", lambda path, start=None, seconds=3, video=None: False)
|
||||
path = tmp_path / "video.mp4"
|
||||
path.write_bytes(b"cipher")
|
||||
drm = FakeDRM({KID: vaults.get_key(KID)[0]})
|
||||
@@ -300,6 +304,158 @@ def test_flag_on_a_read_only_vault_keeps_its_rows(tmp_path: Path) -> None:
|
||||
assert list(ro.get_keys("SVC")) == [(KID.hex, BAD)] # but the row itself is untouched
|
||||
|
||||
|
||||
def test_flag_moves_the_row_and_blocks_it_coming_back(tmp_path: Path) -> None:
|
||||
local = SQLite("local", tmp_path / "local.db")
|
||||
local.add_key("SVC", KID, BAD)
|
||||
local.flag_bad_key("SVC", KID, BAD, "poisoned")
|
||||
assert local.is_bad_key(KID, BAD)
|
||||
assert list(local.get_keys("SVC")) == []
|
||||
assert local.add_key("SVC", KID, BAD) is False
|
||||
assert local.add_keys("SVC", {KID: BAD}) == 0
|
||||
assert list(local.get_keys("SVC")) == []
|
||||
assert local.add_key("SVC", KID, GOOD) # the KID itself is not blocked, only the pair
|
||||
|
||||
|
||||
def test_bulk_add_refuses_a_flagged_pair_in_any_case_or_kid_form(tmp_path: Path) -> None:
|
||||
local = SQLite("local", tmp_path / "local.db")
|
||||
local.flag_bad_key("SVC", KID, BAD, "poisoned")
|
||||
for kid, key in ((KID.hex.upper(), BAD), (str(KID), BAD.upper()), (KID, BAD)):
|
||||
assert local.add_keys("SVC", {kid: key}) == 0
|
||||
assert list(local.get_keys("SVC")) == []
|
||||
|
||||
|
||||
def test_only_the_kid_whose_window_fails_is_flagged(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""A lead under one KID and the rest under another: the lead's key is right and stays."""
|
||||
lead = UUID(int=7)
|
||||
local = SQLite("local", tmp_path / "local.db")
|
||||
poisoned = Remote("poisoned", tmp_path / "poisoned.db")
|
||||
poisoned.add_key("SVC", lead, BAD) # the same wrong key under both KIDs, as the NF title had
|
||||
poisoned.add_key("SVC", KID, BAD)
|
||||
other = Remote("other", tmp_path / "other.db")
|
||||
other.add_key("SVC", KID, GOOD)
|
||||
vaults = Vaults("SVC")
|
||||
vaults.vaults = [local, poisoned]
|
||||
cmd = make_cmd(monkeypatch, vaults)
|
||||
drm = FakeDRM({lead: vaults.get_key(lead)[0], KID: vaults.get_key(KID)[0]})
|
||||
right = {lead: BAD, KID: GOOD}
|
||||
monkeypatch.setattr(
|
||||
dl_module.verify,
|
||||
"kid_windows",
|
||||
lambda path: dl_module.verify.KidMap({lead: [(0.0, 2.0)], KID: [(120.0, 122.0)]}, True),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
dl_module,
|
||||
"ffmpeg_decodes",
|
||||
lambda path, start=None, seconds=3, video=None: (
|
||||
drm.decrypted_with[-1][lead if start == 0.0 else KID] == right[lead if start == 0.0 else KID]
|
||||
),
|
||||
)
|
||||
path = tmp_path / "video.mp4"
|
||||
path.write_bytes(b"cipher")
|
||||
|
||||
def licence(drm: FakeDRM, track_kid: UUID) -> None:
|
||||
vaults.vaults.append(other)
|
||||
drm.content_keys[KID] = vaults.get_key(KID)[0]
|
||||
|
||||
cmd.decrypt_verified(drm, path, licence, KID)
|
||||
cmd.wait_vault_writes()
|
||||
assert drm.content_keys == right
|
||||
assert local.is_bad_key(KID, BAD)
|
||||
assert not local.is_bad_key(lead, BAD)
|
||||
|
||||
|
||||
def test_a_kid_absent_from_the_map_is_judged_by_the_fallback(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""A KID the map does not know (a zero-KID tenc, a licence KID the file does not carry)
|
||||
shares the whole-file verdict, and a wrong key there is flagged."""
|
||||
lead = UUID(int=7)
|
||||
local = SQLite("local", tmp_path / "local.db")
|
||||
poisoned = Remote("poisoned", tmp_path / "poisoned.db")
|
||||
poisoned.add_key("SVC", lead, GOOD)
|
||||
poisoned.add_key("SVC", KID, BAD)
|
||||
other = Remote("other", tmp_path / "other.db")
|
||||
other.add_key("SVC", KID, GOOD)
|
||||
vaults = Vaults("SVC")
|
||||
vaults.vaults = [local, poisoned]
|
||||
cmd = make_cmd(monkeypatch, vaults)
|
||||
drm = FakeDRM({lead: vaults.get_key(lead)[0], KID: vaults.get_key(KID)[0]})
|
||||
calls: list[tuple] = []
|
||||
|
||||
def decodes(path: Path, start: object = None, seconds: float = 3, video: object = None) -> bool:
|
||||
calls.append((start, seconds, video))
|
||||
return drm.decrypted_with[-1][lead if start is not None else KID] == GOOD
|
||||
|
||||
monkeypatch.setattr(
|
||||
dl_module.verify, "kid_windows", lambda path: dl_module.verify.KidMap({lead: [(4.0, 6.0)]}, True)
|
||||
)
|
||||
monkeypatch.setattr(dl_module, "ffmpeg_decodes", decodes)
|
||||
path = tmp_path / "video.mp4"
|
||||
path.write_bytes(b"cipher")
|
||||
|
||||
def licence(drm: FakeDRM, track_kid: UUID) -> None:
|
||||
vaults.vaults.append(other)
|
||||
drm.content_keys[KID] = vaults.get_key(KID)[0]
|
||||
|
||||
cmd.decrypt_verified(drm, path, licence, KID)
|
||||
cmd.wait_vault_writes()
|
||||
assert drm.content_keys == {lead: GOOD, KID: GOOD}
|
||||
assert local.is_bad_key(KID, BAD)
|
||||
assert not local.is_bad_key(lead, GOOD)
|
||||
assert (4.0, 2.0, True) in calls # the window is capped at the fragment's end
|
||||
assert (None, 3, True) in calls # the fallback knows the track is video
|
||||
|
||||
|
||||
def test_only_checked_kids_are_copied_to_other_vaults(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Both the mapped KID and the absent one are proven before they reach another vault."""
|
||||
lead = UUID(int=7)
|
||||
local = SQLite("local", tmp_path / "local.db")
|
||||
remote = Remote("remote", tmp_path / "remote.db")
|
||||
remote.add_key("SVC", lead, GOOD)
|
||||
remote.add_key("SVC", KID, GOOD)
|
||||
vaults = Vaults("SVC")
|
||||
vaults.vaults = [local, remote]
|
||||
cmd = make_cmd(monkeypatch, vaults)
|
||||
drm = FakeDRM({lead: vaults.get_key(lead)[0], KID: vaults.get_key(KID)[0]})
|
||||
starts: list = []
|
||||
|
||||
def decodes(path: Path, start: object = None, seconds: float = 3, video: object = None) -> bool:
|
||||
starts.append(start)
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(
|
||||
dl_module.verify, "kid_windows", lambda path: dl_module.verify.KidMap({lead: [(4.0, 6.0)]}, True)
|
||||
)
|
||||
monkeypatch.setattr(dl_module, "ffmpeg_decodes", decodes)
|
||||
path = tmp_path / "video.mp4"
|
||||
path.write_bytes(b"cipher")
|
||||
|
||||
cmd.decrypt_verified(drm, path, None, KID)
|
||||
cmd.wait_vault_writes()
|
||||
assert starts.count(None) == 1 # the absent KID went through the fallback before its copy
|
||||
assert local.get_key(lead, "SVC") == GOOD
|
||||
assert local.get_key(KID, "SVC") == GOOD
|
||||
|
||||
|
||||
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")
|
||||
poisoned.add_key("SVC", KID, BAD)
|
||||
local.flag_bad_key("SVC", KID, BAD, "poisoned")
|
||||
vaults = Vaults("SVC")
|
||||
vaults.vaults = [local, poisoned]
|
||||
with caplog.at_level(logging.WARNING, logger="unshackle.core.vaults"):
|
||||
for _ in range(3):
|
||||
assert vaults.query_vault(poisoned, KID) is None
|
||||
assert sum("flagged bad, skipping" in r.getMessage() for r in caplog.records) == 1
|
||||
|
||||
|
||||
def test_add_key_refuses_a_flagged_pair_with_a_dashed_kid(tmp_path: Path) -> None:
|
||||
local = SQLite("local", tmp_path / "local.db")
|
||||
local.flag_bad_key("SVC", KID, BAD, "poisoned")
|
||||
assert local.add_key("SVC", str(KID), BAD) is False
|
||||
assert local.add_key("SVC", str(KID).upper(), BAD.upper()) is False
|
||||
assert list(local.get_keys("SVC")) == []
|
||||
|
||||
|
||||
def test_a_sibling_verdict_during_the_decrypt_still_triggers_the_retry(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
|
||||
Reference in New Issue
Block a user