Skip to content

New Scanner

Jump to bottom Edit New page
BlackSnufkin edited this page May 20, 2025 · 5 revisions

Adding Custom Scanners to LitterBox

LitterBox is a modular malware analysis framework that provides both static and dynamic analysis capabilities. This guide will walk you through the process of adding your own custom scanner (analyzer) to the project.

Table of Contents

  1. Architecture Overview
  2. Analyzer Inheritance Structure
  3. Creating a New Analyzer
  4. Registering Your Analyzer
  5. Configuration
  6. Best Practices
  7. Examples
  8. Special Analyzer Types

Architecture Overview

LitterBox uses a hierarchical, modular architecture for its analysis capabilities:

app/
└── analyzers/
    ├── base.py         # Base abstract class for all analyzers
    ├── dynamic/        # Analyzers for running processes
    │   ├── base.py     # Base class for dynamic analyzers
    │   ├── hollows_hunter_analyzer.py
    │   ├── hsb_analyzer.py
    │   ├── moneta_analyzer.py
    │   ├── patriot_analyzer.py
    │   ├── pe_sieve_analyzer.py
    │   ├── rededr_analyzer.py
    │   └── yara_analyzer.py
    ├── static/         # Analyzers for files
    │   ├── base.py     # Base class for static analyzers
    │   ├── checkplz_analyzer.py
    │   ├── stringnalyzer_analyzer.py
    │   └── yara_analyzer.py
    └── manager.py      # Manages and coordinates analyzers

The AnalysisManager in manager.py is responsible for initializing, running, and coordinating all analyzer modules.

Analyzer Inheritance Structure

LitterBox uses a 3-level inheritance structure:

  1. BaseAnalyzer (abstract): Defines the basic interface and functionality for all analyzers

    • Initializes with a configuration object
    • Provides shared functionality

  2. StaticAnalyzer/DynamicAnalyzer: Inherit from BaseAnalyzer and specialize for the type of analysis

    • StaticAnalyzer: For analyzing files without execution
    • DynamicAnalyzer: For analyzing running processes

  3. Specific Analyzers: Inherit from StaticAnalyzer or DynamicAnalyzer and implement tool-specific logic

    • Example: YaraStaticAnalyzer, PESieveAnalyzer, etc.

Creating a New Analyzer

All analyzers must implement the following key methods:

  • analyze(target): Perform the actual analysis
  • get_results(): Return the results of the analysis (automatically inherited if you store results in self.results)
  • cleanup(): Release any resources used during analysis

Step 1: Choose the Analyzer Type

Determine whether your analyzer is static (file-based) or dynamic (process-based).

Step 2: Create a New File

Create a new Python file in the appropriate directory:

  • Static analyzers: app/analyzers/static/your_analyzer_name.py
  • Dynamic analyzers: app/analyzers/dynamic/your_analyzer_name.py

Step 3: Implement the Analyzer Class

For Static Analyzers:

# app/analyzers/static/your_analyzer_name.py
import subprocess
import os
from .base import StaticAnalyzer

class YourStaticAnalyzer(StaticAnalyzer):
    def __init__(self, config):
        super().__init__(config)
        # Initialize any analyzer-specific attributes
        # Optional: set up logging
        # self.logger = logging.getLogger("LitterBox")
    
    def analyze(self, file_path):
        """
        Analyze the given file and store results.
        
        Args:
            file_path (str): Path to the file to analyze
        """
        try:
            # Get configuration from the global config
            tool_config = self.config['analysis']['static']['your_analyzer_name']
            
            # Build and execute your analysis command
            command = tool_config['command'].format(
                tool_path=os.path.abspath(tool_config['tool_path']),
                file_path=os.path.abspath(file_path)
                # Add any other parameters you need
            )
            
            # Execute the command
            process = subprocess.Popen(
                command,
                shell=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                universal_newlines=True
            )
            
            stdout, stderr = process.communicate(timeout=tool_config.get('timeout', 300))
            
            # Parse the output
            parsed_results = self._parse_output(stdout)
            
            # Store the results
            self.results = {
                'status': 'completed' if process.returncode == 0 else 'failed',
                'scan_info': {
                    'target': file_path,
                    'tool': 'YourAnalyzer'
                },
                'findings': parsed_results,
                'errors': stderr if stderr else None
            }
            
        except Exception as e:
            self.results = {
                'status': 'error',
                'error': str(e)
            }
    
    def _parse_output(self, output):
        """
        Parse the tool output into structured data.
        
        Args:
            output (str): Raw output from your analysis tool
            
        Returns:
            dict: Structured results
        """
        # Implement your parsing logic here
        parsed_data = {}
        # Your parsing logic
        return parsed_data
    
    def cleanup(self):
        """
        Clean up any resources if needed.
        Most static analyzers don't need cleanup.
        """
        pass

For Dynamic Analyzers:

# app/analyzers/dynamic/your_analyzer_name.py
import subprocess
from .base import DynamicAnalyzer

class YourDynamicAnalyzer(DynamicAnalyzer):
    def __init__(self, config):
        super().__init__(config)
        # Initialize any analyzer-specific attributes
        # self.pid will be set in analyze()
    
    def analyze(self, pid):
        """
        Analyze the process with the given PID.
        
        Args:
            pid (int): Process ID to analyze
        """
        self.pid = pid  # Store the PID
        try:
            # Get configuration
            tool_config = self.config['analysis']['dynamic']['your_analyzer_name']
            
            # Build command
            command = tool_config['command'].format(
                tool_path=tool_config['tool_path'],
                pid=pid
                # Add other parameters as needed
            )
            
            # Execute command
            process = subprocess.Popen(
                command,
                shell=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                universal_newlines=True
            )
            
            stdout, stderr = process.communicate(timeout=tool_config.get('timeout', 300))
            
            # Parse output
            parsed_results = self._parse_output(stdout)
            
            # Store results
            self.results = {
                'status': 'completed' if process.returncode == 0 else 'failed',
                'findings': parsed_results,
                'errors': stderr if stderr else None
            }
            
        except Exception as e:
            self.results = {
                'status': 'error',
                'error': str(e)
            }
    
    def _parse_output(self, output):
        """Parse the tool output into structured data"""
        # Implement your parsing logic
        return {}
    
    def cleanup(self):
        """
        Clean up resources if needed (e.g., stopping processes).
        This is important for dynamic analyzers that may leave
        process artifacts.
        """
        pass

Step 4: Create a Robust Parser

The _parse_output() method is crucial - it converts your tool's raw output into a structured format that LitterBox can use. Study the existing analyzers for examples of how to parse different types of output:

  • Text-based outputs (lines, sections): See HSBAnalyzer._parse_output()
  • JSON outputs: See HollowsHunterAnalyzer.analyze()
  • Complex text parsing: See MonetaAnalyzer._parse_output()

Registering Your Analyzer

To make LitterBox aware of your analyzer, you need to register it in the AnalysisManager class in app/analyzers/manager.py:

  1. Import your analyzer class at the top of the file:
# Import your analyzer
from .static.your_analyzer_name import YourStaticAnalyzer  # For static analyzers
# OR
from .dynamic.your_analyzer_name import YourDynamicAnalyzer  # For dynamic analyzers
  1. Add your analyzer to the appropriate dictionary in the AnalysisManager class:
class AnalysisManager:
    # Define analyzer mappings
    STATIC_ANALYZERS = {
        'yara': YaraStaticAnalyzer,
        'checkplz': CheckPlzAnalyzer,
        'stringnalyzer': StringsAnalyzer,
        'your_analyzer_name': YourStaticAnalyzer  # Add your analyzer here
    }

    DYNAMIC_ANALYZERS = {
        'yara': YaraDynamicAnalyzer,
        'pe_sieve': PESieveAnalyzer,
        'moneta': MonetaAnalyzer,
        'patriot': PatriotAnalyzer,
        'hsb': HSBAnalyzer,
        'rededr': RedEdrAnalyzer,
        'your_analyzer_name': YourDynamicAnalyzer  # Add your analyzer here
    }

The analyzer's key in these dictionaries ('your_analyzer_name') must match the configuration section name you'll create in the next step.

Configuration

Your analyzer needs configuration entries in the LitterBox config file. The structure should match other analyzers:

Static Analyzer Configuration:

{
  "analysis": {
    "static": {
      "your_analyzer_name": {
        "enabled": true,
        "tool_path": "/path/to/your/tool",
        "command": "{tool_path} -argument1 value1 -file {file_path}",
        "timeout": 300,
        "additional_option": "value"
      }
    }
  }
}

Dynamic Analyzer Configuration:

{
  "analysis": {
    "dynamic": {
      "your_analyzer_name": {
        "enabled": true,
        "tool_path": "/path/to/your/tool",
        "command": "{tool_path} -pid {pid} -other_options",
        "timeout": 300,
        "additional_option": "value"
      }
    }
  }
}

The command field is particularly important - it's a template string that gets formatted with values like tool_path and file_path or pid. You can add other parameters as needed.

Best Practices

Based on the existing analyzers in LitterBox, follow these practices for consistent, robust analyzers:

1. Robust Error Handling

Always wrap your primary code in try-except blocks to prevent analyzer failures from affecting other components:

try:
    # Analysis code here
except Exception as e:
    self.results = {
        'status': 'error',
        'error': str(e)
    }

2. Proper Cleanup

Ensure your cleanup() method properly releases any resources:

def cleanup(self):
    # For example, terminating processes
    if hasattr(self, 'tool_process') and self.tool_process:
        try:
            self.tool_process.terminate()
            self.tool_process.wait(timeout=5)
        except:
            # Force kill if necessary
            self.tool_process.kill()

3. Consistent Result Structure

Follow the result structure pattern seen in existing analyzers:

self.results = {
    'status': 'completed',  # or 'failed', 'error'
    'scan_info': {          # Context about the scan
        'target': target,
        'tool': 'YourTool',
        # Other scan metadata
    },
    'findings': {           # Actual analysis results
        # Tool-specific structure, but include:
        'summary': {},      # Summary stats when applicable
        'details': []       # Detailed findings
    },
    'errors': stderr if stderr else None
}

4. Detailed Parsing

Invest time in the _parse_output() method to provide high-quality structured data. Use appropriate techniques:

def _parse_output(self, output):
    # Example parsing approach for structured tool output
    parsed_results = {
        'summary': {
            'total_findings': 0,
            'critical_findings': 0
        },
        'findings': []
    }
    
    # Parse sections, structured data, etc.
    for line in output.splitlines():
        if line.startswith('FINDING:'):
            finding = self._parse_finding_line(line)
            parsed_results['findings'].append(finding)
            parsed_results['summary']['total_findings'] += 1
            if finding.get('severity') == 'critical':
                parsed_results['summary']['critical_findings'] += 1
    
    return parsed_results

5. Documentation

Add detailed docstrings and comments to explain your analyzer's functionality:

class YourAnalyzer(StaticAnalyzer):
    """
    YourAnalyzer implements analysis of files using YourTool.
    
    This analyzer detects [specific behaviors/patterns] by executing
    YourTool against the target file and parsing the structured output.
    
    Capabilities:
    - Feature 1
    - Feature 2
    """

Examples

Let's look at some examples based on the actual implementation patterns in LitterBox:

Example 1: Simple Static File Scanner

This example demonstrates a simple static analyzer that checks files for suspicious API imports:

# app/analyzers/static/api_scanner.py
import subprocess
import re
import os
from .base import StaticAnalyzer

class ApiScannerAnalyzer(StaticAnalyzer):
    """
    Analyzer that detects suspicious Win32 API imports in executables.
    """
    
    def __init__(self, config):
        super().__init__(config)
        # Define suspicious API categories
        self.api_categories = {
            'process_injection': [
                'VirtualAllocEx', 'WriteProcessMemory', 'CreateRemoteThread',
                'NtCreateThreadEx', 'QueueUserAPC'
            ],
            'keylogging': [
                'GetAsyncKeyState', 'GetKeyState', 'SetWindowsHookEx'
            ],
            'persistence': [
                'RegSetValueEx', 'RegCreateKeyEx'
            ]
        }
    
    def analyze(self, file_path):
        try:
            tool_config = self.config['analysis']['static']['api_scanner']
            
            # Use a tool to extract imports (e.g., dumpbin or objdump)
            command = tool_config['command'].format(
                tool_path=os.path.abspath(tool_config['tool_path']),
                file_path=os.path.abspath(file_path)
            )
            
            process = subprocess.Popen(
                command,
                shell=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                universal_newlines=True
            )
            
            stdout, stderr = process.communicate(timeout=tool_config.get('timeout', 300))
            
            # Parse the output to find API imports
            findings = self._parse_output(stdout)
            
            self.results = {
                'status': 'completed' if process.returncode == 0 else 'failed',
                'scan_info': {
                    'target': file_path,
                    'tool': 'ApiScanner'
                },
                'findings': findings,
                'errors': stderr if stderr else None
            }
            
        except Exception as e:
            self.results = {
                'status': 'error',
                'error': str(e)
            }
    
    def _parse_output(self, output):
        findings = {
            'summary': {
                'total_suspicious_apis': 0,
                'categories': {}
            },
            'suspicious_apis': []
        }
        
        # Initialize category counters
        for category in self.api_categories:
            findings['summary']['categories'][category] = 0
        
        # Extract all API calls from the output
        apis = []
        for line in output.splitlines():
            if "Import:" in line or "Imports:" in line:
                # Extract API name, might be in various formats depending on the tool
                match = re.search(r'[A-Za-z0-9_]+\.(dll|DLL|Dll)!([A-Za-z0-9_]+)', line)
                if match:
                    apis.append(match.group(2))
        
        # Check if any APIs are in our suspicious list
        for api in apis:
            for category, api_list in self.api_categories.items():
                if api in api_list:
                    findings['suspicious_apis'].append({
                        'api': api,
                        'category': category,
                        'description': f"Potential {category} functionality"
                    })
                    findings['summary']['total_suspicious_apis'] += 1
                    findings['summary']['categories'][category] += 1
        
        return findings

Example 2: Dynamic Memory Scanner

This example demonstrates a dynamic analyzer that scans a process's memory for suspicious patterns:

# app/analyzers/dynamic/memory_scanner.py
import subprocess
import json
import logging
from .base import DynamicAnalyzer

class MemoryScannerAnalyzer(DynamicAnalyzer):
    def __init__(self, config):
        super().__init__(config)
        self.logger = logging.getLogger("LitterBox")
    
    def analyze(self, pid):
        self.pid = pid
        try:
            tool_config = self.config['analysis']['dynamic']['memory_scanner']
            command = tool_config['command'].format(
                tool_path=tool_config['tool_path'],
                pid=pid,
                scan_depth=tool_config.get('scan_depth', 'normal')
            )
            
            process = subprocess.Popen(
                command,
                shell=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                universal_newlines=True
            )
            
            stdout, stderr = process.communicate(timeout=tool_config.get('timeout', 300))
            
            # Try to parse JSON output, fall back to text parsing if needed
            try:
                results_json = json.loads(stdout)
                self.results = {
                    'status': 'completed',
                    'findings': results_json,
                    'errors': stderr if stderr else None
                }
            except json.JSONDecodeError:
                # If output isn't JSON, parse it as text
                parsed_results = self._parse_text_output(stdout)
                self.results = {
                    'status': 'completed',
                    'findings': parsed_results,
                    'errors': stderr if stderr else None
                }
                
        except Exception as e:
            self.logger.error(f"Error in MemoryScanner analysis: {str(e)}")
            self.results = {
                'status': 'error',
                'error': str(e)
            }
    
    def _parse_text_output(self, output):
        findings = {
            'memory_regions': [],
            'suspicious_regions': [],
            'summary': {
                'total_regions': 0,
                'suspicious_count': 0,
                'rwx_regions': 0,
                'hidden_regions': 0
            }
        }
        
        current_region = None
        
        for line in output.splitlines():
            line = line.strip()
            
            if line.startswith('Region:'):
                # Start of new memory region
                if current_region:
                    findings['memory_regions'].append(current_region)
                    if current_region.get('is_suspicious'):
                        findings['suspicious_regions'].append(current_region)
                
                # Parse region address
                region_data = line.replace('Region:', '').strip()
                address_match = re.search(r'(0x[0-9A-Fa-f]+)', region_data)
                
                current_region = {
                    'address': address_match.group(1) if address_match else 'unknown',
                    'flags': [],
                    'is_suspicious': False,
                    'details': []
                }
                
                findings['summary']['total_regions'] += 1
                
            elif line.startswith('Flags:') and current_region:
                # Parse memory protection flags
                flags = line.replace('Flags:', '').strip()
                current_region['flags'] = [f.strip() for f in flags.split(',')]
                
                # Check for suspicious flags
                if 'RWX' in flags:
                    current_region['is_suspicious'] = True
                    current_region['details'].append('Read-Write-Execute memory region')
                    findings['summary']['rwx_regions'] += 1
                    findings['summary']['suspicious_count'] += 1
            
            elif line.startswith('Hidden:') and current_region:
                if 'Yes' in line:
                    current_region['is_suspicious'] = True
                    current_region['details'].append('Hidden memory region')
                    findings['summary']['hidden_regions'] += 1
                    findings['summary']['suspicious_count'] += 1
        
        # Add the last region if exists
        if current_region:
            findings['memory_regions'].append(current_region)
            if current_region.get('is_suspicious'):
                findings['suspicious_regions'].append(current_region)
        
        return findings
    
    def cleanup(self):
        # No cleanup needed for this analyzer
        pass

Special Analyzer Types

Some analyzers in LitterBox have special functionality that requires additional consideration:

1. RedEdr Analyzer

The RedEdr analyzer is different from other dynamic analyzers because it uses a continuous monitoring approach instead of a one-shot analysis. If you want to create a similar monitoring tool:

  1. Implement a start_tool() method to begin monitoring
  2. Use a separate thread to collect output continuously
  3. Provide a specific get_results() implementation that processes all collected data
# Simplified example of a monitoring analyzer
def start_tool(self, target_name):
    """Start the monitoring tool"""
    self._stop_monitoring = False
    self.monitoring_thread = threading.Thread(target=self._monitor_thread)
    self.monitoring_thread.daemon = True
    self.monitoring_thread.start()
    return True

def _monitor_thread(self):
    """Background thread to monitor and collect data"""
    while not self._stop_monitoring:
        # Collect data
        new_data = self._collect_new_data()
        if new_data:
            with self._lock:
                self.collected_data.append(new_data)
        time.sleep(0.1)  # Polling interval

def get_results(self):
    """Process all collected data and return results"""
    with self._lock:
        processed_data = self._process_data(self.collected_data)
    return processed_data

def cleanup(self):
    """Stop monitoring and clean up resources"""
    self._stop_monitoring = True
    if self.monitoring_thread:
        self.monitoring_thread.join(timeout=2)
    # Clean up other resources

2. YARA Analyzers

YARA is used in both static and dynamic analysis. The key differences are:

  • Static: Scans a file on disk
  • Dynamic: Scans a process's memory

If you're creating a YARA-style analyzer that uses pattern matching:

  1. Create two separate analyzers (static and dynamic) that share common parsing code
  2. Ensure your rule files are properly loaded and parsed
  3. Provide detailed metadata about matched patterns, including severity and descriptions

Repo · Releases · CHANGELOG · Issues · GPLv3

📌 LitterBox · self-hosted payload analysis sandbox

Release

🚀 Getting Started

📊 Pipelines & Pages

🔬 Scanners · 4 modules

🛰️ EDR Integration
🔌 API & Clients
⚙️ Configuration & Dev

Clone this wiki locally