drm-pentesting-toolkit

---
name: drm-pentesting-toolkit
description: Plugin-based DRM pentesting toolkit with GUI for device management, PSSH extraction, license key retrieval, and content downloading
triggers:
  - extract DRM keys from streaming content
  - test Widevine or PlayReady device files
  - parse PSSH from MPD manifest
  - download encrypted streaming content
  - validate DRM device credentials
  - create DRM pentesting plugin
  - chain DRM pentesting operations
  - extract license server keys
---

# DRM Pentesting Toolkit Skill

> Skill by [ara.so](https://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

```bash
# 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

```bash
# 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:
```python
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

```python
# 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

```python
# 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

```python
# 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

```python
# 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:

```ini
[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`:

```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:

```python
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

```python
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

```python
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:

```python
# 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:

```python
# 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:

```python
# 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:

```python
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:

```python
# 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:

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

## Advanced Patterns

### Multi-Key Content

```python
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

```python
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.