mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-13 12:35:05 +00:00
refactor(drm): drop the hardcoded service branch in the license flow
One service needed session_id in its license call, which a near-duplicate of get_content_keys supplied behind a service-tag test. service_licence now passes each service only the arguments its own license function declares, so the duplicate and the branch are gone. The REST API license paths and the base PlayReady delegate get the same filtering, replacing their fixed argument set. Two behaviour changes ride along: an empty license challenge now raises EmptyLicense for every service, not just the one, and a service that declares only challenge now receives a working call instead of a TypeError.
This commit is contained in:
imSp4rky
@@ -0,0 +1,34 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from unshackle.core.utilities import declared_kwargs
|
||||
|
||||
SENT = {"challenge": b"c", "title": "t", "track": "tr", "pssh": "P", "session_id": "S"}
|
||||
|
||||
|
||||
def test_keeps_only_declared_arguments() -> None:
|
||||
def licence(*, challenge: bytes, title: str, track: str) -> None: ...
|
||||
|
||||
assert sorted(declared_kwargs(licence, SENT)) == ["challenge", "title", "track"]
|
||||
|
||||
|
||||
def test_passes_session_id_to_a_function_that_asks_for_it() -> None:
|
||||
def licence(*, challenge: bytes, title: str, track: str, session_id: str | None = None) -> None: ...
|
||||
|
||||
assert sorted(declared_kwargs(licence, SENT)) == ["challenge", "session_id", "title", "track"]
|
||||
|
||||
|
||||
def test_a_var_keyword_function_gets_everything() -> None:
|
||||
def licence(*, challenge: bytes, **kwargs: object) -> None: ...
|
||||
|
||||
assert declared_kwargs(licence, SENT) == SENT
|
||||
|
||||
|
||||
def test_an_unreadable_signature_gets_everything() -> None:
|
||||
# min() has no readable signature, and a compiled service may not either
|
||||
assert declared_kwargs(min, SENT) == SENT
|
||||
|
||||
|
||||
def test_a_signature_we_cannot_call_by_keyword_gets_everything() -> None:
|
||||
def licence(challenge, /, **kwargs: object) -> None: ...
|
||||
|
||||
assert declared_kwargs(licence, SENT) == SENT
|
||||
@@ -0,0 +1,95 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
from uuid import UUID
|
||||
|
||||
import pytest
|
||||
from pywidevine.pssh import PSSH
|
||||
|
||||
from unshackle.core.drm import Widevine
|
||||
from unshackle.core.utilities import declared_kwargs
|
||||
|
||||
KID = UUID("00112233-4455-6677-8899-aabbccddeeff")
|
||||
KEY = "00000000000000000000000000000001"
|
||||
|
||||
|
||||
class FakeKey:
|
||||
def __init__(self, kid: UUID, key: str) -> None:
|
||||
self.kid = kid
|
||||
self.key = bytes.fromhex(key)
|
||||
self.type = "CONTENT"
|
||||
|
||||
|
||||
class FakeCdm:
|
||||
"""The smallest CDM the licensing flow accepts; records what it was given."""
|
||||
|
||||
service_certificate_challenge = b"cert-challenge"
|
||||
|
||||
def __init__(self, challenge: bytes = b"challenge") -> None:
|
||||
self._challenge = challenge
|
||||
self.parsed: list[Any] = []
|
||||
|
||||
def open(self) -> str:
|
||||
return "session-1"
|
||||
|
||||
def close(self, session_id: str) -> None:
|
||||
pass
|
||||
|
||||
def get_license_challenge(self, session_id: str, pssh: PSSH) -> bytes:
|
||||
return self._challenge
|
||||
|
||||
def parse_license(self, session_id: str, response: Any) -> None:
|
||||
self.parsed.append(response)
|
||||
|
||||
def get_keys(self, session_id: str, type_: str) -> list[FakeKey]:
|
||||
return [FakeKey(KID, KEY)]
|
||||
|
||||
|
||||
def _drm() -> Widevine:
|
||||
return Widevine(pssh=PSSH.new(system_id=PSSH.SystemId.Widevine, key_ids=[KID]), kid=KID)
|
||||
|
||||
|
||||
def _licence_for(fn: Any) -> Any:
|
||||
"""Build the callable dl.py gives the DRM object, filtered to what the service declares."""
|
||||
return lambda **kw: fn(**declared_kwargs(fn, {**kw, "title": "t", "track": "tr"}))
|
||||
|
||||
|
||||
def test_a_session_aware_service_gets_the_open_session_id() -> None:
|
||||
seen: dict[str, Any] = {}
|
||||
|
||||
def get_widevine_license(*, challenge: bytes, title: str, track: str, session_id: Any = None) -> bytes:
|
||||
seen.update(challenge=challenge, title=title, track=track, session_id=session_id)
|
||||
return b"license"
|
||||
|
||||
drm = _drm()
|
||||
cdm = FakeCdm()
|
||||
drm.get_content_keys(cdm=cdm, certificate=lambda **_: None, licence=_licence_for(get_widevine_license))
|
||||
|
||||
assert seen["session_id"] == "session-1"
|
||||
assert "pssh" not in seen
|
||||
assert drm.content_keys == {KID: KEY}
|
||||
|
||||
|
||||
def test_a_plain_service_does_not_get_pssh_or_session_id() -> None:
|
||||
seen: dict[str, Any] = {}
|
||||
|
||||
def get_widevine_license(*, challenge: bytes, title: str, track: str) -> bytes:
|
||||
seen.update(challenge=challenge, title=title, track=track)
|
||||
return b"license"
|
||||
|
||||
drm = _drm()
|
||||
drm.get_content_keys(cdm=FakeCdm(), certificate=lambda **_: None, licence=_licence_for(get_widevine_license))
|
||||
|
||||
assert set(seen) == {"challenge", "title", "track"}
|
||||
assert drm.content_keys == {KID: KEY}
|
||||
|
||||
|
||||
def test_an_empty_challenge_raises_instead_of_licensing() -> None:
|
||||
def get_widevine_license(*, challenge: bytes, title: str, track: str) -> bytes:
|
||||
raise AssertionError("the licence request must not go out for an empty challenge")
|
||||
|
||||
drm = _drm()
|
||||
with pytest.raises(Widevine.Exceptions.EmptyLicense):
|
||||
drm.get_content_keys(
|
||||
cdm=FakeCdm(challenge=b""), certificate=lambda **_: None, licence=_licence_for(get_widevine_license)
|
||||
)
|
||||
@@ -0,0 +1,381 @@
|
||||
"""The REST API licence paths must give a service only the arguments it declares.
|
||||
|
||||
`serve` reaches a service's licence function through three call sites that bypass the
|
||||
`dl` command: server_cdm Widevine, server_cdm PlayReady, and the proxy-license forward.
|
||||
Each one now filters with `declared_kwargs`, so a service that never asked for `pssh` or
|
||||
`session_id` keeps working while a session-aware service finally receives `session_id`
|
||||
on the server_cdm path.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
from types import SimpleNamespace
|
||||
from typing import Any, Optional
|
||||
from uuid import UUID
|
||||
|
||||
import pyplayready.system.pssh as pr_pssh_mod
|
||||
import pytest
|
||||
import pywidevine.pssh as wv_pssh_mod
|
||||
|
||||
import unshackle.core.cdm as cdm_mod
|
||||
import unshackle.core.cdm.detect as detect_mod
|
||||
import unshackle.core.drm as drm_mod
|
||||
from unshackle.core.api import handlers
|
||||
from unshackle.core.drm import PlayReady, Widevine
|
||||
from unshackle.core.service import Service
|
||||
|
||||
pytestmark = pytest.mark.unit
|
||||
|
||||
KID = UUID("11111111111111111111111111111111")
|
||||
KEY = bytes.fromhex("22222222222222222222222222222222")
|
||||
PSSH_B64 = base64.b64encode(b"pssh").decode()
|
||||
SESSION_ID = b"sid-bytes"
|
||||
|
||||
|
||||
# --- fake services, one per licence signature shape -------------------------------------
|
||||
|
||||
|
||||
class _Recorder:
|
||||
"""Base for the fake services; every licence call lands in `calls`."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.calls: list[dict[str, Any]] = []
|
||||
|
||||
def get_widevine_service_certificate(self, *, challenge: Any, title: Any, track: Any) -> None:
|
||||
return None
|
||||
|
||||
|
||||
class ClassicService(_Recorder):
|
||||
"""EXAMPLE: the pre-`session_id` signature every older service uses."""
|
||||
|
||||
def get_widevine_license(self, *, challenge: bytes, title: Any, track: Any) -> bytes:
|
||||
self.calls.append({"challenge": challenge, "title": title, "track": track})
|
||||
return b"LIC"
|
||||
|
||||
def get_playready_license(self, *, challenge: bytes, title: Any, track: Any) -> str:
|
||||
self.calls.append({"challenge": challenge, "title": title, "track": track})
|
||||
return "<License>ok</License>"
|
||||
|
||||
|
||||
class SessionService(_Recorder):
|
||||
"""DEMO: a session-aware service, the shape that needs the open CDM session id."""
|
||||
|
||||
def get_widevine_license(
|
||||
self, *, challenge: bytes, title: Any, track: Any, session_id: Optional[bytes] = None
|
||||
) -> bytes:
|
||||
self.calls.append({"challenge": challenge, "title": title, "track": track, "session_id": session_id})
|
||||
return b"LIC"
|
||||
|
||||
def get_playready_license(
|
||||
self, *, challenge: bytes, title: Any, track: Any, session_id: Optional[bytes] = None
|
||||
) -> str:
|
||||
self.calls.append({"challenge": challenge, "title": title, "track": track, "session_id": session_id})
|
||||
return "<License>ok</License>"
|
||||
|
||||
|
||||
class VarKwService(_Recorder):
|
||||
"""DEMO2: a `**kwargs` signature takes whatever the framework offers."""
|
||||
|
||||
def get_widevine_license(self, *, challenge: bytes, **kwargs: Any) -> bytes:
|
||||
self.calls.append({"challenge": challenge, **kwargs})
|
||||
return b"LIC"
|
||||
|
||||
def get_playready_license(self, *, challenge: bytes, **kwargs: Any) -> str:
|
||||
self.calls.append({"challenge": challenge, **kwargs})
|
||||
return "<License>ok</License>"
|
||||
|
||||
|
||||
# --- stub DRM objects that keep the real get_content_keys call shape ---------------------
|
||||
|
||||
|
||||
class StubWidevine(Widevine):
|
||||
"""Real `Widevine.get_content_keys`, with the PSSH parsing skipped."""
|
||||
|
||||
# pssh and kids are read-only properties on the real class; shadow them so a stub can set them
|
||||
pssh: Any = None
|
||||
kids: Any = None
|
||||
|
||||
def __init__(self, pssh: Any = None, **kwargs: Any) -> None:
|
||||
self.pssh = pssh
|
||||
self.kids = [KID]
|
||||
self.content_keys: dict[UUID, str] = {}
|
||||
|
||||
|
||||
class StubPlayReady(PlayReady):
|
||||
"""Real `PlayReady.get_content_keys`, with the WRMHEADER parsing skipped."""
|
||||
|
||||
# pssh, pssh_b64 and kids are read-only properties on the real class
|
||||
pssh: Any = None
|
||||
pssh_b64: Any = None
|
||||
kids: Any = None
|
||||
|
||||
def __init__(self, pssh: Any = None, pssh_b64: Optional[str] = None, **kwargs: Any) -> None:
|
||||
self.pssh = SimpleNamespace(wrm_headers=["<WRMHEADER/>"])
|
||||
self.pssh_b64 = pssh_b64
|
||||
self.kids = [KID]
|
||||
self.content_keys: dict[UUID, str] = {}
|
||||
|
||||
def extract_keys_from_cdm(self, cdm: Any, session_id: Any) -> dict[UUID, str]:
|
||||
return {KID: KEY.hex()}
|
||||
|
||||
|
||||
class FakeWidevineCdm:
|
||||
service_certificate_challenge = b"cert-chal"
|
||||
|
||||
def open(self) -> bytes:
|
||||
return SESSION_ID
|
||||
|
||||
def get_license_challenge(self, session_id: bytes, pssh: Any) -> bytes:
|
||||
return b"CHAL"
|
||||
|
||||
def parse_license(self, session_id: bytes, license_res: Any) -> None:
|
||||
return None
|
||||
|
||||
def get_keys(self, session_id: bytes, type_: str) -> list[SimpleNamespace]:
|
||||
return [SimpleNamespace(kid=KID, key=KEY)]
|
||||
|
||||
def close(self, session_id: bytes) -> None:
|
||||
return None
|
||||
|
||||
|
||||
class FakePlayReadyCdm:
|
||||
def open(self) -> bytes:
|
||||
return SESSION_ID
|
||||
|
||||
def get_license_challenge(self, session_id: bytes, wrm_header: Any) -> str:
|
||||
return "CHAL"
|
||||
|
||||
def parse_license(self, session_id: bytes, license_str: str) -> None:
|
||||
return None
|
||||
|
||||
def close(self, session_id: bytes) -> None:
|
||||
return None
|
||||
|
||||
|
||||
# --- fixtures ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def widevine_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(wv_pssh_mod, "PSSH", lambda b64: SimpleNamespace(b64=b64))
|
||||
monkeypatch.setattr(drm_mod, "Widevine", StubWidevine)
|
||||
monkeypatch.setattr(cdm_mod, "load_cdm", lambda *a, **k: FakeWidevineCdm())
|
||||
monkeypatch.setattr(detect_mod, "is_widevine_cdm", lambda cdm: True)
|
||||
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: None)
|
||||
monkeypatch.setattr(handlers, "cache_to_vaults", lambda keys, name: None)
|
||||
monkeypatch.setattr(handlers.config, "serve", {"users": {}}, raising=False)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def playready_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(pr_pssh_mod, "PSSH", lambda data: SimpleNamespace(data=data))
|
||||
monkeypatch.setattr(drm_mod, "PlayReady", StubPlayReady)
|
||||
monkeypatch.setattr(cdm_mod, "load_cdm", lambda *a, **k: FakePlayReadyCdm())
|
||||
monkeypatch.setattr(detect_mod, "is_playready_cdm", lambda cdm: True)
|
||||
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, "cache_to_vaults", lambda keys, name: None)
|
||||
monkeypatch.setattr(handlers.config, "serve", {"users": {}}, raising=False)
|
||||
|
||||
|
||||
def _server_cdm(service: Any, drm_type: str) -> dict[str, str]:
|
||||
return handlers.handle_single_server_cdm(
|
||||
service=service,
|
||||
title=SimpleNamespace(),
|
||||
track=SimpleNamespace(),
|
||||
pssh_b64=PSSH_B64,
|
||||
drm_type=drm_type,
|
||||
request=None,
|
||||
)
|
||||
|
||||
|
||||
# --- server_cdm, Widevine ---------------------------------------------------------------
|
||||
|
||||
|
||||
def test_server_cdm_widevine_classic_signature_gets_three_arguments(widevine_env: None) -> None:
|
||||
service = ClassicService()
|
||||
keys = _server_cdm(service, "widevine")
|
||||
|
||||
assert keys == {KID.hex: KEY.hex()}
|
||||
assert len(service.calls) == 1
|
||||
assert set(service.calls[0]) == {"challenge", "title", "track"}
|
||||
assert service.calls[0]["challenge"] == b"CHAL"
|
||||
|
||||
|
||||
def test_server_cdm_widevine_session_service_receives_the_open_session_id(widevine_env: None) -> None:
|
||||
service = SessionService()
|
||||
_server_cdm(service, "widevine")
|
||||
|
||||
assert set(service.calls[0]) == {"challenge", "title", "track", "session_id"}
|
||||
assert service.calls[0]["session_id"] == SESSION_ID
|
||||
|
||||
|
||||
def test_server_cdm_widevine_var_kwargs_service_receives_pssh_and_session_id(widevine_env: None) -> None:
|
||||
service = VarKwService()
|
||||
_server_cdm(service, "widevine")
|
||||
|
||||
assert set(service.calls[0]) == {"challenge", "title", "track", "pssh", "session_id"}
|
||||
assert service.calls[0]["session_id"] == SESSION_ID
|
||||
|
||||
|
||||
def test_server_cdm_widevine_unnarrowable_signature_retries_without_the_extras(widevine_env: None) -> None:
|
||||
# declared_kwargs cannot narrow a signature it cannot read (a compiled service) so it offers
|
||||
# everything. The call raises TypeError and Widevine retries with `challenge` alone, which
|
||||
# the lambda turns back into the classic challenge/title/track set the service does accept.
|
||||
service = ClassicService()
|
||||
calls: list[dict[str, Any]] = []
|
||||
|
||||
def compiled(**kw: Any) -> bytes:
|
||||
calls.append(kw)
|
||||
if {"pssh", "session_id"} & set(kw):
|
||||
raise TypeError("unexpected keyword argument")
|
||||
return b"LIC"
|
||||
|
||||
service.get_widevine_license = compiled # type: ignore[method-assign]
|
||||
_server_cdm(service, "widevine")
|
||||
|
||||
assert len(calls) == 2
|
||||
assert set(calls[0]) == {"challenge", "title", "track", "pssh", "session_id"}
|
||||
assert set(calls[1]) == {"challenge", "title", "track"}
|
||||
|
||||
|
||||
def test_server_cdm_widevine_lambda_is_never_itself_filtered(widevine_env: None) -> None:
|
||||
# The `(**kw)` lambda would look like a VAR_KEYWORD service if declared_kwargs were ever
|
||||
# applied to it, and a classic service would then be handed `pssh` and `session_id`.
|
||||
service = ClassicService()
|
||||
_server_cdm(service, "widevine")
|
||||
|
||||
assert "pssh" not in service.calls[0]
|
||||
assert "session_id" not in service.calls[0]
|
||||
|
||||
|
||||
# --- server_cdm, PlayReady --------------------------------------------------------------
|
||||
|
||||
|
||||
def test_server_cdm_playready_classic_signature_gets_three_arguments(playready_env: None) -> None:
|
||||
service = ClassicService()
|
||||
keys = _server_cdm(service, "playready")
|
||||
|
||||
assert keys == {KID.hex: KEY.hex()}
|
||||
assert set(service.calls[0]) == {"challenge", "title", "track"}
|
||||
|
||||
|
||||
def test_server_cdm_playready_offers_pssh_b64_not_pssh_or_session_id(playready_env: None) -> None:
|
||||
# PlayReady.get_content_keys calls licence(challenge=..., pssh_b64=...); it has no
|
||||
# session_id to give, so a session-aware service sees the None default there.
|
||||
service = VarKwService()
|
||||
_server_cdm(service, "playready")
|
||||
|
||||
assert set(service.calls[0]) == {"challenge", "title", "track", "pssh_b64"}
|
||||
assert service.calls[0]["pssh_b64"] == PSSH_B64
|
||||
|
||||
|
||||
def test_server_cdm_playready_session_service_is_not_passed_session_id(playready_env: None) -> None:
|
||||
service = SessionService()
|
||||
_server_cdm(service, "playready")
|
||||
|
||||
assert set(service.calls[0]) == {"challenge", "title", "track", "session_id"}
|
||||
assert service.calls[0]["session_id"] is None # the signature default, never an offered value
|
||||
|
||||
|
||||
# --- proxy license ----------------------------------------------------------------------
|
||||
|
||||
|
||||
def _proxy(service: Any, drm_type: str) -> Any:
|
||||
return handlers.handle_proxy_license(
|
||||
service=service,
|
||||
title=SimpleNamespace(),
|
||||
track=SimpleNamespace(),
|
||||
challenge_b64=base64.b64encode(b"CHAL").decode(),
|
||||
drm_type=drm_type,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("drm_type", ["widevine", "playready"])
|
||||
def test_proxy_license_classic_signature(drm_type: str) -> None:
|
||||
service = ClassicService()
|
||||
response = _proxy(service, drm_type)
|
||||
|
||||
assert response.status == 200
|
||||
assert set(service.calls[0]) == {"challenge", "title", "track"}
|
||||
assert service.calls[0]["challenge"] == b"CHAL"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("drm_type", ["widevine", "playready"])
|
||||
def test_proxy_license_var_kwargs_service_gets_no_session_id(drm_type: str) -> None:
|
||||
# The client holds the CDM, so there is no server-side session to name. A `**kwargs`
|
||||
# service must not see a session_id key at all.
|
||||
service = VarKwService()
|
||||
_proxy(service, drm_type)
|
||||
|
||||
assert set(service.calls[0]) == {"challenge", "title", "track"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("drm_type", ["widevine", "playready"])
|
||||
def test_proxy_license_session_service_sees_its_own_default(drm_type: str) -> None:
|
||||
# A session-aware service is called without session_id, so its signature default applies.
|
||||
# A service that then uses session_id unconditionally fails here, before and after the
|
||||
# declared_kwargs change alike.
|
||||
service = SessionService()
|
||||
_proxy(service, drm_type)
|
||||
|
||||
assert set(service.calls[0]) == {"challenge", "title", "track", "session_id"}
|
||||
assert service.calls[0]["session_id"] is None
|
||||
|
||||
|
||||
def test_proxy_license_rejects_an_unknown_drm_type() -> None:
|
||||
from unshackle.core.api.errors import APIError
|
||||
|
||||
with pytest.raises(APIError):
|
||||
_proxy(ClassicService(), "monalisa")
|
||||
|
||||
|
||||
# --- the base-class PlayReady delegate --------------------------------------------------
|
||||
|
||||
|
||||
class _DelegatingService(Service):
|
||||
"""A service with no PlayReady implementation: the base class delegates to Widevine."""
|
||||
|
||||
def __init__(self) -> None: # deliberately skips Service.__init__
|
||||
self.calls: list[dict[str, Any]] = []
|
||||
|
||||
def get_titles(self) -> Any: ... # pragma: no cover - abstract stub
|
||||
|
||||
def get_tracks(self, title: Any) -> Any: ... # pragma: no cover - abstract stub
|
||||
|
||||
def get_chapters(self, title: Any) -> Any: ... # pragma: no cover - abstract stub
|
||||
|
||||
def get_widevine_license(self, *, challenge: bytes, title: Any, track: Any) -> bytes:
|
||||
self.calls.append({"challenge": challenge, "title": title, "track": track})
|
||||
return b"LIC"
|
||||
|
||||
|
||||
class _DelegatingVarKwService(_DelegatingService):
|
||||
def get_widevine_license(self, *, challenge: bytes, **kwargs: Any) -> bytes: # type: ignore[override]
|
||||
self.calls.append({"challenge": challenge, **kwargs})
|
||||
return b"LIC"
|
||||
|
||||
|
||||
def test_base_playready_delegate_filters_to_the_widevine_signature() -> None:
|
||||
service = _DelegatingService()
|
||||
assert service.get_playready_license(challenge=b"C", title="T", track="TR") == b"LIC"
|
||||
assert set(service.calls[0]) == {"challenge", "title", "track"}
|
||||
|
||||
|
||||
def test_base_playready_delegate_passes_everything_to_a_var_kwargs_service() -> None:
|
||||
service = _DelegatingVarKwService()
|
||||
service.get_playready_license(challenge=b"C", title="T", track="TR")
|
||||
assert set(service.calls[0]) == {"challenge", "title", "track"}
|
||||
|
||||
|
||||
def test_base_playready_delegate_cannot_forward_session_id() -> None:
|
||||
# Service.get_playready_license declares (challenge, title, track) only, so the server_cdm
|
||||
# filter drops session_id before the delegate ever runs. A session-aware service that wants
|
||||
# PlayReady must implement get_playready_license itself.
|
||||
from unshackle.core.utilities import declared_kwargs
|
||||
|
||||
offered = {"challenge": b"C", "title": "T", "track": "TR", "session_id": SESSION_ID}
|
||||
assert "session_id" not in declared_kwargs(Service.get_playready_license, offered)
|
||||
@@ -67,6 +67,7 @@ from unshackle.core.tracks.hybrid import Hybrid
|
||||
from unshackle.core.tracks.track import assert_fragments_decrypted, has_encrypted_sample_entry
|
||||
from unshackle.core.utilities import (
|
||||
as_requested,
|
||||
declared_kwargs,
|
||||
embedded_audio_langs,
|
||||
excluded_language_tags,
|
||||
find_font_with_fallbacks,
|
||||
@@ -4007,14 +4008,16 @@ class dl:
|
||||
def service_licence(self, service: Service, drm_system: Optional[str] = None, **kwargs: Any) -> Any:
|
||||
"""Call the service's licence function for the track's DRM system.
|
||||
|
||||
Arguments go through declared_kwargs, so a service receives only the ones its own
|
||||
licence function declares.
|
||||
|
||||
prepare_drm passes drm_system so the choice follows the track, not the loaded CDM:
|
||||
a track downloading at the same time can swap self.cdm.
|
||||
"""
|
||||
if drm_system is None:
|
||||
drm_system = "playready" if is_playready_cdm(self.cdm) else "widevine"
|
||||
if drm_system == "playready":
|
||||
return service.get_playready_license(**kwargs)
|
||||
return service.get_widevine_license(**kwargs)
|
||||
fn = service.get_playready_license if drm_system == "playready" else service.get_widevine_license
|
||||
return fn(**declared_kwargs(fn, kwargs))
|
||||
|
||||
def prepare_drm(
|
||||
self,
|
||||
@@ -4281,10 +4284,7 @@ class dl:
|
||||
)
|
||||
|
||||
try:
|
||||
if self.service == "NF":
|
||||
drm.get_NF_content_keys(cdm=track_cdm, licence=licence, certificate=certificate)
|
||||
else:
|
||||
drm.get_content_keys(cdm=track_cdm, licence=licence, certificate=certificate)
|
||||
drm.get_content_keys(cdm=track_cdm, licence=licence, certificate=certificate)
|
||||
except Exception as e:
|
||||
if drm.content_keys:
|
||||
self.log.debug(f"License call failed but keys already in content_keys: {e}")
|
||||
|
||||
@@ -27,6 +27,7 @@ from unshackle.core.proxies.resolve import initialize_proxy_providers, resolve_p
|
||||
from unshackle.core.services import Services
|
||||
from unshackle.core.titles import Episode, Movie, Song, Title_T
|
||||
from unshackle.core.tracks import Audio, Subtitle, Tracks, Video
|
||||
from unshackle.core.utilities import declared_kwargs
|
||||
from unshackle.core.utils.collections import ci_get
|
||||
from unshackle.core.utils.redact import REDACTED, URL_USERINFO_RE, redact_text
|
||||
|
||||
@@ -3678,7 +3679,9 @@ def handle_single_server_cdm(
|
||||
pr_drm.get_content_keys(
|
||||
cdm=cdm,
|
||||
certificate=lambda challenge, **_: None,
|
||||
licence=lambda challenge, **_: service.get_playready_license(challenge=challenge, title=title, track=track),
|
||||
licence=lambda **kw: service.get_playready_license(
|
||||
**declared_kwargs(service.get_playready_license, {**kw, "title": title, "track": track})
|
||||
),
|
||||
)
|
||||
keys = {kid.hex: key for kid, key in pr_drm.content_keys.items()}
|
||||
elif drm_type == "widevine":
|
||||
@@ -3705,7 +3708,9 @@ def handle_single_server_cdm(
|
||||
certificate=lambda challenge, **_: service.get_widevine_service_certificate(
|
||||
challenge=challenge, title=title, track=track
|
||||
),
|
||||
licence=lambda challenge, **_: service.get_widevine_license(challenge=challenge, title=title, track=track),
|
||||
licence=lambda **kw: service.get_widevine_license(
|
||||
**declared_kwargs(service.get_widevine_license, {**kw, "title": title, "track": track})
|
||||
),
|
||||
)
|
||||
keys = {kid.hex: key for kid, key in wv_drm.content_keys.items()}
|
||||
else:
|
||||
@@ -3736,9 +3741,17 @@ def handle_proxy_license(
|
||||
challenge_bytes = base64.b64decode(challenge_b64)
|
||||
|
||||
if drm_type == "widevine":
|
||||
license_response = service.get_widevine_license(challenge=challenge_bytes, title=title, track=track)
|
||||
license_response = service.get_widevine_license(
|
||||
**declared_kwargs(
|
||||
service.get_widevine_license, {"challenge": challenge_bytes, "title": title, "track": track}
|
||||
)
|
||||
)
|
||||
elif drm_type == "playready":
|
||||
license_response = service.get_playready_license(challenge=challenge_bytes, title=title, track=track)
|
||||
license_response = service.get_playready_license(
|
||||
**declared_kwargs(
|
||||
service.get_playready_license, {"challenge": challenge_bytes, "title": title, "track": track}
|
||||
)
|
||||
)
|
||||
else:
|
||||
raise APIError(
|
||||
APIErrorCode.INVALID_PARAMETERS,
|
||||
|
||||
@@ -220,6 +220,10 @@ class Widevine:
|
||||
|
||||
if hasattr(cdm, "has_cached_keys") and cdm.has_cached_keys(session_id):
|
||||
pass
|
||||
elif not challenge:
|
||||
raise Widevine.Exceptions.EmptyLicense(
|
||||
"The CDM returned an empty licence challenge and has no cached keys"
|
||||
)
|
||||
else:
|
||||
log_event(
|
||||
"drm_license_request",
|
||||
@@ -231,7 +235,7 @@ class Widevine:
|
||||
kid_count=len(self.kids),
|
||||
)
|
||||
try:
|
||||
license_res = licence(challenge=challenge, pssh=self.pssh)
|
||||
license_res = licence(challenge=challenge, pssh=self.pssh, session_id=session_id)
|
||||
except TypeError:
|
||||
license_res = licence(challenge=challenge)
|
||||
cdm.parse_license(session_id, license_res)
|
||||
@@ -265,52 +269,6 @@ class Widevine:
|
||||
finally:
|
||||
cdm.close(session_id)
|
||||
|
||||
def get_NF_content_keys(self, cdm: WidevineCdm, certificate: Callable, licence: Callable) -> None:
|
||||
"""
|
||||
Make a CDM Session and get Content Keys for this DRM Instance.
|
||||
The certificate and license params are functions. unshackle gives them
|
||||
the challenge and the session ID.
|
||||
"""
|
||||
for kid in self.kids:
|
||||
if kid in self.content_keys:
|
||||
continue
|
||||
|
||||
session_id = cdm.open()
|
||||
|
||||
try:
|
||||
cert = certificate(challenge=cdm.service_certificate_challenge)
|
||||
if cert and hasattr(cdm, "set_service_certificate"):
|
||||
cdm.set_service_certificate(session_id, cert)
|
||||
|
||||
if hasattr(cdm, "set_required_kids"):
|
||||
try:
|
||||
cdm.set_required_kids(self.kids, session_id=session_id)
|
||||
except TypeError: # CDM predating the per-session signature
|
||||
cdm.set_required_kids(self.kids)
|
||||
|
||||
challenge = cdm.get_license_challenge(session_id, self.pssh)
|
||||
|
||||
if hasattr(cdm, "has_cached_keys") and cdm.has_cached_keys(session_id):
|
||||
pass
|
||||
elif not challenge:
|
||||
raise Widevine.Exceptions.EmptyLicense(
|
||||
"The CDM returned an empty license challenge and has no cached keys"
|
||||
)
|
||||
else:
|
||||
cdm.parse_license(
|
||||
session_id,
|
||||
licence(session_id=session_id, challenge=challenge),
|
||||
)
|
||||
|
||||
self.content_keys = {key.kid: key.key.hex() for key in cdm.get_keys(session_id, "CONTENT")}
|
||||
if not self.content_keys:
|
||||
raise Widevine.Exceptions.EmptyLicense("No Content Keys were within the License")
|
||||
|
||||
if kid not in self.content_keys:
|
||||
raise Widevine.Exceptions.CEKNotFound(f"No Content Key for KID {kid.hex} within the License")
|
||||
finally:
|
||||
cdm.close(session_id)
|
||||
|
||||
def decrypt(self, path: Path) -> None:
|
||||
"""
|
||||
Decrypt a Track with Widevine DRM.
|
||||
|
||||
@@ -38,6 +38,7 @@ from unshackle.core.title_cacher import TitleCacher, get_account_hash, get_regio
|
||||
from unshackle.core.titles import Title_T, Titles_T, remap_titles
|
||||
from unshackle.core.tracks import Chapters, Tracks
|
||||
from unshackle.core.tracks.video import Video
|
||||
from unshackle.core.utilities import declared_kwargs
|
||||
from unshackle.core.utils.ip_info import get_ip_info
|
||||
from unshackle.core.utils.redact import mask_proxy
|
||||
|
||||
@@ -512,8 +513,11 @@ class Service(metaclass=ABCMeta):
|
||||
:return: The License response as Bytes or a Base64 string. Do not Base64 Encode or
|
||||
Decode the data, return as is to reduce unnecessary computations.
|
||||
"""
|
||||
# Delegates license handling to the Widevine license method by default if a service-specific PlayReady implementation is not provided.
|
||||
return self.get_widevine_license(challenge=challenge, title=title, track=track)
|
||||
# A service that implements Widevine licensing only still gets PlayReady, with the
|
||||
# arguments its own signature declares.
|
||||
return self.get_widevine_license(
|
||||
**declared_kwargs(self.get_widevine_license, {"challenge": challenge, "title": title, "track": track})
|
||||
)
|
||||
|
||||
def get_clearkey_license(
|
||||
self, *, challenge: bytes, title: Title_T, track: AnyTrack
|
||||
|
||||
@@ -2,6 +2,7 @@ import ast
|
||||
import contextlib
|
||||
import gzip
|
||||
import importlib.util
|
||||
import inspect
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
@@ -16,7 +17,7 @@ from collections import defaultdict
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from types import ModuleType
|
||||
from typing import Any, Optional, Sequence, Union
|
||||
from typing import Any, Callable, Optional, Sequence, Union
|
||||
from urllib.parse import ParseResult, urlparse
|
||||
from uuid import uuid4
|
||||
|
||||
@@ -1254,6 +1255,23 @@ def close_debug_logger():
|
||||
debug_logger = None
|
||||
|
||||
|
||||
def declared_kwargs(fn: Callable[..., Any], kwargs: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Keep only the arguments ``fn`` declares, so an extra one never raises TypeError.
|
||||
|
||||
A service licence function that never asked for ``pssh`` or ``session_id`` must not fail
|
||||
on them. Filter only against a signature this function can read and call by keyword.
|
||||
Anything else, such as the licence function of a compiled service, gets every argument,
|
||||
and the caller handles the TypeError.
|
||||
"""
|
||||
try:
|
||||
params = inspect.signature(fn).parameters
|
||||
except (TypeError, ValueError):
|
||||
return kwargs
|
||||
if "challenge" not in params or any(p.kind is p.VAR_KEYWORD for p in params.values()):
|
||||
return kwargs
|
||||
return {k: v for k, v in kwargs.items() if k in params}
|
||||
|
||||
|
||||
__all__ = (
|
||||
"DebugLogger",
|
||||
"get_debug_logger",
|
||||
|
||||
Reference in New Issue
Block a user