fix(vaults): stop false bad-key flags from a failed decode check

Judge a vault KID only on a file whose KID map holds it, tell remote vaults of a bad pair only once another key replaces it, and let the server CDM unflag a pair it returns; the server never flags a CDM-licensed pair. Log FFmpeg stderr on a failed check and name the missing CDM when none is set.

docs/dev/rest-api/endpoints.md

@@ -1218,7 +1218,7 @@ A slow answer arrives as a [slow response](#slow-responses).
 
 ### `POST /api/session/{session_id}/keys/bad`
 
-Report a content key from `vault_keys` that did not decrypt the track. The server flags the pair in its local vaults and reports it to the vault that served it, and the next licence for that KID reaches the CDM. The client never learns the server vault names.
+Report a content key from `vault_keys` that did not decrypt the track. The server flags the pair in its local vaults and reports it to the vault that served it, and the next licence for that KID reaches the CDM. When the CDM returns the same pair, the server removes the flag. The client never learns the server vault names.
 
 ```json
 { "kid": "<kid_hex>", "key": "<key_hex>" }
@@ -1226,7 +1226,7 @@ Report a content key from `vault_keys` that did not decrypt the track. The serve
 
 | Status | Error code | Meaning |
 | --- | --- | --- |
-| `200` | - | The pair is flagged. |
+| `200` | - | `{"flagged": true}`: the pair is flagged. `{"flagged": false}`: the CDM licensed the pair, so it is not flagged. |
 | `400` | `INVALID_INPUT` | This remote session was not served that pair. |
 | `404` | `SESSION_NOT_FOUND` | Unknown remote session. |
 

docs/dev/rest-api/remote-sessions.md

@@ -303,9 +303,11 @@ There are two ways DRM keys get resolved, chosen by the client's `server_cdm` fl
       decodes, and a wrong pair goes back to `POST /api/session/{id}/keys/bad`. The
       server flags the pair in every local vault it holds, reports it to the vault
       that served it, and the next licence for that KID skips the pair and reaches
-      the CDM. A server with no local SQLite vault cannot store the flag. The
-      server accepts a report only for a pair it served to that remote session. The
-      client never learns the server vault names.
+      the CDM. When the CDM returns the same pair, the server removes the flag,
+      because then the client's check was wrong. A server with no local SQLite
+      vault cannot store the flag. The server accepts a report only for a pair it
+      served to that remote session, and does not flag a pair its CDM licensed.
+      The client never learns the server vault names.
     - A content key in the client's own vaults goes first, even when the server
       already answered. The client proves it, and a server key waits as the next
       candidate.

docs/guide/vaults.md

@@ -34,11 +34,11 @@ 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. 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 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:
 
-1. Flags the KID:KEY in the `bad_keys` table of every local SQLite vault, with the service and the name of the vault it came from. The pair also goes out of the service table, except in a vault marked `no_push`, where only the flag is written. A flagged pair cannot go back into a local vault from another vault, a later run, or `kv copy`. A licence that returns the pair clears the flag and stores it. An `API` vault that served the pair also gets a report; `HTTP` and `MySQL` vaults have no report route. A pair from a local vault goes to no remote vault, even when a remote vault holds the same pair.
+1. Flags the KID:KEY in the `bad_keys` table of every local SQLite vault, with the service and the name of the vault it came from. The pair also goes out of the service table, except in a vault marked `no_push`, where only the flag is written. A flagged pair cannot go back into a local vault from another vault, a later run, or `kv copy`. A licence that returns the pair clears the flag and stores it. An `API` vault that served the pair gets a report only when a different content key for that KID is accepted. When the CDM returns the same content key, that vault gets no report. `HTTP` and `MySQL` vaults have no report route. A pair from a local vault goes to no remote vault, even when a remote vault holds the same pair.
 2. Restores the ciphertext from a hard link it made before decryption, then asks for the content key again. Every vault lookup skips a flagged pair, so the request reaches the remaining vaults and then the CDM. A remote answer that arrived after the winning one is reused instead of queried again.
 3. Removes the flag when the CDM returns the same content key, because then the decode check was wrong, not the content key. A content key from another vault is checked in its turn, and flagged as well if it also fails.
 

tests/core/test_vault_bad_keys.py

@@ -287,6 +287,26 @@ def test_cdm_returning_the_flagged_key_clears_the_flag(tmp_path: Path, monkeypat
     assert not local.is_bad_key(KID, BAD)
 
 
+def test_a_licence_key_that_replaces_a_bad_local_pair_is_stored(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The local vault held the bad pair when the licence key arrived, so it skipped the KID; store it now."""
+    local = SQLite("local", tmp_path / "local.db")
+    local.add_key("SVC", KID, BAD)
+    vaults = Vaults("SVC")
+    vaults.vaults = [local]
+    cmd = make_cmd(monkeypatch, vaults)
+    drm = FakeDRM({KID: vaults.get_key(KID)[0]})
+    path = tmp_path / "video.mp4"
+    path.write_bytes(b"cipher")
+    assert local.add_keys("SVC", {KID: GOOD}) == 0
+
+    cmd.decrypt_verified(drm, path, lambda drm, track_kid: drm.content_keys.update({KID: GOOD}), KID)
+    cmd.wait_vault_writes()
+    assert local.is_bad_key(KID, BAD)
+    assert local.get_key(KID, "SVC") == GOOD
+
+
 def test_stale_backup_from_a_killed_run_is_not_restored(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     poisoned = Remote("poisoned", tmp_path / "poisoned.db")
     poisoned.add_key("SVC", KID, BAD)
@@ -426,7 +446,7 @@ def test_only_the_kid_whose_window_fails_is_flagged(tmp_path: Path, monkeypatch:
 
 
 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)
+    """A map that holds an all-zero tenc KID is not complete, so a KID it does not know
     shares the whole-file verdict, and a wrong key there is flagged."""
     lead = UUID(int=7)
     local = SQLite("local", tmp_path / "local.db")
@@ -446,7 +466,9 @@ def test_a_kid_absent_from_the_map_is_judged_by_the_fallback(tmp_path: Path, mon
         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)
+        dl_module.verify,
+        "kid_windows",
+        lambda path: dl_module.verify.KidMap({lead: [(4.0, 6.0)], UUID(int=0): [(8.0, 10.0)]}, True),
     )
     monkeypatch.setattr(dl_module, "ffmpeg_decodes", decodes)
     path = tmp_path / "video.mp4"
@@ -501,11 +523,93 @@ def test_a_wrong_mapped_key_does_not_flag_a_kid_the_file_does_not_carry(
     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
+    assert local.get_key(audio, "SVC") is None
+
+
+def test_a_kid_the_file_does_not_carry_is_not_judged_by_it(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """The CDM licensed the video KID and a vault served the audio KID. An FFmpeg that fails
+    every decode of the video file must not flag the audio KID, nor copy it to another vault."""
+    audio = UUID(int=7)
+    local = SQLite("local", tmp_path / "local.db")
+    remote = Remote("remote", tmp_path / "remote.db")
+    remote.add_key("SVC", audio, GOOD)
+    vaults = Vaults("SVC")
+    vaults.vaults = [local, remote]
+    cmd = make_cmd(monkeypatch, vaults)
+    drm = FakeDRM({audio: vaults.get_key(audio)[0], KID: GOOD})
+    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: False)
+    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)
+    cmd.wait_vault_writes()
+    assert drm.content_keys == {audio: GOOD, KID: GOOD}
+    assert not licences
+    assert not local.is_bad_key(audio, GOOD)
+    assert local.get_key(audio, "SVC") is None
+
+
+@pytest.mark.parametrize(
+    ("served", "decodes", "told"),
+    [
+        (GOOD, False, False),  # broken FFmpeg, the CDM returns the same key
+        (BAD, True, True),  # the CDM returns another key
+    ],
+)
+def test_a_remote_vault_is_told_only_when_another_key_replaces_its_pair(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, served: str, decodes: bool, told: bool
+) -> None:
+    remote = Remote("remote", tmp_path / "remote.db")
+    remote.add_key("SVC", KID, served)
+    vaults = Vaults("SVC")
+    vaults.vaults = [SQLite("local", tmp_path / "local.db"), remote]
+    cmd = make_cmd(monkeypatch, vaults)
+    if not decodes:
+        monkeypatch.setattr(dl_module, "ffmpeg_decodes", lambda path, start=None, seconds=3, video=None: False)
+    drm = FakeDRM({KID: vaults.get_key(KID)[0]})
+    path = tmp_path / "video.mp4"
+    path.write_bytes(b"cipher")
+
+    def licence(drm: FakeDRM, track_kid: UUID) -> None:
+        assert not remote.is_bad_key(KID, served)
+        drm.content_keys[KID] = GOOD
+
+    cmd.decrypt_verified(drm, path, licence, KID)
+    assert drm.content_keys == {KID: GOOD}
+    assert remote.is_bad_key(KID, served) is told
+
+
+def test_the_server_vault_is_told_at_once(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Over --remote the report is what makes the server skip the pair on the next licence."""
+    reports: list[tuple[UUID, str]] = []
+    server = SimpleNamespace(
+        name="server_cdm",
+        local=False,
+        report_now=True,
+        flag_bad_key=lambda service, kid, key, source: reports.append((kid, key)),
+    )
+    vaults = Vaults("SVC")
+    vaults.vaults = [SQLite("local", tmp_path / "local.db")]
+    vaults.sources[KID] = (BAD, server)
+    cmd = make_cmd(monkeypatch, vaults)
+    drm = FakeDRM({KID: BAD})
+    path = tmp_path / "video.mp4"
+    path.write_bytes(b"cipher")
+
+    def licence(drm: FakeDRM, track_kid: UUID) -> None:
+        assert reports == [(KID, BAD)]
+        drm.content_keys[KID] = GOOD
+
+    cmd.decrypt_verified(drm, path, licence, KID)
+    assert drm.content_keys == {KID: GOOD}
 
 
 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."""
+    """Only the mapped KID is proven and copied; the KID this file does not carry waits for its own file."""
     lead = UUID(int=7)
     local = SQLite("local", tmp_path / "local.db")
     remote = Remote("remote", tmp_path / "remote.db")
@@ -530,9 +634,9 @@ def test_only_checked_kids_are_copied_to_other_vaults(tmp_path: Path, monkeypatc
 
     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 None not in starts
     assert local.get_key(lead, "SVC") == GOOD
-    assert local.get_key(KID, "SVC") == GOOD
+    assert local.get_key(KID, "SVC") is None
 
 
 def test_flagged_pair_warns_once_per_vault(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
@@ -602,6 +706,8 @@ def test_api_vault_reports_the_pair_upstream(monkeypatch: pytest.MonkeyPatch, ca
     vaults.vaults = [api]
     vaults.sources[KID] = (BAD, api)
     vaults.flag_bad_key(KID, BAD)
+    assert not calls
+    vaults.report_replaced({KID: GOOD})
     vaults.unflag_bad_key(KID, BAD)
 
     assert calls == [
@@ -609,28 +715,31 @@ def test_api_vault_reports_the_pair_upstream(monkeypatch: pytest.MonkeyPatch, ca
         ("DELETE", f"https://vault.example/bad-keys/{KID.hex}/{BAD}", None),
     ]
 
+    def flag_and_replace(source: object) -> None:
+        vaults.sources[KID] = (BAD, source)
+        vaults.flag_bad_key(KID, BAD)
+        vaults.report_replaced({KID: GOOD})
+
     calls.clear()
-    vaults.sources[KID] = (BAD, SimpleNamespace(name="other"))
-    vaults.flag_bad_key(KID, BAD)
+    flag_and_replace(SimpleNamespace(name="other", local=False, report_now=False))
     assert calls == []
-    vaults.sources[KID] = (BAD, api)
 
     monkeypatch.setattr(api.session, "post", lambda url, **kw: (_ for _ in ()).throw(ConnectionError("down")))
     caplog.clear()
     with caplog.at_level(logging.DEBUG, logger="unshackle.core.vaults"):
-        vaults.flag_bad_key(KID, BAD)
+        flag_and_replace(api)
     assert [r.levelname for r in caplog.records] == ["DEBUG"]
 
     gone = SimpleNamespace(status_code=404, raise_for_status=lambda: (_ for _ in ()).throw(RuntimeError("404")))
     monkeypatch.setattr(api.session, "post", lambda url, **kw: gone)
     caplog.clear()
-    vaults.flag_bad_key(KID, BAD)
+    flag_and_replace(api)
     assert not caplog.records
 
     denied = SimpleNamespace(status_code=200, raise_for_status=lambda: None, json=lambda: {"code": 1, "message": "no"})
     monkeypatch.setattr(api.session, "post", lambda url, **kw: denied)
     monkeypatch.setattr(api.session, "delete", lambda url, **kw: denied)
-    vaults.flag_bad_key(KID, BAD)
+    flag_and_replace(api)
     vaults.unflag_bad_key(KID, BAD)
     assert not caplog.records
 

tests/core/test_vault_push_retry.py

@@ -39,6 +39,9 @@ def fake_server_vaults(pushes: list, vault_list: list[FakeVault]) -> Any:
         def __len__(self) -> int:
             return len(vault_list)
 
+        def clear_licensed_flags(self, content_keys: dict) -> list:
+            return []
+
         def add_keys(self, key_map: dict) -> int:
             pushes.append(key_map)
             return sum(vault.ok for vault in vault_list)

tests/remote/unit/test_server_cdm_key_verification.py

@@ -43,11 +43,12 @@ def test_client_marks_vault_keys_and_forgets_cdm_keys():
     assert svc.server_vault_keys == {}
 
 
-def test_report_bad_posts_only_the_pair():
+def test_server_vault_flag_posts_only_the_pair():
     svc = remote_service()
     posted: list = []
     svc.client = SimpleNamespace(post_optional=lambda ep, data: posted.append((ep, data)) or {})
-    svc.server_vault.report_bad(KID, "v")
+    assert svc.server_vault.report_now
+    svc.server_vault.flag_bad_key("SVC", KID, "v", "server_cdm")
     assert posted == [("/api/session/sess/keys/bad", {"kid": KID.hex, "key": "v"})]
 
 
@@ -77,18 +78,27 @@ def session(monkeypatch):
     return entry, vault
 
 
-async def test_server_flags_a_served_pair_with_its_own_vault_name(session):
+async def test_server_flags_a_served_pair_with_its_own_vault_name(session, monkeypatch):
     entry, vault = session
+    monkeypatch.setattr(handlers, "CACHED_PAIRS", {("SimpleNamespace", KID.hex, "v")})
     resp = await handlers.session_bad_key_handler({"kid": KID.hex, "key": "v"}, "sess")
     assert resp.status == 200
     assert vault.flagged == [("SVC", KID, "v", "sqlite")]
     assert KID.hex not in entry.served_keys
+    assert not handlers.CACHED_PAIRS
 
     with pytest.raises(APIError):
         await handlers.session_bad_key_handler({"kid": CDM_KID.hex, "key": "nope"}, "sess")
     assert len(vault.flagged) == 1
 
 
+async def test_server_does_not_flag_a_pair_its_cdm_licensed(session):
+    entry, vault = session
+    resp = await handlers.session_bad_key_handler({"kid": CDM_KID.hex, "key": "c"}, "sess")
+    assert json.loads(resp.body) == {"flagged": False}
+    assert vault.flagged == []
+
+
 class _WV:
     def __init__(self, b64, kids):
         self._pssh = SimpleNamespace(dumps=lambda: b64)
@@ -218,3 +228,20 @@ def test_reconstruct_drm_keeps_the_default_kid_of_a_pssh_without_kids():
     (drm,) = reconstruct_drm([{"type": "widevine", "pssh": content_id_only.dumps(), "kids": [str(KID)]}])
 
     assert real_kids(drm) == [KID]
+
+
+def test_a_cdm_licence_unflags_a_pair_a_client_reported(monkeypatch, tmp_path):
+    """The CDM returned the reported pair, so the client's decode check was wrong."""
+    from unshackle.core.vaults import Vaults
+    from unshackle.vaults.SQLite import SQLite
+
+    local = SQLite("local", tmp_path / "local.db")
+    vaults = Vaults("SVC")
+    vaults.vaults = [local]
+    vaults.flag_bad_key(KID, "ab" * 16)
+    monkeypatch.setattr(handlers, "load_server_vaults", lambda service: vaults)
+    monkeypatch.setattr(handlers, "CACHED_PAIRS", set())
+
+    handlers.cache_to_vaults({KID.hex: "ab" * 16}, "SVC")
+    assert not local.is_bad_key(KID, "ab" * 16)
+    assert local.get_key(KID, "SVC") == "ab" * 16

tests/remote/unit/test_server_cdm_vault_write.py

@@ -42,6 +42,9 @@ class FakeVaults:
     def is_flagged(self, kid, key):
         return False
 
+    def clear_licensed_flags(self, content_keys):
+        return []
+
     def add_keys(self, kid_keys):
         self.added.update(kid_keys)
         self.pushes.append(dict(kid_keys))
@@ -227,3 +230,21 @@ def test_cdm_only_trusts_server_vault_keys():
 
     assert vaults.sources == {}
     assert dl.LICENSE_KEY_CACHE == {CACHED: "k_cached"}
+
+
+@pytest.mark.parametrize("system", ["widevine", "playready"])
+def test_no_cdm_and_no_server_cdm_names_the_missing_cdm(system):
+    """The server does not license this service for the API key, and the client has no CDM for it."""
+    import click
+
+    from unshackle.core.drm import PlayReady, Widevine
+
+    cmd = make_cmd(FakeVaults())
+    cmd._remote_service = SimpleNamespace(_server_cdm=False)
+    cmd.cdm = None
+    cmd.cdm_for = lambda system, quality=None: None
+    drm_class = PlayReady if system == "playready" else Widevine
+    drm = drm_class.__new__(drm_class)
+    drm.content_keys = {}
+    with pytest.raises(click.ClickException, match="No CDM configured for TEST"):
+        prepare(cmd, drm, lambda **kwargs: None, NEW)

unshackle/commands/dl.py

@@ -4507,10 +4507,11 @@ class dl:
         then copied to the other vaults, once FFmpeg can decode the result. Without FFmpeg
         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 KID that is not in the map, and every KID
-        of any other file, gets a start and end check of the whole file, and a failure there flags
-        each of those KIDs. On failure the pair is flagged in the local vaults and reported
-        to the vault that served it, the ciphertext is restored, and the licence runs again. Every vault skips a
+        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
+        start and end check of the whole file, and a failure there flags each of those KIDs. On
+        failure the pair is flagged in the local vaults, the ciphertext is restored, and the
+        licence runs again; ``Vaults`` tells the vault that served it. Every vault skips a
         flagged pair, so each pass burns at most one vault; when the vaults run out the CDM
         answers, and a CDM key is not checked. A CDM that returns the flagged key clears the
         flag and stores the content key again, because then the decode check was wrong.
@@ -4574,6 +4575,13 @@ class dl:
         def window_decodes(start: float, end: float) -> bool:
             return ffmpeg_decodes(path, start=start, seconds=end - start if end > start else 0.5, video=video)
 
+        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):
+                return {kid: vault for kid, vault in kids.items() if kid in windows}
+            return kids
+
         def failed_kids(kids: dict[UUID, Vault]) -> set[UUID]:
             """The vault KIDs whose content key did not decode.
 
@@ -4604,20 +4612,21 @@ class dl:
                 done += 1
                 if decrypt:
                     drm.decrypt(path)
-                kids = vault_kids()
+                kids = judged_kids()
                 stale = flagged_kids()
                 failed: set[UUID] = set()
                 if not stale:
-                    if not kids or not binaries.FFMPEG or not path.exists():
-                        return
-                    failed = failed_kids(kids)
+                    checked = bool(kids and binaries.FFMPEG and path.exists())
+                    failed = failed_kids(kids) if checked else set()
                     if not failed:
-                        self.flush_vault_writes(
-                            [
-                                partial(self.vaults.add_key, kid, keys[kid], excluding=vault)
-                                for kid, vault in kids.items()
-                            ]
-                        )
+                        if checked:
+                            self.flush_vault_writes(
+                                [
+                                    partial(self.vaults.add_key, kid, keys[kid], excluding=vault)
+                                    for kid, vault in kids.items()
+                                ]
+                            )
+                        self.vaults.report_replaced(keys)
                         return
                 bad = {kid: drop(kid) for kid in {*failed, *stale}}
                 for kid, key in bad.items():
@@ -4625,21 +4634,23 @@ class dl:
                         continue
                     self.log.warning(f"{key} from {kids[kid].name} was bad, trying other vaults")
                     self.vaults.flag_bad_key(kid, key)
-                    if report := getattr(kids[kid], "report_bad", None):
-                        report(kid, key)
                 if not backup.exists() or not licence:
                     raise ValueError("The content key from the vault did not decrypt the track; run again")
                 path.unlink()
                 os.link(backup, path)
                 relicense(set(bad))
                 for kid, key in bad.items():
-                    if keys.get(kid) != key:
+                    new = keys.get(kid)
+                    if new is None:
                         continue
                     source = self.vaults.sources.get(kid)
                     if source is None:
-                        self.vaults.unflag_bad_key(kid, key)
-                        self.flush_vault_writes([partial(self.vaults.add_key, kid, key)])
-                    elif kid in kids and source[1] is kids[kid]:
+                        if new == key:
+                            self.vaults.unflag_bad_key(kid, key)
+                            self.flush_vault_writes([partial(self.vaults.add_key, kid, key)])
+                        elif not self.vaults.held_locally(kid, new):
+                            self.flush_vault_writes([partial(self.vaults.add_key, kid, new)])
+                    elif new == key and kid in kids and source[1] is kids[kid]:
                         raise ValueError(f"{key} from {kids[kid].name} was bad and no other source has the key")
             raise ValueError("No vault or CDM produced a content key that decrypts the track")
         finally:
@@ -4728,9 +4739,7 @@ class dl:
         flag came from a decode check that was wrong, and a local vault refuses a flagged pair."""
         if not content_keys or not self.vaults:
             return
-        for kid, key in content_keys.items():
-            if self.vaults.is_flagged(kid, key):
-                self.vaults.unflag_bad_key(kid, key)
+        self.vaults.clear_licensed_flags(content_keys)
         successful_caches = self.vaults.add_keys(content_keys)
         kids, caches = self.vault_cache_tally or (set(), successful_caches)
         self.vault_cache_tally = (kids | content_keys.keys(), min(caches, successful_caches))
@@ -5034,34 +5043,22 @@ class dl:
                     drm.content_keys.update(local_drm.content_keys)
                 return
 
-        licence = partial(licence, drm_system="playready" if isinstance(drm, PlayReady) else "widevine")
+        drm_system = "playready" if isinstance(drm, PlayReady) else "widevine"
+        licence = partial(licence, drm_system=drm_system)
 
-        if not server_cdm:
-            if isinstance(drm, Widevine):
-                if not is_widevine_cdm(self.cdm):
-                    widevine_cdm = self.cdm_for("widevine", track_quality)
-                    if widevine_cdm and is_widevine_cdm(widevine_cdm):
-                        if track_quality:
-                            self.log.info(f"Switching to Widevine CDM for Widevine {track_quality}p content")
-                        else:
-                            self.log.info("Switching to Widevine CDM for Widevine content")
-                        self.cdm = widevine_cdm
-                        track_cdm = widevine_cdm
-                    else:
-                        raise ValueError(f"Title needs a Widevine CDM but {self.service} is configured with PlayReady.")
-
-            elif isinstance(drm, PlayReady):
-                if not is_playready_cdm(self.cdm):
-                    playready_cdm = self.cdm_for("playready", track_quality)
-                    if playready_cdm and is_playready_cdm(playready_cdm):
-                        if track_quality:
-                            self.log.info(f"Switching to PlayReady CDM for PlayReady {track_quality}p content")
-                        else:
-                            self.log.info("Switching to PlayReady CDM for PlayReady content")
-                        self.cdm = playready_cdm
-                        track_cdm = playready_cdm
-                    else:
-                        raise ValueError(f"Title needs a PlayReady CDM but {self.service} is configured with Widevine.")
+        if not server_cdm and isinstance(drm, (Widevine, PlayReady)):
+            matches = is_playready_cdm if drm_system == "playready" else is_widevine_cdm
+            if not matches(self.cdm):
+                name, other = ("PlayReady", "Widevine") if drm_system == "playready" else ("Widevine", "PlayReady")
+                system_cdm = self.cdm_for(drm_system, track_quality)
+                if system_cdm and matches(system_cdm):
+                    self.log.info(f"Switching to the {name} CDM")
+                    self.cdm = system_cdm
+                    track_cdm = system_cdm
+                elif self.cdm is None:
+                    raise click.ClickException(f"No CDM configured for {self.service} and no default CDM")
+                else:
+                    raise ValueError(f"Title needs a {name} CDM but {self.service} is configured with {other}.")
 
         if isinstance(drm, Widevine):
             if self.debug_logger:

unshackle/core/api/handlers.py

@@ -4928,11 +4928,12 @@ CACHED_PAIRS: set[tuple[str, str, str]] = set()
 
 
 def cache_to_vaults(keys: Dict[str, str], service_name: str) -> None:
-    """Cache newly obtained keys to server vaults.
+    """Cache the content keys of a CDM licence to the server vaults.
 
     A PlayReady licence always runs, and a Widevine licence runs when one PSSH KID has no content
     key in a vault, so the same content keys come back on every request for a title. This process
-    does not send a pair again after every vault accepted it.
+    does not send a pair again after every vault accepted it. A pair a client reported as bad
+    leaves that set, and is unflagged when the CDM returns it.
     """
     from uuid import UUID
 
@@ -4945,6 +4946,8 @@ def cache_to_vaults(keys: Dict[str, str], service_name: str) -> None:
             return
 
         key_map = {UUID(hex=kid): key for kid, key in new.items()}
+        for kid in vaults.clear_licensed_flags(key_map):
+            log.info(f"Unflagged {kid.hex}:{key_map[kid]}, the CDM returned it")
         cached = vaults.add_keys(key_map)
         if cached == len([vault for vault in vaults.vaults if not vault.no_push]):
             CACHED_PAIRS.update((service_name, kid, key) for kid, key in new.items())
@@ -5617,7 +5620,8 @@ async def session_bad_key_handler(
 
     The server never sees a segment, so the client is the only side that can test a content key.
     Only a pair this remote session served can be flagged, which keeps a client from
-    poisoning the bad-key table for content keys it never received.
+    poisoning the bad-key table for content keys it never received. A pair the CDM licensed is
+    not flagged, because the CDM is the authority.
     """
     from uuid import UUID
 
@@ -5630,8 +5634,12 @@ async def session_bad_key_handler(
     if not key or served_key.lower() != key:
         log.warning(f"Session {sanitize_log(session_id[:12])} reported a bad content key it was never served: {kid}")
         raise APIError(APIErrorCode.INVALID_INPUT, "This session was not served that KID:KEY pair")
+    if source == "cdm":
+        log.warning(f"Session {sanitize_log(session_id[:12])} reported a CDM key as bad, not flagged: {kid}")
+        return web.json_response({"flagged": False})
 
-    vaults = load_server_vaults(session.service_instance.__class__.__name__)
+    service_name = session.service_instance.__class__.__name__
+    vaults = load_server_vaults(service_name)
 
     def flag() -> None:
         for vault in vaults.vaults:
@@ -5644,6 +5652,7 @@ async def session_bad_key_handler(
 
     await asyncio.to_thread(flag)
     session.served_keys.pop(kid, None)
+    CACHED_PAIRS.discard((service_name, kid, served_key))
     log.warning(
         f"Client proved {kid}:{served_key} from vault {sanitize_log(source)} wrong, flagged in the server vaults"
     )

unshackle/core/api/routes.py

@@ -2421,7 +2421,7 @@ async def session_bad_key(request: web.Request) -> web.Response:
                 description: Content key as hex
     responses:
       '200':
-        description: The pair is flagged
+        description: "flagged true when the pair is flagged; false when the CDM licensed it"
       '400':
         description: The remote session was not served that pair
       '404':

unshackle/core/remote_service.py

@@ -1045,9 +1045,11 @@ class ServerVault(Vault):
     ``decrypt_verified`` needs a Vault to hold in ``Vaults.sources`` so a server-supplied
     content key gets the same decode check as a local vault key. The client never learns
     which server vault answered and can neither read nor write it, so every method returns
-    nothing and a bad verdict goes to the server, which flags the row itself.
+    nothing. A bad verdict goes to the server at once, so its next licence skips the pair.
     """
 
+    report_now = True
+
     def __init__(self, service: "RemoteService"):
         super().__init__("server_cdm", no_push=True)
         self.service = service
@@ -1067,8 +1069,8 @@ class ServerVault(Vault):
     def get_services(self) -> Iterator[str]:
         return iter(())
 
-    def report_bad(self, kid: UUID, key: str) -> None:
-        self.service.report_bad_key(kid, key)
+    def flag_bad_key(self, service: str, kid: Union[UUID, str], key: str, source: str) -> None:
+        self.service.report_bad_key(UUID(str(kid)), key)
 
 
 class RemoteService:

unshackle/core/utils/subprocess.py

@@ -131,6 +131,7 @@ def ffmpeg_decodes(path: Path, start: Optional[float] = None, seconds: float = 3
             return True
         if not keyframes and video is not False and any(m in err for m in FFMPEG_MISSING_REFERENCE):
             return decode(window, keyframes=True)
+        log_tool_run("FFmpeg decode check", ffmpeg, ff.returncode, window=window, stderr=ff.stderr[-2000:])
         return False
 
     tail = ["-sseof", f"-{FFMPEG_CHECK_TAIL}", "-t", str(FFMPEG_CHECK_TAIL)]

unshackle/core/vault.py

@@ -51,6 +51,7 @@ def vault_session(timeout: float) -> Session:
 class Vault(metaclass=ABCMeta):
     # a vault on this machine; Vaults.get_key tries these first, before the network ones
     local: bool = False
+    report_now: bool = False
 
     def __init__(self, name: str, no_push: bool = False):
         self.name = name

unshackle/core/vaults.py

@@ -30,6 +30,7 @@ class Vaults:
         self.candidates: dict[Union[UUID, str], list[tuple[str, Vault]]] = {}
         self.warned: set[tuple[str, Union[UUID, str], str]] = set()
         self.unreachable: set[Vault] = set()
+        self.unreported: dict[tuple[Union[UUID, str], str], Vault] = {}
 
     def __iter__(self) -> Iterator[Vault]:
         return iter(self.vaults)
@@ -153,22 +154,42 @@ class Vaults:
         return key if found else None
 
     def flag_bad_key(self, kid: Union[UUID, str], key: str) -> None:
-        """Flag a KID:KEY that failed decryption, naming the Vault it came from.
-
-        Every local vault records it. A remote vault is told only when it served the pair,
-        so it stops serving it too, and a vault that never held it is not bothered. A vault
-        that cannot take the flag is skipped with a debug line: the local flag already
-        protects this run, and a vault without the route is not a fault.
-        """
+        """Flag a KID:KEY in every local vault. A failed decode is no proof, so the remote vault that
+        served it hears of it from ``report_replaced``, or at once when it sets ``report_now``."""
         self.flagged.add((kid, key))
         source = self.sources.pop(kid, None)
+        name = source[1].name if source else "unknown"
         for vault in self.live():
-            if not vault.local and (source is None or vault is not source[1]):
-                continue
-            try:
-                vault.flag_bad_key(self.service, kid, key, source[1].name if source else "unknown")
-            except Exception as e:
-                log.debug(f"Could not flag {key} as bad in Vault '{vault.name}': {e!r}")
+            if vault.local:
+                self.tell(vault, kid, key, name)
+        if source and not source[1].local:
+            if source[1].report_now:
+                self.tell(source[1], kid, key, name)
+            else:
+                self.unreported[kid, key] = source[1]
+
+    def report_replaced(self, content_keys: dict[Union[UUID, str], str]) -> None:
+        """Tell each remote vault about its flagged pair once ``content_keys`` holds another key for the KID."""
+        live = self.live()
+        for (kid, key), vault in list(self.unreported.items()):
+            if content_keys.get(kid) not in (None, key):
+                del self.unreported[kid, key]
+                if vault in live:
+                    self.tell(vault, kid, key, vault.name)
+
+    def tell(self, vault: Vault, kid: Union[UUID, str], key: str, source: str) -> None:
+        """Flag one pair in one vault; a failure is only a debug line."""
+        try:
+            vault.flag_bad_key(self.service, kid, key, source)
+        except Exception as e:
+            log.debug(f"Could not flag {key} as bad in Vault '{vault.name}': {e!r}")
+
+    def clear_licensed_flags(self, content_keys: dict[UUID, str]) -> list[UUID]:
+        """Unflag each pair a licence returned, since the decode check was wrong. Returns their KIDs."""
+        cleared = [kid for kid, key in content_keys.items() if self.is_flagged(kid, key)]
+        for kid in cleared:
+            self.unflag_bad_key(kid, content_keys[kid])
+        return cleared
 
     def unflag_bad_key(self, kid: Union[UUID, str], key: str) -> None:
         """Remove a flag from every Vault."""