Skip to content

plugins.md

Latest commit

 

History

History
1347 lines (1064 loc) · 42.3 KB

plugins.md

File metadata and controls

1347 lines (1064 loc) · 42.3 KB

DRM Pentesting Toolkit – Plugin Development Guide

Overview

The DRM Pentesting Toolkit uses a sophisticated, modular plugin-based architecture for maximum flexibility and extensibility. Each plugin is a Python script that performs a specific task in the DRM workflow—such as device loading, content discovery, license extraction, or output generation. Plugins are auto-discovered from the plugins/ directory and can be executed individually or chained together in a stack for complex workflows.

Key Features

  • Automatic Function Introspection: The plugin manager automatically detects plugin arguments through Python's inspect module
  • Dynamic Context Passing: Plugin outputs are merged into a shared context and passed to subsequent plugins
  • Credential Integration: Automatic mapping of credentials based on environment.ini configuration
  • Search Query Support: Built-in search functionality for content discovery and custom filename support
  • Separated Data Structure: Distinct handling of MPD and License data with HTTP method support
  • Proxy Support: Full proxy support with authenticated and non-authenticated configurations
  • Error Handling: Comprehensive error management with detailed logging
  • Hot Reloading: Plugin rescanning without application restart
  • Subfolder Organization: Support for organizing plugins in subdirectories

Architecture Benefits

  • Zero Configuration: New plugins work immediately without modifying core code
  • Type Safety: Automatic argument validation and type checking
  • State Management: Automatic state passing between plugin executions
  • Error Isolation: Plugin failures don't crash the entire application
  • Extensibility: Easy to add new DRM systems, platforms, and workflows

Quick Start: Build Your First Plugin

1. Create a New Plugin File

Create a new Python file in the plugins/ directory. Use descriptive names and optional numeric prefixes for execution order:

# Simple naming
# plugins/my_content_search.py

# Ordered naming (recommended for workflows)
# plugins/9_my_license_plugin.py

# Subfolder organization
# plugins/custom/my_platform_plugin.py

2. Define the run() Function

Your plugin must define a run() function as the entry point. The plugin manager uses function introspection to automatically determine and provide the required arguments:

def run(search_query=None, USER=None, PASS=None, mpd_url=None, license_url=None, **kwargs):
    """
    Plugin documentation here.
    
    Args:
        search_query: Search term from UI (optional) - also used for custom filenames
        USER: Username from credentials (optional) 
        PASS: Password from credentials (optional)
        mpd_url: MPD URL from MPD Data tab (optional)
        license_url: License URL from License Data tab (optional)
        **kwargs: Additional context variables from previous plugins
    
    Returns:
        Dictionary with results and preview
    """
    # Your plugin logic here
    pass

3. Implement Your Plugin Logic

def run(search_query=None, USER=None, PASS=None, mpd_url=None, license_url=None, **kwargs):
    """Example content search plugin with custom filename support."""
    if not search_query:
        return {
            "error": "No search query provided",
            "preview": "Search query required"
        }
    
    try:
        # Use search_query for both content discovery and custom filename
        sanitized_filename = search_query.replace(" ", "_").replace(":", "")
        
        # Simulate content search
        results = [
            {"title": f"{search_query} - Episode 1", "id": "12345"},
            {"title": f"{search_query} - Episode 2", "id": "12346"}
        ]
        
        return {
            "search_results": results,
            "search_query": search_query,
            "preview": f"Found {len(results)} results for '{search_query}'"
        }
    except Exception as e:
        return {
            "error": str(e),
            "preview": f"Search failed: {e}"
        }

Available Built-in Plugins

Core System Plugins

1_load_devices.py

Device file loading and CDM initialization

  • Loads Widevine (.wvd) and PlayReady (.prd) device files
  • Creates CDM instances and opens sessions
  • Parameters: device_file
  • Returns: device, cdm, session_id, file_path
  • Dependencies: pywidevine, pyplayready

2_parse_pssh.py

MPD manifest parsing and PSSH extraction

  • Fetches MPD files from URLs with configurable HTTP methods
  • Extracts Widevine PSSH data using regex patterns
  • Supports custom headers, cookies, and POST data from MPD Data tab
  • Parameters: mpd_url, mpd_headers, mpd_cookies, mpd_post_data, mpd_method
  • Returns: pssh, mpd_content
  • Dependencies: requests

4_send_license.py

License server communication and key extraction

  • Sends license challenges to license servers
  • Extracts decryption keys from license responses
  • Supports custom headers, cookies, and POST data from License Data tab
  • Parameters: license_url, license_headers, license_cookies, license_post_data, license_method, pssh, cdm, session_id
  • Returns: keys, license_response
  • Dependencies: requests

5_write_m3u.py

M3U playlist generation

  • Creates M3U playlists with extracted keys and stream URLs
  • Handles file I/O with proper directory creation
  • Parameters: keys, mpd_url
  • Returns: m3u_content, m3u_file, output_path

Download & Processing Plugins

6_n_m3u8dl_re.py

Advanced content download using N_m3u8DL-RE

  • Downloads encrypted content using the N_m3u8DL-RE tool with decryption keys
  • Custom Filename Support: Uses search_query parameter for automatic filename generation
  • Supports multiple key formats and header configurations
  • Automatically uses Shaka Packager for optimal output quality
  • Parameters: keys, mpd_url, search_query (for custom filename), headers, cookies
  • Returns: download_command, output_path, custom_filename
  • Dependencies: N_m3u8DL-RE executable, Shaka Packager
  • Parameters: mpd_url, keys, headers, output_format
  • Returns: command, downloaded_files, output_directory, return_code
  • Features:
    • Multiple header format support (JSON, dictionary, multi-line)
    • Comprehensive error handling and logging
    • Automatic working directory management
    • Support for multiple decryption keys
    • File size reporting and download verification
    • Background process management
    • Progress monitoring and ETA calculation
  • Dependencies: subprocess, psutil, json

Validation & Testing Plugins

7_wvd_validation.py

Widevine device validation

  • Validates Widevine device file integrity
  • Tests CDM functionality and key extraction capability
  • Parameters: device_file, cdm, session_id
  • Returns: validation_result, tests_passed, device_info
  • Dependencies: pywidevine, requests

8_prd_validation.py

PlayReady device validation

  • Validates PlayReady device file integrity
  • Tests device functionality and license handling
  • Parameters: device_file, cdm, session_id
  • Returns: validation_result, tests_passed, device_info
  • Dependencies: pyplayready, requests

Platform-Specific Plugins (Examples)

3_shahid_license.py

Shahid platform license generation

  • Generates signed license URLs for Shahid VIP platform
  • Implements HMAC-SHA256 authentication
  • Parameters: assetId, country_code
  • Returns: license_url, signature, timestamp
  • Dependencies: requests, hmac, hashlib
def run(search_query=None, USER=None, PASS=None, **kwargs):
    """
    Your plugin documentation here.
    
    Args:
        search_query: Search term from UI (optional)
        USER: Username from credentials (optional)
        PASS: Password from credentials (optional)
        **kwargs: Additional context variables
    
    Returns:
        Dictionary with results and preview
    """
    # Your plugin logic here
    pass

3. Implement Your Plugin Logic

def run(search_query=None, USER=None, PASS=None, **kwargs):
    """Example content search plugin."""
    if not search_query:
        return {
            "error": "No search query provided",
            "preview": "Search query required"
        }
    
    try:
        # Simulate content search
        results = [
            {"title": f"{search_query} - Episode 1", "id": "12345"},
            {"title": f"{search_query} - Episode 2", "id": "12346"}
        ]
        
        return {
            "search_results": results,
            "search_query": search_query,
            "preview": f"Found {len(results)} results for '{search_query}'"
        }
    except Exception as e:
        return {
            "error": str(e),
            "preview": f"Search failed: {e}"
        }

Available Context Variables

The following variables are commonly available in the plugin context and can be used as parameters in your run() function:

Core System Variables

  • search_query: Search term entered in the UI (also used for custom filenames)
  • device_file: Selected device file path (.wvd or .prd)
  • USER, PASS: Credentials from selected profile
  • proxy_host, proxy_port, proxy_user, proxy_pass: Proxy configuration

MPD Data Tab Variables (New Structure)

  • mpd_url: MPD manifest URL
  • mpd_headers: Headers for MPD requests
  • mpd_cookies: Cookies for MPD requests
  • mpd_post_data: POST data for MPD requests
  • mpd_method: HTTP method for MPD requests (GET, POST, PUT, etc.)

License Data Tab Variables (New Structure)

  • license_url: License server URL
  • license_headers: Headers for license requests
  • license_cookies: Cookies for license requests
  • license_post_data: POST data for license requests
  • license_method: HTTP method for license requests (GET, POST, PUT, etc.)

Plugin Output Variables (Context Passed Between Plugins)

  • device: Loaded device object from device loading plugins
  • cdm: CDM instance from device loading plugins
  • session_id: CDM session ID from device loading plugins
  • pssh: PSSH data extracted from MPD parsing plugins
  • keys: Decryption keys from license plugins
  • mpd_content: Raw MPD content from parsing plugins
  • license_response: License server response data
  • m3u_content: Generated M3U playlist content
  • m3u_file: Path to generated M3U file
  • output_path: Download output directory path
  • custom_filename: Sanitized filename from search_query

Legacy Variables (Backward Compatibility)

  • headers: Fallback to mpd_headers if not specified
  • cookies: Fallback to mpd_cookies if not specified
  • postdata: Fallback to mpd_post_data if not specified
  • method: Fallback to mpd_method if not specified

HTTP Configuration

  • headers: HTTP headers as string (multi-line) or dictionary
  • postdata: POST data for license requests (JSON or raw)
  • cookies: Cookie data for session management

Credentials (Automatically Mapped from environment.ini)

  • USER / PASS: Username/password credentials
  • EMAIL / PASSWORD: Email/password credentials
  • MAC: MAC address for device identification
  • APIKEY: API key for service authentication
  • Any other fields defined in SlicesAlias configuration

Plugin Chain Data

  • Any keys returned by previous plugins are automatically added to the context
  • Context accumulates as plugins execute in sequence
  • Downstream plugins can access all previous plugin outputs

Search Query Integration

The toolkit includes built-in search query functionality that allows users to enter search terms in the main UI for content discovery, filtering, and custom filename generation.

Accessing Search Queries

def run(search_query=None, **kwargs):
    """Plugin that uses search query from UI for both search and custom filename."""
    if not search_query:
        return {
            "preview": "No search query provided",
            "search_performed": False
        }
    
    # Use search query for content discovery
    print(f"Searching for: {search_query}")
    
    # Use search query for custom filename generation
    import re
    sanitized_filename = re.sub(r'[<>:"/\\|?*]', '', search_query.replace(' ', '_'))
    
    # Example: Search multiple platforms
    results = []
    for platform in ["Platform A", "Platform B", "Platform C"]:
        # Simulate platform-specific search
        platform_results = search_platform(platform, search_query)
        results.extend(platform_results)
    
    return {
        "search_query": search_query,
        "custom_filename": sanitized_filename,
        "search_results": results,
        "total_results": len(results),
        "preview": f"Found {len(results)} results for '{search_query}'"
    }

Custom Filename Support

The search_query parameter can be used to generate custom filenames for downloaded content:

def run(search_query=None, keys=None, mpd_url=None, **kwargs):
    """Download plugin with custom filename support."""
    import re
    import os
    
    # Generate custom filename from search query
    if search_query:
        # Sanitize the search query for filesystem compatibility
        custom_filename = re.sub(r'[<>:"/\\|?*]', '', search_query)
        custom_filename = custom_filename.replace(' ', '_').replace(':', '')
        
        # Use in download command
        output_file = f"OUTPUT/{custom_filename}"
    else:
        # Use default filename
        output_file = "OUTPUT/downloaded_content"
    
    # Example download command with custom filename
    download_command = [
        "N_m3u8DL-RE/N_m3u8DL-RE.exe",
        mpd_url,
        "--save-name", custom_filename,
        "--save-dir", "OUTPUT"
    ]
    
    return {
        "custom_filename": custom_filename,
        "output_path": output_file,
        "download_command": " ".join(download_command),
        "preview": f"Download configured with filename: {custom_filename}"
    }

Search Query Best Practices

  • Dual Purpose: Use search_query for both content discovery and custom filename generation
  • Validation: Always check if search_query is provided before using it
  • Sanitization: Sanitize search queries for safe use in HTTP requests and file paths
  • Context Passing: Include the search query and custom filename in your return dictionary
  • Error Handling: Provide meaningful error messages when search queries are required but missing
  • Matching: Consider case-insensitive matching and partial string matching

Proxy Support System

The toolkit provides comprehensive proxy support for enhanced privacy and access to geo-restricted content.

Proxy Configuration Files

Proxy files are stored in the proxies/ directory with .txt extension:

# Non-authenticated proxies (host:port)
128.199.202.122:3128
198.199.86.11:8080
185.125.204.174:80

# Authenticated proxies (host:port:username:password)
proxy.example.com:8080:username:password
premium.proxy.com:3128:user123:pass456
rp.proxyscrape.com:6060:51qmsxyvtf8cmxb:jtx76zocdfgyzwq

# Mixed configurations are supported

Using Proxies in Plugins

def run(proxies=None, use_proxies=False, **kwargs):
    """Plugin with proxy support."""
    
    proxy_config = None
    if use_proxies and proxies:
        # Select first available proxy (or implement selection logic)
        proxy = proxies[0]
        
        if len(proxy.split(':')) == 4:
            # Authenticated proxy
            host, port, username, password = proxy.split(':')
            proxy_config = {
                'http': f'http://{username}:{password}@{host}:{port}',
                'https': f'https://{username}:{password}@{host}:{port}'
            }
        else:
            # Non-authenticated proxy
            host, port = proxy.split(':')
            proxy_config = {
                'http': f'http://{host}:{port}',
                'https': f'https://{host}:{port}'
            }
    
    # Use proxy in requests
    response = requests.get(url, proxies=proxy_config)
    
    return {
        "proxy_used": proxy_config is not None,
        "response_status": response.status_code,
        "preview": f"Request completed via {'proxy' if proxy_config else 'direct connection'}"
    }

Proxy Best Practices

  • Error Handling: Always handle proxy connection failures gracefully
  • Fallback: Implement fallback to direct connection if proxy fails
  • Testing: Test proxy connectivity before making important requests
  • Security: Don't log proxy credentials in plaintext
"preview": f"Found {len(results)} results for '{search_query}'"

Search Query Best Practices

  • Always check if search_query is provided before using it
  • Include the search query in your return dictionary for downstream plugins
  • Provide meaningful error messages when search queries are required but missing
  • Consider case-insensitive matching and partial string matching

HTTP Methods and Tab Structure

The toolkit now supports a separated tab structure with distinct configuration for MPD and License requests, each with customizable HTTP methods.

MPD Data Tab Configuration

def run(mpd_url=None, mpd_headers=None, mpd_cookies=None, mpd_post_data=None, mpd_method='GET', **kwargs):
    """Plugin that uses MPD data with custom HTTP method."""
    import requests
    
    # Prepare headers
    headers = {}
    if mpd_headers:
        for line in mpd_headers.split('\n'):
            if ':' in line:
                key, value = line.split(':', 1)
                headers[key.strip()] = value.strip()
    
    # Prepare cookies
    cookies = {}
    if mpd_cookies:
        for line in mpd_cookies.split('\n'):
            if '=' in line:
                key, value = line.split('=', 1)
                cookies[key.strip()] = value.strip()
    
    # Make request with specified method
    if mpd_method.upper() == 'POST':
        response = requests.post(mpd_url, headers=headers, cookies=cookies, data=mpd_post_data)
    elif mpd_method.upper() == 'PUT':
        response = requests.put(mpd_url, headers=headers, cookies=cookies, data=mpd_post_data)
    else:  # Default to GET
        response = requests.get(mpd_url, headers=headers, cookies=cookies)
    
    return {
        "mpd_content": response.text,
        "status_code": response.status_code,
        "preview": f"MPD fetched via {mpd_method.upper()} - {response.status_code}"
    }

License Data Tab Configuration

def run(license_url=None, license_headers=None, license_cookies=None, 
        license_post_data=None, license_method='POST', pssh=None, cdm=None, session_id=None, **kwargs):
    """Plugin that handles license requests with custom HTTP method."""
    import requests
    
    if not all([pssh, cdm, session_id]):
        return {"error": "Missing CDM data", "preview": "CDM session required"}
    
    # Generate license challenge
    challenge = cdm.get_license_challenge(session_id, pssh)
    
    # Prepare headers
    headers = {'Content-Type': 'application/octet-stream'}
    if license_headers:
        for line in license_headers.split('\n'):
            if ':' in line:
                key, value = line.split(':', 1)
                headers[key.strip()] = value.strip()
    
    # Prepare cookies
    cookies = {}
    if license_cookies:
        for line in license_cookies.split('\n'):
            if '=' in line:
                key, value = line.split('=', 1)
                cookies[key.strip()] = value.strip()
    
    # Make license request with specified method
    data = license_post_data if license_post_data else challenge
    
    if license_method.upper() == 'GET':
        response = requests.get(license_url, headers=headers, cookies=cookies)
    elif license_method.upper() == 'PUT':
        response = requests.put(license_url, headers=headers, cookies=cookies, data=data)
    else:  # Default to POST
        response = requests.post(license_url, headers=headers, cookies=cookies, data=data)
    
    # Parse license and extract keys
    cdm.parse_license(session_id, response.content)
    keys = cdm.get_keys(session_id)
    
    return {
        "keys": keys,
        "license_response": response.content,
        "status_code": response.status_code,
        "preview": f"License via {license_method.upper()} - {len(keys)} keys extracted"
    }

Backward Compatibility

The plugin manager maintains backward compatibility with legacy plugins using the old parameter names:

def run(headers=None, cookies=None, postdata=None, method='GET', **kwargs):
    """Legacy plugin - automatically mapped to new structure."""
    # This plugin will receive:
    # - headers -> mpd_headers (if mpd context) or license_headers (if license context)
    # - cookies -> mpd_cookies or license_cookies
    # - postdata -> mpd_post_data or license_post_data
    # - method -> mpd_method or license_method

Proxy Support

The toolkit provides comprehensive proxy support for enhanced privacy and access to geo-restricted content.

Proxy Configuration

Proxy files are stored in the proxies/ directory with .txt extension:

# Non-authenticated proxies (host:port)
128.199.202.122:3128
198.199.86.11:8080
185.125.204.174:80

# Authenticated proxies (host:port:username:password)
proxy.example.com:8080:username:password
premium.proxy.com:3128:user123:pass456
rp.proxyscrape.com:6060:51qmsxyvtf8cmxb-country-vg:jtx76zocdfgyzwq

# Mixed configurations
free.proxy.com:8080
authenticated.proxy.net:3128:myuser:mypass

Using Proxies in Plugins

def run(use_proxies=False, proxies=None, search_query=None, **kwargs):
    """Plugin demonstrating proxy usage."""
    import requests
    import random
    
    if not search_query:
        return {"error": "Search query required", "preview": "No search query"}
    
    session = requests.Session()
    proxy_info = "Direct connection"
    
    # Configure proxy if enabled
    if use_proxies and proxies:
        # Select a random proxy for load balancing
        proxy = random.choice(proxies)
        
        # Format proxy URL
        if proxy.get("username") and proxy.get("password"):
            proxy_url = f"http://{proxy['username']}:{proxy['password']}@{proxy['host']}:{proxy['port']}"
        else:
            proxy_url = f"http://{proxy['host']}:{proxy['port']}"
        
        # Configure session with proxy
        session.proxies = {
            'http': proxy_url,
            'https': proxy_url
        }
        
        proxy_info = f"Proxy: {proxy['host']}:{proxy['port']}"
        if proxy.get("username"):
            proxy_info += " (authenticated)"
    
    try:
        # Make request with or without proxy
        response = session.get(f"https://api.example.com/search?q={search_query}")
        
        return {
            "search_query": search_query,
            "proxy_used": proxy_info,
            "status_code": response.status_code,
            "preview": f"Search completed via {proxy_info}"
        }
    except Exception as e:
        return {
            "error": str(e),
            "preview": f"Request failed: {e}"
        }
    finally:
        session.close()

Proxy Data Structure

Each proxy in the proxies list is a dictionary:

{
    "host": "proxy.example.com",
    "port": "8080",
    "username": "user123",      # None for non-authenticated proxies
    "password": "pass456",      # None for non-authenticated proxies
    "type": "http"              # Currently only HTTP proxies supported
}

Proxy Best Practices

  • Always check if use_proxies is enabled before using proxy configuration
  • Handle proxy connection failures gracefully with fallbacks
  • Use session.close() in a finally block to clean up connections
  • Consider proxy rotation for load balancing and redundancy
  • Test proxy functionality with connection timeouts

Credential System Integration

The toolkit features a sophisticated credential management system that automatically maps credentials to plugin parameters based on configuration patterns.

Credential Types Configuration

Configure credential types in settings/environment.ini:

[WORDLIST TYPE]
Name=Credentials
Regex=^.*:.*$
Verify=True
Separator=:
Slices=USERNAME,PASSWORD
SlicesAlias=USER,PASS

[EMAIL TYPE]
Name=Email
Regex=^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}:.*$
Verify=True
Separator=:
Slices=EMAIL,PASSWORD
SlicesAlias=EMAIL,PASS

[API TYPE]
Name=API
Regex=^[A-Za-z0-9_-]{20,}$
Verify=True
Separator=
Slices=APIKEY
SlicesAlias=APIKEY

[MAC TYPE]
Name=MAC
Regex=^([0-9A-Fa-f]{2}[:-]){5}([0-9A-Fa-f]{2})$
Verify=True
Separator=
Slices=MAC
SlicesAlias=MAC

Using Credentials in Plugins

def run(USER=None, PASS=None, EMAIL=None, APIKEY=None, **kwargs):
    """Plugin with credential support."""
    
    # Check for different credential types
    if USER and PASS:
        # Username/password authentication
        auth_data = {"username": USER, "password": PASS}
        return authenticate_with_credentials(auth_data)
    
    elif EMAIL and PASS:
        # Email/password authentication
        auth_data = {"email": EMAIL, "password": PASS}
        return authenticate_with_email(auth_data)
    
    elif APIKEY:
        # API key authentication
        headers = {"Authorization": f"Bearer {APIKEY}"}
        return authenticate_with_api_key(headers)
    
    else:
        return {
            "error": "No valid credentials provided",
            "preview": "Authentication failed - no credentials"
        }

Credential File Formats

Create credential files in the credentials/ directory:

# credentials/username_passwords.txt
user1:password123
admin:secretpass
testuser:mypassword

# credentials/email_accounts.txt  
user@example.com:password123
admin@company.com:adminpass

# credentials/api_keys.txt
AIzaSyBf8v9k2Xp7Q3m4R5t6Y7u8I9o0P1a2S3d4F5g6
sk-abcd1234efgh5678ijkl9012mnop3456qrst

Advanced Plugin Development

Error Handling Best Practices

def run(**kwargs):
    """Plugin with comprehensive error handling."""
    try:
        # Main plugin logic
        result = perform_operation()
        
        # Validate result
        if not result:
            return {
                "error": "Operation returned no results",
                "preview": "No data available"
            }
        
        return {
            "data": result,
            "preview": f"Successfully processed {len(result)} items"
        }
        
    except requests.RequestException as e:
        return {
            "error": f"Network error: {str(e)}",
            "preview": "Network request failed"
        }
    
    except ValueError as e:
        return {
            "error": f"Data validation error: {str(e)}",
            "preview": "Invalid data format"
        }
    
    except Exception as e:
        return {
            "error": f"Unexpected error: {str(e)}",
            "preview": "Plugin execution failed"
        }

Logging and Debug Output

import logging

def run(**kwargs):
    """Plugin with logging support."""
    
    # Set up logging (optional - for development)
    logging.basicConfig(level=logging.DEBUG)
    logger = logging.getLogger(__name__)
    
    logger.debug(f"Plugin started with args: {kwargs}")
    
    try:
        # Your plugin logic
        result = process_data()
        logger.info(f"Processing completed: {len(result)} items")
        
        return {
            "data": result,
            "preview": f"Processed {len(result)} items successfully"
        }
    
    except Exception as e:
        logger.error(f"Plugin failed: {e}")
        return {
            "error": str(e),
            "preview": "Processing failed"
        }

Resource Management

def run(cdm=None, session_id=None, **kwargs):
    """Plugin with proper resource cleanup."""
    
    temp_files = []
    
    try:
        # Create temporary files
        temp_file = create_temp_file()
        temp_files.append(temp_file)
        
        # Process data
        result = process_with_temp_file(temp_file)
        
        return {
            "result": result,
            "preview": "Processing completed successfully"
        }
    
    except Exception as e:
        return {
            "error": str(e),
            "preview": "Processing failed"
        }
    
    finally:
        # Cleanup resources
        for temp_file in temp_files:
            try:
                os.remove(temp_file)
            except:
                pass
        
        # Close CDM session if created
        if cdm and session_id:
            try:
                cdm.close(session_id)
            except:
                pass

Plugin Testing Framework

def run(**kwargs):
    """Plugin with built-in testing capabilities."""
    
    # Test mode detection
    test_mode = kwargs.get('_test_mode', False)
    
    if test_mode:
        # Return test data
        return {
            "test_data": generate_test_data(),
            "preview": "Test mode - generated sample data"
        }
    
    # Normal execution
    try:
        result = normal_operation(**kwargs)
        return {
            "data": result,
            "preview": "Operation completed successfully"
        }
    except Exception as e:
        return {
            "error": str(e),
            "preview": "Operation failed"
        }

def generate_test_data():
    """Generate test data for plugin validation."""
    return {
        "sample_keys": ["key1", "key2", "key3"],
        "sample_urls": ["http://example.com/content1.mpd"],
        "timestamp": datetime.now().isoformat()
    }

Plugin Performance Guidelines

Efficient Data Handling

def run(large_dataset=None, **kwargs):
    """Plugin optimized for large datasets."""
    
    if not large_dataset:
        return {
            "error": "No dataset provided",
            "preview": "Missing input data"
        }
    
    # Process data in chunks for memory efficiency
    chunk_size = 1000
    results = []
    
    for i in range(0, len(large_dataset), chunk_size):
        chunk = large_dataset[i:i + chunk_size]
        processed_chunk = process_chunk(chunk)
        results.extend(processed_chunk)
        
        # Optional progress reporting
        progress = (i + chunk_size) / len(large_dataset) * 100
        print(f"Progress: {progress:.1f}%")
    
    return {
        "processed_data": results,
        "total_processed": len(results),
        "preview": f"Processed {len(results)} items in {len(large_dataset)//chunk_size + 1} chunks"
    }

Async Operations Support

import asyncio
import aiohttp

def run(**kwargs):
    """Plugin with async HTTP requests."""
    
    urls = kwargs.get('urls', [])
    if not urls:
        return {
            "error": "No URLs provided",
            "preview": "Missing URL list"
        }
    
    # Run async operations
    results = asyncio.run(fetch_multiple_urls(urls))
    
    return {
        "responses": results,
        "total_requests": len(urls),
        "preview": f"Completed {len(results)} async requests"
    }

async def fetch_multiple_urls(urls):
    """Fetch multiple URLs concurrently."""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        return await asyncio.gather(*tasks)

async def fetch_url(session, url):
    """Fetch single URL."""
    try:
        async with session.get(url) as response:
            return {
                "url": url,
                "status": response.status,
                "content": await response.text()
            }
    except Exception as e:
        return {
            "url": url,
            "error": str(e)
        }

Plugin Examples by Use Case

Content Discovery Plugin

def run(search_query=None, USER=None, PASS=None, **kwargs):
    """Discover content across multiple platforms."""
    
    if not search_query:
        return {
            "error": "Search query required",
            "preview": "No search term provided"
        }
    
    platforms = {
        "netflix": search_netflix,
        "amazon": search_amazon, 
        "hulu": search_hulu
    }
    
    all_results = []
    
    for platform_name, search_func in platforms.items():
        try:
            results = search_func(search_query, USER, PASS)
            for result in results:
                result['platform'] = platform_name
            all_results.extend(results)
        except Exception as e:
            print(f"Failed to search {platform_name}: {e}")
    
    return {
        "search_results": all_results,
        "total_found": len(all_results),
        "platforms_searched": list(platforms.keys()),
        "preview": f"Found {len(all_results)} results across {len(platforms)} platforms"
    }

License Server Plugin

def run(license_url=None, pssh=None, headers=None, **kwargs):
    """Custom license server interaction."""
    
    if not license_url or not pssh:
        return {
            "error": "License URL and PSSH required",
            "preview": "Missing required parameters"
        }
    
    # Parse headers
    headers_dict = {}
    if headers:
        if isinstance(headers, str):
            for line in headers.split('\n'):
                if ':' in line:
                    key, value = line.split(':', 1)
                    headers_dict[key.strip()] = value.strip()
        elif isinstance(headers, dict):
            headers_dict = headers
    
    # Prepare license request
    license_data = {
        "pssh": pssh,
        "challenge": generate_challenge(pssh)
    }
    
    try:
        response = requests.post(
            license_url,
            json=license_data,
            headers=headers_dict,
            timeout=30
        )
        
        if response.status_code == 200:
            keys = extract_keys_from_response(response.content)
            return {
                "keys": keys,
                "license_response": response.content,
                "preview": f"Extracted {len(keys)} keys successfully"
            }
        else:
            return {
                "error": f"License server returned {response.status_code}",
                "preview": "License request failed"
            }
    
    except Exception as e:
        return {
            "error": str(e),
            "preview": "License request failed"
        }

Data Validation Plugin

def run(keys=None, mpd_url=None, **kwargs):
    """Validate extracted keys and content data."""
    
    validation_results = {
        "keys_valid": False,
        "mpd_accessible": False,
        "content_playable": False
    }
    
    # Validate keys
    if keys:
        validation_results["keys_valid"] = validate_keys(keys)
        validation_results["key_count"] = len(keys)
    
    # Validate MPD
    if mpd_url:
        validation_results["mpd_accessible"] = test_mpd_access(mpd_url)
    
    # Test content playability (if both keys and MPD available)
    if keys and mpd_url and validation_results["keys_valid"] and validation_results["mpd_accessible"]:
        validation_results["content_playable"] = test_content_decryption(keys, mpd_url)
    
    # Generate summary
    passed_tests = sum(1 for result in validation_results.values() if result is True)
    total_tests = sum(1 for result in validation_results.values() if isinstance(result, bool))
    
    return {
        "validation_results": validation_results,
        "tests_passed": passed_tests,
        "total_tests": total_tests,
        "preview": f"Validation: {passed_tests}/{total_tests} tests passed"
    }

def validate_keys(keys):
    """Validate key format and content."""
    if not keys or not isinstance(keys, list):
        return False
    
    for key in keys:
        if not isinstance(key, str) or len(key) < 32:
            return False
    
    return True

def test_mpd_access(mpd_url):
    """Test if MPD is accessible."""
    try:
        response = requests.head(mpd_url, timeout=10)
        return response.status_code == 200
    except:
        return False

Testing and Debugging

Plugin Testing Strategy

  1. Unit Testing: Test individual plugin functions with mock data
  2. Integration Testing: Test plugins within the full application context
  3. Error Case Testing: Verify proper error handling and recovery
  4. Performance Testing: Monitor execution time and resource usage

Debugging Techniques

Debug Parameter Plugin

Use the included debug_params.py plugin to inspect available context:

# Add debug_params.py to your plugin stack to see all available variables
def run(**kwargs):
    """Debug plugin that displays all available context variables."""
    
    preview_lines = ["Available Context Variables:"]
    for key, value in sorted(kwargs.items()):
        value_preview = str(value)[:50] + "..." if len(str(value)) > 50 else str(value)
        preview_lines.append(f"  {key}: {value_preview}")
    
    return {
        "all_context": kwargs,
        "context_keys": list(kwargs.keys()),
        "preview": "\n".join(preview_lines)
    }

Logging and Output

def run(debug_mode=False, **kwargs):
    """Plugin with comprehensive logging."""
    
    if debug_mode:
        print(f"Plugin started with {len(kwargs)} context variables")
        for key in kwargs.keys():
            print(f"  Available: {key}")
    
    try:
        # Plugin logic here
        result = perform_operation()
        
        if debug_mode:
            print(f"Operation completed successfully: {result}")
        
        return {
            "result": result,
            "debug_info": kwargs if debug_mode else None,
            "preview": f"Operation completed {'(debug mode)' if debug_mode else ''}"
        }
    
    except Exception as e:
        error_msg = f"Error in plugin: {e}"
        print(error_msg)  # Always log errors
        
        return {
            "error": error_msg,
            "debug_context": kwargs if debug_mode else None,
            "preview": "Plugin execution failed"
        }

Plugin Development Guidelines

Code Style and Structure

  1. Documentation: Include comprehensive docstrings with parameter descriptions
  2. Error Handling: Always handle exceptions gracefully with meaningful error messages
  3. Resource Cleanup: Use try/finally blocks or context managers for resource management
  4. Parameter Validation: Check required parameters and provide clear error messages
  5. Return Standards: Always include a preview key in return dictionaries

Security Considerations

  1. Input Validation: Sanitize and validate all input parameters
  2. Credential Handling: Never log or expose credentials in plain text
  3. Network Security: Use HTTPS where possible and validate SSL certificates
  4. File Access: Restrict file operations to intended directories
  5. Injection Prevention: Avoid using user input directly in system commands

Performance Guidelines

  1. Lazy Loading: Load heavy resources only when needed
  2. Connection Reuse: Use session objects for multiple HTTP requests
  3. Parallel Processing: Use threading for I/O-bound operations
  4. Memory Management: Clean up large objects and close file handles
  5. Timeout Settings: Always set reasonable timeouts for network operations

Troubleshooting Common Issues

Plugin Not Loading

Problem: Plugin doesn't appear in the GUI plugin list Solutions:

  • Ensure the file is in the plugins/ directory
  • Check that the filename has a .py extension
  • Verify the plugin has a run() function
  • Use the "Rescan" button to refresh the plugin list
  • Check for Python syntax errors in the plugin file

Missing Arguments

Problem: Plugin receives None values for expected parameters Solutions:

  • Check parameter names match available context variables
  • Verify credentials are properly configured in environment.ini
  • Ensure previous plugins in the chain provide required data
  • Use the debug_params.py plugin to inspect available context

Authentication Failures

Problem: Plugins fail to authenticate with services Solutions:

  • Verify credential format matches regex patterns in environment.ini
  • Check that "Expose Credentials to Plugins" is enabled in the GUI
  • Test credentials manually before using in plugins
  • Check for API key/token expiration

Network and Proxy Issues

Problem: Network requests fail or proxy connections timeout Solutions:

  • Verify proxy configuration and format
  • Test proxy connectivity with check_proxies.py plugin
  • Implement proper timeout settings in requests
  • Use fallback mechanisms for proxy failures
  • Check firewall and network security settings

Memory and Performance Issues

Problem: Plugins consume excessive memory or run slowly Solutions:

  • Implement proper resource cleanup with finally blocks
  • Use streaming for large file operations
  • Limit concurrent operations and use appropriate thread pools
  • Implement caching for repeated operations
  • Monitor and profile plugin execution times

Best Practices Summary

  • Import only what you need to minimize startup time and memory usage
  • Handle errors gracefully and always return a preview string for the UI
  • Close sessions/clean up in a finally or except block if needed
  • Do not use global state—all state should be passed via arguments/context
  • Test your plugin by adding it to the stack and running it from the GUI
  • Use meaningful parameter names that match the context variables
  • Provide default values for optional parameters
  • Document your plugins with clear docstrings and parameter descriptions
  • Validate inputs and provide helpful error messages
  • Follow naming conventions and use numeric prefixes for execution order

Additional Resources

  • Project Repository: Check the main repository for updates and examples
  • Plugin Examples: Examine existing plugins in the plugins/ directory
  • Debug Tools: Use debug_params.py to understand available context
  • Community Plugins: Look for community-contributed plugins and examples
  • Issue Tracker: Report bugs and request features through the issue tracker

This documentation is continuously updated to reflect the latest features and best practices. For the most current information, always refer to the latest version of this file and examine the actual plugin implementations in the codebase.