feat(remote): verify server-CDM vault keys on the client and report bad pairs

The licence response lists KIDs a server vault supplied, the client proves them with the decode check, and a bad pair goes to the new session keys/bad route so the server flags its own row. Client vault keys go first over --remote, and the dashboard health warns when no SQLite vault can store flags.

docs/dev/rest-api/dashboard.md

@@ -210,8 +210,10 @@ missing or a vault's credentials expired. This finds that before a job does.
 `status` is `ok`, `degraded` (an optional dependency is missing) or `failing` (a required one
 is). Player binaries are not checked: a headless server never has one and it plays nothing.
 Each check is `ok`, `warn` or `fail`. Check ids are the binary names, `cdm`,
-`vault:<name>` (the vault's configured `name`, so two vaults of the same type stay apart), and
+`vault:<name>` (the vault's configured `name`, so two vaults of the same type stay apart),
-`proxies`.
+`bad_keys` and `proxies`. `bad_keys` warns when the vaults hold no `SQLite` entry, because only
+that backend stores the flag for a content key a remote client proved wrong; without it the
+server serves the same bad pair again.
 
 The server caches the result for 30 seconds and keeps every probe shallow: a vault answers one
 lookup that should miss, and no probe allocates anything. A failing probe's `detail` has the

docs/dev/rest-api/endpoints.md

@@ -849,6 +849,7 @@ POST /api/session/{id}/tracks   → tracks + chapters (with URLs) for one title
 POST /api/session/{id}/segments → resolve segment URLs / DRM / headers per track
 POST /api/session/{id}/segment_filter → HLS segments the service drops (ads, bumpers)
 POST /api/session/{id}/license  → proxy or server-CDM DRM licensing
+POST /api/session/{id}/keys/bad → report a server-vault content key that did not decrypt
 DELETE /api/session/{id}        → tear down, harvest updated cache
 ```
 
@@ -1058,13 +1059,13 @@ Get the content keys for the DRM. The `mode` field selects one of two modes.
 === "Single: Response `200`"
 
     ```json
-    { "keys": { "<kid_hex>": "<key_hex>" } }
+    { "keys": { "<kid_hex>": "<key_hex>" }, "vault_keys": ["<kid_hex>"] }
     ```
 
 === "Batch: Response `200`"
 
     ```json
-    { "keys": { "v-1": { "<kid_hex>": "<key_hex>" } }, "drm_type": "widevine" }
+    { "keys": { "v-1": { "<kid_hex>": "<key_hex>" } }, "vault_keys": ["<kid_hex>"], "drm_type": "widevine" }
     ```
 
 | Field | Description |
@@ -1074,6 +1075,7 @@ Get the content keys for the DRM. The `mode` field selects one of two modes.
 | `pssh` | Base64 PSSH (server-CDM mode). |
 | `drm_type` | `widevine` (default) or `playready`. |
 | `mode` | `proxy` (default) or `server_cdm`. |
+| `vault_keys` (response) | KIDs whose content key came from a server vault, not the CDM. The client proves such a content key with a decode before it trusts it; the field is absent when every content key came from the CDM. |
 
 | Status | Error code | Meaning |
 | --- | --- | --- |
@@ -1084,6 +1086,20 @@ Get the content keys for the DRM. The `mode` field selects one of two modes.
 | `404` | `NO_CONTENT` | Server CDM produced no keys. |
 | `502` | `SERVICE_ERROR` | The service aborted during licensing. |
 
+### `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, under the name of the vault that served it, and the next licence for that KID reaches the CDM. The client never learns the server vault names.
+
+```json
+{ "kid": "<kid_hex>", "key": "<key_hex>" }
+```
+
+| Status | Error code | Meaning |
+| --- | --- | --- |
+| `200` | - | The pair is flagged. |
+| `400` | `INVALID_INPUT` | This remote session was not served that pair. |
+| `404` | `SESSION_NOT_FOUND` | Unknown remote session. |
+
 ### `GET /api/session/{session_id}`
 
 Return remote session metadata and validity.

docs/dev/rest-api/index.md

@@ -197,6 +197,7 @@ request and response in full.
 | `POST` | `/api/session/{session_id}/segments` | Per-track segment/download details. |
 | `POST` | `/api/session/{session_id}/segment_filter` | Server-side HLS segment filter results for one track. |
 | `POST` | `/api/session/{session_id}/license` | Proxy or server-side DRM licensing. |
+| `POST` | `/api/session/{session_id}/keys/bad` | Report a server-vault content key that did not decrypt. |
 | `GET` | `/api/session/{session_id}/logs` | Drain the service's server-side log output. |
 | `GET` | `/api/session/{session_id}/prompt` | Poll for interactive auth status/prompt. |
 | `POST` | `/api/session/{session_id}/prompt` | Submit an answer to an auth prompt. |

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

@@ -236,6 +236,17 @@ There are two ways DRM keys get resolved, chosen by the client's `server_cdm` fl
     - Enable with `server_cdm: true` in the server entry.
     - The server tells the client which DRM type it used
       (`widevine` or `playready`).
+    - A content key the server took from its own vault is unproven. The response
+      lists its KID in `vault_keys`, the client decrypts and checks that the output
+      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, records the name of 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.
+    - 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.
 
 ---
 
@@ -313,6 +324,7 @@ The server mounts all these routes, even in `--remote-only` mode. Paths use the
 | `POST` | `/api/session/{id}/segments` | Resolve per-segment/track download descriptors |
 | `POST` | `/api/session/{id}/segment_filter` | Unwanted HLS segment URIs for one track (ads, bumpers) |
 | `POST` | `/api/session/{id}/license` | DRM licensing (proxy or server CDM) |
+| `POST` | `/api/session/{id}/keys/bad` | Report a server-vault content key that did not decrypt |
 | `GET` | `/api/session/{id}/logs` | Drain the service's server-side log output |
 | `GET` | `/api/session/{id}/prompt` | Poll interactive auth status / pending prompt |
 | `POST` | `/api/session/{id}/prompt` | Submit an answer to a pending prompt |

docs/guide/downloading.md

@@ -762,7 +762,7 @@ unshackle dl -w S01 --slow 30-60 EXAMPLE 81234567
 By default, unshackle checks your **key vaults** first and only asks a **CDM** to license
 a content key when the vault misses. You can force one side or the other:
 
-- `--cdm-only`: only use the CDM (skip vaults).
+- `--cdm-only`: only use the CDM (skip vaults). Over a `server_cdm` remote session the flag does not reach the server, whose vault shortcut still runs; it only skips the client's decode check on a server key.
 - `--cdm <name>`: use the named CDM device for this run, ignoring the `cdm` config mapping (including its quality and Widevine/PlayReady sub-entries).
 - `--vaults-only`: only use key vaults and never license through the CDM. A missing
   content key fails.

docs/guide/drm-and-cdm.md

@@ -497,7 +497,7 @@ control this trade-off on the `dl` command:
 | Flag | Effect |
 |------|--------|
 | `--vaults-only` | Never send a license request; use only cached/vault keys. Fails if a content key is missing. |
-| `--cdm-only` | Ignore vaults for sourcing keys and always perform a fresh license request. |
+| `--cdm-only` | Ignore vaults for sourcing keys and always perform a fresh license request. Over a `server_cdm` remote session the server still reads its own vaults; the flag only skips the client's decode check on the key it gets back. |
 | `--cdm <name>` | Use the named CDM device for this run instead of the one the `cdm` config maps for the service. |
 
 See [key vaults](vaults.md) for how to configure vault backends.

docs/reference/configuration/vaults.md

@@ -39,6 +39,14 @@ key_vaults:
     - unshackle treats an all-zero content key (32 zeros) as "no key" everywhere and never
       stores it.
 
+!!! note "Bad content keys"
+    A content key from a vault is proven by a short FFmpeg decode of the decrypted output.
+    A content key that fails the decode is written to the `bad_keys` table of every `SQLite`
+    vault with the name of the vault that supplied it, and every vault lookup skips a flagged
+    pair from then on. Only the `SQLite` backend stores flags. A `serve` instance with
+    `server_cdm` flags a pair the same way when a remote client reports it, so a server whose
+    `key_vaults` hold no `SQLite` entry cannot remember a bad content key.
+
 ## `vault_timeout`
 
 - **Type:** `float`  ·  **Default:** `10.0`

tests/core/test_vault_bad_keys.py

@@ -296,3 +296,38 @@ def test_flag_on_a_read_only_vault_keeps_its_rows(tmp_path: Path) -> None:
     assert ro.is_bad_key(KID, BAD)
     assert ro.get_key(KID, "SVC") is None  # the lookup still skips the flagged pair
     assert list(ro.get_keys("SVC")) == [(KID.hex, BAD)]  # but the row itself is untouched
+
+
+def test_a_sibling_verdict_during_the_decrypt_still_triggers_the_retry(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Two tracks share a KID and decrypt at the same time; the first verdict pops the shared
+    source, so the second track must read the flag instead of returning unchecked."""
+    local = SQLite("local", tmp_path / "local.db")
+    vaults = Vaults("SVC")
+    vaults.vaults = [local]
+    vaults.sources[KID] = (BAD, local)
+    cmd = make_cmd(monkeypatch, vaults)
+    dl.LICENSE_KEY_CACHE[KID] = GOOD  # the sibling's retry already licensed the good key
+
+    path = tmp_path / "audio.mp4"
+    path.write_bytes(b"cipher")
+
+    class SiblingFlagsMeanwhile(FakeDRM):
+        def decrypt(self, path: Path) -> None:
+            super().decrypt(path)
+            if self.content_keys[KID] == BAD:
+                vaults.flag_bad_key(KID, BAD)
+
+    drm = SiblingFlagsMeanwhile({KID: BAD})
+    licences: list = []
+
+    def licence(drm: FakeDRM, track_kid: UUID) -> None:
+        licences.append(dl.LICENSE_KEY_CACHE.get(KID))
+        drm.content_keys[KID] = dl.LICENSE_KEY_CACHE[KID]
+
+    cmd.decrypt_verified(drm, path, licence, KID)
+
+    assert path.read_bytes() == b"plain:" + GOOD.encode()
+    assert drm.decrypted_with == [{KID: BAD}, {KID: GOOD}]
+    assert licences == [GOOD]  # the good key in the run cache survives the flag of the bad one

tests/remote/unit/test_dashboard.py

@@ -351,6 +351,19 @@ async def test_health_vault_probe_tolerates_api_vault_rejecting_the_probe(aiohtt
     assert checks["vault:t"]["status"] == "ok", checks["vault:t"]
 
 
+def test_health_warns_without_a_sqlite_vault_to_store_flags(monkeypatch) -> None:
+    from unshackle.core.api.handlers import run_health_checks
+
+    monkeypatch.setattr(config, "proxy_providers", {})
+    monkeypatch.setattr(
+        config, "key_vaults", [{"type": "API", "name": "shared", "uri": "http://127.0.0.1:9", "token": "t"}]
+    )
+    assert {c["id"]: c for c in run_health_checks()}["bad_keys"]["status"] == "warn"
+
+    monkeypatch.setattr(config, "key_vaults", [])
+    assert {c["id"]: c for c in run_health_checks()}["bad_keys"]["status"] == "ok"
+
+
 def test_health_proxy_probe_reports_a_provider_that_fails_to_build(monkeypatch) -> None:
     from unshackle.core.api.handlers import run_health_checks
 

tests/remote/unit/test_extract_pssh_strict.py

@@ -46,13 +46,14 @@ async def _run_batch(monkeypatch, tracks, cdm_type, calls):
     session = SimpleNamespace(
         service_tag="EXAMPLE",
         service_instance=SimpleNamespace(),
+        served_keys={},
         tracks=SimpleNamespace(get=tracks.get),
     )
 
     async def fake_get_session(session_id, request):
         return session
 
-    def fake_single(service, title, track, pssh_str, drm_type, request):
+    def fake_single(service, title, track, pssh_str, drm_type, request, sources=None):
         calls.append((pssh_str, drm_type))
         return {"00" * 16: "key"}
 

tests/remote/unit/test_server_cdm_absorb_routes.py

@@ -79,6 +79,7 @@ async def test_single_track_path_keeps_siblings(env, monkeypatch):
     session = SimpleNamespace(
         service_tag="EX",
         service_instance=SimpleNamespace(),
+        served_keys={},
         tracks=SimpleNamespace(get={"vid": track}.get),
         log_buffer=None,
     )
@@ -112,6 +113,7 @@ async def test_batch_cache_does_not_starve_second_track(env, monkeypatch):
     session = SimpleNamespace(
         service_tag="EX",
         service_instance=SimpleNamespace(),
+        served_keys={},
         tracks=SimpleNamespace(get=tracks.get),
         log_buffer=None,
     )
@@ -142,6 +144,7 @@ async def test_no_pssh_tells_the_client_why(env, monkeypatch):
     session = SimpleNamespace(
         service_tag="EX",
         service_instance=SimpleNamespace(),
+        served_keys={},
         tracks=SimpleNamespace(get={"vid": track}.get),
         log_buffer=buf,
     )

tests/remote/unit/test_server_cdm_key_verification.py

@@ -0,0 +1,134 @@
+"""A server vault key is unproven until the client decodes with it.
+
+Over --remote with server_cdm the server never sees a segment, so the client holds a
+vault-sourced key back from its own vaults, checks the decode, and reports a bad pair to
+the server. The client never learns which server vault answered: the server keeps the
+name on its served-key record and flags the row itself, only for a pair it served.
+"""
+
+import json
+from types import SimpleNamespace
+from uuid import UUID
+
+import pytest
+
+from unshackle.core.api import handlers
+from unshackle.core.api.errors import APIError
+from unshackle.core.api.session_store import SessionEntry
+from unshackle.core.remote_service import RemoteService, ServerVault
+
+pytestmark = pytest.mark.unit
+
+KID = UUID("161219ec3df64e0eadbb3827c0ccc46d")
+CDM_KID = UUID("43de497c6d1945f49acc2951696ec628")
+
+
+def remote_service() -> RemoteService:
+    svc = RemoteService.__new__(RemoteService)
+    svc.server_vault_keys = {}
+    svc.server_vault = ServerVault(svc)
+    svc._session_id = "sess"
+    svc.log = SimpleNamespace(warning=lambda *a, **k: None)
+    return svc
+
+
+def test_client_marks_vault_keys_and_forgets_cdm_keys():
+    svc = remote_service()
+    svc.note_vault_keys({KID.hex: "v", CDM_KID.hex: "c"}, {KID.hex})
+    assert svc.server_vault_keys == {KID: "v"}
+
+    # a re-licence that came from the CDM clears the marker
+    svc.note_vault_keys({KID.hex: "v2"}, set())
+    assert svc.server_vault_keys == {}
+
+
+def test_report_bad_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 posted == [("/api/session/sess/keys/bad", {"kid": KID.hex, "key": "v"})]
+
+
+class _Vault:
+    local = True
+
+    def __init__(self):
+        self.flagged: list = []
+
+    def flag_bad_key(self, service, kid, key, source):
+        self.flagged.append((service, kid, key, source))
+
+
+@pytest.fixture
+def session(monkeypatch):
+    entry = SessionEntry(session_id="sess", service_tag="SVC", service_instance=SimpleNamespace())
+    handlers.note_served_keys(entry, {KID.hex: "v", CDM_KID.hex: "c"}, {KID.hex: "sqlite"})
+    assert entry.served_keys == {KID.hex: ("v", "sqlite"), CDM_KID.hex: ("c", "cdm")}
+
+    async def validated(session_id, request):
+        return entry
+
+    monkeypatch.setattr(handlers, "get_validated_session", validated)
+    monkeypatch.setattr(handlers, "require_authenticated", lambda s: None)
+    vault = _Vault()
+    monkeypatch.setattr(handlers, "load_server_vaults", lambda name: SimpleNamespace(vaults=[vault], service="SVC"))
+    return entry, vault
+
+
+async def test_server_flags_a_served_pair_with_its_own_vault_name(session):
+    entry, vault = session
+    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
+
+    with pytest.raises(APIError):
+        await handlers.session_bad_key_handler({"kid": CDM_KID.hex, "key": "nope"}, "sess")
+    assert len(vault.flagged) == 1
+
+
+class _WV:
+    def __init__(self, b64, kids):
+        self._pssh = SimpleNamespace(dumps=lambda: b64)
+        self.kids = kids
+
+
+_WV.__name__ = "Widevine"
+
+
+async def test_batch_reports_vault_keys_from_the_init_segment_pssh(monkeypatch):
+    """A vault hit on the init-segment PSSH retry must reach `vault_keys` like a manifest one."""
+    track = SimpleNamespace(id="vid", drm=[_WV("manifest", [KID])], get_key_id=lambda init: CDM_KID)
+    session = SessionEntry(session_id="sess", service_tag="SVC", service_instance=SimpleNamespace())
+    session.tracks = {"vid": track}
+
+    async def validated(sid, req):
+        return session
+
+    def fake_single(service, title, track, pssh_str, drm_type, request, sources=None):
+        if pssh_str == "init":
+            sources[CDM_KID.hex] = "sqlite"
+            return {CDM_KID.hex: "v"}
+        return {KID.hex: "c"}
+
+    monkeypatch.setattr(handlers, "get_validated_session", validated)
+    monkeypatch.setattr(handlers, "server_cdm_allowed", lambda request, tag: True)
+    monkeypatch.setattr(handlers, "ensure_track_drm", lambda track, session=None, init_data=None: None)
+    monkeypatch.setattr(handlers, "find_title_for_track", lambda tid, session: SimpleNamespace())
+    monkeypatch.setattr(handlers, "detect_cdm_type_for_service", lambda tag, cfg: "widevine")
+    monkeypatch.setattr(handlers, "fetch_init_segment", lambda track, sess: b"init")
+    monkeypatch.setattr(
+        handlers, "drm_from_init_segment", lambda track, session=None, init_data=None: [_WV("init", [CDM_KID])]
+    )
+    monkeypatch.setattr(handlers, "handle_single_server_cdm", fake_single)
+    monkeypatch.setattr(handlers.config, "serve", {"users": {}}, raising=False)
+
+    resp = await handlers.session_license_handler(
+        {"mode": "server_cdm", "track_ids": ["vid"], "drm_type": "widevine"}, "sess", None
+    )
+
+    payload = json.loads(resp.body)
+    assert payload["keys"] == {"vid": {CDM_KID.hex: "v"}}
+    assert payload["vault_keys"] == [CDM_KID.hex]
+    assert session.served_keys == {CDM_KID.hex: ("v", "sqlite")}

tests/remote/unit/test_server_cdm_manifest_kid.py

@@ -102,6 +102,7 @@ def _session(tracks: dict, buf=None):
     session = SimpleNamespace(
         service_tag="EX",
         service_instance=SimpleNamespace(),
+        served_keys={},
         tracks=SimpleNamespace(get=tracks.get),
         log_buffer=buf,
     )

tests/remote/unit/test_server_cdm_multikey.py

@@ -98,13 +98,15 @@ def test_widevine_still_uses_vault_shortcut(monkeypatch):
     monkeypatch.setattr(drm_mod, "Widevine", _FakeWidevine)
     monkeypatch.setattr(handlers, "ensure_track_drm", lambda track, session=None, init_data=None: None)
     monkeypatch.setattr(handlers, "resolve_device_name", lambda *a, **k: "dev")
-    monkeypatch.setattr(handlers, "check_vaults", lambda kids, name: {kid.hex: "vaultkey"})
+    monkeypatch.setattr(handlers, "check_vaults", lambda kids, name: ({kid.hex: "vaultkey"}, {kid.hex: "sqlite"}))
     monkeypatch.setattr(handlers.config, "serve", {"users": {}}, raising=False)
 
+    sources: dict = {}
     keys = handlers.handle_single_server_cdm(
-        SimpleNamespace(), SimpleNamespace(), SimpleNamespace(), PSSH_B64, "widevine", None
+        SimpleNamespace(), SimpleNamespace(), SimpleNamespace(), PSSH_B64, "widevine", None, sources
     )
     assert keys == {kid.hex: "vaultkey"}
+    assert sources == {kid.hex: "sqlite"}  # the server keeps the name; the client only learns it is vault output
 
 
 async def test_batch_shares_full_bundle_per_pssh(monkeypatch):
@@ -118,6 +120,7 @@ async def test_batch_shares_full_bundle_per_pssh(monkeypatch):
     session = SimpleNamespace(
         service_tag="EXAMPLE",
         service_instance=SimpleNamespace(),
+        served_keys={},
         tracks=SimpleNamespace(get=tracks.get),
     )
 
@@ -128,7 +131,7 @@ async def test_batch_shares_full_bundle_per_pssh(monkeypatch):
     bundle_a = {REAL.hex: "k_real", PSSH_KID.hex: "k_pssh", OTHER.hex: "k_other"}
     bundle_b = {"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa": "k_b"}
 
-    def fake_single(service, title, track, pssh_str, drm_type, request):
+    def fake_single(service, title, track, pssh_str, drm_type, request, sources=None):
         calls.append(pssh_str)
         return dict(bundle_a) if pssh_str == pssh_a else dict(bundle_b)
 

tests/remote/unit/test_server_cdm_vault_write.py

@@ -28,6 +28,7 @@ class FakeVaults:
         self.added: dict = {}
         self.pushes: list = []
         self.replicated: list = []
+        self.sources: dict = {}
 
     def __len__(self):
         return 1
@@ -130,6 +131,37 @@ def test_vault_hits_wait_for_the_decrypt_before_replicating():
     assert vaults.added == {}
 
 
+def test_server_vault_keys_are_held_back_until_verified():
+    """A key the server took from its vault is sourced like a local vault hit and not pushed yet."""
+    from unshackle.core.remote_service import ServerVault
+
+    vaults = FakeVaults()
+    cmd = make_cmd(vaults)
+    remote = cmd._remote_service
+    remote.server_vault_keys = {CACHED: "k_cached"}
+    stub = ServerVault(remote)
+    remote.server_vault = stub
+
+    drm = FakeWidevine([CACHED, NEW])
+
+    def licence(**kwargs):
+        drm.content_keys.update({CACHED: "k_cached", NEW: "k_new"})
+
+    prepare(cmd, drm, licence, NEW)
+    cmd.wait_vault_writes()
+
+    assert vaults.added == {NEW: "k_new"}
+    assert vaults.sources == {CACHED: ("k_cached", stub)}
+
+    # the batch path caches at its own call site and must hold the same key back
+    drm2 = FakeWidevine([CACHED])
+    drm2.content_keys = {CACHED: "k_cached"}
+    dl.LICENSE_KEY_CACHE.clear()
+    cmd.cache_resolved_keys(SimpleNamespace(tracks=[SimpleNamespace(drm=[drm2])]))
+    cmd.wait_vault_writes()
+    assert vaults.added == {NEW: "k_new"}
+
+
 def test_tracks_sharing_one_drm_licence_and_push_once():
     """Two tracks on the same DRM run at the same time; the second must find the first one's keys."""
     vaults = FakeVaults()
@@ -171,3 +203,21 @@ def test_cdm_only_still_reuses_keys_from_the_run_cache():
 
     assert len(calls) == 1
     assert vaults.pushes == [{NEW: "k_new"}]
+
+
+def test_cdm_only_trusts_server_vault_keys():
+    """--cdm-only skips the vault checks, so a server vault key is not sourced for a decode check."""
+    from unshackle.core.remote_service import ServerVault
+
+    vaults = FakeVaults()
+    cmd = make_cmd(vaults)
+    remote = cmd._remote_service
+    remote.server_vault_keys = {CACHED: "k_cached"}
+    remote.server_vault = ServerVault(remote)
+
+    drm = FakeWidevine([CACHED])
+    prepare(cmd, drm, lambda **kw: drm.content_keys.update({CACHED: "k_cached"}), CACHED, cdm_only=True)
+    cmd.wait_vault_writes()
+
+    assert vaults.sources == {}
+    assert dl.LICENSE_KEY_CACHE == {CACHED: "k_cached"}

unshackle/commands/dl.py

@@ -2921,6 +2921,8 @@ class dl:
 
             if hasattr(service, "resolve_server_keys"):
                 service.resolve_server_keys(title)
+                if not cdm_only:
+                    self.prefer_vault_keys(title)
                 self.cache_resolved_keys(title)
 
             dl_start_time = time.time()
@@ -3997,11 +3999,21 @@ class dl:
             if decrypt:
                 drm.decrypt(path)
             return
-        known_bad = [kid for kid, key in keys.items() if (kid, key) in self.vaults.flagged]
+
+        def flagged_kids() -> list[UUID]:
+            """The KIDs on this track whose content key another track already proved wrong."""
+            return [kid for kid, key in keys.items() if (kid, key) in self.vaults.flagged]
+
+        def drop(kid: UUID) -> str:
+            key = keys.pop(kid)
+            if self.LICENSE_KEY_CACHE.get(kid) == key:
+                self.LICENSE_KEY_CACHE.pop(kid)
+            return key
+
+        known_bad = flagged_kids()
         if known_bad and licence:
             for kid in known_bad:
-                keys.pop(kid)
+                drop(kid)
-                self.LICENSE_KEY_CACHE.pop(kid, None)
             licence(drm, track_kid=track_kid)
 
         def vault_kids() -> dict[UUID, Vault]:
@@ -4016,23 +4028,40 @@ class dl:
                 os.link(path, backup)
             except OSError as e:
                 self.log.debug(f"Cannot keep the ciphertext for a decrypt retry: {e!r}")
+
+        def passes_left(done: int) -> bool:
+            """One pass per source that can still answer: every loaded vault, the server's vaults as
+            one source that walks them itself over --remote, and one for the CDM."""
+            remote = 1 if getattr(getattr(self, "_remote_service", None), "_server_cdm", False) else 0
+            return done < len(self.vaults) + remote + 1
+
         try:
-            for _ in range(len(self.vaults) + 1):
+            done = 0
+            while passes_left(done):
+                done += 1
                 if decrypt:
                     drm.decrypt(path)
                 kids = vault_kids()
-                if not kids or not binaries.FFMPEG or not path.exists():
+                stale = flagged_kids()
-                    return
+                if not stale:
-                if ffmpeg_decodes(path):
+                    if not kids or not binaries.FFMPEG or not path.exists():
-                    self.flush_vault_writes(
+                        return
-                        [partial(self.vaults.add_key, kid, keys[kid], excluding=vault) for kid, vault in kids.items()]
+                    if ffmpeg_decodes(path):
-                    )
+                        self.flush_vault_writes(
-                    return
+                            [
-                bad = {kid: keys.pop(kid) for kid in kids}
+                                partial(self.vaults.add_key, kid, keys[kid], excluding=vault)
+                                for kid, vault in kids.items()
+                            ]
+                        )
+                        return
+                bad = {kid: drop(kid) for kid in {*kids, *stale}}
                 for kid, key in bad.items():
+                    if kid not in kids:
+                        continue
                     self.log.warning(f"{key} from {kids[kid].name} was bad, trying other vaults")
                     self.vaults.flag_bad_key(kid, key)
-                    self.LICENSE_KEY_CACHE.pop(kid, None)
+                    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()
@@ -4044,7 +4073,7 @@ class dl:
                     source = self.vaults.sources.get(kid)
                     if source is None:
                         self.vaults.unflag_bad_key(kid, key)
-                    elif source[1] is kids[kid]:
+                    elif 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:
@@ -4070,18 +4099,49 @@ class dl:
         self.vault_cache_tally = None
         self.log.info(f"Cached {keys} Key{'' if keys == 1 else 's'} to {successful_caches}/{len(self.vaults)} Vaults")
 
+    def prefer_vault_keys(self, title: Title_T) -> None:
+        """Let the client vaults answer before a content key the server batch licence returned.
+
+        The client only learns the KIDs from the batch response, so the server has already
+        answered. A vault key still goes first, so ``decrypt_verified`` proves it and flags a
+        poisoned row; the server key waits as the next candidate. A server CDM key is trusted
+        like a local CDM key: it goes to the run cache and the vaults, and a retry after a
+        flagged vault key takes it before any vault. A server vault key is unproven, so it
+        waits in ``Vaults.candidates`` and gets the same check when its turn comes.
+        """
+        server_vault_keys = getattr(self._remote_service, "server_vault_keys", {})
+        for track in title.tracks:
+            for drm in getattr(track, "drm", None) or []:
+                for kid, server_key in list(getattr(drm, "content_keys", {}).items()):
+                    vault_key, vault = self.vaults.get_key(kid)
+                    if not vault_key or vault_key == server_key:
+                        continue
+                    drm.content_keys[kid] = vault_key
+                    if server_vault_keys.get(kid) == server_key:
+                        source = self._remote_service.server_vault
+                        self.vaults.candidates.setdefault(kid, []).append((server_key, source))
+                    elif kid not in self.LICENSE_KEY_CACHE:
+                        self.LICENSE_KEY_CACHE[kid] = server_key
+                        self.flush_vault_writes([partial(self.cache_keys_to_vaults, {kid: server_key})])
+                    self.log.debug(
+                        f"{vault.name} holds a different key for {kid.hex} than the server, testing it first"
+                    )
+
     def cache_resolved_keys(self, title: Title_T) -> None:
         """Cache the keys a batch licence or an import put on the tracks.
 
         Both fill track.drm before any prepare_drm call, so this is the only point where those
         keys can still reach the local vaults.
         """
+        server_vault_keys = getattr(getattr(self, "_remote_service", None), "server_vault_keys", {})
         keys = {
             kid: key
             for track in title.tracks
             for drm in (getattr(track, "drm", None) or [])
             for kid, key in getattr(drm, "content_keys", {}).items()
             if kid not in self.LICENSE_KEY_CACHE
+            and server_vault_keys.get(kid) != key
+            and self.vaults.sources.get(kid, (None,))[0] != key
         }
         if keys:
             self.LICENSE_KEY_CACHE.update(keys)
@@ -4181,7 +4241,15 @@ class dl:
                     except Exception as e:
                         self.log.debug(f"Server CDM licence with an empty challenge failed: {e!r}")
 
-                new_keys = {kid: key for kid, key in drm.content_keys.items() if kid not in known_keys}
+                server_vault_keys = {} if cdm_only else getattr(svc_for_cdm, "server_vault_keys", {})
+                for kid, key in drm.content_keys.items():
+                    if server_vault_keys.get(kid) == key:
+                        self.vaults.sources[kid] = (key, svc_for_cdm.server_vault)
+                new_keys = {
+                    kid: key
+                    for kid, key in drm.content_keys.items()
+                    if kid not in known_keys and server_vault_keys.get(kid) != key
+                }
                 if new_keys:
                     self.LICENSE_KEY_CACHE.update(new_keys)
                     pending_vault_writes.append(partial(self.cache_keys_to_vaults, new_keys))

unshackle/commands/serve.py

@@ -267,6 +267,8 @@ def serve(
             regions = server_account_regions(tag) or {}
             covered = list(regions.get("regions") or []) + (["global"] if regions.get("global") else [])
             log.info(f"Server accounts for {tag}: {', '.join(covered)}")
+        if config.key_vaults and not any(v.get("type") == "SQLite" for v in config.key_vaults):
+            log.warning("No SQLite key vault configured: a content key a remote client proves wrong cannot be flagged")
         users = config.serve.get("users", {})
         if isinstance(users, dict):
             # yaml keys can parse as int; hmac.compare_digest and allowlist lookups need str

unshackle/core/api/handlers.py

@@ -2279,6 +2279,16 @@ def run_health_checks(checks: Optional[List[Dict[str, Any]]] = None) -> List[Dic
 
         checks.append(health_check(f"vault:{name}", f"vault {name}", vault_probe, config_secrets(vault_config)))
 
+    def bad_keys_probe() -> tuple[str, str]:
+        if not config.key_vaults:
+            return "ok", "no vaults configured"
+        local = [v.get("name") or v["type"] for v in config.key_vaults if v.get("type") == "SQLite"]
+        if local:
+            return "ok", f"flags stored in {', '.join(local)}"
+        return "warn", "no SQLite vault, so a content key a client proves wrong cannot be flagged"
+
+    checks.append(health_check("bad_keys", "bad content key flags", bad_keys_probe))
+
     def proxy_probe() -> tuple[str, str]:
         providers = initialize_proxy_providers(raise_errors=True, quiet=True)
         if not providers:
@@ -3780,10 +3790,11 @@ def load_server_vaults(service_name: str) -> Any:
     return vaults
 
 
-def check_vaults(kids: list, service_name: str) -> Optional[Dict[str, str]]:
+def check_vaults(kids: list, service_name: str) -> Optional[tuple[Dict[str, str], Dict[str, str]]]:
     """Examine the server vaults for existing content keys that match all KIDs.
 
-    Returns a `KID:KEY` dict if ALL KIDs are found, None otherwise.
+    Returns `(KID:KEY, KID:vault name)` if ALL KIDs are found, None otherwise. The vault
+    name travels to the client, which is the only side that can prove the content key decrypts.
     """
     from uuid import UUID
 
@@ -3792,16 +3803,18 @@ def check_vaults(kids: list, service_name: str) -> Optional[Dict[str, str]]:
         if not vaults.vaults:
             return None
         keys: Dict[str, str] = {}
+        sources: Dict[str, str] = {}
         for kid in kids:
             kid_uuid = kid if isinstance(kid, UUID) else UUID(hex=str(kid))
             content_key, vault_used = vaults.get_key(kid_uuid)
             if content_key:
                 keys[kid_uuid.hex] = content_key
+                sources[kid_uuid.hex] = vault_used.name if vault_used else "unknown"
             else:
                 return None
         if keys:
             log.info(f"Vault hit: {len(keys)} key(s) from server vaults, skipping CDM")
-            return keys
+            return keys, sources
     # vault lookup is a best-effort shortcut before the CDM; any failure falls through to licensing
     except Exception as e:
         log.debug(f"Server vault lookup failed: {e!r}")
@@ -3899,8 +3912,13 @@ def handle_single_server_cdm(
     pssh_b64: Optional[str],
     drm_type: str,
     request: Optional[web.Request],
+    sources: Optional[Dict[str, str]] = None,
 ) -> Dict[str, str]:
-    """Do the single-track server_cdm licensing with the DRM class get_content_keys() flow."""
+    """Do the single-track server_cdm licensing with the DRM class get_content_keys() flow.
+
+    ``sources`` is filled with the vault name for every returned content key a server vault
+    supplied. A content key the CDM licensed gets no entry.
+    """
     import base64
 
     from unshackle.core.cdm import load_cdm
@@ -3955,8 +3973,11 @@ def handle_single_server_cdm(
         # harvest server-side keys from the vault fallback below.
         device_name = resolve_device_name(user_config, drm_type, service.__class__.__name__)
 
-        vault_keys = check_vaults(wv_drm.kids, service.__class__.__name__)
+        vault_hit = check_vaults(wv_drm.kids, service.__class__.__name__)
-        if vault_keys:
+        if vault_hit:
+            vault_keys, vault_sources = vault_hit
+            if sources is not None:
+                sources.update(vault_sources)
             return vault_keys
 
         cdm = load_cdm(device_name, service_name=service.__class__.__name__)
@@ -4113,8 +4134,10 @@ async def session_license_handler(
         config_cdm_type = detect_cdm_type_for_service(service_tag, app_config)
 
         all_keys: Dict[str, Dict[str, str]] = {}
+        vault_keys: list[str] = []
         drm_types: Dict[str, str] = {}
         keys_by_pssh: Dict[tuple, Dict[str, str]] = {}
+        sources_by_pssh: Dict[tuple, Dict[str, str]] = {}
         drm_type_by_pssh: Dict[tuple, str] = {}
         actual_drm_type: Optional[str] = None
 
@@ -4162,8 +4185,11 @@ async def session_license_handler(
             cache_key = (pssh_str, pssh_set(track))
             if cache_key not in keys_by_pssh:
                 keys_by_pssh[cache_key] = {}
+                sources_by_pssh[cache_key] = {}
                 try:
-                    keys = handle_single_server_cdm(service, title, track, pssh_str, candidate, request)
+                    keys = handle_single_server_cdm(
+                        service, title, track, pssh_str, candidate, request, sources_by_pssh[cache_key]
+                    )
                     if keys:
                         keys_by_pssh[cache_key] = keys
                         drm_type_by_pssh[cache_key] = candidate
@@ -4227,18 +4253,24 @@ async def session_license_handler(
                         f"{sanitize_log(tid[:12])}, tried the init segment PSSH"
                     )
                 if track_kid.hex in init_keys:
-                    keys, track_drm_type = init_keys, init_drm_type
+                    keys, track_drm_type, pssh_str = init_keys, init_drm_type, init_pssh
                 else:
                     track.drm = manifest_drm
                     warn(f"No content key for KID {track_kid.hex} of track {sanitize_log(tid[:12])}")
 
             if keys:
                 all_keys[tid] = keys
+                sources = sources_by_pssh.get((pssh_str, pssh_set(track)), {}) if pssh_str is not None else {}
+                note_served_keys(session, keys, sources)
+                if sources:
+                    vault_keys.extend(sources)
             if keys and track_drm_type:
                 drm_types[tid] = track_drm_type
                 actual_drm_type = track_drm_type
 
         response: Dict[str, Any] = {"keys": all_keys}
+        if vault_keys:
+            response["vault_keys"] = vault_keys
         if actual_drm_type:
             response["drm_type"] = actual_drm_type
         if drm_types:
@@ -4268,9 +4300,11 @@ async def session_license_handler(
                 track.pr_pssh = pssh_b64
 
         if mode == "server_cdm":
-            keys = handle_single_server_cdm(service, title, track, pssh_b64, drm_type, request)
+            key_sources: Dict[str, str] = {}
+            keys = handle_single_server_cdm(service, title, track, pssh_b64, drm_type, request, key_sources)
             log.info(f"Server CDM resolved {len(keys)} key(s) for track {sanitize_log(track_id[:12])}")
-            return web.json_response({"keys": keys})
+            note_served_keys(session, keys, key_sources)
+            return web.json_response({"keys": keys, "vault_keys": list(key_sources)})
 
         return handle_proxy_license(service, title, track, challenge_b64, drm_type)
 
@@ -4293,6 +4327,48 @@ async def session_license_handler(
         )
 
 
+def note_served_keys(session: Any, keys: Dict[str, str], sources: Dict[str, str]) -> None:
+    """Remember every KID:KEY the remote session handed out and which server vault supplied it.
+
+    The vault name stays on the server: a bad-key report names only the pair, and the
+    server looks the source up here to flag its own row.
+    """
+    for kid, key in keys.items():
+        session.served_keys[kid] = (key, sources.get(kid, "cdm"))
+
+
+async def session_bad_key_handler(
+    data: Dict[str, Any], session_id: str, request: Optional[web.Request] = None
+) -> web.Response:
+    """Flag a server-vault content key the client proved wrong, so the next licence reaches the CDM.
+
+    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.
+    """
+    from uuid import UUID
+
+    session = await get_validated_session(session_id, request)
+    require_authenticated(session)
+
+    kid = str(data.get("kid") or "").replace("-", "").lower()
+    key = str(data.get("key") or "").lower()
+    served_key, source = session.served_keys.get(kid, ("", ""))
+    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")
+
+    vaults = load_server_vaults(session.service_instance.__class__.__name__)
+    for vault in vaults.vaults:
+        if vault.local:
+            vault.flag_bad_key(vaults.service, UUID(hex=kid), served_key, source)
+    session.served_keys.pop(kid, None)
+    log.warning(
+        f"Client proved {kid}:{served_key} from vault {sanitize_log(source)} wrong, flagged in the server vaults"
+    )
+    return web.json_response({"flagged": True})
+
+
 async def session_info_handler(session_id: str, request: Optional[web.Request] = None) -> web.Response:
     """Make sure that the remote session is valid, and get the remote session info."""
     session = await get_validated_session(session_id, request)

unshackle/core/api/routes.py

@@ -47,6 +47,7 @@ from unshackle.core.api.handlers import (
     server_account_regions,
     server_accounts_allowed,
     server_config_handler,
+    session_bad_key_handler,
     session_create_handler,
     session_delete_handler,
     session_info_handler,
@@ -1819,7 +1820,11 @@ async def session_license(request: web.Request) -> web.Response:
                 description: DRM type (default widevine)
     responses:
       '200':
-        description: License response
+        description: >-
+          License response. In server_cdm mode `keys` maps KID to content key and `vault_keys`,
+          an array of KID hex strings that may be absent and may repeat a KID shared by several
+          tracks, lists the content keys a server vault supplied, which the client has to prove
+          before it trusts them.
       '404':
         description: Remote session or track not found
     """
@@ -1840,6 +1845,58 @@ async def session_license(request: web.Request) -> web.Response:
         )
 
 
+@api_handler
+async def session_bad_key(request: web.Request) -> web.Response:
+    """
+    Flag a server-vault content key the client proved wrong.
+    ---
+    summary: Report a bad content key
+    description: >-
+      The client decrypted with a content key the server took from its vault and the output did
+      not decode. The server flags the pair in its local vaults, so the next licence for that KID
+      reaches the CDM when a local vault stores the flag. The server accepts only a pair it served
+      to this remote session.
+    parameters:
+      - name: session_id
+        in: path
+        required: true
+        schema:
+          type: string
+    requestBody:
+      required: true
+      content:
+        application/json:
+          schema:
+            type: object
+            required:
+              - kid
+              - key
+            properties:
+              kid:
+                type: string
+                description: KID as hex
+              key:
+                type: string
+                description: Content key as hex
+    responses:
+      '200':
+        description: The pair is flagged
+      '400':
+        description: The remote session was not served that pair
+      '404':
+        description: Remote session not found
+    """
+    session_id = request.match_info["session_id"]
+    try:
+        data = await request.json()
+    except Exception as e:
+        return build_error_response(
+            APIError(APIErrorCode.INVALID_INPUT, "Invalid JSON request body", details={"error": str(e)}),
+            request.app.get("debug_api", False),
+        )
+    return await session_bad_key_handler(data, session_id, request)
+
+
 @api_handler
 async def session_info(request: web.Request) -> web.Response:
     """
@@ -2548,6 +2605,7 @@ ROUTES: list[tuple[str, str, Handler, bool]] = [
     ("POST", "/api/session/{session_id}/segments", session_segments, True),
     ("POST", "/api/session/{session_id}/segment_filter", session_segment_filter, True),
     ("POST", "/api/session/{session_id}/license", session_license, True),
+    ("POST", "/api/session/{session_id}/keys/bad", session_bad_key, True),
     ("GET", "/api/session/{session_id}/logs", session_logs, True),
     ("GET", "/api/session/{session_id}/prompt", session_prompt_get, True),
     ("POST", "/api/session/{session_id}/prompt", session_prompt_submit, True),

unshackle/core/api/session_store.py

@@ -36,6 +36,9 @@ class SessionEntry:
     title_map: Dict[str, Any] = field(default_factory=dict)
     current_title_id: Optional[str] = None  # title the client last asked tracks for
     tracks: Dict[str, Track] = field(default_factory=dict)
+    served_keys: Dict[str, tuple[str, str]] = field(
+        default_factory=dict
+    )  # KID -> (KEY, serving vault name or "cdm") from server_cdm
     tracks_by_title: Dict[str, Dict[str, Track]] = field(default_factory=dict)
     chapters_by_title: Dict[str, List[Any]] = field(default_factory=dict)
     lock: asyncio.Lock = field(default_factory=asyncio.Lock, repr=False, compare=False)

unshackle/core/remote_service.py

@@ -19,7 +19,8 @@ from enum import Enum
 from http.cookiejar import CookieJar
 from pathlib import Path
 from threading import Lock
-from typing import Any, Callable, Dict, Optional, Union
+from typing import Any, Callable, Dict, Iterator, Optional, Union
+from uuid import UUID
 
 import click
 import requests
@@ -41,6 +42,7 @@ from unshackle.core.tracks import Audio, Chapter, Chapters, Subtitle, Tracks, Vi
 from unshackle.core.tracks.attachment import Attachment
 from unshackle.core.tracks.track import Track
 from unshackle.core.utils.redact import redact_path, redact_text, safe_display_url
+from unshackle.core.vault import Vault
 
 log = logging.getLogger("remote_service")
 
@@ -717,6 +719,38 @@ def cache_stem_is_relevant(
     return not matched_foreign
 
 
+class ServerVault(Vault):
+    """The server's vaults as one source seen from the client: it can only be reported against.
+
+    ``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.
+    """
+
+    def __init__(self, service: "RemoteService"):
+        super().__init__("server_cdm", no_push=True)
+        self.service = service
+
+    def get_key(self, kid: Union[UUID, str], service: str) -> Optional[str]:
+        return None
+
+    def get_keys(self, service: str) -> Iterator[tuple[str, str]]:
+        return iter(())
+
+    def add_key(self, service: str, kid: Union[UUID, str], key: str) -> bool:
+        return False
+
+    def add_keys(self, service: str, kid_keys: dict[Union[UUID, str], str]) -> int:
+        return 0
+
+    def get_services(self) -> Iterator[str]:
+        return iter(())
+
+    def report_bad(self, kid: UUID, key: str) -> None:
+        self.service.report_bad_key(kid, key)
+
+
 class RemoteService:
     """Service adapter that proxies to a remote unshackle server.
 
@@ -757,6 +791,8 @@ class RemoteService:
         self._chapters_by_title: Dict[str, list] = {}
         self._session_id: Optional[str] = None
         self._server_cdm_type: str = "widevine"
+        self.server_vault_keys: Dict[UUID, str] = {}
+        self.server_vault = ServerVault(self)
         self._segment_filters: Dict[str, tuple[set[str], set[str]]] = {}
         self._log_seq = 0
         self._log_drain_lock = Lock()
@@ -1117,6 +1153,9 @@ class RemoteService:
                 )
             self.drain_server_logs()
             keys_by_track = resp.get("keys", {})
+            vault_kids = set(resp.get("vault_keys", []))
+            for track_keys in keys_by_track.values():
+                self.note_vault_keys(track_keys, vault_kids)
             server_drm_type = resp.get("drm_type", drm_type)
             drm_types_by_track = resp.get("drm_types", {})
             self._server_cdm_type = server_drm_type
@@ -1144,6 +1183,22 @@ class RemoteService:
         except Exception as e:
             self.log.warning("Failed to resolve server CDM keys: %s", e)
 
+    def note_vault_keys(self, keys: Dict[str, str], vault_kids: set[str]) -> None:
+        """Remember which of the served keys a server vault supplied, and forget the rest."""
+        for kid_hex, key in keys.items():
+            kid = UUID(hex=kid_hex)
+            if kid_hex in vault_kids:
+                self.server_vault_keys[kid] = key
+            else:
+                self.server_vault_keys.pop(kid, None)
+
+    def report_bad_key(self, kid: UUID, key: str) -> None:
+        """Tell the server a vault key it served did not decrypt, so it flags its own row."""
+        try:
+            self.client.post_optional(f"/api/session/{self._session_id}/keys/bad", {"kid": kid.hex, "key": key})
+        except Exception as e:
+            self.log.warning(f"Could not report the bad content key to the server: {e!r}")
+
     @staticmethod
     def create_drm_stub(drm_type: str, kid_hexes: list[str]) -> Any:
         """Make a DRM object stub that matches the type the server used.
@@ -1263,6 +1318,7 @@ class RemoteService:
                         },
                     )
                     keys = resp.get("keys", {})
+                    self.note_vault_keys(keys, set(resp.get("vault_keys", [])))
                     if keys and track.drm:
                         for drm_obj in track.drm:
                             if hasattr(drm_obj, "content_keys"):