thegame_1980/KeyDive
1
0
Fork 0
mirror of https://github.com/hyugogirubato/KeyDive.git synced 2026-07-15 18:40:02 +02:00
Code Issues Packages Projects Releases Wiki Activity

module separation + fix provisioning response decrypt

Browse Source
This commit is contained in:
hyugogirubato
2025-06-14 16:06:02 +02:00
parent d27758b7dc
commit 8ed3eccf01
5 changed files with 840 additions and 677 deletions
Download Patch File Download Diff File Expand all files Collapse all files
keydive/drm/cdm.py
+15 -677
View File
@@ -1,46 +1,19 @@
import json
import logging
import re
from enum import Enum
from json.encoder import encode_basestring_ascii
from typing import Union, Dict, List, Optional
from typing import Union, Dict
from pathlib import Path
from zlib import crc32
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.padding import OAEP, MGF1
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey
from cryptography.hazmat.primitives.ciphers import Cipher
from cryptography.hazmat.primitives.ciphers.algorithms import AES
from cryptography.hazmat.primitives.ciphers.modes import CBC
from cryptography.hazmat.primitives.hashes import SHA1
from cryptography.hazmat.primitives.padding import PKCS7
from cryptography.hazmat.primitives.serialization import pkcs7
from cryptography.x509 import Certificate
from asn1crypto.core import Integer
from keydive.drm import OEM_CRYPTO_API, KEYBOX_MAX_CLEAR_API
from keydive.drm import KEYBOX_MAX_CLEAR_API
from keydive.drm.device import Device, DeviceTypes
from keydive.drm.keybox import KeyBox
from keydive.utils import dumps, b64enc, b64dec, unidec
from keydive.drm.protocol.license_pb2 import (
LicenseRequest, SignedMessage, ProvisioningResponse, SignedProvisioningMessage, ClientIdentification,
EncryptedClientIdentification, DrmCertificate, SignedDrmCertificate
)
kWidevineSystemIdExtensionOid = '1.3.6.1.4.1.11129.4.1.1'
class OEMCrypto_ProvisioningMethod(Enum):
"""
Enum representing different OEMCrypto provisioning methods.
"""
ProvisioningError = 0 # Device cannot be provisioned.
DrmCertificate = 1 # Device has baked-in DRM certificate (level 3 only).
Keybox = 2 # Device has factory-installed unique keybox.
OEMCertificate = 3 # Device has factory-installed OEM certificate.
from keydive.drm.modules.client import get_client_info
from keydive.drm.modules.provisioning import Provisioning
from keydive.utils import unidec
from keydive.drm.protocol.license_pb2 import DrmCertificate, SignedDrmCertificate
def CryptoSession_ExtractSystemIdFromOemCert(cert: Certificate) -> int:
@@ -61,6 +34,8 @@ def CryptoSession_ExtractSystemIdFromOemCert(cert: Certificate) -> int:
Raises:
ValueError: If the certificate does not contain the Widevine System ID extension.
"""
kWidevineSystemIdExtensionOid = '1.3.6.1.4.1.11129.4.1.1'
# Iterate through all extensions in the certificate to locate the custom Widevine System ID
for ext in cert.extensions:
# Match against the known OID for the Widevine System ID
@@ -100,7 +75,7 @@ def CryptoSession_GetSecurityLevel(data: Union[Certificate, str]) -> int:
return next((l for l in range(1, 4) if pattern.format(l) in data), 3)
class Cdm:
class Cdm(Provisioning):
"""
Content Decryption Module (CDM) helper class for managing DRM provisioning
and cryptographic assets such as keys, certificates, and client identity
@@ -121,646 +96,9 @@ class Cdm:
sdk (int): The Android SDK version (used to determine feature compatibility or restrictions).
disabler (bool, optional): If True, applies logic related to L1 disabling behavior. Defaults to True.
"""
self.logger = logging.getLogger('Cdm')
self._sdk = sdk
self._disabler = disabler
self._device_aes_key: List[bytes] = []
self._keybox: Dict[bytes, KeyBox] = {} # stable_id -> keybox
self._client_id: Dict[int, ClientIdentification] = {} # public_key.n -> client_id
self._certificate: Dict[int, List[Certificate]] = {} # public_key.n -> oem_certificate
self._private_key: Dict[int, RSAPrivateKey] = {} # public_key.n -> private_key
# Cached ClientIdentification instance representing the device’s current provisioning context
# This may be reused across multiple requests to avoid re-parsing
self._provisioning: Optional[ClientIdentification] = None
@staticmethod
def __client_info(client_id: ClientIdentification) -> dict:
"""
Converts a ClientIdentification object into a dictionary containing its core information,
and optionally includes capability details if logging is set to DEBUG level.
This is useful for structured logging or inspection of client identity during
the provisioning process.
Args:
client_id (ClientIdentification): The client identification object that contains
basic client info and optional client capability fields.
Returns:
dict: A dictionary of client info fields. If logging is set to DEBUG,
a 'capabilities' key is included with detailed capability values.
"""
# Extract base client_info fields and store them in a dictionary
infos = {e.name: e.value for e in client_id.client_info}
# Check the global logger level; only include capabilities in DEBUG mode
level = logging.getLogger().getEffectiveLevel()
if level != logging.DEBUG:
return infos # Return only core info if not debugging
capabilities = {}
# Iterate over all explicitly set fields in client_capabilities
for field, value in client_id.client_capabilities.ListFields():
# Handle fields of enum type
if field.type == field.TYPE_ENUM:
if field.label == field.LABEL_REPEATED:
# Convert each enum value (integer) to its named representation
value = [field.enum_type.values_by_number[v].name for v in value]
else:
# Convert single enum value to its name
value = field.enum_type.values_by_number[value].name
# Add the field and its (possibly transformed) value to the capabilities dictionary
capabilities[field.name] = value
# Merge basic client info and capabilities into a single dictionary and return
return {**infos, 'capabilities': capabilities}
@staticmethod
def __client_generic(client_id: ClientIdentification) -> ClientIdentification:
"""
Produces a simplified version of a ClientIdentification object by removing
select non-essential fields from the `client_info` list. This is useful
when generating a generic client profile that omits sensitive or dynamic data,
such as application-specific metadata.
Args:
client_id (ClientIdentification): The original, full client identification
object containing metadata and capabilities.
Returns:
ClientIdentification: A new instance with reduced client_info fields, preserving
core identity and capabilities but omitting values like app name, origin, and cert hash.
Notes:
The fields being removed are typically:
- 'application_name': Identifies the calling app; often dynamic or sensitive.
- 'origin': Source domain or app origin; may vary per request or client.
- 'package_certificate_hash_bytes': May be tied to APK signing identity.
This function ensures that the returned object retains its structure and utility
while discarding non-critical identifiers that might affect reproducibility or
comparability across sessions.
"""
# Define the fields that are considered dynamic or non-essential for generic use
excluded_fields = {'application_name', 'origin', 'package_certificate_hash_bytes'}
# Create a filtered list excluding the above fields from client_info
filtered_client_info = [
info for info in client_id.client_info
if info.name not in excluded_fields
]
# Return a new ClientIdentification object with the cleaned client_info list
return ClientIdentification(
type=client_id.type,
token=client_id.token,
client_info=filtered_client_info,
# provider_client_token=client_id.provider_client_token,
# license_counter=client_id.license_counter,
client_capabilities=client_id.client_capabilities,
# vmp_data=client_id.vmp_data,
# device_credentials=client_id.device_credentials
)
def set_challenge(self, data: Union[bytes, List[Path]]) -> None:
"""
Parses a Widevine license challenge to extract and register client identification data.
This method accepts either raw protobuf bytes representing a SignedMessage or a list
of file paths containing such data. It supports recursive processing of multiple
challenge files and handles both encrypted and unencrypted client IDs.
Args:
data (Union[bytes, List[Path]]):
- Raw protobuf data as bytes representing a SignedMessage, or
- A list of Path objects pointing to files containing SignedMessage data.
"""
# If input is a list of file paths, process each file individually
if isinstance(data, list):
for path in data:
try:
# Read file content as bytes and recursively process it
self.set_challenge(data=path.read_bytes())
except Exception as e:
self.logger.error('Could not load challenge from file %s: %s', path, e)
# Proceed only if data is raw bytes after potential recursive calls
if not isinstance(data, bytes):
return
# https://integration.widevine.com/diagnostics
try:
# Parse the SignedMessage protobuf from raw data
signed_message = SignedMessage()
signed_message.ParseFromString(data)
# Extract and parse the embedded LicenseRequest message
license_request = LicenseRequest()
license_request.ParseFromString(signed_message.msg)
# Attempt to extract the encrypted client ID (if present)
encrypted_client_id: EncryptedClientIdentification = license_request.encrypted_client_id
if encrypted_client_id.SerializeToString():
# Encrypted client ID found - log its details for debugging
"""
self.logger.info(
'Received encrypted client ID: \n\n%s\n',
dumps({
f.name: v for f, v in encrypted_client_id.ListFields()
}, beauty=True)
)
"""
self.logger.warning('The client ID in this challenge is encrypted and cannot be processed directly.')
else:
# Unencrypted client ID is available - register it immediately
self.set_client_id(data=license_request.client_id)
except Exception as e:
# Log any parsing errors at debug level to avoid cluttering standard logs
self.logger.debug('Unable to register challenge data: %s', e)
def set_client_id(self, data: Union[ClientIdentification, bytes]) -> None:
"""
Loads and registers a client ID from either a parsed ClientIdentification object
or raw protobuf bytes. Handles different client ID token types, such as DRM device
certificates, keyboxes, and OEM device certificates.
Args:
data (Union[ClientIdentification, bytes]):
Either a ClientIdentification protobuf object or raw protobuf bytes representing it.
Raises:
ValueError: If the client type is unsupported or certificate chain length is invalid.
"""
try:
# If already a parsed ClientIdentification object, use it directly
if isinstance(data, ClientIdentification):
client_id = data
else:
# Otherwise, parse the protobuf bytes into a ClientIdentification object
client_id = ClientIdentification()
client_id.ParseFromString(data)
# Cache this client ID for provisioning context usage elsewhere
self._provisioning = client_id
# Process based on the token type of the client ID
if client_id.type == ClientIdentification.TokenType.DRM_DEVICE_CERTIFICATE:
# Clean unnecessary fields to simplify the client representation
client_id = self.__client_generic(client_id)
# Initialize protobuf objects to parse DRM certificates
signed_drm_certificate = SignedDrmCertificate()
drm_certificate = DrmCertificate()
# Parse signed DRM certificate from client token
signed_drm_certificate.ParseFromString(client_id.token)
drm_certificate.ParseFromString(signed_drm_certificate.drm_certificate)
# Load the public key from the DRM certificate (DER format)
public_key = serialization.load_der_public_key(
data=drm_certificate.public_key,
backend=default_backend()
)
# Extract the RSA modulus as a unique client identifier
modulus = public_key.public_numbers().n
# Register new or changed client IDs keyed by modulus
if modulus not in self._client_id or self._client_id[modulus] != client_id:
self.logger.info(
'Received client ID: \n\n%s\n',
dumps(self.__client_info(client_id), beauty=True)
)
self._client_id[modulus] = client_id
elif client_id.type == ClientIdentification.TokenType.KEYBOX:
# For keybox-based provisioning, register the token as the device ID
self.set_device_id(client_id.token)
elif client_id.type == ClientIdentification.TokenType.OEM_DEVICE_CERTIFICATE:
# Load the certificate chain from DER-encoded PKCS7 token
certs = pkcs7.load_der_pkcs7_certificates(data=client_id.token)
# Validate the certificate chain length (expect exactly two certs)
if len(certs) != 2:
raise ValueError(f'Invalid certificate chain length: expected 2, got {len(certs)}')
# Extract RSA modulus from the intermediate certificate's public key
modulus = certs[1].public_key().public_numbers().n
# Register new or updated OEM certificates keyed by modulus
if modulus not in self._certificate:
# openssl asn1parse -inform der -in oem_certificate.der
# openssl pkcs7 -inform der -in oem_certificate.der -print_certs -out output.pem
self.logger.info(
'Received OEM certificate: \n\n%s\n',
dumps([{
'subject': c.subject.rfc4514_string(),
'issuer': c.subject.rfc4514_string(),
'serial_number': c.serial_number,
'valid_from': c.not_valid_before_utc,
'valid_until': c.not_valid_after_utc
} for c in certs], beauty=True)
)
self._certificate[modulus] = certs
else:
raise ValueError(f'Unsupported client type: {client_id.type}')
except Exception as e:
# Log exceptions at debug level to keep regular logs clean
self.logger.debug('Unable to register client ID: %s', e)
def set_private_key(self, data: Union[bytes, List[Path]], name: Optional[str] = None) -> None:
"""
Registers an OEM private RSA key used for Widevine decryption.
Supports loading from raw PEM/DER bytes or multiple files containing keys.
Args:
data (Union[bytes, List[Path]]):
Raw private key bytes in PEM or DER format, or a list of file paths to key files.
name (Optional[str]):
Optional identifier for the function or implementation related to this key.
"""
# If input is a list of file paths, process each file recursively
if isinstance(data, list):
for path in data:
try:
# Read the file's bytes and call this method again with raw bytes
self.set_private_key(data=path.read_bytes(), name=None)
except Exception as e:
self.logger.error('Could not load RSA key from file %s: %s', path, e)
# If the input is not bytes at this point, do nothing (invalid input)
if not isinstance(data, bytes):
return
try:
# Attempt to load the private key assuming PEM (Base64 encoded) format first
try:
private_key = serialization.load_pem_private_key(
data=data,
password=None,
backend=default_backend()
)
except ValueError:
# If PEM loading fails, try DER (binary) format instead
private_key = serialization.load_der_private_key(
data=data,
password=None,
backend=default_backend()
)
# Extract the RSA modulus (n) from the public key for unique identification
modulus = private_key.public_key().public_numbers().n
# Only register the key if this modulus is not already present
if modulus not in self._private_key:
self.logger.info(
'Received RSA private key: \n\n%s\n',
private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption()
).decode('utf-8'))
self._private_key[modulus] = private_key
# If a name is provided but unrecognized, log a warning for developer attention
if name and name not in OEM_CRYPTO_API:
self.logger.warning(
'Unrecognized function name "%s". Please notify the developers to improve this tool.',
name
)
except Exception as e:
# Log loading failures at debug level to avoid cluttering normal logs
self.logger.debug('Unable to register RSA private key: %s', e)
def set_keybox(self, data: bytes) -> None:
"""
Parses and registers a KeyBox structure used for device provisioning.
The KeyBox contains cryptographic and device-specific data necessary for DRM.
Args:
data (bytes): Raw binary data containing the serialized KeyBox protobuf.
"""
try:
# Initialize a new KeyBox instance and parse the protobuf data from bytes
keybox = KeyBox()
keybox.ParseFromString(data) # Parse the raw data into the KeyBox object
# Check if the stable_id is either not present in the dictionary,
# or if the existing KeyBox with this stable_id does not have a device_id set.
if keybox.stable_id not in self._keybox or not self._keybox[keybox.stable_id].device_id:
# Extract keybox metadata info for logging
infos = keybox.keybox_info
# Determine label based on presence of system_id in the keybox info
label = 'keybox' if infos.get('system_id') else 'encrypted keybox'
self.logger.info(
'Received %s: \n\n%s\n',
label, dumps(infos, beauty=True)
)
# Store or update the internal keybox dictionary using stable_id as the key
self._keybox[keybox.stable_id] = keybox
except Exception as e:
# Log failure details at debug level to avoid noise in normal logs
self.logger.debug('Unable to register KeyBox data: %s', e)
def set_stable_id(self, data: bytes) -> None:
"""
Registers or updates the stable ID for a KeyBox instance.
The stable ID uniquely identifies a device and acts as the key
in the internal KeyBox dictionary.
Args:
data (bytes): Stable device identifier as raw bytes,
used as the unique key to identify the keybox.
Notes:
- If a KeyBox already exists with an empty stable_id, it will be reused;
otherwise, a new KeyBox instance is created.
"""
try:
# Only proceed if this stable ID is not already present in the keybox dictionary
if data not in self._keybox:
# Retrieve existing KeyBox with empty stable_id or create a new instance
keybox = self._keybox.get(b'', KeyBox())
# Assign the provided stable ID to this KeyBox instance
keybox.stable_id = data
# Log the stable ID in a human-readable ASCII-safe format
self.logger.info(
'Received stable ID: \n\n%s\n',
encode_basestring_ascii(keybox.stable_id.decode('utf-8')))
# Store or update the KeyBox in the dictionary with stable_id as the key
self._keybox[keybox.stable_id] = keybox
except Exception as e:
# Log exception details at debug level without interrupting flow
self.logger.debug('Unable to register KeyBox stable ID: %s', e)
def set_device_id(self, data: bytes) -> None:
"""
Registers or updates the device ID within a KeyBox instance.
The device ID is used to uniquely identify a device for cryptographic
provisioning and licensing purposes.
Args:
data (bytes): Raw device identifier bytes, used to associate a device ID
with a KeyBox object.
Notes:
- If the device ID is already associated with a KeyBox, this method does nothing.
- If a KeyBox without a device ID exists, it will be reused; otherwise,
a new KeyBox instance will be created.
"""
try:
# Check if this device ID is already registered in any existing KeyBox
if not any(k for k in self._keybox.values() if k.device_id == data):
# Find a KeyBox without a device ID to reuse, or create a new KeyBox
keybox = next((k for k in self._keybox.values() if not k.device_id), KeyBox())
# Assign the provided device ID to the KeyBox
keybox.device_id = data
# Retrieve device-related metadata for logging purposes
infos = keybox.device_info
# Choose label based on whether a system_id is present in metadata
label = 'device ID' if infos.get('system_id') else 'encrypted device ID'
self.logger.info(
'Received %s: \n\n%s\n',
label, dumps(infos, beauty=True) if infos else b64enc(keybox.device_id))
# Store or update the KeyBox in the dictionary using its stable_id as the key
self._keybox[keybox.stable_id] = keybox
except Exception as e:
# Log exceptions with context at debug level without raising
self.logger.debug('Unable to register KeyBox device ID: %s', e)
def set_device_aes_key(self, data: Union[bytes, List[str]]) -> None:
"""
Imports and stores AES device keys used for decrypting protected content.
This method accepts either a single AES key as raw bytes or a list of strings
representing AES keys in various formats, including:
- Hexadecimal strings
- Base64-encoded strings (standard or URL-safe)
- File paths pointing to binary key files
Args:
data (bytes or List[str]): The AES key(s) to import. Either a single
16-byte AES key (bytes) or a list of strings
representing keys in different encoded forms or file paths.
Notes:
- Only keys exactly 16 bytes in length are accepted as valid AES keys.
- If a list of strings is given, the method tries each parser (hex, base64, file)
until one succeeds for each entry.
"""
if isinstance(data, list):
# Iterate over all string inputs, attempt parsing with multiple strategies
for value in data:
for parser in (bytes.fromhex, b64dec, lambda v: Path(v).read_bytes()):
try:
# Recursively call with successfully decoded raw bytes
self.set_device_aes_key(parser(value))
break # Exit parser loop once successful
except Exception:
# Ignore failures and try next parser
pass
else:
# If no parser succeeded, log the failure with the raw input value
self.logger.error('Could not import AES key from input: %s', value)
# From here on, ensure input is raw bytes representing the AES key
if not isinstance(data, bytes):
return
try:
# Confirm the AES key length matches 16 bytes (AES-128 standard)
assert len(data) == 16, f'Invalid AES key length: expected 16 bytes, got {len(data)}'
# Store the valid AES key into internal cache for later use
self._device_aes_key.append(data)
# Log success message with the hex-encoded key for clarity
self.logger.info('Received device AES key: %s', data.hex())
except Exception as e:
# Log any exceptions encountered during key validation or storage
self.logger.debug('Unable to register device AES key: %s', e)
def set_provisioning_method(self, data: bytes) -> None:
"""
Determines and logs the provisioning method used by the Content Decryption Module (CDM).
This method decodes the given byte data to an integer, which maps to an
OEMCrypto_ProvisioningMethod enumeration value. It helps identify the provisioning
mechanism in use and logs relevant diagnostic information, especially when L1
provisioning appears to be disabled improperly.
Args:
data (bytes): UTF-8 encoded string representing an integer corresponding to
an OEMCrypto_ProvisioningMethod enum value.
Exception:
Catches and logs all exceptions encountered during decoding or enum conversion.
"""
try:
# Decode bytes to UTF-8 string, convert to int, and map to provisioning method enum
method = OEMCrypto_ProvisioningMethod(int(data.decode('utf-8')))
if method == OEMCrypto_ProvisioningMethod.Keybox and self._disabler:
# Warn user if L1 provisioning is enabled but disabling procedure incomplete
self.logger.warning(
'L1 provisioning deactivation appears incomplete. '
'Consider using a web dump or forcibly terminating the process to ensure proper disabling.'
)
else:
# Log the provisioning method name for informational purposes
self.logger.debug('Receive provisioning method: %s', method.name)
except Exception as e:
# Log any errors during decoding or mapping to enum with debug severity
self.logger.debug('Unable to parse provisioning method: %s', e)
def set_provisioning_response(self, data: bytes) -> None:
"""
Parses and applies a provisioning response from the Google Widevine provisioning service.
Supports both Keybox-based provisioning and Provisioning 3.0 OTA PKI formats.
This method extracts and decrypts the device RSA private key using known AES session keys,
which may be retrieved from keyboxes or OEM provisioning keys. It also supports
setting up the device certificate and updating client identification.
Args:
data (bytes): JSON-encoded provisioning response as received from the provisioning server.
Exception:
Catches and logs all exceptions raised during parsing, decryption, or client ID setup.
"""
try:
# Extract and decode the base64-encoded signed provisioning message from the JSON response
b64_signed_data = json.loads(data.split(b'\x00')[0])['signedResponse']
signed_data = b64dec(b64_signed_data, safe=True)
# Parse the SignedProvisioningMessage protobuf
signed_response = SignedProvisioningMessage()
signed_response.ParseFromString(signed_data)
# Extract the ProvisioningResponse payload embedded in the signed message
provisioning_response = ProvisioningResponse()
provisioning_response.ParseFromString(signed_response.message)
# Gather all known AES session keys from stored keyboxes (OEM Keybox keys)
session_enc_keys = [k.device_aes_key for k in self._keybox.values() if k.device_aes_key]
# Optionally, add additional AES keys extracted via reverse engineering/TEE exploit (if any)
session_enc_keys += self._device_aes_key
if provisioning_response.wrapping_key:
# OTA PKI-Based provisioning (Provisioning 3.0)
self.logger.info(
'Received OTA provisioning response: \n\n%s\n',
dumps({
'signature': {'type': 'RSASSA-PSS', 'data': signed_response.signature},
'nonce': provisioning_response.nonce,
'wrapping_key': provisioning_response.wrapping_key
}, beauty=True)
)
# Attempt to decrypt the AES wrapping key using all stored OEM private RSA keys
for oem_cert_priv_key in self._private_key.values():
try:
session_enc_key = oem_cert_priv_key.decrypt(
ciphertext=provisioning_response.wrapping_key,
padding=OAEP(
mgf=MGF1(algorithm=SHA1()),
algorithm=SHA1(),
label=None
)
)
session_enc_keys.append(session_enc_key)
except Exception as e:
self.logger.debug('Unable to decrypt OTA session key: %s', e)
else:
# Keybox-based provisioning (Provisioning 2.0)
self.logger.info(
'Receive Keybox provisioning response: \n\n%s\n',
dumps({
'signature': {'type': 'HMAC-SHA256', 'data': signed_response.signature},
'nonce': provisioning_response.nonce
}, beauty=True)
)
# Attempt to decrypt the device RSA private key using all gathered AES session keys
for session_enc_key in session_enc_keys:
try:
# TODO: complete the keybox with the AES key
# Initialize AES-CBC cipher with the session key and provided IV
cipher = Cipher(
algorithm=AES(session_enc_key),
mode=CBC(provisioning_response.device_rsa_key_iv),
backend=default_backend()
)
decryptor = cipher.decryptor()
dec_padded_data = decryptor.update(provisioning_response.device_rsa_key) + decryptor.finalize()
# Remove PKCS7 padding to obtain the original private key bytes
unpadder = PKCS7(AES.block_size).unpadder()
dec_data = unpadder.update(dec_padded_data) + unpadder.finalize()
# Register the decrypted RSA private key internally
self.set_private_key(dec_data, None)
# Successfully decrypted and stored the RSA key, no need to try further keys
# break
except Exception as e:
# If no decryption succeeded, OEM private key remains unset
# At this point, OTA provisioning using Provisioning 3.0 (PKI-based) might apply
# But this flow assumes Provisioning 2.0 (Keybox-based), so no further action here
self.logger.debug('Failed to decrypt RSA private key using AES key: %s', session_enc_key.hex())
# If a provisioning context exists, update the ClientIdentification with new capabilities and certificates
if self._provisioning:
"""
client_capabilities {
client_token: true
session_token: true
max_hdcp_version: HDCP_V2_2
oem_crypto_api_version: 15
anti_rollback_usage_table: false
srm_version: 0
can_update_srm: false
supported_certificate_key_type: RSA_2048
analog_output_capabilities: ANALOG_OUTPUT_NONE
can_disable_analog_output: false
}
"""
# Update client capabilities with typical fields
client_capabilities = self._provisioning.client_capabilities
client_capabilities.session_token = True
client_capabilities.max_hdcp_version = ClientIdentification.ClientCapabilities.HdcpVersion.HDCP_NONE
client_capabilities.anti_rollback_usage_table = False
client_capabilities.can_update_srm = False
# Construct a new ClientIdentification token with the provisioned device certificate
client_id = ClientIdentification(
type=ClientIdentification.TokenType.DRM_DEVICE_CERTIFICATE,
token=provisioning_response.device_certificate,
client_info=self._provisioning.client_info,
provider_client_token=self._provisioning.provider_client_token,
license_counter=self._provisioning.license_counter,
client_capabilities=client_capabilities,
vmp_data=self._provisioning.vmp_data,
device_credentials=self._provisioning.device_credentials
)
# Register the updated client identification token
self.set_client_id(client_id)
except Exception as e:
# Log any unexpected error encountered during the provisioning response handling
self.logger.debug('Unable to process provisioning response: %s', e)
super().__init__()
self.sdk = sdk
self.disabler = disabler
def __resolve(self) -> Dict[int, dict]:
"""
@@ -790,7 +128,7 @@ class Cdm:
# Process all client IDs and their associated private keys
for key, client_id in self._client_id.items():
client_info = self.__client_info(client_id)
client_info = get_client_info(client_id)
# Parse DRM certificate from the client token protobuf data
signed_drm_certificate = SignedDrmCertificate()
@@ -866,7 +204,7 @@ class Cdm:
if system_id:
# Determine security level based on SDK version and keybox state
level = 1 if self._sdk > KEYBOX_MAX_CLEAR_API else 3
level = 1 if self.sdk > KEYBOX_MAX_CLEAR_API else 3
# Initialize or retrieve record for this system ID
item = items.get(system_id, {
@@ -958,7 +296,7 @@ class Cdm:
device_serialized = device.dumps()
# Create a descriptive and unique filename based on device info and a CRC32 hash
client_info = self.__client_info(client_id)
client_info = get_client_info(client_id)
name = f"{client_info['company_name']} {client_info['model_name']}"
if client_info.get('widevine_cdm_version'):
name += f" {client_info['widevine_cdm_version']}"
keydive/drm/modules/__init__.py
+28
View File
@@ -0,0 +1,28 @@
import logging
from typing import List, Dict, Optional
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey
from cryptography.x509 import Certificate
from keydive.drm.keybox import KeyBox
from keydive.drm.protocol.license_pb2 import ClientIdentification
class BaseCdm:
def __init__(self):
self.logger = logging.getLogger(self.__class__.__name__)
self.sdk = 36
self.disabler = True
self._device_aes_key: List[bytes] = []
self._keybox: Dict[bytes, KeyBox] = {} # stable_id -> keybox
self._client_id: Dict[int, ClientIdentification] = {} # public_key.n -> client_id
self._certificate: Dict[int, List[Certificate]] = {} # public_key.n -> oem_certificate
self._private_key: Dict[int, RSAPrivateKey] = {} # public_key.n -> private_key
# Cached ClientIdentification instance representing the device’s current provisioning context
# This may be reused across multiple requests to avoid re-parsing
self._provisioning: Optional[ClientIdentification] = None
self._context: Optional[bytes] = None
keydive/drm/modules/client.py
+316
View File
@@ -0,0 +1,316 @@
import logging
from pathlib import Path
from typing import Union, List, Optional
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.serialization import pkcs7
from keydive.drm.modules.oemcrypto import OEMCrypto
from keydive.drm import OEM_CRYPTO_API
from keydive.utils import dumps
from keydive.drm.protocol.license_pb2 import (
ClientIdentification, SignedMessage, LicenseRequest, EncryptedClientIdentification, SignedDrmCertificate,
DrmCertificate)
def get_client_info(client_id: ClientIdentification) -> dict:
"""
Converts a ClientIdentification object into a dictionary containing its core information,
and optionally includes capability details if logging is set to DEBUG level.
This is useful for structured logging or inspection of client identity during
the provisioning process.
Args:
client_id (ClientIdentification): The client identification object that contains
basic client info and optional client capability fields.
Returns:
dict: A dictionary of client info fields. If logging is set to DEBUG,
a 'capabilities' key is included with detailed capability values.
"""
# Extract base client_info fields and store them in a dictionary
infos = {e.name: e.value for e in client_id.client_info}
# Check the global logger level; only include capabilities in DEBUG mode
level = logging.getLogger().getEffectiveLevel()
if level != logging.DEBUG:
return infos # Return only core info if not debugging
capabilities = {}
# Iterate over all explicitly set fields in client_capabilities
for field, value in client_id.client_capabilities.ListFields():
# Handle fields of enum type
if field.type == field.TYPE_ENUM:
if field.label == field.LABEL_REPEATED:
# Convert each enum value (integer) to its named representation
value = [field.enum_type.values_by_number[v].name for v in value]
else:
# Convert single enum value to its name
value = field.enum_type.values_by_number[value].name
# Add the field and its (possibly transformed) value to the capabilities dictionary
capabilities[field.name] = value
# Merge basic client info and capabilities into a single dictionary and return
return {**infos, 'capabilities': capabilities}
def get_client_generic(client_id: ClientIdentification) -> ClientIdentification:
"""
Produces a simplified version of a ClientIdentification object by removing
select non-essential fields from the `client_info` list. This is useful
when generating a generic client profile that omits sensitive or dynamic data,
such as application-specific metadata.
Args:
client_id (ClientIdentification): The original, full client identification
object containing metadata and capabilities.
Returns:
ClientIdentification: A new instance with reduced client_info fields, preserving
core identity and capabilities but omitting values like app name, origin, and cert hash.
Notes:
The fields being removed are typically:
- 'application_name': Identifies the calling app; often dynamic or sensitive.
- 'origin': Source domain or app origin; may vary per request or client.
- 'package_certificate_hash_bytes': May be tied to APK signing identity.
This function ensures that the returned object retains its structure and utility
while discarding non-critical identifiers that might affect reproducibility or
comparability across sessions.
"""
# Define the fields that are considered dynamic or non-essential for generic use
excluded_fields = {'application_name', 'origin', 'package_certificate_hash_bytes'}
# Create a filtered list excluding the above fields from client_info
filtered_client_info = [
info for info in client_id.client_info
if info.name not in excluded_fields
]
# Return a new ClientIdentification object with the cleaned client_info list
return ClientIdentification(
type=client_id.type,
token=client_id.token,
client_info=filtered_client_info,
# provider_client_token=client_id.provider_client_token,
# license_counter=client_id.license_counter,
client_capabilities=client_id.client_capabilities,
# vmp_data=client_id.vmp_data,
# device_credentials=client_id.device_credentials
)
class Client(OEMCrypto):
def set_challenge(self, data: Union[bytes, List[Path]]) -> None:
"""
Parses a Widevine license challenge to extract and register client identification data.
This method accepts either raw protobuf bytes representing a SignedMessage or a list
of file paths containing such data. It supports recursive processing of multiple
challenge files and handles both encrypted and unencrypted client IDs.
Args:
data (Union[bytes, List[Path]]):
- Raw protobuf data as bytes representing a SignedMessage, or
- A list of Path objects pointing to files containing SignedMessage data.
"""
# If input is a list of file paths, process each file individually
if isinstance(data, list):
for path in data:
try:
# Read file content as bytes and recursively process it
self.set_challenge(data=path.read_bytes())
except Exception as e:
self.logger.error('Could not load challenge from file %s: %s', path, e)
# Proceed only if data is raw bytes after potential recursive calls
if not isinstance(data, bytes):
return
# https://integration.widevine.com/diagnostics
try:
# Parse the SignedMessage protobuf from raw data
signed_message = SignedMessage()
signed_message.ParseFromString(data)
# Extract and parse the embedded LicenseRequest message
license_request = LicenseRequest()
license_request.ParseFromString(signed_message.msg)
# Attempt to extract the encrypted client ID (if present)
encrypted_client_id: EncryptedClientIdentification = license_request.encrypted_client_id
if encrypted_client_id.SerializeToString():
# Encrypted client ID found - log its details for debugging
"""
self.logger.info(
'Received encrypted client ID: \n\n%s\n',
dumps({
f.name: v for f, v in encrypted_client_id.ListFields()
}, beauty=True)
)
"""
self.logger.warning('The client ID in this challenge is encrypted and cannot be processed directly.')
else:
# Unencrypted client ID is available - register it immediately
self.set_client_id(data=license_request.client_id)
except Exception as e:
# Log any parsing errors at debug level to avoid cluttering standard logs
self.logger.debug('Unable to register challenge data: %s', e)
def set_client_id(self, data: Union[ClientIdentification, bytes]) -> None:
"""
Loads and registers a client ID from either a parsed ClientIdentification object
or raw protobuf bytes. Handles different client ID token types, such as DRM device
certificates, keyboxes, and OEM device certificates.
Args:
data (Union[ClientIdentification, bytes]):
Either a ClientIdentification protobuf object or raw protobuf bytes representing it.
Raises:
ValueError: If the client type is unsupported or certificate chain length is invalid.
"""
try:
# If already a parsed ClientIdentification object, use it directly
if isinstance(data, ClientIdentification):
client_id = data
else:
# Otherwise, parse the protobuf bytes into a ClientIdentification object
client_id = ClientIdentification()
client_id.ParseFromString(data)
# Cache this client ID for provisioning context usage elsewhere
self._provisioning = client_id
# Process based on the token type of the client ID
if client_id.type == ClientIdentification.TokenType.DRM_DEVICE_CERTIFICATE:
# Clean unnecessary fields to simplify the client representation
client_id = get_client_generic(client_id)
# Initialize protobuf objects to parse DRM certificates
signed_drm_certificate = SignedDrmCertificate()
drm_certificate = DrmCertificate()
# Parse signed DRM certificate from client token
signed_drm_certificate.ParseFromString(client_id.token)
drm_certificate.ParseFromString(signed_drm_certificate.drm_certificate)
# Load the public key from the DRM certificate (DER format)
public_key = serialization.load_der_public_key(
data=drm_certificate.public_key,
backend=default_backend()
)
# Extract the RSA modulus as a unique client identifier
modulus = public_key.public_numbers().n
# Register new or changed client IDs keyed by modulus
if modulus not in self._client_id or self._client_id[modulus] != client_id:
self.logger.info(
'Received client ID: \n\n%s\n',
dumps(get_client_info(client_id), beauty=True)
)
self._client_id[modulus] = client_id
elif client_id.type == ClientIdentification.TokenType.KEYBOX:
# For keybox-based provisioning, register the token as the device ID
self.set_device_id(client_id.token)
elif client_id.type == ClientIdentification.TokenType.OEM_DEVICE_CERTIFICATE:
# Load the certificate chain from DER-encoded PKCS7 token
certs = pkcs7.load_der_pkcs7_certificates(data=client_id.token)
# Validate the certificate chain length (expect exactly two certs)
if len(certs) != 2:
raise ValueError(f'Invalid certificate chain length: expected 2, got {len(certs)}')
# Extract RSA modulus from the intermediate certificate's public key
modulus = certs[1].public_key().public_numbers().n
# Register new or updated OEM certificates keyed by modulus
if modulus not in self._certificate:
# openssl asn1parse -inform der -in oem_certificate.der
# openssl pkcs7 -inform der -in oem_certificate.der -print_certs -out output.pem
self.logger.info(
'Received OEM certificate: \n\n%s\n',
dumps([{
'subject': c.subject.rfc4514_string(),
'issuer': c.subject.rfc4514_string(),
'serial_number': c.serial_number,
'valid_from': c.not_valid_before_utc,
'valid_until': c.not_valid_after_utc
} for c in certs], beauty=True)
)
self._certificate[modulus] = certs
else:
raise ValueError(f'Unsupported client type: {client_id.type}')
except Exception as e:
# Log exceptions at debug level to keep regular logs clean
self.logger.debug('Unable to register client ID: %s', e)
def set_private_key(self, data: Union[bytes, List[Path]], name: Optional[str] = None) -> None:
"""
Registers an OEM private RSA key used for Widevine decryption.
Supports loading from raw PEM/DER bytes or multiple files containing keys.
Args:
data (Union[bytes, List[Path]]):
Raw private key bytes in PEM or DER format, or a list of file paths to key files.
name (Optional[str]):
Optional identifier for the function or implementation related to this key.
"""
# If input is a list of file paths, process each file recursively
if isinstance(data, list):
for path in data:
try:
# Read the file's bytes and call this method again with raw bytes
self.set_private_key(data=path.read_bytes(), name=None)
except Exception as e:
self.logger.error('Could not load RSA key from file %s: %s', path, e)
# If the input is not bytes at this point, do nothing (invalid input)
if not isinstance(data, bytes):
return
try:
# Attempt to load the private key assuming PEM (Base64 encoded) format first
try:
private_key = serialization.load_pem_private_key(
data=data,
password=None,
backend=default_backend()
)
except ValueError:
# If PEM loading fails, try DER (binary) format instead
private_key = serialization.load_der_private_key(
data=data,
password=None,
backend=default_backend()
)
# Extract the RSA modulus (n) from the public key for unique identification
modulus = private_key.public_key().public_numbers().n
# Only register the key if this modulus is not already present
if modulus not in self._private_key:
self.logger.info(
'Received RSA private key: \n\n%s\n',
private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption()
).decode('utf-8'))
self._private_key[modulus] = private_key
# If a name is provided but unrecognized, log a warning for developer attention
if name and name not in OEM_CRYPTO_API:
self.logger.warning(
'Unrecognized function name "%s". Please notify the developers to improve this tool.',
name
)
except Exception as e:
# Log loading failures at debug level to avoid cluttering normal logs
self.logger.debug('Unable to register RSA private key: %s', e)
keydive/drm/modules/oemcrypto.py
+207
View File
@@ -0,0 +1,207 @@
from json.encoder import encode_basestring_ascii
from pathlib import Path
from typing import Union, List
from keydive.drm.keybox import KeyBox
from keydive.drm.modules import BaseCdm
from keydive.utils import dumps, b64enc, b64dec
class OEMCrypto(BaseCdm):
def set_keybox(self, data: bytes) -> None:
"""
Parses and registers a KeyBox structure used for device provisioning.
The KeyBox contains cryptographic and device-specific data necessary for DRM.
Args:
data (bytes): Raw binary data containing the serialized KeyBox protobuf.
"""
try:
# Initialize a new KeyBox instance and parse the protobuf data from bytes
keybox = KeyBox()
keybox.ParseFromString(data) # Parse the raw data into the KeyBox object
# Check if the stable_id is either not present in the dictionary,
# or if the existing KeyBox with this stable_id does not have a device_id set.
if keybox.stable_id not in self._keybox or not self._keybox[keybox.stable_id].device_id:
# Extract keybox metadata info for logging
infos = keybox.keybox_info
# Determine label based on presence of system_id in the keybox info
label = 'keybox' if infos.get('system_id') else 'encrypted keybox'
self.logger.info(
'Received %s: \n\n%s\n',
label, dumps(infos, beauty=True)
)
# Store or update the internal keybox dictionary using stable_id as the key
self._keybox[keybox.stable_id] = keybox
except Exception as e:
# Log failure details at debug level to avoid noise in normal logs
self.logger.debug('Unable to register KeyBox data: %s', e)
def set_stable_id(self, data: bytes) -> None:
"""
Registers or updates the stable ID for a KeyBox instance.
The stable ID uniquely identifies a device and acts as the key
in the internal KeyBox dictionary.
Args:
data (bytes): Stable device identifier as raw bytes,
used as the unique key to identify the keybox.
Notes:
- If a KeyBox already exists with an empty stable_id, it will be reused;
otherwise, a new KeyBox instance is created.
"""
try:
# Only proceed if this stable ID is not already present in the keybox dictionary
if data not in self._keybox:
# Retrieve existing KeyBox with empty stable_id or create a new instance
keybox = self._keybox.get(b'', KeyBox())
# Assign the provided stable ID to this KeyBox instance
keybox.stable_id = data
# Log the stable ID in a human-readable ASCII-safe format
self.logger.info(
'Received stable ID: \n\n%s\n',
encode_basestring_ascii(keybox.stable_id.decode('utf-8')))
# Store or update the KeyBox in the dictionary with stable_id as the key
self._keybox[keybox.stable_id] = keybox
except Exception as e:
# Log exception details at debug level without interrupting flow
self.logger.debug('Unable to register KeyBox stable ID: %s', e)
def set_device_id(self, data: bytes) -> None:
"""
Registers or updates the device ID within a KeyBox instance.
The device ID is used to uniquely identify a device for cryptographic
provisioning and licensing purposes.
Args:
data (bytes): Raw device identifier bytes, used to associate a device ID
with a KeyBox object.
Notes:
- If the device ID is already associated with a KeyBox, this method does nothing.
- If a KeyBox without a device ID exists, it will be reused; otherwise,
a new KeyBox instance will be created.
"""
try:
# Check if this device ID is already registered in any existing KeyBox
if not any(k for k in self._keybox.values() if k.device_id == data):
# Find a KeyBox without a device ID to reuse, or create a new KeyBox
keybox = next((k for k in self._keybox.values() if not k.device_id), KeyBox())
# Assign the provided device ID to the KeyBox
keybox.device_id = data
# Retrieve device-related metadata for logging purposes
infos = keybox.device_info
# Choose label based on whether a system_id is present in metadata
label = 'device ID' if infos.get('system_id') else 'encrypted device ID'
self.logger.info(
'Received %s: \n\n%s\n',
label, dumps(infos, beauty=True) if infos else b64enc(keybox.device_id))
# Store or update the KeyBox in the dictionary using its stable_id as the key
self._keybox[keybox.stable_id] = keybox
except Exception as e:
# Log exceptions with context at debug level without raising
self.logger.debug('Unable to register KeyBox device ID: %s', e)
def set_device_aes_key(self, data: Union[bytes, List[str]]) -> None:
"""
Imports and stores AES device keys used for decrypting protected content.
This method accepts either a single AES key as raw bytes or a list of strings
representing AES keys in various formats, including:
- Hexadecimal strings
- Base64-encoded strings (standard or URL-safe)
- File paths pointing to binary key files
Args:
data (bytes or List[str]): The AES key(s) to import. Either a single
16-byte AES key (bytes) or a list of strings
representing keys in different encoded forms or file paths.
Notes:
- Only keys exactly 16 bytes in length are accepted as valid AES keys.
- If a list of strings is given, the method tries each parser (hex, base64, file)
until one succeeds for each entry.
"""
if isinstance(data, list):
# Iterate over all string inputs, attempt parsing with multiple strategies
for value in data:
for parser in (bytes.fromhex, b64dec, lambda v: Path(v).read_bytes()):
try:
# Recursively call with successfully decoded raw bytes
self.set_device_aes_key(parser(value))
break # Exit parser loop once successful
except Exception:
# Ignore failures and try next parser
pass
else:
# If no parser succeeded, log the failure with the raw input value
self.logger.error('Could not import AES key from input: %s', value)
# From here on, ensure input is raw bytes representing the AES key
if not isinstance(data, bytes):
return
try:
# Confirm the AES key length matches 16 bytes (AES-128 standard)
assert len(data) == 16, f'Invalid AES key length: expected 16 bytes, got {len(data)}'
# Store the valid AES key into internal cache for later use
self._device_aes_key.append(data)
# Log success message with the hex-encoded key for clarity
self.logger.info('Received device AES key: %s', data.hex())
except Exception as e:
# Log any exceptions encountered during key validation or storage
self.logger.debug('Unable to register device AES key: %s', e)
def set_encryption_context(self, data: bytes) -> None:
"""
Registers the encryption context used in content key derivation.
The encryption context is a blob passed to OEMCrypto or CDM implementations,
typically including metadata like the label ('ENCRYPTION') and key size.
It is essential for deriving secure session keys used in content encryption/decryption.
This function validates and stores the encryption context for use in session key derivation.
Args:
data (bytes): A binary blob representing the encryption derivation context.
This must begin with a label and end with the encoded key size.
Notes:
- Expected structure: b'ENCRYPTION\\x00' + context-specific payload + b'\\x00\\x00\\x00\\x80'
- The tail 'b"\\x00\\x00\\x00\\x80"' corresponds to 128-bit (0x80) key size in big-endian form.
- The label is a null-terminated UTF-8 string used in OEMCrypto derivation contexts.
- Invalid contexts are ignored but logged for debugging.
"""
try:
# Constant prefix label used for AES encryption key derivation (null-terminated)
kEncryptionKeyLabel = b'ENCRYPTION\000'
# Expected suffix: 128 bits (0x80) in big-endian encoding (for AES-128)
kEncryptionKeySizeBits = b'\0\0\0\x80'
# Validate the encryption context starts with the correct label
assert data.startswith(kEncryptionKeyLabel), 'Context missing expected label: ENCRYPTION\\x00'
# Validate the encryption context ends with the correct encoded key size
assert data.endswith(kEncryptionKeySizeBits), 'Context missing expected AES key size specifier'
# Log that the encryption context was successfully received and registered
self.logger.info('Received encryption context: %s bytes', len(data))
# Store the validated encryption context for future use (e.g., key derivation)
self._context = data
except Exception as e:
self.logger.debug('Unable to register encryption context: %s', e)
keydive/drm/modules/provisioning.py
+274
View File
@@ -0,0 +1,274 @@
import json
from enum import Enum
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import cmac
from cryptography.hazmat.primitives.asymmetric.padding import OAEP, MGF1
from cryptography.hazmat.primitives.ciphers import Cipher
from cryptography.hazmat.primitives.ciphers.algorithms import AES
from cryptography.hazmat.primitives.ciphers.modes import CBC
from cryptography.hazmat.primitives.hashes import SHA1
from cryptography.hazmat.primitives.padding import PKCS7
from keydive.drm.modules.client import Client
from keydive.drm.protocol.license_pb2 import SignedProvisioningMessage, ProvisioningResponse, ClientIdentification
from keydive.utils import b64dec, dumps
class OEMCrypto_ProvisioningMethod(Enum):
"""
Enum representing different OEMCrypto provisioning methods.
"""
ProvisioningError = 0 # Device cannot be provisioned.
DrmCertificate = 1 # Device has baked-in DRM certificate (level 3 only).
Keybox = 2 # Device has factory-installed unique keybox.
OEMCertificate = 3 # Device has factory-installed OEM certificate.
def ContentKeySession_GenerateDerivedKeys(enc_key_base: bytes, key: bytes) -> bytes:
"""
Derives a single CMAC-based encryption key from the given base key.
This is a minimal implementation of key derivation used in Widevine's content key session setup.
It returns only the 'enc' key from the full provisioning/session derivation pipeline.
Args:
enc_key_base (bytes): Context string for encryption key derivation (e.g., from provisioning nonce or session context).
key (bytes): Base AES key (e.g., device AES key or session key).
Returns:
bytes: Derived encryption key.
"""
cipher = cmac.CMAC(
algorithm=AES(key),
backend=default_backend()
)
cipher.update(b'\x01' + enc_key_base)
return cipher.finalize()
class Provisioning(Client):
def __unwrap_rsa_key(self, key: bytes, iv: bytes, enc_data: bytes) -> bytes:
"""
Attempts to decrypt an RSA private key blob using AES-CBC and PKCS7 unpadding.
This method is used in Widevine provisioning flows to unwrap the encrypted RSA device key
from the provisioning response using a derived or session AES key.
Args:
key (bytes): The AES key used to decrypt the RSA private key (128/256-bit).
iv (bytes): Initialization Vector used during the AES encryption.
enc_data (bytes): Encrypted RSA private key data (typically from the provisioning response).
Returns:
bytes: The unwrapped (decrypted) RSA private key if successful, otherwise an empty byte string.
"""
dec_data = b''
try:
# Initialize AES-CBC cipher with the session key and provided IV
cipher = Cipher(
algorithm=AES(key),
mode=CBC(iv),
backend=default_backend()
)
decryptor = cipher.decryptor()
dec_padded_data = decryptor.update(enc_data) + decryptor.finalize()
# Remove PKCS7 padding to obtain the original private key bytes
unpadder = PKCS7(AES.block_size).unpadder()
dec_data = unpadder.update(dec_padded_data) + unpadder.finalize()
except Exception as e:
if key:
self.logger.debug('Failed to decrypt RSA private key using AES key: %s', key.hex())
# Return decrypted RSA private key (if successful); otherwise, return empty bytes
return dec_data
def set_provisioning_method(self, data: bytes) -> None:
"""
Determines and logs the provisioning method used by the Content Decryption Module (CDM).
This method decodes the given byte data to an integer, which maps to an
OEMCrypto_ProvisioningMethod enumeration value. It helps identify the provisioning
mechanism in use and logs relevant diagnostic information, especially when L1
provisioning appears to be disabled improperly.
Args:
data (bytes): UTF-8 encoded string representing an integer corresponding to
an OEMCrypto_ProvisioningMethod enum value.
Exception:
Catches and logs all exceptions encountered during decoding or enum conversion.
"""
try:
# Decode bytes to UTF-8 string, convert to int, and map to provisioning method enum
method = OEMCrypto_ProvisioningMethod(int(data.decode('utf-8')))
if method == OEMCrypto_ProvisioningMethod.Keybox and self.disabler:
# Warn user if L1 provisioning is enabled but disabling procedure incomplete
self.logger.warning(
'L1 provisioning deactivation appears incomplete. '
'Consider using a web dump or forcibly terminating the process to ensure proper disabling.'
)
else:
# Log the provisioning method name for informational purposes
self.logger.debug('Receive provisioning method: %s', method.name)
except Exception as e:
# Log any errors during decoding or mapping to enum with debug severity
self.logger.debug('Unable to parse provisioning method: %s', e)
def set_provisioning_response(self, data: bytes) -> None:
"""
Parses and applies a provisioning response from the Google Widevine provisioning service.
Supports both Keybox-based provisioning and Provisioning 3.0 OTA PKI formats.
This method extracts and decrypts the device RSA private key using known AES session keys,
which may be retrieved from keyboxes or OEM provisioning keys. It also supports
setting up the device certificate and updating client identification.
Args:
data (bytes): JSON-encoded provisioning response as received from the provisioning server.
Exception:
Catches and logs all exceptions raised during parsing, decryption, or client ID setup.
"""
try:
# Extract and decode the base64-encoded signed provisioning message from the JSON response
b64_signed_data = json.loads(data.split(b'\x00')[0])['signedResponse']
signed_data = b64dec(b64_signed_data, safe=True)
# Parse the SignedProvisioningMessage protobuf
signed_response = SignedProvisioningMessage()
signed_response.ParseFromString(signed_data)
# Extract the ProvisioningResponse payload embedded in the signed message
provisioning_response = ProvisioningResponse()
provisioning_response.ParseFromString(signed_response.message)
# Gather all known AES session keys from stored keyboxes (OEM Keybox keys)
session_enc_keys = [k.device_aes_key for k in self._keybox.values() if k.device_aes_key]
# Optionally, add additional AES keys extracted via reverse engineering/TEE exploit (if any)
session_enc_keys += self._device_aes_key
if provisioning_response.wrapping_key:
# OTA PKI-Based provisioning (Provisioning 3.0)
self.logger.info(
'Received OTA provisioning response: \n\n%s\n',
dumps({
'signature': {'type': 'RSASSA-PSS', 'data': signed_response.signature},
'nonce': provisioning_response.nonce,
'wrapping_key': provisioning_response.wrapping_key
}, beauty=True)
)
# Attempt to decrypt the AES wrapping key using all stored OEM private RSA keys
for oem_cert_priv_key in self._private_key.values():
try:
session_enc_key = oem_cert_priv_key.decrypt(
ciphertext=provisioning_response.wrapping_key,
padding=OAEP(
mgf=MGF1(algorithm=SHA1()),
algorithm=SHA1(),
label=None
)
)
session_enc_keys.append(session_enc_key)
except Exception as e:
self.logger.debug('Unable to decrypt OTA session key: %s', e)
else:
# Keybox-based provisioning (Provisioning 2.0)
self.logger.info(
'Receive Keybox provisioning response: \n\n%s\n',
dumps({
'signature': {'type': 'HMAC-SHA256', 'data': signed_response.signature},
'nonce': provisioning_response.nonce
}, beauty=True)
)
# Derive session encryption keys and map each original AES key to its derived encryption key
key_pairs = {
key: ContentKeySession_GenerateDerivedKeys(self._context, key) if self._context else None
for key in session_enc_keys
}
# Attempt to decrypt the device RSA private key using all gathered AES session keys
for aes_key, derived_key in key_pairs.items():
iv = provisioning_response.device_rsa_key_iv
enc_data = provisioning_response.device_rsa_key
# Attempt to decrypt using the derived key (from ContentKeySession_GenerateDerivedKeys)
dec_data = self.__unwrap_rsa_key(derived_key, iv, enc_data)
if dec_data:
self.logger.info('Provisioning from AES key derivation: %s', derived_key.hex())
# Try to find the corresponding Keybox which doesn't yet have an AES key assigned
keybox = next((k for k in self._keybox.values() if k.stable_id and k.device_id and not k.device_aes_key), None)
if keybox:
# Check if system_id is available to ensure the Keybox is valid and useful
if keybox.keybox_info.get('system_id'):
# Link this AES key back to the Keybox for future use
keybox.device_aes_key = aes_key
self.logger.info(
'Completed keybox with corresponding AES key:\n\n%s\n',
dumps(keybox.keybox_info, beauty=True)
)
# Update the stored keybox with the completed information
self._keybox[keybox.stable_id] = keybox
# Set the now-decrypted RSA private key as the active OEM private key
self.set_private_key(dec_data, None)
# Attempt to decrypt the RSA key using the raw session AES key (not derived)
dec_data = self.__unwrap_rsa_key(aes_key, iv, enc_data)
if dec_data:
self.logger.info('Provisioning from OTA AES key: %s', aes_key.hex())
# Set decrypted RSA private key from OTA flow
self.set_private_key(dec_data, None)
# If no decryption succeeded, OEM private key remains unset
# At this point, OTA provisioning using Provisioning 3.0 (PKI-based) might apply
# But this flow assumes Provisioning 2.0 (Keybox-based), so no further action here
# If a provisioning context exists, update the ClientIdentification with new capabilities and certificates
if self._provisioning:
"""
client_capabilities {
client_token: true
session_token: true
max_hdcp_version: HDCP_V2_2
oem_crypto_api_version: 15
anti_rollback_usage_table: false
srm_version: 0
can_update_srm: false
supported_certificate_key_type: RSA_2048
analog_output_capabilities: ANALOG_OUTPUT_NONE
can_disable_analog_output: false
}
"""
# Update client capabilities with typical fields
client_capabilities = self._provisioning.client_capabilities
client_capabilities.session_token = True
client_capabilities.max_hdcp_version = ClientIdentification.ClientCapabilities.HdcpVersion.HDCP_NONE
client_capabilities.anti_rollback_usage_table = False
client_capabilities.can_update_srm = False
# Construct a new ClientIdentification token with the provisioned device certificate
client_id = ClientIdentification(
type=ClientIdentification.TokenType.DRM_DEVICE_CERTIFICATE,
token=provisioning_response.device_certificate,
client_info=self._provisioning.client_info,
provider_client_token=self._provisioning.provider_client_token,
license_counter=self._provisioning.license_counter,
client_capabilities=client_capabilities,
vmp_data=self._provisioning.vmp_data,
device_credentials=self._provisioning.device_credentials
)
# Register the updated client identification token
self.set_client_id(client_id)
except Exception as e:
# Log any unexpected error encountered during the provisioning response handling
self.logger.debug('Unable to process provisioning response: %s', e)