From c2f7fd87a314df93beacaeefbdd309c957ddf7eb Mon Sep 17 00:00:00 2001 From: imSp4rky Date: Thu, 23 Jul 2026 08:48:51 -0600 Subject: [PATCH] feat(drm): detect revoked PlayReady devices from license responses Add is_revocation()/REVOCATION_CODES and centrally raise DeviceRevoked when a license fault carries a device/cert revocation HRESULT, so any service surfaces it without its own decoding. --- unshackle/core/drm/playready.py | 29 +++++++++++++++++++++++++- unshackle/core/drm/playready_errors.py | 20 ++++++++++++++++++ 2 files changed, 48 insertions(+), 1 deletion(-) diff --git a/unshackle/core/drm/playready.py b/unshackle/core/drm/playready.py index 6fdd46d9..58685361 100644 --- a/unshackle/core/drm/playready.py +++ b/unshackle/core/drm/playready.py @@ -1,6 +1,7 @@ from __future__ import annotations import base64 +import re import shutil import subprocess import textwrap @@ -307,6 +308,7 @@ class PlayReady: challenge = cdm.get_license_challenge(session_id, self.pssh.wrm_headers[0]) if challenge: + license_str = "" try: log_event( "drm_license_request", @@ -339,7 +341,12 @@ class PlayReady: drm_type="PlayReady", response_size=len(license_str), ) - except Exception: + except PlayReady.Exceptions.DeviceRevoked: + raise + except Exception as e: + revoked = self._detect_revocation(license_str) or self._detect_revocation(str(e)) + if revoked: + raise PlayReady.Exceptions.DeviceRevoked(revoked) from e raise keys = self._extract_keys_from_cdm(cdm, session_id) @@ -360,6 +367,23 @@ class PlayReady: if not self.content_keys: raise PlayReady.Exceptions.EmptyLicense("No Content Keys were within the License") + @staticmethod + def _detect_revocation(text: str) -> Optional[str]: + """Return the decoded error string if a revocation HRESULT is present, else None. + + Reads the code from the raw SOAP body (0x8004C065) + or from an exception message that carries it, so any service is covered. + """ + from unshackle.core.drm.playready_errors import describe, is_revocation + + if not text: + return None + for token in re.findall(r"0x[0-9A-Fa-f]{8}|-?\d{7,}", text): + code = int(token, 16) if token.lower().startswith("0x") else int(token) + if is_revocation(code): + return describe(code) or f"0x{code & 0xFFFFFFFF:08X}" + return None + def decrypt(self, path: Path) -> None: """ Decrypt a Track with PlayReady DRM. @@ -548,5 +572,8 @@ class PlayReady: class EmptyLicense(Exception): pass + class DeviceRevoked(Exception): + pass + __all__ = ("PlayReady",) diff --git a/unshackle/core/drm/playready_errors.py b/unshackle/core/drm/playready_errors.py index 0560eaed..9890ea89 100644 --- a/unshackle/core/drm/playready_errors.py +++ b/unshackle/core/drm/playready_errors.py @@ -841,6 +841,20 @@ ERRORS: dict[int, tuple[str, str]] = { } +REVOCATION_CODES = frozenset( + { + 0x8004C065, # DRM_E_DEVCERT_REVOKED + 0x8004C053, # DRM_E_CERTIFICATE_REVOKED + 0x8004C07D, # DRM_E_ACTIVATION_GROUP_CERT_REVOKED_ERROR + } +) + + +def is_revocation(code: int) -> bool: + """True if the HRESULT means a device or certificate was revoked (signed or unsigned).""" + return code is not None and (code & 0xFFFFFFFF) in REVOCATION_CODES + + def describe(code: int) -> str | None: """Human string for a PlayReady/DRM HRESULT, or None if unknown. @@ -862,4 +876,10 @@ if __name__ == "__main__": assert describe(0x8004C065) == describe(-2147172251) assert "DEVCERT_REVOKED" in describe(-2147172251) assert describe(-1021) is None # non-DRM_E service code + assert is_revocation(0x8004C065) and is_revocation(-2147172251) # DEVCERT_REVOKED, both forms + assert not is_revocation(0x8004C006) # INVALIDLICENSE is not revocation + assert is_revocation(0x8004C07D) # activation group cert revoked + assert not is_revocation(0x8004A013) # metering sig, not revocation + assert not is_revocation(0x8004C0C4) # rev-list-not-available, not revocation + assert not is_revocation(0x8004C0C2) # content license revoked, not device print("ok")