fix(vaults): harden the KID window check and clear a stale flag on licence

tests/core/test_kid_windows.py

@@ -90,17 +90,37 @@ def test_each_kid_maps_to_its_own_fragments(tmp_path: Path) -> None:
     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:
+def test_windows_run_through_a_kid_but_stop_at_another_or_after_the_longest() -> 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)]
+    assert pick_windows(long_run, longest=10.0) == [(0.0, 10.0), (20.0, 30.0), (30.0, 40.0)]
+    assert pick_windows(long_run) == [(0.0, 4.0), (20.0, 24.0), (36.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)]
+    assert pick_windows([(0, 2), (2, 4), (10, 12), (12, 14), (14, 16)], longest=10.0) == [(0, 4), (10, 16)]
+    assert pick_windows([(0, 2), (2, 4), (10, 12), (12, 14), (14, 16)]) == [(0, 4), (10, 14.0), (12, 16)]
+
+
+def test_fragments_longer_than_the_longest_and_empty_spans() -> None:
+    assert pick_windows([(0.0, 60.0)]) == [(0.0, 4.0)]
+    assert pick_windows([(0, 15), (15, 30), (30, 45), (45, 60)]) == [(0, 4), (30, 34), (45, 49)]
+    assert pick_windows([(0.0, 5.0), (5.0, 5.0)]) == [(0.0, 4.0)]
+    assert pick_windows([(5.0, 5.0)]) == []
+
+
+def test_nested_box_bomb_and_oversized_sbgp(tmp_path: Path) -> None:
+    body = b""
+    for _ in range(5000):
+        body = box(b"trak", body)
+    (tmp_path / "bomb.mp4").write_bytes(box(b"moov", body))
+    assert kid_windows(tmp_path / "bomb.mp4") is None
+    huge = full(b"sbgp", b"seig" + struct.pack(">I", 0x00FFFFFF) + bytes(16))
+    (tmp_path / "sbgp.mp4").write_bytes(movie(encv(A)) + moof(1, 0, huge))
+    assert kid_windows(tmp_path / "sbgp.mp4") is None
 
 
 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)
+    assert kid_windows(path) == KidMap({A: [(0.0, 4.0), (2.0, 6.0), (3.0, 7.0)]}, True)
 
 
 def test_last_fragment_ends_after_its_sample_durations(tmp_path: Path) -> None:
@@ -145,7 +165,7 @@ def test_samples_past_the_sbgp_runs_take_the_sample_entry_kid(tmp_path: Path) ->
 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)
+    assert kid_windows(path) == KidMap({A: [(0.0, 4.0), (2.0, 5.0)]}, True)
 
 
 def test_oversized_box(tmp_path: Path) -> None:

tests/core/test_vault_bad_keys.py

@@ -316,6 +316,19 @@ def test_flag_moves_the_row_and_blocks_it_coming_back(tmp_path: Path) -> None:
     assert local.add_key("SVC", KID, GOOD)  # the KID itself is not blocked, only the pair
 
 
+def test_a_licence_key_clears_a_flag_from_an_earlier_run(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """A false flag must not outlive the run: the next licence returns the pair and it is stored again."""
+    local = SQLite("local", tmp_path / "local.db")
+    local.flag_bad_key("SVC", KID, GOOD, "poisoned")
+    vaults = Vaults("SVC")
+    vaults.vaults = [local]
+    cmd = make_cmd(monkeypatch, vaults)
+    cmd.vault_cache_tally = None
+    cmd.cache_keys_to_vaults({KID: GOOD})
+    assert not local.is_bad_key(KID, GOOD)
+    assert local.get_key(KID, "SVC") == GOOD
+
+
 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")
@@ -404,6 +417,45 @@ def test_a_kid_absent_from_the_map_is_judged_by_the_fallback(tmp_path: Path, mon
     assert (None, 3, True) in calls  # the fallback knows the track is video
 
 
+def test_a_wrong_mapped_key_does_not_flag_a_kid_the_file_does_not_carry(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The audio KID is in the licence and the vault but not in this video file: the whole-file
+    check fails on the wrong video key's noise, and that verdict must not reach the audio KID."""
+    audio = UUID(int=7)
+    local = SQLite("local", tmp_path / "local.db")
+    poisoned = Remote("poisoned", tmp_path / "poisoned.db")
+    poisoned.add_key("SVC", audio, 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({audio: vaults.get_key(audio)[0], KID: vaults.get_key(KID)[0]})
+    monkeypatch.setattr(
+        dl_module.verify, "kid_windows", lambda path: dl_module.verify.KidMap({KID: [(4.0, 6.0)]}, True)
+    )
+    monkeypatch.setattr(
+        dl_module,
+        "ffmpeg_decodes",
+        lambda path, start=None, seconds=3, video=None: drm.decrypted_with[-1][KID] == GOOD,
+    )
+    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 == {audio: GOOD, KID: GOOD}
+    assert local.is_bad_key(KID, BAD)
+    assert not local.is_bad_key(audio, GOOD)
+    assert local.get_key(audio, "SVC") == GOOD  # judged on the retry and copied
+
+
 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)