Merge last minute changes from dev

Co-authored-by: CodeName393 <62503817+CodeName393@users.noreply.github.com>

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/custom_remote_cdm.py

@@ -27,7 +27,13 @@ class MockCertificateChain:
 class Key:
     """`Key` object compatible with pywidevine."""
 
-    def __init__(self, kid: str, key: str, type_: str = "CONTENT"):
+    def __init__(
+        self,
+        kid: Union[str, UUID],
+        key: Union[str, bytes],
+        type_: str = "CONTENT",
+        permissions: Optional[List[str]] = None,
+    ):
         if isinstance(kid, str):
             clean_kid = kid.replace("-", "")
             if len(clean_kid) == 32:
@@ -43,6 +49,7 @@ class Key:
             self.key = key
 
         self.type = type_
+        self.permissions = permissions or []
 
 
 class CustomRemoteCDMExceptions:
@@ -641,7 +648,14 @@ class CustomRemoteCDM:
                     key_type = key_obj.get(key_fields.get("type", "type"), "CONTENT")
 
                     if kid and key:
-                        keys.append({"kid": str(kid), "key": str(key), "type": str(key_type)})
+                        keys.append(
+                            {
+                                "kid": str(kid),
+                                "key": str(key),
+                                "type": str(key_type),
+                                "permissions": key_obj.get("permissions", []),
+                            }
+                        )
 
         elif isinstance(keys_data, str):
             keys = self.apply_transform(keys_data, "parse_key_string")
@@ -779,8 +793,6 @@ class CustomRemoteCDM:
             InvalidSession: If session ID is invalid
             requests.RequestException: If API request fails
         """
-        _ = license_type, privacy_mode
-
         if session_id not in self._sessions:
             raise CustomRemoteCDMExceptions.InvalidSession(f"Invalid session ID: {session_id.hex()}")
 
@@ -818,6 +830,8 @@ class CustomRemoteCDM:
         base_params = {
             "scheme": self.device_name,
             "init_data": init_data,
+            "license_type": license_type,
+            "privacy_mode": privacy_mode,
         }
 
         if self.service_name:
@@ -1028,7 +1042,25 @@ class CustomRemoteCDM:
             raise CustomRemoteCDMExceptions.InvalidSession(f"Invalid session ID: {session_id.hex()}")
 
         key_dicts = self._sessions[session_id]["keys"]
-        keys = [Key(kid=k["kid"], key=k["key"], type_=k["type"]) for k in key_dicts]
+        keys = []
+        for k in key_dicts:
+            kid_val = k["kid"]
+            if isinstance(kid_val, str):
+                clean_kid = kid_val.replace("-", "")
+                kid_obj = UUID(hex=clean_kid) if len(clean_kid) == 32 else UUID(hex=clean_kid.ljust(32, "0"))
+            else:
+                kid_obj = kid_val
+
+            key_bytes = bytes.fromhex(k["key"].replace(":", "")) if isinstance(k["key"], str) else k["key"]
+
+            keys.append(
+                Key(
+                    type_=k.get("type", "CONTENT"),
+                    kid=kid_obj,
+                    key=key_bytes,
+                    permissions=k.get("permissions", []),
+                )
+            )
 
         if type_:
             keys = [key for key in keys if key.type == type_]

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)