feat(serve): log CDM calls for the dashboard

Add GET /api/dashboard/cdm-logs: one record per pywidevine/pyplayready call plus open sessions per device, with no key or licence material.

docs/dev/rest-api/dashboard.md

@@ -47,6 +47,7 @@ All are `GET`.
 | `/api/dashboard/sessions` | Every live remote session: `id`, `owner` (username or masked key), `creator_ip`, `service`, `title_id`, `titles`, `tracks`, `title` (the display name of the title the client is on: the last one it asked tracks for, else the first resolved title), `auth_status`, `auth_error`, `server_account` (the server profile lent to the remote session, else `null`), `server_cdm` (`false` when the client's own device licenses the remote session), `server_cdm_max_height` (the tallest video the server CDM licenses live, else `null`), `client`, `actions`, `created_at` and `last_accessed` (ISO 8601) with `created_ts` and `last_accessed_ts` (Unix epoch, the same clock as log `ts`), `age_seconds`, `idle_seconds`, `log_seq`. `client` is whatever the remote client sent as `client` in its session create request. The CLI sends `version`, `code_hash` (the commit the client runs, `null` when its source cannot be read), `platform` and `argv`. `argv` is the command line the user ran: proxy and URL userinfo, secret query parameters and credential values become `***`, home and install paths shorten as in the logs, and the line is cut at 3000 characters. An older client sends less or nothing and the field is `{}`. `log_seq` is the last sequence number in the remote session's service log - poll `/api/dashboard/sessions/{id}/logs` when it changes. `actions` is the session's request log, newest last, capped at 500: `ts`, `method`, `action` (`titles`, `tracks`, `segments`, `license`, `prompt`, …), `query`, `status`, `ms`, `bytes_in`, `bytes_out`. |
 | `/api/dashboard/jobs` | Every download job with full detail, regardless of owner. Empty in `--remote-only` mode. |
 | `/api/dashboard/logs` | `{"seq": N, "records": [...]}`: the last 1000 log records. `?since=<seq>` returns only newer records; `?level=WARNING` sets the minimum level; `?logger=serve` keeps one logger and its children (`aiohttp.access` is the noisiest). `ts` is a Unix epoch in seconds. |
+| `/api/dashboard/cdm-logs` | `{"seq": N, "records": [...], "sessions": [...]}`: the last 5000 calls to the Widevine and PlayReady CDM device routes, and the open sessions of each device. `?since=<seq>` returns only newer records. See [CDM calls](#cdm-calls). |
 | `/api/dashboard/sessions/{id}/logs` | One remote session's service log: `{"session_id", "last_seq", "records": [...]}`. Each record has `seq`, `ts`, `level`, `message`. `?since=<seq>` returns only newer records. 404 when the remote session is unknown. |
 | `/api/dashboard/keys` | Every configured API key: what it may do and what it has done. One row per key in `serve.users`, plus `serve.api_secret` when it is set. |
 | `/api/dashboard/services` | Every service configured on this server and its load state, including the ones that failed to import. See [Services](#services) for what configured means. |
@@ -113,6 +114,52 @@ never keeps it alive or makes it look active.
     publishes nothing. Poll this route when `log_seq` changes, and on a timer while a drawer
     is open.
 
+## CDM calls
+
+A server in `full` mode also mounts the Widevine CDM device routes at `/{device}/...` and the
+PlayReady ones at `/playready/{device}/...`. The server records each call to these routes in
+a 5000-record buffer of its own, apart from the log records of `/api/dashboard/logs`.
+
+```
+GET /api/dashboard/cdm-logs?since=1230
+{"seq": 1231,
+ "records": [{"seq": 1231, "ts": 1759750000.123, "drm": "widevine", "op": "get_license_challenge",
+              "device": "chrome_l3", "key_id": "ab12cd34ef56", "user": "tier1",
+              "session_id": "0a1b2c3d4e5f60718293a4b5c6d7e8f9",
+              "status": 200, "ok": true, "error": null, "ms": 12.4}],
+ "sessions": [{"drm": "widevine", "device": "chrome_l3", "key_id": "ab12cd34ef56", "user": "tier1",
+               "open": 3, "max": 17}]}
+```
+
+`since` and `seq` work as they do for `/api/dashboard/logs`: `records` holds the calls with a
+`seq` above `since`, oldest first. `seq` starts again at 1 when the server restarts, so a `seq`
+below your cursor means a restart: read again from 0.
+
+| Field | Meaning |
+|-------|---------|
+| `ts` | Unix epoch in seconds, when the call arrived. |
+| `drm` | `widevine`, or `playready` for a route below `/playready/`. |
+| `op` | `open`, `close`, `set_service_certificate`, `get_service_certificate`, `get_license_challenge`, `parse_license` or `get_keys`. The licence type or content key type that follows in the path is not part of it. |
+| `device` | The device name from the path, cut at 255 characters. |
+| `key_id`, `user` | The `id` and `label` of the caller's row in `/api/dashboard/keys`. Both are `anonymous` when the API key is missing or unknown. |
+| `session_id` | The CDM session, lowercase hexadecimal, so the calls of one licence exchange group together. `null` when there is none, for example a refused `open`. |
+| `status`, `ok` | The HTTP status of the reply. `ok` is `true` below 400. |
+| `error` | `null` when `ok`. Otherwise the `message` of the reply, cut at 300 characters. A handler that raised an error gives the error's class name only. |
+| `ms` | Time the server took to answer, in milliseconds. |
+
+A record never holds the API key, `init_data`, a challenge, a certificate, a licence body or a
+content key. A refused call is in the log too, a 401 for a missing or unknown API key included.
+
+`sessions` is a snapshot at read time, not a history. It has one row for each device an API
+key has opened since the server started: `open` is the number of sessions open now, and `max`
+is the number of sessions the device permits at one time. A PlayReady session older than the
+library's session timeout does not count: the library drops it on the next `open`. The
+list is `[]` until the first `open`, and on a server that mounts no CDM device routes.
+
+!!! note "Polled only"
+    The event stream does not carry these records. One licence exchange is five calls, and that
+    volume would push `session` events out of the stream's history.
+
 ## Keys
 
 ```

tests/remote/unit/test_cdm_call_log.py

@@ -0,0 +1,301 @@
+"""The CDM call log of `unshackle serve` and its `/api/dashboard/cdm-logs` route."""
+
+from __future__ import annotations
+
+import json
+import time
+from pathlib import Path
+from typing import Any
+
+import pytest
+from aiohttp import web
+from click.testing import CliRunner
+
+from unshackle.commands.serve import serve
+from unshackle.core.api.stats import SeqRing, cdm_calls, cdm_log_middleware, cdm_sessions, key_id
+from unshackle.core.config import config
+
+pytestmark = pytest.mark.unit
+
+USER = {"X-Secret-Key": "tier-key"}
+DASH = {"X-Secret-Key": "dash-secret"}
+INIT_DATA = "AAAAaW5pdC1kYXRhLW1hcmtlcg=="
+CHALLENGE_B64_PREFIX = "Y2hhbGxlbmdl"
+LICENCE = "bGljZW5jZS1tYXJrZXI="
+CONTENT_KEY = "00112233445566778899aabbccddeeff"
+
+
+class FakeSession:
+    def __init__(self) -> None:
+        self.opened_at = time.time()
+
+
+class Cdm:
+    """A stand-in for both libraries' Cdm. The class name keeps the ``_Cdm__sessions`` mangling."""
+
+    MAX_NUM_OF_SESSIONS = 16
+    system_id = 1234
+    security_level = 3
+
+    def __init__(self) -> None:
+        self.__sessions: dict[bytes, FakeSession] = {}
+
+    @classmethod
+    def from_device(cls, _device: Any) -> Cdm:
+        return cls()
+
+    def open(self) -> bytes:
+        session_id = bytes([len(self.__sessions) + 1]) * 16
+        self.__sessions[session_id] = FakeSession()
+        return session_id
+
+    def close(self, session_id: bytes) -> None:
+        del self.__sessions[session_id]
+
+    def get_service_certificate(self, session_id: bytes) -> None:
+        return None
+
+    def get_license_challenge(self, session_id: bytes, **_: Any) -> Any:
+        return b"challenge-marker" * 40
+
+    def parse_license(self, session_id: bytes, _message: Any) -> None:
+        return None
+
+    def age(self, seconds: float) -> None:
+        for session in self.__sessions.values():
+            session.opened_at -= seconds
+
+
+class PlayReadyCdm(Cdm):
+    SESSION_TIMEOUT = 30
+
+    def get_license_challenge(self, session_id: bytes, **_: Any) -> Any:
+        return "challenge-marker" * 40
+
+
+@pytest.fixture
+def serve_app(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> web.Application:
+    """The app a keyed, full-mode `unshackle serve` builds, with both CDM libraries stubbed."""
+    from pyplayready.remote import serve as pyplayready_serve
+    from pywidevine import serve as pywidevine_serve
+
+    (tmp_path / "chrome_l3.wvd").touch()
+    (tmp_path / "sl2000.prd").touch()
+    monkeypatch.setattr(config.directories, "wvds", tmp_path)
+    monkeypatch.setattr(config.directories, "prds", tmp_path)
+    monkeypatch.setattr(
+        config,
+        "serve",
+        {
+            "api_secret": "master-key",
+            "dashboard": {"key": "dash-secret"},
+            "users": {
+                "tier-key": {"username": "tier1", "devices": ["chrome_l3"], "playready_devices": ["sl2000"]},
+            },
+        },
+    )
+    for module, cdm in ((pywidevine_serve, Cdm), (pyplayready_serve, PlayReadyCdm)):
+        monkeypatch.setattr(module, "Cdm", cdm)
+        monkeypatch.setattr(module.Device, "load", staticmethod(lambda _path: object()))
+    monkeypatch.setattr(pywidevine_serve, "PSSH", lambda data: data)
+    monkeypatch.setattr(pyplayready_serve, "PSSH", lambda _data: type("P", (), {"wrm_headers": ["header"]})())
+
+    captured: dict[str, web.Application] = {}
+    monkeypatch.setattr(web, "run_app", lambda app, **kwargs: captured.update(app=app))
+    result = CliRunner().invoke(serve, [])
+    assert result.exit_code == 0, result.output
+    cdm_calls.records.clear()
+    cdm_calls.seq = 0
+    return captured["app"]
+
+
+async def cdm_log(client: Any, since: int = 0) -> dict[str, Any]:
+    resp = await client.get(f"/api/dashboard/cdm-logs?since={since}", headers=DASH)
+    assert resp.status == 200
+    return await resp.json()
+
+
+async def test_widevine_exchange_is_recorded_per_call(aiohttp_client, serve_app) -> None:
+    client = await aiohttp_client(serve_app)
+    opened = await (await client.get("/chrome_l3/open", headers=USER)).json()
+    session_id = opened["data"]["session_id"]
+    assert (await cdm_log(client))["sessions"] == [
+        {"drm": "widevine", "device": "chrome_l3", "key_id": key_id("tier-key"), "user": "tier1", "open": 1, "max": 17}
+    ]
+
+    # Accept-Encoding makes compression_middleware gzip the long challenge reply.
+    challenge = await client.post(
+        "/chrome_l3/get_license_challenge/STREAMING",
+        headers={**USER, "Accept-Encoding": "gzip"},
+        json={"session_id": session_id.upper(), "init_data": INIT_DATA},
+    )
+    assert challenge.status == 200, await challenge.text()
+    parsed = await client.post(
+        "/chrome_l3/parse_license", headers=USER, json={"session_id": session_id, "license_message": LICENCE}
+    )
+    assert parsed.status == 200, await parsed.text()
+    assert (await client.get(f"/chrome_l3/close/{session_id}", headers=USER)).status == 200
+
+    page = await cdm_log(client)
+    records = page["records"]
+    assert [r["op"] for r in records] == ["open", "get_license_challenge", "parse_license", "close"]
+    assert [r["seq"] for r in records] == [1, 2, 3, 4] and page["seq"] == 4
+    for record in records:
+        assert record["drm"] == "widevine" and record["device"] == "chrome_l3"
+        assert record["session_id"] == session_id
+        assert record["user"] == "tier1" and record["key_id"] == key_id("tier-key")
+        assert record["status"] == 200 and record["ok"] is True and record["error"] is None
+        assert isinstance(record["ms"], float) and abs(record["ts"] - time.time()) < 60
+        assert set(record) == {
+            "seq", "ts", "drm", "op", "device", "key_id", "user", "session_id", "status", "ok", "error", "ms",
+        }  # fmt: skip
+    assert page["sessions"][0]["open"] == 0
+
+    dump = json.dumps(page)
+    for secret in ("tier-key", INIT_DATA, CHALLENGE_B64_PREFIX, "challenge-marker", LICENCE, CONTENT_KEY):
+        assert secret not in dump
+
+
+async def test_playready_sub_app_calls_are_tagged_playready(aiohttp_client, serve_app) -> None:
+    client = await aiohttp_client(serve_app)
+    opened = await (await client.get("/playready/sl2000/open", headers=USER)).json()
+    session_id = opened["data"]["session_id"]
+    challenge = await client.post(
+        "/playready/sl2000/get_license_challenge", headers=USER, json={"session_id": session_id, "init_data": INIT_DATA}
+    )
+    assert challenge.status == 200, await challenge.text()
+
+    page = await cdm_log(client)
+    assert [(r["drm"], r["op"], r["device"], r["session_id"]) for r in page["records"]] == [
+        ("playready", "open", "sl2000", session_id),
+        ("playready", "get_license_challenge", "sl2000", session_id),
+    ]
+    assert page["sessions"] == [
+        {"drm": "playready", "device": "sl2000", "key_id": key_id("tier-key"), "user": "tier1", "open": 1, "max": 17}
+    ]
+    # A session past SESSION_TIMEOUT is one the library drops on its next open.
+    serve_app["playready_app"]["cdms"][("tier-key", "sl2000")].age(PlayReadyCdm.SESSION_TIMEOUT + 1)
+    assert (await cdm_log(client))["sessions"][0]["open"] == 0
+
+
+async def test_refusals_are_recorded_with_their_message(aiohttp_client, serve_app) -> None:
+    client = await aiohttp_client(serve_app)
+    assert (await client.get("/other_device/open", headers=USER)).status == 403
+    assert (await client.get("/chrome_l3/open", headers={"X-Secret-Key": "wrong-key"})).status == 401
+    assert (await client.get("/chrome_l3/open")).status == 401
+    assert (await client.get("/playready/sl2000/open", headers={"X-Secret-Key": "wrong-key"})).status == 401
+    unopened = await client.post("/chrome_l3/get_keys/CONTENT", headers=USER, json={"session_id": "AB" * 16})
+    assert unopened.status == 400
+    assert (await client.post("/chrome_l3/parse_license", headers=USER, json={})).status == 400
+
+    page = await cdm_log(client)
+    forbidden, invalid, empty, playready, no_cdm, no_field = page["records"]
+    assert forbidden["status"] == 403 and forbidden["ok"] is False and forbidden["user"] == "tier1"
+    assert forbidden["device"] == "other_device" and "not authorized" in forbidden["error"]
+    assert forbidden["session_id"] is None
+    assert invalid["status"] == 401 and invalid["ok"] is False and "Invalid" in invalid["error"]
+    assert invalid["user"] == "anonymous" and invalid["key_id"] == "anonymous"
+    assert empty["status"] == 401 and "Empty" in empty["error"] and empty["user"] == "anonymous"
+    assert playready["drm"] == "playready" and playready["status"] == 401 and "Invalid" in playready["error"]
+    assert no_cdm["op"] == "get_keys" and no_cdm["status"] == 400 and no_cdm["session_id"] == "ab" * 16
+    assert no_field["op"] == "parse_license" and no_field["session_id"] is None and "session_id" in no_field["error"]
+    assert page["sessions"] == [] and "wrong-key" not in json.dumps(page)
+
+
+async def test_gzipped_refusal_is_read_and_capped(aiohttp_client, serve_app) -> None:
+    """compression_middleware runs inside the log middleware, so a long refusal arrives as gzip."""
+    client = await aiohttp_client(serve_app)
+    device = "d" * 400
+    resp = await client.get(f"/{device}/open", headers={**USER, "Accept-Encoding": "gzip"})
+    assert resp.status == 403 and resp.headers["Content-Encoding"] == "gzip"
+    (record,) = (await cdm_log(client))["records"]
+    assert record["device"] == device[:255] and len(record["error"]) == 300
+    assert record["error"].startswith("Device 'ddd")
+
+
+async def test_refused_call_leaves_the_request_body_unread(aiohttp_client, serve_app) -> None:
+    """A 401 is decided before the handler, so the body stays unread: no cost, and no session id, for an unknown key."""
+    client = await aiohttp_client(serve_app)
+    body = {"session_id": "ab" * 16, "license_message": LICENCE}
+    assert (
+        await client.post("/chrome_l3/parse_license", headers={"X-Secret-Key": "wrong-key"}, json=body)
+    ).status == 401
+    assert (await client.post("/chrome_l3/parse_license", headers=USER, json=body)).status == 400
+    refused, unopened = (await cdm_log(client))["records"]
+    assert refused["status"] == 401 and refused["session_id"] is None
+    assert unopened["status"] == 400 and unopened["session_id"] == "ab" * 16
+
+
+async def test_raised_http_exception_is_recorded_with_its_status(aiohttp_client) -> None:
+    """aiohttp answers a raised ``HTTPException`` itself, so the record carries that status, not 500."""
+
+    async def too_large(_request: web.Request) -> web.Response:
+        raise web.HTTPRequestEntityTooLarge(max_size=1, actual_size=2)
+
+    app = web.Application(middlewares=[cdm_log_middleware])
+    app.router.add_post("/{device}/parse_license", too_large)
+    before = cdm_calls.seq
+    client = await aiohttp_client(app)
+    resp = await client.post("/chrome_l3/parse_license", json={"session_id": "ab" * 16})
+    (record,) = cdm_calls.since(before)
+    assert resp.status == 413 and record["status"] == 413 and record["ok"] is False
+    assert record["error"] == resp.reason
+
+
+async def test_since_cursor_and_dashboard_key(aiohttp_client, serve_app) -> None:
+    client = await aiohttp_client(serve_app)
+    for _ in range(3):
+        await client.get("/chrome_l3/open", headers=USER)
+    assert (await client.get("/api/dashboard/cdm-logs")).status == 401
+    assert (await client.get("/api/dashboard/cdm-logs", headers=USER)).status == 401
+    # Only the device routes are CDM calls: not the API, not the ping, not this route.
+    await client.get("/api/health")
+    await client.get("/", headers=USER)
+
+    page = await cdm_log(client, since=1)
+    assert [r["seq"] for r in page["records"]] == [2, 3] and page["seq"] == 3
+    assert await cdm_log(client, since=3) == {"seq": 3, "records": [], "sessions": page["sessions"]}
+    resp = await client.get("/api/dashboard/cdm-logs?since=junk", headers=DASH)
+    assert len((await resp.json())["records"]) == 3
+    assert page["sessions"][0]["open"] == 3
+
+
+async def test_handler_exception_is_recorded_without_its_text(aiohttp_client) -> None:
+    """An unhandled error still answers 500, and its text (it can hold the API key) stays out."""
+
+    async def broken(request: web.Request) -> web.Response:
+        raise KeyError(request.headers["X-Secret-Key"])
+
+    app = web.Application(middlewares=[cdm_log_middleware])
+    app.router.add_get("/{device}/open", broken)
+    before = cdm_calls.seq
+    client = await aiohttp_client(app)
+    assert (await client.get("/chrome_l3/open", headers={"X-Secret-Key": "leaky-key"})).status == 500
+    (record,) = cdm_calls.since(before)
+    assert record["status"] == 500 and record["ok"] is False and record["error"] == "KeyError"
+
+
+async def test_recording_failure_never_fails_the_cdm_call(aiohttp_client, monkeypatch) -> None:
+    async def opened(_request: web.Request) -> web.Response:
+        return web.json_response({"message": "Success"})
+
+    def refuse(_record: dict[str, Any]) -> None:
+        raise RuntimeError("log is broken")
+
+    monkeypatch.setattr(cdm_calls, "append", refuse)
+    app = web.Application(middlewares=[cdm_log_middleware])
+    app.router.add_get("/{device}/open", opened)
+    client = await aiohttp_client(app)
+    assert (await client.get("/chrome_l3/open")).status == 200
+
+
+def test_call_log_keeps_the_newest_and_counts_on() -> None:
+    log = SeqRing(maxlen=2)
+    for op in ("open", "get_keys", "close"):
+        log.append({"op": op})
+    assert [(r["seq"], r["op"]) for r in log.since()] == [(2, "get_keys"), (3, "close")]
+    assert log.since(2) == [{"seq": 3, "op": "close"}] and log.seq == 3
+
+
+def test_sessions_snapshot_is_empty_without_a_cdm() -> None:
+    assert cdm_sessions(web.Application()) == []

unshackle/commands/serve.py

@@ -25,7 +25,14 @@ from unshackle.core.api.handlers import (
     server_cdm_max_height_error,
     validate_server_accounts,
 )
-from unshackle.core.api.stats import key_rate_limit, rate_limit_error, ring, stats, stats_middleware
+from unshackle.core.api.stats import (
+    cdm_log_middleware,
+    key_rate_limit,
+    rate_limit_error,
+    ring,
+    stats,
+    stats_middleware,
+)
 from unshackle.core.config import config
 from unshackle.core.console import console
 from unshackle.core.constants import context_settings
@@ -424,7 +431,13 @@ def serve(
 
             if no_key:
                 app = web.Application(
-                    middlewares=[cors_middleware, stats_middleware, dashboard_authentication, compression_middleware]
+                    middlewares=[
+                        cors_middleware,
+                        stats_middleware,
+                        cdm_log_middleware,
+                        dashboard_authentication,
+                        compression_middleware,
+                    ]
                 )
             else:
                 serve_auth = create_serve_authentication(serve_playready and bool(prd_devices))
@@ -432,6 +445,7 @@ def serve(
                     middlewares=[
                         cors_middleware,
                         stats_middleware,
+                        cdm_log_middleware,
                         dashboard_authentication,
                         serve_auth,
                         compression_middleware,
@@ -495,6 +509,7 @@ def serve(
                 app.router.add_route("*", "/playready", playready_ping)
 
                 app.add_subapp("/playready", playready_app)
+                app["playready_app"] = playready_app
                 log.info(f"PlayReady CDM endpoints available at http://{host}:{port}/playready/")
             elif serve_playready:
                 log.info("No PlayReady devices found, skipping PlayReady CDM endpoints")

unshackle/core/api/handlers.py

@@ -2342,19 +2342,36 @@ def _poll_cursor(before: int, items: List[Dict[str, Any]]) -> int:
     return max(before, items[-1]["seq"]) if items else before
 
 
+def _since(request: web.Request) -> int:
+    """The ``since`` cursor of a poll; 0 (everything) when it is missing or not a number."""
+    try:
+        return int(request.query.get("since", 0))
+    except ValueError:
+        return 0
+
+
 async def dashboard_logs_handler(request: web.Request) -> web.Response:
     """Recent log records; `since` (seq) and `level` filter the ring buffer."""
     from unshackle.core.api.stats import ring
 
-    try:
-        since = int(request.query.get("since", 0))
-    except ValueError:
-        since = 0
+    since = _since(request)
     before = ring.seq
     records = ring.since(since, request.query.get("level"), request.query.get("logger"))
     return web.json_response({"seq": _poll_cursor(before, records), "records": records})
 
 
+async def dashboard_cdm_logs_handler(request: web.Request) -> web.Response:
+    """CDM calls after `since` (seq), plus a snapshot of the open sessions of each live Cdm."""
+    from unshackle.core.api.stats import cdm_calls, cdm_sessions
+
+    since = _since(request)
+    before = cdm_calls.seq
+    records = cdm_calls.since(since)
+    return web.json_response(
+        {"seq": _poll_cursor(before, records), "records": records, "sessions": cdm_sessions(request.app)}
+    )
+
+
 async def dashboard_session_logs_handler(request: web.Request) -> web.Response:
     """One remote session's mirrored service log, for an operator watching a failing auth.
 
@@ -2368,10 +2385,7 @@ async def dashboard_session_logs_handler(request: web.Request) -> web.Response:
     session = get_session_store().peek(session_id)
     if session is None:
         raise APIError(APIErrorCode.SESSION_NOT_FOUND, f"Remote session not found: {session_id}")
-    try:
-        since = int(request.query.get("since", 0))
-    except ValueError:
-        since = 0
+    since = _since(request)
     buffer = session.log_buffer
     before = buffer.last_seq if buffer else 0
     records = buffer.since(since) if buffer else []

unshackle/core/api/routes.py

@@ -21,6 +21,7 @@ from unshackle.core.api.handlers import (
     clear_cache_handler,
     clear_finished_download_jobs_handler,
     clear_temp_handler,
+    dashboard_cdm_logs_handler,
     dashboard_events_handler,
     dashboard_health_handler,
     dashboard_jobs_handler,
@@ -2644,6 +2645,91 @@ async def dashboard_logs(request: web.Request) -> web.Response:
     return await dashboard_logs_handler(request)
 
 
+@api_handler
+async def dashboard_cdm_logs(request: web.Request) -> web.Response:
+    """
+    Dashboard: recent CDM calls and open CDM sessions.
+    ---
+    summary: Dashboard CDM call log
+    description: >
+      The last 5000 calls to the Widevine and PlayReady CDM device routes, one record for each
+      call, with a snapshot of the open sessions of each live Cdm. A record never holds an API
+      key, `init_data`, a challenge, a certificate, a licence body or a content key.
+      Poll with `since` set to the `seq` of the last reply. The server does not publish these
+      records on the event stream.
+    tags: [Dashboard]
+    parameters:
+      - name: since
+        in: query
+        required: false
+        schema:
+          type: integer
+        description: Return only records with seq greater than this value
+    responses:
+      '200':
+        description: Records plus the current seq and the open sessions
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                seq:
+                  type: integer
+                records:
+                  type: array
+                  items:
+                    type: object
+                    properties:
+                      seq:
+                        type: integer
+                      ts:
+                        type: number
+                      drm:
+                        type: string
+                        enum: [widevine, playready]
+                      op:
+                        type: string
+                      device:
+                        type: string
+                      key_id:
+                        type: string
+                      user:
+                        type: string
+                      session_id:
+                        type: string
+                        nullable: true
+                      status:
+                        type: integer
+                      ok:
+                        type: boolean
+                      error:
+                        type: string
+                        nullable: true
+                      ms:
+                        type: number
+                sessions:
+                  type: array
+                  items:
+                    type: object
+                    properties:
+                      drm:
+                        type: string
+                      device:
+                        type: string
+                      key_id:
+                        type: string
+                      user:
+                        type: string
+                      open:
+                        type: integer
+                      max:
+                        type: integer
+      '401':
+        description: Dashboard key missing or invalid
+    """
+    return await dashboard_cdm_logs_handler(request)
+
+
 @api_handler
 async def dashboard_session_logs(request: web.Request) -> web.Response:
     """
@@ -2986,6 +3072,7 @@ DASHBOARD_ROUTES: list[tuple[str, str, Handler, bool]] = [
     ("GET", DASHBOARD_PREFIX + "sessions", dashboard_sessions, True),
     ("GET", DASHBOARD_PREFIX + "jobs", dashboard_jobs, True),
     ("GET", DASHBOARD_PREFIX + "logs", dashboard_logs, True),
+    ("GET", DASHBOARD_PREFIX + "cdm-logs", dashboard_cdm_logs, True),
     ("GET", DASHBOARD_PREFIX + "keys", dashboard_keys, True),
     ("GET", DASHBOARD_PREFIX + "services", dashboard_services, True),
     ("GET", DASHBOARD_PREFIX + "health", dashboard_health, True),

unshackle/core/api/stats.py

@@ -2,12 +2,15 @@
 
 from __future__ import annotations
 
+import gzip
 import hashlib
+import json
 import logging
 import time
 from collections import Counter, deque
 from dataclasses import dataclass, field
-from typing import Any, Deque, Dict, List, Optional
+from functools import lru_cache
+from typing import Any, Deque, Dict, List, Optional, Tuple
 
 from aiohttp import web
 
@@ -17,6 +20,20 @@ from unshackle.core.config import config
 from unshackle.core.utils.redact import redact_secrets
 
 RATE_LIMIT_WINDOW = 3600.0
+CDM_OPS = frozenset(
+    {
+        "open",
+        "close",
+        "set_service_certificate",
+        "get_service_certificate",
+        "get_license_challenge",
+        "parse_license",
+        "get_keys",
+    }
+)
+CDM_ERROR_LIMIT = 300
+CDM_SESSION_ID_LIMIT = 64
+CDM_DEVICE_LIMIT = 255
 
 
 def _users() -> Dict[str, Any]:
@@ -215,40 +232,220 @@ async def stats_middleware(request: web.Request, handler: Any) -> web.StreamResp
     return response
 
 
-class RingLogHandler(logging.Handler):
-    """Keep the last N log records in memory and publish each one on the event bus."""
+class SeqRing:
+    """Keep the last N records in memory, each with a ``seq`` one above the record before it."""
 
-    def __init__(self, maxlen: int = 1000) -> None:
-        super().__init__()
+    def __init__(self, maxlen: int) -> None:
         self.records: Deque[Dict[str, Any]] = deque(maxlen=maxlen)
         self.seq = 0
 
-    def emit(self, record: logging.LogRecord) -> None:
-        """Buffer and publish one record; ``seq`` moves only after the append, as in ``EventBus``."""
+    def append(self, record: Dict[str, Any]) -> Dict[str, Any]:
+        """Buffer one record; ``seq`` moves only after the append, as in ``EventBus``."""
         seq = self.seq + 1
-        item = {
-            "seq": seq,
-            "ts": record.created,
-            "level": record.levelname,
-            "logger": record.name,
-            "msg": redact_secrets(self.format(record)),
-        }
+        item = {"seq": seq, **record}
         self.records.append(item)
         self.seq = seq
+        return item
+
+    def since(self, seq: int = 0) -> List[Dict[str, Any]]:
+        """The records after ``seq``, oldest first, from a copy: ``append()`` can run on a worker thread."""
+        return [r for r in self.records.copy() if r["seq"] > seq]
+
+
+class RingLogHandler(SeqRing, logging.Handler):
+    """Keep the last N log records in memory and publish each one on the event bus."""
+
+    def __init__(self, maxlen: int = 1000) -> None:
+        logging.Handler.__init__(self)
+        SeqRing.__init__(self, maxlen)
+
+    def emit(self, record: logging.LogRecord) -> None:
+        """Buffer and publish one record."""
+        item = self.append(
+            {
+                "ts": record.created,
+                "level": record.levelname,
+                "logger": record.name,
+                "msg": redact_secrets(self.format(record)),
+            }
+        )
         bus.publish("log", item)
 
     def since(self, seq: int = 0, level: Optional[str] = None, logger: Optional[str] = None) -> List[Dict[str, Any]]:
-        """Filter a copy of the records: ``emit()`` appends from worker threads."""
+        """The records after ``seq`` at or above ``level``, from ``logger`` or one of its children."""
         min_level = logging.getLevelName(level.upper()) if level else 0
         if not isinstance(min_level, int):
             min_level = 0
         return [
             r
-            for r in self.records.copy()
-            if r["seq"] > seq
-            and logging.getLevelName(r["level"]) >= min_level
+            for r in super().since(seq)
+            if logging.getLevelName(r["level"]) >= min_level
             and (not logger or r["logger"] == logger or r["logger"].startswith(logger + "."))
         ]
 
 
 ring = RingLogHandler()
+
+cdm_calls = SeqRing(maxlen=5000)
+
+
+@lru_cache(maxsize=None)
+def _cdm_op(pattern: str) -> Optional[Tuple[str, str]]:
+    """The ``(drm, op)`` of a route pattern, else None. Cached: the route table does not change."""
+    parts = pattern.strip("/").split("/")
+    drm = "widevine"
+    if parts[:1] == ["playready"]:
+        drm, parts = "playready", parts[1:]
+    if len(parts) < 2 or parts[0] != "{device}" or parts[1] not in CDM_OPS:
+        return None
+    return drm, parts[1]
+
+
+def _cdm_route(request: web.Request) -> Optional[Tuple[str, str]]:
+    """The ``(drm, op)`` of a CDM device route, else None.
+
+    Reads the matched route pattern, not the path, so a device name cannot pass for an operation
+    or for the ``/playready`` prefix.
+    """
+    resource = request.match_info.route.resource
+    return _cdm_op(resource.canonical) if resource is not None else None
+
+
+def _hex_id(value: Any) -> Optional[str]:
+    """``value`` in lowercase when it is a hexadecimal string of a plausible length, else None."""
+    if not isinstance(value, str) or not value or len(value) > CDM_SESSION_ID_LIMIT:
+        return None
+    try:
+        bytes.fromhex(value)
+    except ValueError:
+        return None
+    return value.lower()
+
+
+def _key_fields(secret_key: Optional[str]) -> Dict[str, str]:
+    """The ``key_id`` and ``user`` of an API key: the ``id`` and ``label`` of its ``stats.keys`` entry."""
+    key = configured_key(secret_key)
+    return {"key_id": key_id(key), "user": mask_key(key)}
+
+
+def _reply_json(response: Any) -> Dict[str, Any]:
+    """The JSON object a handler answered, or ``{}``. ``compression_middleware`` runs inside the
+    CDM log middleware, so the body can arrive as gzip."""
+    try:
+        body = response.body
+        if response.headers.get("Content-Encoding") == "gzip":
+            body = gzip.decompress(body)
+        data = json.loads(body)
+    except Exception:
+        return {}
+    return data if isinstance(data, dict) else {}
+
+
+async def _request_session_id(request: web.Request) -> Any:
+    """The ``session_id`` of the JSON request body, when the handler consumed it: aiohttp caches that read.
+
+    A body nobody read belongs to a call refused before the handler, a 401 or a 429. It stays
+    unread: reading it would spend time and memory on an unauthenticated caller, and wait on a
+    body that never finishes arriving.
+    """
+    if not request.content.at_eof():
+        return None
+    try:
+        body = await request.json()
+    except Exception:
+        return None
+    return body.get("session_id") if isinstance(body, dict) else None
+
+
+async def _cdm_record(request: web.Request, drm: str, op: str, outcome: Any) -> Dict[str, Any]:
+    """The record of one call. ``outcome`` is the response, or the exception the handler raised.
+
+    An exception that is not an HTTP one gives its class name only: its text can hold the
+    presented API key. The reply body is read only for a refusal and for the session id of an
+    ``open``, so a challenge, a licence or a content key never passes through here.
+    """
+    from unshackle.core.api.handlers import request_secret_key
+
+    reply: Dict[str, Any] = {}
+    if isinstance(outcome, web.StreamResponse):
+        status = outcome.status
+        if status >= 400 or op == "open":
+            reply = _reply_json(outcome)
+        message = reply.get("message")
+        error = message if isinstance(message, str) else outcome.reason
+    else:
+        status, error = 500, type(outcome).__name__
+    ok = status < 400
+    if op == "open":
+        data = reply.get("data") if ok else None
+        session_id = data.get("session_id") if isinstance(data, dict) else None
+    elif op == "close":
+        session_id = request.match_info.get("session_id")
+    else:
+        session_id = await _request_session_id(request)
+    return {
+        "drm": drm,
+        "op": op,
+        "device": request.match_info["device"][:CDM_DEVICE_LIMIT],
+        **_key_fields(request_secret_key(request)),
+        "session_id": _hex_id(session_id),
+        "status": status,
+        "ok": ok,
+        "error": None if ok else str(error)[:CDM_ERROR_LIMIT],
+    }
+
+
+async def _record_cdm_call(
+    request: web.Request, route: Tuple[str, str], ts: float, started: float, outcome: Any
+) -> None:
+    """Append one call to ``cdm_calls``. Logs a failure and does not raise it: it must never fail the CDM call."""
+    ms = round((time.perf_counter() - started) * 1000, 1)
+    try:
+        cdm_calls.append({"ts": ts, **await _cdm_record(request, *route, outcome), "ms": ms})
+    except Exception:
+        logging.getLogger("serve").debug("CDM call not recorded", exc_info=True)
+
+
+@web.middleware
+async def cdm_log_middleware(request: web.Request, handler: Any) -> web.StreamResponse:
+    """Record each call to a CDM device route in ``cdm_calls``.
+
+    Must run outside the serve authentication middleware, so the log holds a 401 too.
+    """
+    route = _cdm_route(request)
+    if route is None:
+        return await handler(request)
+    ts, started = time.time(), time.perf_counter()
+    try:
+        response = await handler(request)
+    except Exception as e:
+        await _record_cdm_call(request, route, ts, started, e)
+        raise
+    await _record_cdm_call(request, route, ts, started, response)
+    return response
+
+
+def cdm_sessions(app: web.Application) -> List[Dict[str, Any]]:
+    """One row per live Cdm object: its open session count and the most it allows.
+
+    Both CDM libraries refuse an ``open`` only above ``MAX_NUM_OF_SESSIONS``, so the true limit
+    is one more. A Cdm with a ``SESSION_TIMEOUT`` drops older sessions on its next ``open``,
+    so those do not count as open.
+    """
+    now = time.time()
+    rows: List[Dict[str, Any]] = []
+    for drm, owner in (("widevine", app), ("playready", app.get("playready_app"))):
+        cdms = owner.get("cdms") if owner is not None else None
+        for (secret_key, device), cdm in (cdms or {}).items():
+            sessions = getattr(cdm, "_Cdm__sessions", {}).values()
+            timeout = getattr(cdm, "SESSION_TIMEOUT", None)
+            rows.append(
+                {
+                    "drm": drm,
+                    "device": device,
+                    **_key_fields(secret_key),
+                    "open": len(sessions) if timeout is None else sum(now - s.opened_at <= timeout for s in sessions),
+                    "max": cdm.MAX_NUM_OF_SESSIONS + 1,
+                }
+            )
+    return rows