diff --git a/tests/remote/unit/test_load_full_cdm_type.py b/tests/remote/unit/test_load_full_cdm_type.py new file mode 100644 index 00000000..05523de4 --- /dev/null +++ b/tests/remote/unit/test_load_full_cdm_type.py @@ -0,0 +1,46 @@ +"""load_full_cdm must let a proxy client's cdm_type decide the DRM system. + +A proxy-mode client holds its own CDM and decrypts locally, so the server never licenses. +When the client's cdm_type differs from the server's configured device, the resolved CDM +must be a type-only stub of the client's type, not the server's device, so the service +requests a matching playout and stores the matching licence URL. +""" + +import pytest + +from unshackle.core.api import handlers +from unshackle.core.cdm.detect import is_playready_cdm, is_widevine_cdm + + +@pytest.fixture +def widevine_configured(monkeypatch): + """config.cdm[service] is a quality-tiered dict of Widevine devices (the real deploy shape).""" + from unshackle.core.config import config as app_config + + monkeypatch.setattr( + app_config, "cdm", {"SVC": {"<=1080": "wv_l3", ">1080": "wv_l1", "default": "wv_l3"}}, raising=False + ) + # every resolved device name is a Widevine device + monkeypatch.setattr(handlers, "detect_cdm_type", lambda name, cfg: "widevine") + + def fail_load(*args, **kwargs): + raise AssertionError("load_cdm must not run for a mismatched type") + + monkeypatch.setattr("unshackle.core.cdm.load_cdm", fail_load) + + +def test_playready_client_gets_a_playready_stub(widevine_configured): + cdm = handlers.load_full_cdm("SVC", None, "playready") + assert is_playready_cdm(cdm) and not is_widevine_cdm(cdm) + + +def test_widevine_client_loads_the_configured_device(monkeypatch): + """A matching type must still load the real device (regression guard, not a stub).""" + from unshackle.core.config import config as app_config + + monkeypatch.setattr(app_config, "cdm", {"SVC": {"default": "wv_l3"}}, raising=False) + monkeypatch.setattr(handlers, "detect_cdm_type", lambda name, cfg: "widevine") + loaded = {} + monkeypatch.setattr("unshackle.core.cdm.load_cdm", lambda name, service_name=None: loaded.setdefault("n", name)) + handlers.load_full_cdm("SVC", None, "widevine") + assert loaded["n"] == "wv_l3" diff --git a/unshackle/core/api/handlers.py b/unshackle/core/api/handlers.py index a9467aa6..20cb5542 100644 --- a/unshackle/core/api/handlers.py +++ b/unshackle/core/api/handlers.py @@ -153,6 +153,13 @@ def load_full_cdm(service: str, profile: Optional[str], cdm_type: Optional[str] if not cdm_name or not isinstance(cdm_name, str): return resolve_server_cdm(service, profile, cdm_type) + if cdm_type: + wanted = {"wv": "widevine", "widevine": "widevine", "pr": "playready", "playready": "playready"}.get( + cdm_type.lower() + ) + if wanted and detect_cdm_type(cdm_name, app_config) not in (None, wanted): + return cdm_type_stub(wanted) + try: return load_cdm(cdm_name, service_name=service) except Exception as exc: # noqa: BLE001 - fall back to stub on load failure @@ -4052,26 +4059,35 @@ def handle_proxy_license( raise APIError(APIErrorCode.INVALID_INPUT, "Missing required parameter: challenge") challenge_bytes = base64.b64decode(challenge_b64) - if drm_type == "widevine": - license_response = service.get_widevine_license( - **declared_kwargs( - service.get_widevine_license, {"challenge": challenge_bytes, "title": title, "track": track} - ) - ) - elif drm_type == "playready": - challenge_str = challenge_bytes.decode("utf-8", errors="replace") - license_response = service.get_playready_license( - **declared_kwargs( - service.get_playready_license, {"challenge": challenge_str, "title": title, "track": track} - ) - ) - else: + if drm_type not in ("widevine", "playready"): raise APIError( APIErrorCode.INVALID_PARAMETERS, f"Unsupported DRM type: {drm_type}", details={"drm_type": drm_type, "supported": ["widevine", "playready"]}, ) + # A service raises when the upstream licence server rejects the challenge. + # Surface it as a structured licence error, not an uncaught 500 the edge turns into a 502. + try: + if drm_type == "widevine": + license_response = service.get_widevine_license( + **declared_kwargs( + service.get_widevine_license, {"challenge": challenge_bytes, "title": title, "track": track} + ) + ) + else: + challenge_str = challenge_bytes.decode("utf-8", errors="replace") + license_response = service.get_playready_license( + **declared_kwargs( + service.get_playready_license, {"challenge": challenge_str, "title": title, "track": track} + ) + ) + except APIError: + raise + except (Exception, SystemExit) as exc: + log.exception(f"{sanitize_log(drm_type)} licence request failed for the proxied challenge") + raise APIError(APIErrorCode.SERVICE_ERROR, f"Licence request failed: {exc}") from None + if isinstance(license_response, str): license_response = license_response.encode("utf-8")