fix(cdm): skip the server header probe for pywidevine and pyplayready remote CDMs

A reverse proxy or CDN rewrites the Server header, so upstream refused a working serve API.

docs/guide/drm-and-cdm.md

@@ -324,6 +324,9 @@ remote_cdm:
     so a host without the suffix points the requests at the wrong location. The symptom
     is `404` responses from the server rather than an obvious configuration error.
 
+A pywidevine or pyplayready serve API can operate behind a reverse proxy or a CDN. Refer to
+[Serve APIs behind a proxy](../reference/configuration/drm.md#remote_cdm).
+
 ### Decrypt Labs (hosted KeyXtractor API)
 
 Set `type: decrypt_labs` to use Decrypt Labs' hosted KeyXtractor service, which

docs/reference/configuration/drm.md

@@ -74,6 +74,13 @@ For the two pywidevine/pyplayready backends, field names are read case-insensiti
 styles: `Device Type`/`device_type`, `System ID`/`system_id`, `Security Level`/`security_level`,
 `Host`/`host`, `Secret`/`secret`, `Device Name`/`device_name`.
 
+!!! note "Serve APIs behind a proxy"
+    unshackle does not send upstream's `HEAD` request or examine the `Server` response
+    header, so the server can operate behind a reverse proxy or a CDN that removes or
+    changes that header. Thus a `host` that is not a serve API fails at the first licence
+    request, not when unshackle loads the CDM. On a redirect to a different host or port,
+    unshackle removes the `secret` from the request.
+
 !!! warning "PlayReady `host` needs a `/playready` suffix"
     pyplayready's `RemoteCdm` treats `host` as a base URL and appends its own endpoint paths,
     so a PlayReady entry's `host` must include the trailing `/playready` segment (for example

tests/core/test_cdm_detect.py

@@ -157,13 +157,13 @@ def test_load_remote_decryptlabs_widevine_is_stamped():
 
 
 def test_load_remote_native_playready_is_stamped(monkeypatch):
-    import pyplayready.remote.remotecdm as prmod
+    import unshackle.core.cdm.unprobed_remote_cdm as unprobed_mod
 
     class FakePlayReadyRemote:
         def __init__(self, **kwargs):
             self.kwargs = kwargs
 
-    monkeypatch.setattr(prmod, "RemoteCdm", FakePlayReadyRemote)
+    monkeypatch.setattr(unprobed_mod, "UnprobedPlayReadyRemoteCdm", FakePlayReadyRemote)
     cdm_api = {"name": "pr", "Device Type": "PLAYREADY", "host": "h", "secret": "s", "device_name": "d"}
     cdm = loader_mod.load_remote_cdm(dict(cdm_api), "pr", "EXAMPLE", None)
     assert isinstance(cdm, FakePlayReadyRemote)
@@ -173,13 +173,13 @@ def test_load_remote_native_playready_is_stamped(monkeypatch):
 
 
 def test_load_remote_native_widevine_is_stamped(monkeypatch):
-    import pywidevine.remotecdm as wvmod
+    import unshackle.core.cdm.unprobed_remote_cdm as unprobed_mod
 
     class FakeWidevineRemote:
         def __init__(self, **kwargs):
             self.kwargs = kwargs
 
-    monkeypatch.setattr(wvmod, "RemoteCdm", FakeWidevineRemote)
+    monkeypatch.setattr(unprobed_mod, "UnprobedWidevineRemoteCdm", FakeWidevineRemote)
     cdm_api = {"name": "wv", "Device Type": "ANDROID", "host": "h", "secret": "s", "device_name": "d"}
     cdm = loader_mod.load_remote_cdm(dict(cdm_api), "wv", "EXAMPLE", None)
     assert isinstance(cdm, FakeWidevineRemote)

tests/core/test_remote_cdm_no_probe.py

@@ -0,0 +1,109 @@
+import json
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Any, Iterator
+
+import pytest
+import requests
+from pyplayready.remote.remotecdm import RemoteCdm as PlayReadyRemoteCdm
+from pywidevine.remotecdm import RemoteCdm as WidevineRemoteCdm
+
+from unshackle.core.cdm.unprobed_remote_cdm import UnprobedPlayReadyRemoteCdm, UnprobedWidevineRemoteCdm
+
+HOST = "http://cdm.invalid"
+
+
+def widevine(**overrides: Any) -> Any:
+    args = dict(
+        device_type="ANDROID", system_id=4464, security_level=3000, host=HOST, secret="s3cret", device_name="dev"
+    )
+    return UnprobedWidevineRemoteCdm(**{**args, **overrides})
+
+
+def playready(**overrides: Any) -> Any:
+    args = dict(security_level=3000, host=HOST, secret="s3cret", device_name="dev")
+    return UnprobedPlayReadyRemoteCdm(**{**args, **overrides})
+
+
+class FakeResponse:
+    status_code = 200
+
+    def json(self) -> dict:
+        device = {"system_id": 4464, "security_level": 3000}
+        return {"status": 200, "data": {"device": device, "session_id": "00ff"}}
+
+
+@pytest.fixture
+def sent(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, Any]]:
+    """Record every GET, so a test sees what the upstream methods send through the session."""
+    calls: list[tuple[str, Any]] = []
+
+    def get(self: requests.Session, url: str, **kwargs: Any) -> FakeResponse:
+        calls.append((url, self.headers.get("X-Secret-Key")))
+        return FakeResponse()
+
+    monkeypatch.setattr(requests.Session, "get", get)
+    monkeypatch.setattr(requests, "head", lambda *a, **k: pytest.fail("the Server header probe ran"))
+    return calls
+
+
+@pytest.mark.parametrize("make, base", [(widevine, WidevineRemoteCdm), (playready, PlayReadyRemoteCdm)])
+def test_open_goes_through_upstream_session_without_probe(make: Any, base: type, sent: list) -> None:
+    cdm = make()
+    assert isinstance(cdm, base)
+    assert cdm.open() == b"\x00\xff"
+    assert sent == [(f"{HOST}/dev/open", "s3cret")]
+
+
+@pytest.mark.parametrize("make", [widevine, playready])
+@pytest.mark.parametrize("field", ["host", "secret", "device_name"])
+def test_missing_connection_field_raises(make: Any, field: str) -> None:
+    with pytest.raises(ValueError):
+        make(**{field: ""})
+
+
+def test_playready_rejects_quoted_security_level() -> None:
+    with pytest.raises(TypeError):
+        playready(security_level="3000")
+
+
+@pytest.fixture
+def redirecting_host() -> Iterator[tuple[int, list[tuple[str, Any]]]]:
+    """Redirect `/dev/open` to the `Location` in the `to` query, and answer the target like a serve API."""
+    seen: list[tuple[str, Any]] = []
+
+    class Handler(BaseHTTPRequestHandler):
+        def do_GET(self) -> None:
+            seen.append((self.path.split("?")[0], self.headers.get("X-Secret-Key")))
+            if self.path.startswith("/dev/open"):
+                self.send_response(307)
+                self.send_header("Location", f"http://{self.headers['X-Redirect-To']}:{port}/moved")
+                self.end_headers()
+                return
+            body = json.dumps(FakeResponse().json()).encode()
+            self.send_response(200)
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def log_message(self, *args: Any) -> None:
+            pass
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    port = server.server_port
+    threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True).start()
+    yield port, seen
+    server.shutdown()
+    server.server_close()
+
+
+@pytest.mark.parametrize("make", [widevine, playready])
+@pytest.mark.parametrize("target, secret", [("127.0.0.1", "s3cret"), ("localhost", None)])
+def test_secret_is_dropped_on_a_redirect_to_another_host(
+    make: Any, target: str, secret: Any, redirecting_host: tuple[int, list[tuple[str, Any]]]
+) -> None:
+    port, seen = redirecting_host
+    cdm = make(host=f"http://127.0.0.1:{port}")
+    cdm._RemoteCdm__session.headers["X-Redirect-To"] = target
+    assert cdm.open() == b"\x00\xff"
+    assert seen == [("/dev/open", "s3cret"), ("/moved", secret)]

unshackle/core/cdm/loader.py

@@ -92,10 +92,10 @@ def load_remote_cdm(
 
     device_type = cdm_api.get("Device Type", cdm_api.get("device_type", ""))
     if str(device_type).upper() == "PLAYREADY":
-        from pyplayready.remote.remotecdm import RemoteCdm as PlayReadyRemoteCdm
+        from unshackle.core.cdm.unprobed_remote_cdm import UnprobedPlayReadyRemoteCdm
 
         return stamp_remote(
-            PlayReadyRemoteCdm(
+            UnprobedPlayReadyRemoteCdm(
                 security_level=cdm_api.get("Security Level", cdm_api.get("security_level", 3000)),
                 host=cdm_api.get("Host", cdm_api.get("host")),
                 secret=cdm_api.get("Secret", cdm_api.get("secret")),
@@ -104,10 +104,10 @@ def load_remote_cdm(
             "playready",
         )
 
-    from pywidevine.remotecdm import RemoteCdm
+    from unshackle.core.cdm.unprobed_remote_cdm import UnprobedWidevineRemoteCdm
 
     return stamp_remote(
-        RemoteCdm(
+        UnprobedWidevineRemoteCdm(
             device_type=cdm_api.get("Device Type", cdm_api.get("device_type", "")),
             system_id=cdm_api.get("System ID", cdm_api.get("system_id", "")),
             security_level=cdm_api.get("Security Level", cdm_api.get("security_level", 3000)),

unshackle/core/cdm/unprobed_remote_cdm.py

@@ -0,0 +1,69 @@
+"""pywidevine and pyplayready remote CDMs without the upstream `Server` header probe.
+
+Upstream sends a HEAD request to the host and reads the `Server` header: pywidevine refuses a host
+that does not name pywidevine serve, and pyplayready logs a warning. A reverse proxy or CDN in front
+of the server strips or rewrites that header, so a working server fails. A host that is not a serve
+API now fails at the first API call instead.
+"""
+
+from __future__ import annotations
+
+import functools
+from typing import Any, Optional
+
+import requests
+from Crypto.PublicKey import RSA
+from pyplayready.cdm import Cdm as PlayReadyCdm
+from pyplayready.remote.remotecdm import RemoteCdm as PlayReadyRemoteCdm
+from pywidevine.cdm import Cdm as WidevineCdm
+from pywidevine.license_protocol_pb2 import ClientIdentification
+from pywidevine.remotecdm import RemoteCdm as WidevineRemoteCdm
+
+
+class SecretSession(requests.Session):
+    """Drop the serve API secret on a redirect where requests would drop `Authorization`."""
+
+    def rebuild_auth(self, prepared_request: requests.PreparedRequest, response: requests.Response) -> None:
+        super().rebuild_auth(prepared_request, response)
+        if self.should_strip_auth(response.request.url, prepared_request.url):
+            prepared_request.headers.pop("X-Secret-Key", None)
+
+
+def _connect(cdm: Any, drm: str, host: Optional[str], secret: Optional[str], device_name: Optional[str]) -> None:
+    """Set what upstream `__init__` sets after the probe, on the name-mangled upstream session."""
+    if not (host and secret and device_name):
+        raise ValueError(f"A {drm} remote CDM needs host, secret and device_name")
+    cdm.host = host
+    cdm.device_name = device_name
+    cdm._RemoteCdm__session = SecretSession()
+    cdm._RemoteCdm__session.headers["X-Secret-Key"] = secret
+
+
+@functools.cache
+def _placeholder_key() -> RSA.RsaKey:
+    """An RSA private key only to satisfy `Cdm.__init__`; the remote CDM never uses it."""
+    return RSA.generate(2048)
+
+
+class UnprobedWidevineRemoteCdm(WidevineRemoteCdm):
+    def __init__(
+        self,
+        device_type: Any,
+        system_id: int,
+        security_level: int,
+        host: Optional[str],
+        secret: Optional[str],
+        device_name: Optional[str],
+    ) -> None:
+        _connect(self, "Widevine", host, secret, device_name)
+        WidevineCdm.__init__(self, device_type, system_id, security_level, ClientIdentification(), _placeholder_key())
+
+
+class UnprobedPlayReadyRemoteCdm(PlayReadyRemoteCdm):
+    def __init__(
+        self, security_level: int, host: Optional[str], secret: Optional[str], device_name: Optional[str]
+    ) -> None:
+        if not isinstance(security_level, int):
+            raise TypeError(f"A PlayReady remote CDM needs an int security_level, not {security_level!r}")
+        _connect(self, "PlayReady", host, secret, device_name)
+        PlayReadyCdm.__init__(self, security_level, None, None, None)