enrichment-module-builder

# Enrichment Module Builder Skill This skill guides the creation of new Nemesis enrichment modules from start to finish. **CRITICAL: At each user approval gate (Steps 2, 3, 4, and 8), prompt for explicit user approval before proceeding. Use `request_user_input` when available; otherwise ask a direct concise question in chat and wait for an explicit approval response. Do NOT continue past a gate until the user has explicitly approved.** ## Overview Enrichment modules analyze files and extract security-relevant information like credentials, hashes, metadata, and indicators of compromise. This skill walks through the complete process: 1. Problem Analysis 2. Module Output Mode (with user approval gate) 3. Library Research (with user approval gate) 4. Sample File Acquisition (with user approval gate) 5. Detection Strategy 6. Module Implementation 7. Standalone Testing 8. Integration Testing (with user approval gate) - **REQUIRED** ## Reference Documentation Before starting, review: - **Development Guide:** `libs/file_enrichment_modules/DEVELOPMENT_GUIDE.md` - **Test Harness:** `libs/file_enrichment_modules/tests/harness/` ## Reference Modules Use these 8 modules as implementation references - they cover all major patterns: | Module | Detection Pattern | Key Feature | |--------|------------------|-------------| | `pe` | Magic + YARA | Complex parsing with lief | | `yara` | All files | YARA rule management | | `chromium_cookies` | Magic + YARA + filename | Database + DPAPI | | `gitcredentials` | Filename + plaintext | Simple text parsing | | `group_policy_preferences` | YARA + plaintext | XML + crypto | | `container` | is_container() | Archive handling | | `keytab` | Extension OR YARA | Binary struct parsing | | `office_doc` | Extension OR magic | Multi-format handling | Paths: `libs/file_enrichment_modules/file_enrichment_modules/{module_name}/` --- ## Step 1: Problem Analysis Gather requirements from the user: 1. **Target file type/format:** What files should this module process? 2. **Data to extract:** What information should be extracted? - Credentials (usernames, passwords, tokens) - Hashes (password hashes, encryption keys) - Metadata (configuration, version info) - Security indicators 3. **Finding categories:** Which apply? - CREDENTIAL, EXTRACTED_HASH, EXTRACTED_DATA, VULNERABILITY, YARA_MATCH, PII, MISC, INFORMATIONAL 4. **Severity level:** 0-10 based on security impact **Questions to ask:** - What file types/extensions/names identify target files? - What specific data fields need extraction? - Are there multiple variants of this file format? - Should the module produce transforms (derived files) in addition to findings? --- ## Step 2: Module Output Mode [GATE 1] Determine what the module should produce as output: ### Output Mode Options 1. **Findings Mode:** The module extracts security-relevant data and generates findings - Use when: Extracting credentials, hashes, vulnerabilities, or other actionable security data - Output: Findings with categories (CREDENTIAL, EXTRACTED_HASH, etc.) and severity levels - Example modules: `chromium_cookies`, `gitcredentials`, `group_policy_preferences` 2. **Parsing-Only Mode:** The module parses the file and stores structured data without generating findings - Use when: Extracting metadata, configuration, or informational data for display/search - Output: Structured results stored in the database, no findings generated - Example modules: `pe` (extracts PE metadata), `office_doc` (extracts document metadata) 3. **Hybrid Mode:** The module parses data AND generates findings for specific conditions - Use when: Most data is informational, but certain patterns warrant findings - Output: Structured results plus conditional findings - Example: Parse all PE metadata, but generate finding only if unsigned or suspicious ### Present to User **Format your recommendation:** ``` ## Module Output Mode for {file_type} Module Based on the data to be extracted, I recommend: ### Recommended: {Findings Mode | Parsing-Only Mode | Hybrid Mode} **Rationale:** {why this mode fits the use case} ### What this means: - {description of what will be produced} - {how data will be stored/displayed} - {whether alerts will be generated} ### Alternative consideration: {brief note on why other modes might or might not apply} **Do you approve this output mode, or would you prefer a different approach?** ``` **STOP: Ask the user to approve one of the three output mode options (Findings Mode, Parsing-Only Mode, Hybrid Mode) before proceeding to Step 3. Prefer `request_user_input` when available.** --- ## Step 3: Library Research [GATE 2] Search for parsing libraries before implementation: ### Research Steps 1. **Search PyPI** for relevant parsing libraries: - Search terms: "{file_format} parser python", "{file_format} python library" - Evaluate: popularity (downloads), maintenance status, API quality 2. **Search GitHub** for reference implementations: - Look for existing parsers, security tools, CTF write-ups - Check for format documentation 3. **Evaluate options:** - Does the library handle the specific format variant? - Is it actively maintained? - Does it have security-relevant features? - What's the API complexity? ### Present to User **Format your recommendation:** ``` ## Library Recommendation for {file_type} Module ### Recommended: {library_name} - **PyPI:** https://pypi.org/project/{library_name}/ - **GitHub:** {github_url} - **Why:** {reasons - API quality, maintenance, features} - **Downloads:** {monthly_downloads} ### Alternatives Considered: 1. {alt_library_1} - {why_not_chosen} 2. {alt_library_2} - {why_not_chosen} ### Manual Parsing If no good library exists, we can implement manual parsing using: - struct module for binary formats - xml.etree for XML - Regular expressions for text patterns **Do you approve this library choice, or would you prefer an alternative?** ``` **STOP: Present the recommended library and alternatives and get explicit user approval before proceeding to Step 4. Prefer `request_user_input` when available.** --- ## Step 4: Sample File Acquisition [GATE 3] Obtain test files for development and testing: ### Search Locations 1. **Public GitHub repos:** Search for sample files (<100MB) - Query: `"{file_extension}" OR "{file_type} sample"` - Look in security research repos, CTF repos, test fixtures 2. **Sample file repositories:** - file-examples.com - filesamples.com - Sample files in related tool repos 3. **Generate synthetic files:** - If no public samples exist, create test files - Document the generation method ### Present to User **Format your recommendation:** ``` ## Sample File for {file_type} Module ### Source: {source_description} - **URL/Location:** {url_or_path} - **File:** {filename} - **Size:** {size} - **Why suitable:** {reasons} ### Alternative sources if needed: 1. {alt_source_1} 2. {alt_source_2} ### Synthetic generation (if no public samples): {description of how to create test file} **Do you approve this sample file source, or do you have an alternative?** ``` **STOP: Present sample file options and get explicit user approval before proceeding to Step 5. Prefer `request_user_input` when available.** --- ## Step 5: Detection Strategy Determine how `should_process()` will identify target files: ### Analyze the Sample File 1. **Check magic type:** Run `file` command on sample 2. **Check MIME type:** What MIME type does Nemesis assign? 3. **Identify binary signatures:** Look for distinctive headers/magic bytes 4. **Check filenames/extensions:** Are there standard naming conventions? ### Choose Detection Method Based on analysis, select from: 1. **Magic/MIME type:** For files with distinctive signatures 2. **File extension:** For convention-based identification 3. **Filename:** For config files with specific names 4. **YARA rule:** For binary patterns 5. **Combined:** For higher confidence ### Generate YARA Rule (if needed) If the file has distinctive binary signatures: ```yara rule {file_type}_file { meta: description = "Detects {file_type} files" strings: $header = { XX XX XX XX } // Magic bytes condition: $header at 0 } ``` --- ## Step 6: Module Implementation Create the module structure: ### 1. Create Directory ```bash mkdir -p libs/file_enrichment_modules/file_enrichment_modules/{module_name} ``` ### 2. Create analyzer.py Use this template, adapting based on the reference module that matches your pattern: ```python # enrichment_modules/{module_name}/analyzer.py from common.logger import get_logger from common.models import EnrichmentResult, FileObject, Finding, FindingCategory, FindingOrigin, Transform from common.state_helpers import get_file_enriched_async from common.storage import StorageS3 from file_enrichment_modules.module_loader import EnrichmentModule logger = get_logger(__name__) class {ModuleName}Analyzer(EnrichmentModule): name: str = "{module_name}_analyzer" dependencies: list[str] = [] def __init__(self): self.storage = StorageS3() self.asyncpg_pool = None self.workflows = ["default"] async def should_process(self, object_id: str, file_path: str | None = None) -> bool: """Determine if this module should process the file.""" file_enriched = await get_file_enriched_async(object_id, self.asyncpg_pool) # TODO: Implement detection logic return False def _analyze_file(self, file_path: str, file_enriched) -> EnrichmentResult | None: """Analyze the file and extract data.""" result = EnrichmentResult(module_name=self.name, dependencies=self.dependencies) try: # TODO: Implement parsing logic # Create findings if relevant data found # Create transforms for derived files return result except Exception: logger.exception(message=f"Error analyzing {file_enriched.file_name}") return None async def process(self, object_id: str, file_path: str | None = None) -> EnrichmentResult | None: """Process the file.""" try: file_enriched = await get_file_enriched_async(object_id, self.asyncpg_pool) if file_path: return self._analyze_file(file_path, file_enriched) else: with self.storage.download(object_id) as temp_file: return self._analyze_file(temp_file.name, file_enriched) except Exception: logger.exception(message="Error in process()") return None def create_enrichment_module() -> EnrichmentModule: return {ModuleName}Analyzer() ``` ### 3. Create pyproject.toml (if custom deps needed) ```toml [project] name = "{module_name}" version = "0.1.0" requires-python = ">=3.12" dependencies = [ "{library_name}>=X.Y.Z", ] [build-system] requires = ["hatchling"] build-backend = "hatchling.build" ``` ### 4. Create YARA rules (if using YARA detection) Create `rules.yar` with detection rules. --- ## Step 7: Standalone Testing Create and run tests using the test harness: ### Create Test File ```python # tests/test_{module_name}.py import pytest from tests.harness import ModuleTestHarness, FileEnrichedFactory from file_enrichment_modules.{module_name}.analyzer import {ModuleName}Analyzer class Test{ModuleName}Analyzer: """Tests for {ModuleName}Analyzer.""" @pytest.mark.asyncio async def test_should_process_target_file(self): """Test that should_process returns True for target files.""" harness = ModuleTestHarness() harness.register_file( object_id="test-uuid", local_path="/path/to/sample/file", file_enriched=FileEnrichedFactory.create( object_id="test-uuid", file_name="sample.ext", magic_type="expected magic type", # ... other fields ), ) async with harness.create_module({ModuleName}Analyzer) as module: result = await module.should_process("test-uuid") assert result is True @pytest.mark.asyncio async def test_should_not_process_unrelated_file(self): """Test that should_process returns False for unrelated files.""" harness = ModuleTestHarness() harness.register_file( object_id="test-uuid", local_path="/path/to/unrelated/file", file_enriched=FileEnrichedFactory.create_plaintext_file( object_id="test-uuid", file_name="readme.txt", ), ) async with harness.create_module({ModuleName}Analyzer) as module: result = await module.should_process("test-uuid") assert result is False @pytest.mark.asyncio async def test_process_extracts_expected_data(self): """Test that process extracts the expected data.""" harness = ModuleTestHarness() harness.register_file( object_id="test-uuid", local_path="/path/to/sample/file", file_enriched=FileEnrichedFactory.create(...), ) async with harness.create_module({ModuleName}Analyzer) as module: result = await module.process("test-uuid") assert result is not None assert result.module_name == "{module_name}_analyzer" # Assert on findings # Assert on transforms # Assert on results ``` Guidelines: - Test both positive(happy path) and negative(unhappy path) cases. - Do this for all result types, findings, transforms, DB writes, and file uploads generated. - Handle edge cases and error conditions ### Run Tests ```bash cd libs/file_enrichment_modules