drm-pentesting-toolkit

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