DRM Pentesting Toolkit Skill

Skill by ara.so — Security Skills collection.

This skill enables AI agents to assist with DRM pentesting using a modular, plugin-based toolkit. The project provides a CustomTkinter GUI for managing DRM devices (Widevine .wvd, PlayReady .prd), extracting PSSH from MPD manifests, communicating with license servers, and downloading protected content.

What It Does

The DRM Pentesting Toolkit is a comprehensive framework for:

• Loading and validating Widevine/PlayReady device files
• Parsing PSSH (Protection System Specific Header) data from DASH/MPD manifests
• Extracting decryption keys from license servers
• Generating M3U playlists with embedded keys
• Downloading and decrypting content via N_m3u8DL-RE
• Managing credentials, proxies, and HTTP configurations
• Chaining multiple operations through a plugin stack system

Installation

# Clone the repository
git clone https://github.com/fairy-root/drm-pentesting-toolkit.git
cd drm-pentesting-toolkit

# Install dependencies
pip install -r requirements.txt

# Set up directory structure
mkdir -p devices/widevine devices/playready credentials proxies N_m3u8DL-RE OUTPUT

Required dependencies (from requirements.txt):

customtkinter>=5.2.0
pywidevine>=1.8.0
pyplayready>=1.3.0
requests>=2.31.0
Brotli>=1.1.0
protobuf>=4.25.0

Additional setup:

1. Place DRM device files in devices/widevine/*.wvd or devices/playready/*.prd
2. Download N_m3u8DL-RE and shaka-packager executables to N_m3u8DL-RE/
3. Create credential files in credentials/ (format: username:password)
4. Create proxy files in proxies/ (format: host:port:user:pass)

Running the Application

# Launch GUI
python main.py

The GUI provides tabs for:

• MPD Data: Configure manifest URLs, headers, cookies
• License Data: Set license server URLs and authentication
• Logs & Output: Monitor plugin execution
• Downloads: Track N_m3u8DL-RE download progress

Core Architecture

Plugin System

Plugins are Python files in plugins/ that define a run() function. The plugin manager uses introspection to:

1. Auto-detect required parameters
2. Map UI context to function arguments
3. Pass results between chained plugins

Plugin context variables available:

device_path: str          # Selected device file path
device_type: str          # 'widevine' or 'playready'
cdm: object              # CDM instance (Widevine/PlayReady)
session_id: bytes        # Active CDM session ID
pssh: str                # Extracted PSSH box
license_url: str         # License server URL
mpd_url: str             # MPD manifest URL
mpd_headers: dict        # HTTP headers for MPD requests
mpd_cookies: dict        # Cookies for MPD requests
license_headers: dict    # HTTP headers for license requests
license_cookies: dict    # Cookies for license requests
license_post_data: str   # POST data for license requests
keys: list               # Extracted decryption keys
credentials: dict        # Exposed credentials (if enabled)
proxy_line: str          # Selected proxy configuration
search_query: str        # User search input

Creating a Plugin

# plugins/my_custom_plugin.py
def run(mpd_url, mpd_headers, log_callback=None):
    """
    Custom plugin that fetches and analyzes MPD manifest
    
    Args:
        mpd_url: The DASH manifest URL
        mpd_headers: Dictionary of HTTP headers
        log_callback: Function to log messages to GUI
    
    Returns:
        dict: Context updates for next plugin
    """
    import requests
    
    if log_callback:
        log_callback(f"Fetching MPD from: {mpd_url}")
    
    try:
        response = requests.get(mpd_url, headers=mpd_headers, timeout=10)
        response.raise_for_status()
        
        # Parse MPD content
        mpd_content = response.text
        
        if log_callback:
            log_callback(f"MPD size: {len(mpd_content)} bytes")
        
        # Return context for next plugin
        return {
            'mpd_content': mpd_content,
            'mpd_size': len(mpd_content)
        }
    
    except Exception as e:
        if log_callback:
            log_callback(f"Error: {str(e)}")
        raise

Common Workflows

Key Extraction Workflow

# Plugin sequence:
# 1. Load device
# 2. Parse PSSH
# 3. Send license request
# 4. Extract keys

# Example: 1_load_devices.py
def run(device_path, device_type, log_callback=None):
    from pywidevine.cdm import Cdm as WidevineCdm
    from pywidevine.device import Device as WidevineDevice
    
    if device_type == 'widevine':
        device = WidevineDevice.load(device_path)
        cdm = WidevineCdm.from_device(device)
        session_id = cdm.open()
        
        if log_callback:
            log_callback(f"Widevine session opened: {session_id.hex()}")
        
        return {
            'cdm': cdm,
            'session_id': session_id,
            'device_security_level': device.security_level
        }

# Example: 2_parse_pssh.py
def run(mpd_url, mpd_headers, mpd_cookies, http_method='GET', log_callback=None):
    import requests
    import re
    from base64 import b64decode
    
    response = requests.request(
        method=http_method,
        url=mpd_url,
        headers=mpd_headers,
        cookies=mpd_cookies,
        timeout=15
    )
    
    # Extract PSSH from MPD
    pssh_pattern = r'<cenc:pssh>([A-Za-z0-9+/=]+)</cenc:pssh>'
    matches = re.findall(pssh_pattern, response.text)
    
    if matches:
        pssh = matches[0]
        if log_callback:
            log_callback(f"PSSH extracted: {pssh[:50]}...")
        return {'pssh': pssh}
    
    raise ValueError("No PSSH found in MPD")

# Example: 4_send_license.py
def run(cdm, session_id, pssh, license_url, license_headers, 
        license_cookies, license_post_data, log_callback=None):
    import requests
    from base64 import b64decode, b64encode
    
    # Generate license challenge
    challenge = cdm.get_license_challenge(session_id, b64decode(pssh))
    
    # Prepare license request
    payload = license_post_data.replace('{CHALLENGE}', b64encode(challenge).decode())
    
    response = requests.post(
        license_url,
        headers=license_headers,
        cookies=license_cookies,
        data=payload
    )
    
    # Parse license
    cdm.parse_license(session_id, response.content)
    
    # Extract keys
    keys = []
    for key in cdm.get_keys(session_id):
        if key.type == 'CONTENT':
            key_str = f"{key.kid.hex()}:{key.key.hex()}"
            keys.append(key_str)
            if log_callback:
                log_callback(f"Key: {key_str}")
    
    return {'keys': keys}

Download Workflow

# Example: 6_n_m3u8dl_re.py
def run(mpd_url, keys, search_query='output', log_callback=None):
    import subprocess
    import os
    
    exe_path = os.path.join('N_m3u8DL-RE', 'N_m3u8DL-RE.exe')
    
    # Build key arguments
    key_args = []
    for key in keys:
        key_args.extend(['--key', key])
    
    # Build command
    cmd = [
        exe_path,
        mpd_url,
        '--save-dir', 'OUTPUT',
        '--save-name', search_query,
        '--auto-select',
        '--no-log'
    ] + key_args
    
    if log_callback:
        log_callback(f"Starting download: {search_query}")
    
    # Execute download
    process = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True
    )
    
    return {
        'download_process': process,
        'output_name': search_query
    }

Credential Integration

# Plugins can access credentials when "Expose Credentials" is enabled

def run(credentials, license_url, log_callback=None):
    """
    Use credentials to authenticate with license server
    
    credentials structure:
    {
        'username': 'user@example.com',
        'password': 'pass123',
        'api_key': 'abc123xyz',
        'mac_address': '00:11:22:33:44:55'
    }
    """
    import requests
    
    # Example: Use credentials for authentication
    if 'api_key' in credentials:
        headers = {
            'Authorization': f"Bearer {credentials['api_key']}"
        }
    elif 'username' in credentials and 'password' in credentials:
        headers = {
            'X-Username': credentials['username'],
            'X-Password': credentials['password']
        }
    else:
        raise ValueError("No valid credentials found")
    
    response = requests.post(license_url, headers=headers)
    
    if log_callback:
        log_callback(f"Authenticated with {list(credentials.keys())}")
    
    return {'auth_headers': headers}

Configuration

Environment Settings

Edit settings/environment.ini to configure credential types:

[CREDENTIALS]
EMAIL_PASSWORD = ^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}:.+$
USERNAME_PASSWORD = ^[a-zA-Z0-9_]+:.+$
API_KEY = ^[a-zA-Z0-9]{20,}$
MAC_ADDRESS = ^([0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}$

State Persistence

Application state is saved to settings/savestate.json:

{
  "mpd_url": "https://example.com/manifest.mpd",
  "mpd_headers": "{\"User-Agent\": \"Mozilla/5.0\"}",
  "license_url": "https://license.example.com/widevine",
  "selected_device": "devices/widevine/device.wvd",
  "selected_credentials": "credentials/account.txt",
  "plugin_stack": ["1_load_devices", "2_parse_pssh", "4_send_license"]
}

Proxy Configuration

Format proxy files as proxies/myproxies.txt:

# Non-authenticated
proxy1.example.com:8080
192.168.1.100:3128

# Authenticated
proxy2.example.com:8080:username:password

Access in plugins:

def run(proxy_line, log_callback=None):
    if proxy_line:
        parts = proxy_line.split(':')
        if len(parts) == 2:
            proxy = f"http://{parts[0]}:{parts[1]}"
        elif len(parts) == 4:
            proxy = f"http://{parts[2]}:{parts[3]}@{parts[0]}:{parts[1]}"
        
        proxies = {
            'http': proxy,
            'https': proxy
        }
        
        return {'proxies': proxies}

Programmatic Usage

Using Plugin Manager Directly

from core.plugin_manager import PluginManager

# Initialize plugin manager
pm = PluginManager(plugins_dir='plugins')
pm.scan_plugins()

# Build context
context = {
    'device_path': 'devices/widevine/l3.wvd',
    'device_type': 'widevine',
    'mpd_url': 'https://example.com/manifest.mpd',
    'mpd_headers': {'User-Agent': 'Mozilla/5.0'},
    'license_url': 'https://license.example.com/widevine'
}

# Execute plugin stack
plugin_stack = ['1_load_devices', '2_parse_pssh', '4_send_license']

for plugin_name in plugin_stack:
    plugin_info = pm.plugins[plugin_name]
    result = pm.run_plugin(plugin_info, context, log_callback=print)
    
    # Merge results into context
    if isinstance(result, dict):
        context.update(result)

# Access extracted keys
print(f"Extracted keys: {context.get('keys', [])}")

Validating Device Files

from pywidevine.cdm import Cdm
from pywidevine.device import Device

def validate_widevine_device(device_path):
    """Validate a Widevine device file"""
    try:
        device = Device.load(device_path)
        cdm = Cdm.from_device(device)
        session_id = cdm.open()
        
        print(f"✓ Valid Widevine device")
        print(f"  Security Level: {device.security_level}")
        print(f"  Client ID: {device.client_id.token[:20].hex()}...")
        
        cdm.close(session_id)
        return True
    except Exception as e:
        print(f"✗ Invalid device: {e}")
        return False

validate_widevine_device('devices/widevine/device.wvd')

Troubleshooting

Plugin Not Detecting Parameters

Issue: Plugin not receiving expected context variables

Solution: Check function signature matches available context:

# Wrong - parameter name doesn't match context
def run(manifest_url, log_callback=None):
    pass

# Correct - matches context key 'mpd_url'
def run(mpd_url, log_callback=None):
    pass

PSSH Extraction Fails

Issue: No PSSH found in MPD manifest

Solution: Check HTTP method and authentication:

# Try different HTTP method
response = requests.request(
    method='POST',  # Some MPDs require POST
    url=mpd_url,
    headers=mpd_headers
)

# Or check for alternative PSSH locations
pssh_patterns = [
    r'<cenc:pssh>([^<]+)</cenc:pssh>',
    r'cenc:default_KID="([^"]+)"',
    r'<widevine:license>([^<]+)</widevine:license>'
]

License Request Fails

Issue: 403/401 errors from license server

Solution: Verify headers and credential handling:

# Check required headers
required_headers = {
    'User-Agent': 'Mozilla/5.0...',
    'Origin': 'https://example.com',
    'Referer': 'https://example.com/player',
    'Content-Type': 'application/octet-stream'  # Often required
}

# Verify POST data format
# Some servers expect base64 challenge, others expect JSON
payload_formats = [
    b64encode(challenge).decode(),  # Raw base64
    json.dumps({'challenge': b64encode(challenge).decode()}),  # JSON
    f'spc={b64encode(challenge).decode()}'  # Form-encoded
]

Download Process Hangs

Issue: N_m3u8DL-RE download not progressing

Solution: Check executable path and permissions:

import os
import stat

exe_path = 'N_m3u8DL-RE/N_m3u8DL-RE.exe'

# Verify file exists
if not os.path.exists(exe_path):
    raise FileNotFoundError(f"N_m3u8DL-RE not found at {exe_path}")

# Check execute permissions (Unix)
if os.name != 'nt':
    st = os.stat(exe_path)
    os.chmod(exe_path, st.st_mode | stat.S_IEXEC)

Device Loading Errors

Issue: CDM fails to initialize

Solution: Verify device file format and dependencies:

# For Widevine
try:
    from pywidevine.device import Device
    device = Device.load(device_path)
    print(f"Client ID length: {len(device.client_id.token)}")
    print(f"Private key present: {device.private_key is not None}")
except Exception as e:
    print(f"Device load error: {e}")
    # May need to update pywidevine version
    # pip install --upgrade pywidevine

Credentials Not Exposing

Issue: credentials parameter not available in plugin

Solution: Ensure "Expose Credentials to Plugins" checkbox is enabled in GUI, and credentials file is properly formatted:

# credentials/account.txt format
user@example.com:password123
# Not: {"user": "...", "pass": "..."}

Advanced Patterns

Multi-Key Content

def run(cdm, session_id, mpd_content, log_callback=None):
    """Extract keys for multi-period content"""
    import re
    from base64 import b64decode
    
    # Find all PSSH boxes
    pssh_list = re.findall(r'<cenc:pssh>([^<]+)</cenc:pssh>', mpd_content)
    
    all_keys = []
    for i, pssh in enumerate(set(pssh_list)):  # Deduplicate
        try:
            challenge = cdm.get_license_challenge(session_id, b64decode(pssh))
            # ... send license request ...
            keys = cdm.get_keys(session_id)
            all_keys.extend(keys)
            
            if log_callback:
                log_callback(f"Period {i+1}: {len(keys)} keys")
        except Exception as e:
            if log_callback:
                log_callback(f"Period {i+1} failed: {e}")
    
    return {'keys': all_keys}

Custom Headers from Environment

def run(mpd_url, log_callback=None):
    """Use environment variables for sensitive headers"""
    import os
    import requests
    
    headers = {
        'User-Agent': os.getenv('USER_AGENT', 'Mozilla/5.0'),
        'Authorization': os.getenv('AUTH_TOKEN'),  # Never hardcode tokens
        'X-Custom-Header': os.getenv('CUSTOM_HEADER')
    }
    
    # Remove None values
    headers = {k: v for k, v in headers.items() if v is not None}
    
    response = requests.get(mpd_url, headers=headers)
    return {'mpd_content': response.text}

This skill enables AI agents to help developers perform DRM security testing, create custom plugins, and automate key extraction workflows using this comprehensive toolkit.