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

@@ -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 (<StatusCode>0x8004C065</StatusCode>)
+        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",)

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")