A memory image taken from a live system is the closest thing to a complete picture of what was happening on that machine at a specific moment in time. Everything that was running, communicating, injected, or hiding is captured in those bytes. The challenge is that extracting useful intelligence from a raw memory image manually is slow. Running Volatility plugins one at a time, copying outputs into notes, correlating findings by hand. A thorough manual analysis of a single 16GB image can take three to four hours for an experienced analyst. When you are hunting across multiple endpoints, or when you need to answer the question “is anything malicious in this image” quickly enough to act on the result, manual analysis does not scale.<\/p>\n
This post covers building a single script that accepts a memory image path and produces a comprehensive, prioritised threat hunting report without any further analyst input. You drop an image in, you get a report out. The script handles plugin selection, parallel execution, result parsing, anomaly detection, IOC extraction, cross-referencing between plugins, and report generation. It handles both Windows and Linux images, uses Volatility 2 where it provides unique capability and Volatility 3 for everything else, and produces both a shareable HTML report and structured JSON for SIEM ingestion.<\/p>\n
The Volatility project has two major versions in active use. Volatility 3 is the current version, written from scratch with a cleaner architecture, better Python 3 support, and no dependency on a manually selected profile. Volatility 2 is the older version, maintained for specific use cases where it still provides capabilities or plugin depth that Volatility 3 does not yet match. Understanding the practical differences between them determines which one to call for each analysis task.<\/p>\n
Volatility 3 should be your default for almost everything. It automatically identifies the OS version and kernel structures from the image without requiring a manually matched profile, which eliminates the most common source of beginner errors (wrong profile). It runs significantly faster on large images because of improved memory mapping. The output format is consistent JSON across all plugins, which makes automation straightforward. The symbol table system is more robust than Volatility 2’s profile system for modern Windows versions.<\/p>\n
Volatility 2 retains specific advantages that keep it relevant for threat hunters:<\/p>\n
# Where Volatility 2 still provides unique value\n\n# 1. Some plugins simply do not exist yet in Volatility 3\n# - mftparser: deep MFT parsing from memory\n# - shellbags: user navigation history from registry in memory\n# - iehistory: Internet Explorer artefacts from process memory\n# - shimcache: application compatibility cache from memory\n# - prefetchparser: prefetch execution records from memory\n\n# 2. Certain Linux plugins are more mature in Volatility 2\n# - linux_psaux: full command line for Linux processes (more reliable)\n# - linux_netstat: Linux network connections with more detail\n# - linux_check_syscall: syscall table hook detection\n\n# 3. Legacy Windows support (XP, Vista, 2003, 2008)\n# Volatility 3 symbol tables may not exist for very old OS versions\n# Volatility 2 with the correct profile covers these reliably\n\n# 4. Certain rootkit detection plugins\n# - modscan: raw kernel module scanner (sometimes finds what pslist misses)\n# - ssdt: SSDT hook detection (useful on older Windows)\n# - apihooks: API hook detection in process memory\n\n# Check what is available in each version\nvol2 --info | grep \"^[A-Z]\" | sort # Volatility 2 plugins\nvol3 --help 2>&1 | grep \"windows\\.\" | sort # Volatility 3 Windows plugins<\/pre>\nThe automation script uses both. Volatility 3 for the core analysis pipeline where it is faster and more reliable. Volatility 2 for specific supplementary plugins where it provides unique value. The script detects which tools are available and gracefully skips plugins that cannot run rather than failing entirely.<\/p>\n
Setting up both versions side by side<\/h3>\n
#!\/bin\/bash\n## setup_volatility_dual.sh\n## Install Volatility 2 and 3 side by side\n## Ubuntu 22.04 - run as root or with sudo\n\nset -euo pipefail\n\necho \"[*] Setting up dual Volatility environment\"\n\n## \u2500\u2500 Volatility 3 \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\npython3 -m venv \/opt\/vol3-env\nsource \/opt\/vol3-env\/bin\/activate\npip install --upgrade pip -q\npip install volatility3 yara-python pefile capstone python-magic requests -q\n\n# Symbol tables for Windows analysis\nmkdir -p \/opt\/vol3-symbols\nif [ ! -d \/opt\/vol3-symbols\/windows ]; then\n echo \"[*] Downloading Windows symbol tables (~500MB)\"\n wget -q --show-progress \\\n https:\/\/downloads.volatilityfoundation.org\/volatility3\/symbols\/windows.zip \\\n -O \/tmp\/windows_syms.zip\n unzip -q \/tmp\/windows_syms.zip -d \/opt\/vol3-symbols\/\n rm \/tmp\/windows_syms.zip\nfi\n\n# Link symbols into Volatility 3\nSITE3=$(python3 -c \"import site; print(site.getsitepackages()[0])\")\nln -sf \/opt\/vol3-symbols\/windows $SITE3\/volatility3\/symbols\/windows 2>\/dev\/null || true\n\n# Linux symbol tables (requires ISF files generated from target kernel)\n# These must be generated for each specific kernel version being analysed\n# See: https:\/\/github.com\/volatilityfoundation\/dwarf2json\nmkdir -p \/opt\/vol3-symbols\/linux\n\necho \"[*] Volatility 3 installed: $(vol)\"\ndeactivate\n\n## \u2500\u2500 Volatility 2 \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n# Volatility 2 requires Python 2.7 (use pyenv or system python2)\napt-get install -y -qq python2 python2-dev python-is-python2 2>\/dev\/null || \\\n apt-get install -y -qq python2.7 python2.7-dev 2>\/dev\/null || true\n\n# If python2 is not available, use Docker instead\nif command -v python2 &>\/dev\/null; then\n pip2 install --quiet distorm3 pycryptodome 2>\/dev\/null || true\n git clone --depth 1 https:\/\/github.com\/volatilityfoundation\/volatility.git \\\n \/opt\/volatility2 2>\/dev\/null || (cd \/opt\/volatility2 && git pull)\n echo \"[+] Volatility 2: available via python2 \/opt\/volatility2\/vol.py\"\nelse\n echo \"[!] Python 2 not available - using Docker for Volatility 2\"\n docker pull remnux\/volatility 2>\/dev\/null || true\n cat > \/usr\/local\/bin\/vol2 << 'VOL2SCRIPT'\n#!\/bin\/bash\ndocker run --rm -v \"$(dirname $(realpath )):\/data\" \\\n remnux\/volatility \"$@\"\nVOL2SCRIPT\n chmod +x \/usr\/local\/bin\/vol2\nfi\n\n## \u2500\u2500 Supporting tools \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\nsource \/opt\/vol3-env\/bin\/activate\npip install -q \\\n jinja2 \\\n requests \\\n psycopg2-binary \\\n colorama \\\n tqdm \\\n tabulate\ndeactivate\n\n# Install yara, foremost, jq from apt\napt-get install -y -qq yara foremost jq\n\necho \"\"\necho \"[+] Setup complete\"\necho \" Volatility 3: vol (in \/opt\/vol3-env\/bin\/)\"\necho \" Volatility 2: python2 \/opt\/volatility2\/vol.py (or vol2 Docker wrapper)\"\necho \" Symbol tables: \/opt\/vol3-symbols\/\"\n<\/pre>\nThe automation architecture<\/h3>\n
The script runs in five sequential phases. Each phase feeds its results into the next. Phase 1 identifies the image and OS type. Phase 2 runs a broad plugin sweep to collect all available data. Phase 3 applies anomaly detection logic across the collected data to score and prioritise findings. Phase 4 extracts and enriches IOCs. Phase 5 generates the output reports. The whole process runs in parallel where possible, with non-dependent plugins executing concurrently, which reduces total analysis time significantly compared to sequential execution.<\/p>\n
## File: analyse.py\n#!\/usr\/bin\/env python3\n## \/opt\/memory-hunter\/analyse.py\n## THE ONE SCRIPT TO RUN\n## Usage: python3 analyse.py \/path\/to\/memory.raw\n## Usage: python3 analyse.py \/path\/to\/memory.raw --workers 16 --no-vol2\n\nimport sys\nimport argparse\nimport logging\nimport time\nimport json\nfrom datetime import datetime\nfrom pathlib import Path\n\n# Add scripts directory to path\nsys.path.insert(0, str(Path(__file__).parent \/ 'scripts'))\n\nfrom phase1_identify import identify_image\nfrom vol_runner import collect_all_plugins\nfrom anomaly_checks import run_all_checks\nfrom phase4_iocs import extract_iocs\nfrom phase5_report import generate_reports\n\ndef setup_logging(log_file: str = None) -> None:\n handlers = [logging.StreamHandler(sys.stdout)]\n if log_file:\n handlers.append(logging.FileHandler(log_file))\n logging.basicConfig(\n level=logging.INFO,\n format='%(asctime)s [%(levelname)s] %(message)s',\n handlers=handlers\n )\n\ndef print_banner():\n print(\"\"\"\n\u2554\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2557\n\u2551 Memory Hunter - Automated Analysis \u2551\n\u2551 Volatility 2 + 3 | Windows + Linux | HTML + JSON \u2551\n\u255a\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u255d\n\"\"\")\n\ndef print_summary(report_info: dict, elapsed: float):\n risk = report_info['risk_label']\n colour = {\n 'CRITICAL': '3[91m', 'HIGH': '3[93m',\n 'MEDIUM': '3[33m', 'LOW': '3[92m'\n }.get(risk, '')\n reset = '3[0m'\n\n print(f\"\"\"\n{'='*60}\n ANALYSIS COMPLETE ({elapsed:.1f} seconds)\n{'='*60}\n Risk Level: {colour}{risk}{reset}\n Risk Score: {report_info['risk_score']}\n Findings: {report_info['finding_count']}\n HTML Report: {report_info['html']}\n JSON Report: {report_info['json']}\n{'='*60}\n\"\"\")\n\ndef main():\n print_banner()\n\n parser = argparse.ArgumentParser(\n description='Automated memory image analysis for threat hunters'\n )\n parser.add_argument('image', help='Path to memory image file')\n parser.add_argument('--workers', type=int, default=8,\n help='Parallel plugin workers (default: 8)')\n parser.add_argument('--no-vol2', action='store_true',\n help='Skip Volatility 2 plugins')\n parser.add_argument('--output-dir', default='\/opt\/memory-hunter\/reports',\n help='Report output directory')\n parser.add_argument('--yara-rules', default='\/opt\/memory-hunter\/yara_rules\/combined.yar',\n help='Path to compiled Yara rules')\n parser.add_argument('--log-file', help='Write log to file')\n parser.add_argument('--quiet', action='store_true',\n help='Reduce output verbosity')\n args = parser.parse_args()\n\n setup_logging(args.log_file)\n log = logging.getLogger(__name__)\n\n start_time = time.time()\n image_path = str(Path(args.image).absolute())\n\n print(f\"[*] Image: {image_path}\")\n print(f\"[*] Workers: {args.workers}\")\n print()\n\n # \u2500\u2500 Phase 1: Identify \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n print(\"[Phase 1\/5] Identifying image...\")\n meta = identify_image(image_path)\n print(f\" OS Type: {meta.get('os_type', 'unknown')}\")\n print(f\" OS Version: {meta.get('os_version', 'unknown')}\")\n print(f\" Vol2 Profile: {meta.get('vol2_profile', 'NOT DETECTED - Vol2 plugins will not run')}\")\n print(f\" Size: {meta.get('size_gb', 0):.1f} GB\")\n print(f\" SHA256: {meta.get('sha256', '')[:32]}...\")\n if not meta.get('vol2_ok'):\n print(f\" WARNING: Vol2 not available. For older images (Win XP\/7\/Vista)\")\n print(f\" install Vol2 and run: vol2 -f <image> imageinfo\")\n print()\n\n if not meta.get('os_type'):\n print(\"[!] Could not identify OS type - check symbol tables\")\n print(\" Windows: ensure \/opt\/vol3-symbols\/windows\/ is populated\")\n print(\" Linux: ensure ISF file exists for this kernel version\")\n sys.exit(1)\n\n # \u2500\u2500 Phase 2: Collect \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n print(f\"[Phase 2\/5] Running plugin collection ({args.workers} workers)...\")\n plugin_results = collect_all_plugins(image_path, meta, args.workers)\n print(f\" Plugins completed: {len(plugin_results)}\")\n print()\n\n # \u2500\u2500 Phase 3: Detect \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n print(\"[Phase 3\/5] Running anomaly detection...\")\n findings, risk_score = run_all_checks(plugin_results)\n critical = sum(1 for f in findings if f.severity == 'CRITICAL')\n high = sum(1 for f in findings if f.severity == 'HIGH')\n print(f\" Findings: {len(findings)} total ({critical} critical, {high} high)\")\n if findings:\n print(\" Top findings:\")\n for f in findings[:5]:\n print(f\" [{f.severity}] {f.title}\")\n print()\n\n # \u2500\u2500 Phase 4: IOCs \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n print(\"[Phase 4\/5] Extracting IOCs...\")\n suspicious_pids = [f.pid for f in findings if f.pid]\n iocs = extract_iocs(plugin_results, image_path, suspicious_pids)\n ioc_count = sum(len(v) for v in iocs.values())\n print(f\" IOCs extracted: {ioc_count}\")\n if iocs.get('yara_hits'):\n print(f\" YARA hits: {len(iocs['yara_hits'])}\")\n for hit in iocs['yara_hits'][:3]:\n print(f\" -> {hit}\")\n print()\n\n # \u2500\u2500 Phase 5: Report \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n print(\"[Phase 5\/5] Generating reports...\")\n report_info = generate_reports(meta, findings, iocs, plugin_results)\n elapsed = time.time() - start_time\n\n print_summary(report_info, elapsed)\n print(f\" summary.txt: {report_info.get('summary', '')}\")\n print(f\" iocs.txt: {report_info.get('iocs', '')}\")\n\n # Exit code reflects risk level for CI\/CD integration\n exit_codes = {'CRITICAL': 3, 'HIGH': 2, 'MEDIUM': 1, 'LOW': 0}\n sys.exit(exit_codes.get(report_info['risk_label'], 0))\n\nif __name__ == '__main__':\n main()\n<\/pre>\nPhase 1: image identification and OS detection<\/h3>\n
## File: phase1_identify.py\n#!\/usr\/bin\/env python3\n\"\"\"\nPhase 1: Identify memory image OS type and version.\nTries Volatility 3 first, then Volatility 2 imageinfo for profile detection.\nVol2 profile is required for older images (XP\/Vista\/Win7) where Vol3 symbols\ndo not fully match, and for enabling Vol2-only plugins.\n\"\"\"\n\nimport subprocess\nimport json\nimport logging\nimport hashlib\nimport shutil\nfrom pathlib import Path\nfrom typing import Dict\n\nlog = logging.getLogger(__name__)\n\nVOL3_CMD = '\/opt\/vol3-env\/bin\/vol'\nVOL2_CMD = '\/opt\/volatility2\/vol.py'\n\ndef sha256_file(path: str) -> str:\n h = hashlib.sha256()\n with open(path, 'rb') as f:\n for chunk in iter(lambda: f.read(65536), b''):\n h.update(chunk)\n return h.hexdigest()\n\ndef run_vol2_imageinfo(image_path: str) -> str:\n \"\"\"\n Run vol2 imageinfo and return the first suggested profile string.\n Handles the exact Vol2 output format:\n INFO : volatility.debug : Suggested Profile(s) : Win7SP1x64, Win7SP0x64\n Output goes to STDERR in Vol2 - we capture both stdout and stderr.\n Returns empty string if profile cannot be determined.\n \"\"\"\n # Find vol2 - check wrapper script first, then direct python2 call\n vol2_bin = shutil.which('vol2')\n if not vol2_bin and Path(VOL2_CMD).exists():\n py2 = shutil.which('python2') or shutil.which('python2.7')\n if not py2:\n log.warning(\"vol2 not found and python2\/python2.7 not in PATH\")\n return ''\n\n if not vol2_bin and not (shutil.which('python2') or shutil.which('python2.7')):\n log.warning(\"vol2 not available\")\n return ''\n\n cmd = ['vol2', '-f', image_path, 'imageinfo'] if vol2_bin else [\n shutil.which('python2') or shutil.which('python2.7'),\n VOL2_CMD, '-f', image_path, 'imageinfo'\n ]\n\n log.info(f\"Running: {' '.join(cmd[:4])} imageinfo\")\n try:\n r = subprocess.run(\n cmd,\n stdout=subprocess.PIPE,\n stderr=subprocess.PIPE,\n text=True,\n timeout=300\n )\n # Vol2 writes the Suggested Profile line to STDERR\n # stdout may contain it too depending on version - check both\n combined = (r.stderr or '') + (r.stdout or '')\n\n for line in combined.splitlines():\n # Match: \" Suggested Profile(s) : Win7SP1x64, Win7SP0x64...\"\n # or: \"INFO : volatility.debug : Suggested Profile(s) : Win7SP1x64\"\n if 'Suggested Profile' in line and ':' in line:\n # rsplit on last colon to get the profile list\n profile_str = line.rsplit(':', 1)[-1].strip()\n # Take first profile from comma-separated list\n profile = profile_str.split(',')[0].strip()\n if profile and len(profile) > 3:\n log.info(f\"Vol2 profile detected: {profile}\")\n return profile\n\n log.warning(\"vol2 imageinfo ran but no Suggested Profile line found\")\n log.debug(f\"Vol2 stderr (first 400 chars): {r.stderr[:400]}\")\n log.debug(f\"Vol2 stdout (first 200 chars): {r.stdout[:200]}\")\n return ''\n\n except subprocess.TimeoutExpired:\n log.warning(\"vol2 imageinfo timed out after 300 seconds\")\n return ''\n except FileNotFoundError:\n log.warning(\"vol2 command not found\")\n return ''\n except Exception as e:\n log.warning(f\"vol2 imageinfo error: {e}\")\n return ''\n\ndef identify_image(image_path: str) -> Dict:\n \"\"\"\n Identify OS type, version, and architecture from a memory image.\n Returns metadata dict used by all subsequent pipeline phases.\n \"\"\"\n path = Path(image_path)\n if not path.exists():\n raise FileNotFoundError(f\"Image not found: {image_path}\")\n\n size_gb = path.stat().st_size \/ (1024 ** 3)\n log.info(f\"Image: {path.name} ({size_gb:.1f} GB)\")\n\n result = {\n 'path': str(path.absolute()),\n 'filename': path.name,\n 'size_gb': round(size_gb, 2),\n 'sha256': sha256_file(image_path),\n 'os_type': None,\n 'os_version': None,\n 'arch': None,\n 'vol3_ok': False,\n 'vol2_ok': False,\n 'vol3_info': {},\n 'vol2_profile': None,\n 'errors': [],\n }\n\n # \u2500\u2500 Volatility 3: attempt OS identification \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n log.info(\"Running windows.info to identify OS\")\n cmd = [VOL3_CMD, '-f', image_path, '--renderer', 'json', 'windows.info']\n try:\n r = subprocess.run(cmd, capture_output=True, text=True, timeout=120)\n if r.returncode == 0 and r.stdout.strip():\n data = json.loads(r.stdout)\n rows = data.get('rows', data) if isinstance(data, dict) else data\n if rows:\n info = {}\n for row in rows:\n if isinstance(row, list) and len(row) >= 2:\n info[row[0]] = row[1]\n elif isinstance(row, dict):\n info.update(row)\n\n if info:\n result['os_type'] = 'windows'\n result['vol3_ok'] = True\n result['vol3_info'] = info\n result['os_version'] = info.get(\n 'NtBuildLab',\n info.get('Kernel Version', 'Unknown')\n )\n result['arch'] = '64-bit' if '64' in str(\n info.get('Kernel Base', '')\n ) else '32\/64-bit'\n log.info(f\"Vol3 identified: Windows {result['os_version']}\")\n except subprocess.TimeoutExpired:\n result['errors'].append(\"windows.info timed out\")\n except json.JSONDecodeError as e:\n result['errors'].append(f\"windows.info JSON parse error: {e}\")\n except Exception as e:\n result['errors'].append(f\"windows.info error: {e}\")\n\n # \u2500\u2500 Try Linux identification if Windows failed \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n if not result['os_type']:\n try:\n cmd_linux = [VOL3_CMD, '-f', image_path, '--renderer', 'json', 'linux.bash']\n r = subprocess.run(cmd_linux, capture_output=True, text=True, timeout=60)\n if r.returncode == 0 and r.stdout.strip():\n result['os_type'] = 'linux'\n result['vol3_ok'] = True\n log.info(\"Identified: Linux image\")\n except Exception:\n pass\n\n # \u2500\u2500 Volatility 2: always run imageinfo to get profile \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n # Required for:\n # - Older images (XP\/Vista\/Win7) where Vol3 symbols partially match\n # - Vol2-only plugins (shimcache, mftparser, prefetchparser etc.)\n # - Stable sequential plugin execution on uncertain images\n profile = run_vol2_imageinfo(image_path)\n if profile:\n result['vol2_ok'] = True\n result['vol2_profile'] = profile\n # If Vol3 did not identify the OS, use the Vol2 profile to set it\n if not result['os_type']:\n result['os_type'] = 'linux' if 'Linux' in profile else 'windows'\n result['os_version'] = profile\n log.info(f\"OS identified via Vol2: {profile}\")\n else:\n log.warning(\"Vol2 profile not detected - Vol2 plugins will be skipped\")\n log.warning(\"Ensure vol2 is installed: which vol2 && vol2 --info\")\n\n if not result['os_type']:\n result['errors'].append(\"Could not identify OS type from image\")\n log.error(\"Image identification failed - check symbol tables and vol2 installation\")\n\n return result\n<\/pre>\nPhase 2: parallel plugin collection<\/h3>\n
## File: vol_runner.py\n#!\/usr\/bin\/env python3\n\"\"\"\nvol_runner.py - Volatility plugin execution abstraction layer.\nHandles Vol2\/Vol3 differences, parallel execution with race condition\nprevention, and result normalisation.\n\"\"\"\n\nimport subprocess\nimport json\nimport logging\nimport concurrent.futures\nimport shutil\nfrom typing import List, Dict, Optional, Tuple\nfrom pathlib import Path\n\nlog = logging.getLogger(__name__)\n\nVOL3 = '\/opt\/vol3-env\/bin\/vol'\nVOL2 = '\/opt\/volatility2\/vol.py'\n\n# (vol3_plugin, vol2_plugin, timeout_seconds, description)\nWINDOWS_PLUGINS = [\n ('windows.pslist', 'pslist', 120, 'Process list'),\n ('windows.pstree', 'pstree', 120, 'Process tree'),\n ('windows.cmdline', 'cmdline', 180, 'Command line arguments'),\n ('windows.netscan', 'netscan', 180, 'Network connections'),\n ('windows.netstat', None, 120, 'Network statistics'),\n ('windows.malfind', 'malfind', 900, 'Injection detection'),\n ('windows.vadinfo', 'vadinfo', 300, 'VAD region analysis'),\n ('windows.dlllist', 'dlllist', 300, 'Loaded DLL list'),\n ('windows.svcscan', 'svcscan', 180, 'Windows services'),\n ('windows.scheduled_tasks', None, 120, 'Scheduled tasks'),\n ('windows.registry.hivelist','hivelist', 120, 'Registry hives'),\n ('windows.registry.printkey',None, 120, 'Registry run keys'),\n ('windows.callbacks', None, 180, 'Kernel callbacks'),\n ('windows.modules', 'modules', 120, 'Loaded kernel modules'),\n ('windows.driverscan', 'driverscan', 180, 'Driver scan'),\n ('windows.ssdt', 'ssdt', 120, 'SSDT entries'),\n ('windows.handles', 'handles', 900, 'Open handles'),\n ('windows.dumpfiles', None, 300, 'Mapped files'),\n ('windows.mftscan', None, 300, 'MFT entries'),\n ('windows.envars', 'envars', 120, 'Environment variables'),\n ('windows.lsadump', None, 120, 'LSA credentials'),\n ('windows.hashdump', 'hashdump', 120, 'Password hashes'),\n]\n\nLINUX_PLUGINS = [\n ('linux.pslist', 'linux_pslist', 120, 'Process list'),\n ('linux.pstree', 'linux_pstree', 120, 'Process tree'),\n ('linux.bash', 'linux_bash', 120, 'Bash history'),\n ('linux.netstat', 'linux_netstat', 120, 'Network connections'),\n ('linux.malfind', None, 600, 'Injection detection'),\n ('linux.lsmod', 'linux_lsmod', 120, 'Loaded kernel modules'),\n ('linux.check_syscall', 'linux_check_syscall', 180, 'Syscall table integrity'),\n ('linux.check_modules', 'linux_check_modules', 120, 'Module integrity'),\n ('linux.envars', None, 120, 'Environment variables'),\n]\n\n# Vol2-only plugins that provide unique value not in Vol3\nVOL2_ONLY_WINDOWS = [\n ('mftparser', 240, 'MFT parser (Vol2)'),\n ('shimcache', 120, 'Shimcache (Vol2)'),\n ('prefetchparser', 120, 'Prefetch (Vol2)'),\n ('iehistory', 120, 'IE history (Vol2)'),\n]\n\nVOL2_ONLY_LINUX = [\n ('linux_psaux', 120, 'Full process args (Vol2)'),\n ('linux_check_afinfo', 120, 'Network hook detection (Vol2)'),\n]\n\ndef run_vol3(image_path: str, plugin: str,\n timeout: int = 300, extra_args: List[str] = None) -> Tuple[str, List]:\n \"\"\"Run a Volatility 3 plugin, return (plugin_name, results_list).\"\"\"\n cmd = [VOL3, '-f', image_path, '--renderer', 'json', plugin]\n if extra_args:\n cmd.extend(extra_args)\n try:\n r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)\n if r.returncode != 0 or not r.stdout.strip():\n return plugin, []\n data = json.loads(r.stdout)\n if isinstance(data, dict):\n return plugin, data.get('rows', [])\n return plugin, data if isinstance(data, list) else []\n except subprocess.TimeoutExpired:\n log.warning(f\"TIMEOUT: {plugin} ({timeout}s)\")\n return plugin, []\n except Exception as e:\n log.debug(f\"Vol3 error in {plugin}: {e}\")\n return plugin, []\n\ndef run_vol2(image_path: str, plugin: str, profile: str,\n timeout: int = 300, extra_args: List[str] = None) -> Tuple[str, str]:\n \"\"\"Run a Volatility 2 plugin, return (plugin_name, raw_text_output).\"\"\"\n # Find vol2 binary\n vol2_bin = shutil.which('vol2')\n if not vol2_bin:\n py2 = shutil.which('python2') or shutil.which('python2.7')\n if py2 and Path(VOL2).exists():\n vol2_bin = py2\n cmd = [py2, VOL2, '-f', image_path, f'--profile={profile}', plugin]\n else:\n return plugin, ''\n else:\n cmd = [vol2_bin, '-f', image_path, f'--profile={profile}', plugin]\n\n if extra_args:\n cmd.extend(extra_args)\n\n try:\n r = subprocess.run(\n cmd,\n stdout=subprocess.PIPE,\n stderr=subprocess.PIPE,\n text=True,\n timeout=timeout\n )\n # Vol2 outputs to stdout for data, stderr for info messages\n return plugin, r.stdout\n except subprocess.TimeoutExpired:\n log.warning(f\"TIMEOUT (vol2): {plugin} ({timeout}s)\")\n return plugin, ''\n except Exception as e:\n log.debug(f\"Vol2 error in {plugin}: {e}\")\n return plugin, ''\n\ndef collect_all_plugins(image_path: str, image_info: Dict,\n max_workers: int = 8) -> Dict:\n \"\"\"\n Run all applicable plugins in parallel (or sequential for uncertain images).\n Returns dict of plugin_name -> results.\n \"\"\"\n os_type = image_info.get('os_type', 'windows')\n profile = image_info.get('vol2_profile') or image_info.get('os_version', '')\n vol3_ok = image_info.get('vol3_ok', False)\n vol2_ok = image_info.get('vol2_ok', False)\n vol3_info = image_info.get('vol3_info', {})\n\n # Vol3 symbols are only reliable when vol3_info is populated\n vol3_working = vol3_ok and bool(vol3_info)\n\n plugins = WINDOWS_PLUGINS if os_type == 'windows' else LINUX_PLUGINS\n vol2_only = VOL2_ONLY_WINDOWS if os_type == 'windows' else VOL2_ONLY_LINUX\n\n # Use sequential execution (1 worker) when Vol3 symbols are uncertain.\n # Parallel execution causes a race condition on partially-identified images\n # where threads compete to resolve the same kernel structures, producing\n # different results on each run.\n if not vol3_working and vol2_ok and profile:\n effective_workers = 1\n log.info(\"Vol3 symbols uncertain and Vol2 available - running sequentially via Vol2\")\n elif not vol3_working:\n effective_workers = 1\n log.info(\"Vol3 symbols uncertain - running sequentially to prevent race condition\")\n else:\n effective_workers = max_workers\n\n total = len(plugins) + (len(vol2_only) if vol2_ok else 0)\n done = 0\n results = {}\n\n log.info(f\"Running {total} plugins with {effective_workers} worker(s)\")\n\n with concurrent.futures.ThreadPoolExecutor(max_workers=effective_workers) as executor:\n futures = {}\n\n for vol3_plugin, vol2_plugin, timeout, desc in plugins:\n if not vol3_working and vol2_ok and vol2_plugin and profile:\n # Vol3 symbols missing - use Vol2 for all plugins that have it\n f = executor.submit(run_vol2, image_path, vol2_plugin, profile, timeout)\n futures[f] = (vol2_plugin, desc)\n elif vol3_ok and vol3_plugin:\n # Vol3 identified the image - use Vol3\n f = executor.submit(run_vol3, image_path, vol3_plugin, timeout)\n futures[f] = (vol3_plugin, desc)\n elif vol2_ok and vol2_plugin and profile:\n # Vol3 not available - fall back to Vol2\n f = executor.submit(run_vol2, image_path, vol2_plugin, profile, timeout)\n futures[f] = (vol2_plugin, desc)\n\n # Vol2-only plugins\n if vol2_ok and profile:\n for vol2_plugin, timeout, desc in vol2_only:\n f = executor.submit(run_vol2, image_path, vol2_plugin, profile, timeout)\n futures[f] = (vol2_plugin, desc)\n\n for future in concurrent.futures.as_completed(futures):\n plugin_name, desc = futures[future]\n try:\n name, data = future.result()\n results[name] = data\n done += 1\n if isinstance(data, list):\n count = len(data)\n else:\n count = len([l for l in data.splitlines() if l.strip()])\n log.info(f\" [{done}\/{total}] {desc}: {count} records\")\n except Exception as e:\n log.error(f\"Plugin {plugin_name} failed: {e}\")\n results[plugin_name] = []\n done += 1\n\n return results\n<\/pre>\nPhase 3: anomaly detection and scoring<\/h3>\n
## File: anomaly_checks.py\n## \/opt\/memory-hunter\/scripts\/anomaly_checks.py\n## Applies detection logic across collected plugin results\n\nimport re\nimport logging\nfrom typing import Dict, List, Tuple\n\nlog = logging.getLogger(__name__)\n\n# Legitimate system processes and their expected parents on Windows\nWINDOWS_PROCESS_RULES = {\n 'system': {'expected_parents': [''], 'expected_path': ''},\n 'smss.exe': {'expected_parents': ['System'], 'expected_path': r'windows\\system32'},\n 'csrss.exe': {'expected_parents': ['smss.exe'], 'expected_path': r'windows\\system32'},\n 'wininit.exe': {'expected_parents': ['smss.exe'], 'expected_path': r'windows\\system32'},\n 'winlogon.exe':{'expected_parents': ['smss.exe'], 'expected_path': r'windows\\system32'},\n 'services.exe':{'expected_parents': ['wininit.exe'],'expected_path': r'windows\\system32'},\n 'lsass.exe': {'expected_parents': ['wininit.exe'],'expected_path': r'windows\\system32'},\n 'svchost.exe': {'expected_parents': ['services.exe','msiexec.exe'],\n 'expected_path': r'windows\\system32'},\n 'explorer.exe':{'expected_parents': ['userinit.exe',''],\n 'expected_path': r'windows'},\n 'taskhost.exe':{'expected_parents': ['services.exe'],'expected_path': r'windows\\system32'},\n 'spoolsv.exe': {'expected_parents': ['services.exe'],'expected_path': r'windows\\system32'},\n}\n\n# Office apps that should never spawn shells\nSUSPICIOUS_PARENT_CHILD = {\n 'winword.exe': ['cmd.exe', 'powershell.exe', 'wscript.exe', 'cscript.exe',\n 'mshta.exe', 'regsvr32.exe', 'rundll32.exe'],\n 'excel.exe': ['cmd.exe', 'powershell.exe', 'wscript.exe', 'cscript.exe'],\n 'powerpnt.exe': ['cmd.exe', 'powershell.exe', 'wscript.exe'],\n 'outlook.exe': ['cmd.exe', 'powershell.exe', 'wscript.exe'],\n 'acrord32.exe': ['cmd.exe', 'powershell.exe', 'wscript.exe'],\n 'wmiprvse.exe': ['cmd.exe', 'powershell.exe'],\n}\n\n# Paths that are always suspicious for system binaries\nSUSPICIOUS_PATHS = [\n r'\\temp\\', r'\\tmp\\', r'\\appdata\\',\n r'\\public\\', r'\\downloads\\', r'\\desktop\\',\n r'\\programdata\\', r'\\recycle'\n]\n\nSUSPICIOUS_PORTS = {4444, 8080, 8443, 1337, 31337, 9001, 6667, 4545}\n\nclass Finding:\n def __init__(self, severity: str, category: str, title: str,\n detail: str, pid: int = None, process: str = None):\n self.severity = severity # CRITICAL \/ HIGH \/ MEDIUM \/ LOW\n self.category = category\n self.title = title\n self.detail = detail\n self.pid = pid\n self.process = process\n self.score = {'CRITICAL': 40, 'HIGH': 20, 'MEDIUM': 10, 'LOW': 5}[severity]\n\n def to_dict(self):\n return {\n 'severity': self.severity,\n 'category': self.category,\n 'title': self.title,\n 'detail': self.detail,\n 'pid': self.pid,\n 'process': self.process,\n 'score': self.score,\n }\n\ndef parse_vol2_pslist(raw_text: str) -> List:\n \"\"\"Parse Volatility 2 pslist text output into a list of process dicts.\"\"\"\n results = []\n for line in raw_text.splitlines():\n parts = line.split()\n if len(parts) >= 7 and parts[0].isdigit():\n try:\n results.append({\n 'PID': int(parts[1]) if len(parts) > 1 else 0,\n 'PPID': int(parts[2]) if len(parts) > 2 else 0,\n 'ImageFileName': parts[0] if parts else '',\n 'Path': '',\n 'CmdLine': ''\n })\n except (ValueError, IndexError):\n pass\n elif len(parts) >= 2 and not line.startswith('Offset'):\n # Format: Name PID PPID ...\n try:\n results.append({\n 'ImageFileName': parts[0],\n 'PID': int(parts[1]) if len(parts) > 1 and parts[1].isdigit() else 0,\n 'PPID': int(parts[2]) if len(parts) > 2 and parts[2].isdigit() else 0,\n 'Path': '',\n 'CmdLine': ''\n })\n except (ValueError, IndexError):\n pass\n return results\n\ndef check_processes(pslist: List, pstree: List, cmdline: List) -> List[Finding]:\n \"\"\"Detect process anomalies: masquerading, suspicious parents, unusual paths.\"\"\"\n findings = []\n\n # Build lookup dicts\n pid_to_name = {}\n pid_to_ppid = {}\n pid_to_path = {}\n pid_to_cmd = {}\n\n # Handle Vol2 text output\n if isinstance(pslist, str):\n pslist = parse_vol2_pslist(pslist)\n\n for proc in pslist:\n if not isinstance(proc, (list, dict)):\n continue\n if isinstance(proc, list):\n # Vol3 returns lists: [PID, PPID, ImageFileName, Offset, Threads, Handles, SessionId, Wow64, CreateTime, ExitTime, File output]\n pid = proc[0] if len(proc) > 0 else 0\n ppid = proc[1] if len(proc) > 1 else 0\n name = (proc[2] if len(proc) > 2 else '').lower()\n path = (proc[10] if len(proc) > 10 else '').lower()\n else:\n pid = proc.get('PID', proc.get('pid', 0))\n ppid = proc.get('PPID', proc.get('ppid', 0))\n name = proc.get('ImageFileName', proc.get('name', '')).lower()\n path = proc.get('Path', proc.get('path', '')).lower()\n\n pid_to_name[pid] = name\n pid_to_ppid[pid] = ppid\n pid_to_path[pid] = path\n\n for cmd_entry in cmdline:\n if isinstance(cmd_entry, list) and len(cmd_entry) >= 3:\n pid_to_cmd[cmd_entry[0]] = cmd_entry[2] or ''\n elif isinstance(cmd_entry, dict):\n pid_to_cmd[cmd_entry.get('PID', 0)] = cmd_entry.get('Args', '')\n\n # Check each process\n for pid, name in pid_to_name.items():\n path = pid_to_path.get(pid, '')\n ppid = pid_to_ppid.get(pid, 0)\n parent = pid_to_name.get(ppid, '').lower()\n cmd = pid_to_cmd.get(pid, '')\n\n # 1. Process in suspicious location\n for sus_path in SUSPICIOUS_PATHS:\n if sus_path in path.replace('\\', '\\\\'):\n if name in [k for k in WINDOWS_PROCESS_RULES.keys()]:\n findings.append(Finding(\n 'CRITICAL', 'process_masquerade',\n f'System process in suspicious location',\n f'{name} (PID {pid}) running from: {path}',\n pid, name\n ))\n elif name.endswith('.exe'):\n findings.append(Finding(\n 'HIGH', 'suspicious_path',\n f'Executable in staging location',\n f'{name} (PID {pid}) at {path}',\n pid, name\n ))\n\n # 2. Suspicious parent-child pairs\n for parent_name, bad_children in SUSPICIOUS_PARENT_CHILD.items():\n if parent == parent_name and name in bad_children:\n findings.append(Finding(\n 'CRITICAL', 'suspicious_spawn',\n f'Office\/PDF app spawned shell',\n f'{parent} (PPID {ppid}) spawned {name} (PID {pid})\\nCmd: {cmd[:200]}',\n pid, name\n ))\n\n # 3. Encoded PowerShell\n if name == 'powershell.exe' and cmd:\n if any(enc in cmd.lower() for enc in ['-enc', '-encodedcommand', '-e ']):\n findings.append(Finding(\n 'HIGH', 'encoded_powershell',\n 'PowerShell with encoded command',\n f'PID {pid}: {cmd[:300]}',\n pid, name\n ))\n if any(sus in cmd.lower() for sus in [\n 'downloadstring', 'downloadfile', 'webclient',\n 'invoke-expression', 'iex ', 'frombase64'\n ]):\n findings.append(Finding(\n 'HIGH', 'ps_download_cradle',\n 'PowerShell download cradle detected',\n f'PID {pid}: {cmd[:300]}',\n pid, name\n ))\n\n # 4. WMI execution chain\n if parent == 'wmiprvse.exe' and name in ['cmd.exe', 'powershell.exe',\n 'wscript.exe', 'cscript.exe']:\n findings.append(Finding(\n 'HIGH', 'wmi_execution',\n 'WMI spawned command interpreter',\n f'WmiPrvSE spawned {name} (PID {pid}): {cmd[:200]}',\n pid, name\n ))\n\n return findings\n\ndef check_malfind(malfind_results: List) -> List[Finding]:\n \"\"\"Score and categorise malfind results.\"\"\"\n findings = []\n\n for region in malfind_results:\n if isinstance(region, list):\n pid = region[0] if len(region) > 0 else 0\n name = region[1] if len(region) > 1 else ''\n start = region[3] if len(region) > 3 else 0\n prot = region[5] if len(region) > 5 else ''\n hexd = str(region[7]) if len(region) > 7 else ''\n elif isinstance(region, dict):\n pid = region.get('PID', region.get('Pid', 0))\n name = region.get('Process', region.get('ImageFileName', ''))\n start = region.get('Start', region.get('VadStart', 0))\n prot = region.get('Protection', '')\n hexd = str(region.get('Hexdump', region.get('Data', '')))\n else:\n continue\n\n has_pe = hexd.strip().startswith('4d 5a') or hexd.strip().startswith('MZ')\n is_rwx = 'EXECUTE_READ_WRITE' in str(prot)\n\n if has_pe and is_rwx:\n findings.append(Finding(\n 'CRITICAL', 'injection',\n 'PE file in RWX anonymous memory (reflective loading)',\n f'PID {pid} ({name}): addr=0x{start:x} protection={prot}',\n pid, name\n ))\n elif has_pe:\n findings.append(Finding(\n 'HIGH', 'injection',\n 'PE header in executable anonymous memory',\n f'PID {pid} ({name}): addr=0x{start:x} protection={prot}',\n pid, name\n ))\n elif is_rwx:\n findings.append(Finding(\n 'HIGH', 'injection',\n 'RWX anonymous memory region (shellcode staging)',\n f'PID {pid} ({name}): addr=0x{start:x}',\n pid, name\n ))\n\n return findings\n\ndef check_network(netscan_results: List) -> List[Finding]:\n \"\"\"Detect suspicious network activity.\"\"\"\n findings = []\n internal = ['10.', '172.16.', '172.17.', '172.18.', '172.19.',\n '172.20.', '172.21.', '172.22.', '172.23.', '172.24.',\n '172.25.', '172.26.', '172.27.', '172.28.', '172.29.',\n '172.30.', '172.31.', '192.168.', '127.', '0.0.0.0']\n\n for conn in netscan_results:\n if isinstance(conn, list):\n proto = str(conn[0]) if len(conn) > 0 else ''\n local = str(conn[1]) if len(conn) > 1 else ''\n remote = str(conn[3]) if len(conn) > 3 else ''\n rport = int(conn[4]) if len(conn) > 4 else 0\n state = str(conn[5]) if len(conn) > 5 else ''\n pid = int(conn[6]) if len(conn) > 6 else 0\n name = str(conn[7]) if len(conn) > 7 else ''\n elif isinstance(conn, dict):\n remote = str(conn.get('ForeignAddr', conn.get('RemoteAddr', '')))\n rport = int(conn.get('ForeignPort', conn.get('RemotePort', 0)))\n state = str(conn.get('State', ''))\n pid = int(conn.get('PID', conn.get('Pid', 0)))\n name = str(conn.get('Owner', conn.get('Process', '')))\n else:\n continue\n\n if 'ESTABLISHED' not in state:\n continue\n\n remote_ip = remote.split(':')[0] if ':' in remote else remote\n is_external = not any(remote_ip.startswith(r) for r in internal)\n\n if is_external:\n if rport in SUSPICIOUS_PORTS:\n findings.append(Finding(\n 'HIGH', 'suspicious_network',\n f'Connection to external IP on known C2 port',\n f'{name} (PID {pid}) -> {remote_ip}:{rport}',\n pid, name\n ))\n elif name.lower() in ['svchost.exe', 'lsass.exe', 'csrss.exe',\n 'winlogon.exe', 'services.exe']:\n findings.append(Finding(\n 'HIGH', 'suspicious_network',\n f'System process with external network connection',\n f'{name} (PID {pid}) -> {remote_ip}:{rport}',\n pid, name\n ))\n\n return findings\n\ndef check_services(svcscan: List) -> List[Finding]:\n \"\"\"Detect suspicious service configurations.\"\"\"\n findings = []\n suspicious_paths = [r'\\temp\\', r'\\tmp\\', r'\\appdata\\',\n r'\\public\\', r'\\programdata\\']\n\n for svc in svcscan:\n if isinstance(svc, list):\n name = str(svc[0]) if len(svc) > 0 else ''\n binary = str(svc[4]) if len(svc) > 4 else ''\n state = str(svc[2]) if len(svc) > 2 else ''\n elif isinstance(svc, dict):\n name = str(svc.get('ServiceName', svc.get('Name', '')))\n binary = str(svc.get('Binary', svc.get('Path', '')))\n state = str(svc.get('State', ''))\n else:\n continue\n\n binary_lower = binary.lower()\n for sus in suspicious_paths:\n if sus in binary_lower:\n findings.append(Finding(\n 'HIGH', 'suspicious_service',\n 'Service binary in staging location',\n f'Service: {name} | Binary: {binary}',\n None, name\n ))\n if 'powershell' in binary_lower or 'cmd.exe \/c' in binary_lower:\n findings.append(Finding(\n 'HIGH', 'suspicious_service',\n 'Service using interpreter as binary',\n f'Service: {name} | Binary: {binary}',\n None, name\n ))\n\n return findings\n\ndef check_kernel_integrity(callbacks: List, modules: List,\n driverscan: List) -> List[Finding]:\n \"\"\"Detect kernel-level tampering indicators.\"\"\"\n findings = []\n\n # Known legitimate callback registrants\n known_callbacks = [\n 'ntoskrnl.exe', 'nt', 'win32k.sys', 'ndis.sys',\n 'tcpip.sys', 'fltmgr.sys', 'ci.dll',\n ]\n\n for cb in callbacks:\n if isinstance(cb, list):\n callback_type = str(cb[0]) if len(cb) > 0 else ''\n module = str(cb[2]) if len(cb) > 2 else ''\n elif isinstance(cb, dict):\n callback_type = str(cb.get('Type', ''))\n module = str(cb.get('Module', ''))\n else:\n continue\n\n module_lower = module.lower()\n is_known = any(known in module_lower for known in known_callbacks)\n if not is_known and module:\n findings.append(Finding(\n 'MEDIUM', 'kernel_callback',\n f'Unknown kernel callback registration',\n f'Type: {callback_type} | Module: {module}',\n ))\n\n # Check for hidden modules (in driverscan but not modules)\n module_bases = set()\n for mod in modules:\n if isinstance(mod, list) and len(mod) > 1:\n module_bases.add(str(mod[1]))\n elif isinstance(mod, dict):\n module_bases.add(str(mod.get('Base', '')))\n\n for drv in driverscan:\n if isinstance(drv, list) and len(drv) > 1:\n base = str(drv[1])\n name = str(drv[0]) if len(drv) > 0 else ''\n elif isinstance(drv, dict):\n base = str(drv.get('Offset', ''))\n name = str(drv.get('Name', ''))\n else:\n continue\n\n if base and base not in module_bases:\n findings.append(Finding(\n 'HIGH', 'hidden_driver',\n 'Driver found by scan not in module list (possible rootkit)',\n f'Name: {name} | Base: {base}',\n ))\n\n return findings\n\ndef run_all_checks(plugin_results: Dict) -> Tuple[List[Finding], int]:\n \"\"\"Run all anomaly checks and return findings with total risk score.\"\"\"\n all_findings = []\n\n all_findings.extend(check_processes(\n plugin_results.get('windows.pslist', []),\n plugin_results.get('windows.pstree', []),\n plugin_results.get('windows.cmdline', []),\n ))\n\n all_findings.extend(check_malfind(\n plugin_results.get('windows.malfind', []) +\n plugin_results.get('linux.malfind', [])\n ))\n\n all_findings.extend(check_network(\n plugin_results.get('windows.netscan', []) +\n plugin_results.get('linux.netstat', [])\n ))\n\n all_findings.extend(check_services(\n plugin_results.get('windows.svcscan', [])\n ))\n\n all_findings.extend(check_kernel_integrity(\n plugin_results.get('windows.callbacks', []),\n plugin_results.get('windows.modules', []),\n plugin_results.get('windows.driverscan', []),\n ))\n\n total_score = sum(f.score for f in all_findings)\n all_findings.sort(key=lambda f: f.score, reverse=True)\n\n return all_findings, total_score\n<\/pre>\nPhase 4: IOC extraction and Yara scanning<\/h3>\n
## File: phase4_iocs.py\n## \/opt\/memory-hunter\/scripts\/phase4_iocs.py\n\nimport re\nimport subprocess\nimport json\nimport logging\nfrom typing import Dict, List, Set\nfrom pathlib import Path\n\nlog = logging.getLogger(__name__)\n\nVOL3 = '\/opt\/vol3-env\/bin\/vol'\nYARA_RULES = '\/opt\/memory-hunter\/yara_rules\/combined.yar'\n\n# Patterns for IOC extraction\nPATTERNS = {\n 'ipv4': re.compile(r'\\b(?:(?:25[0-5]|2[0-4]\\d|[01]?\\d\\d?)\\.){3}(?:25[0-5]|2[0-4]\\d|[01]?\\d\\d?)\\b'),\n 'url': re.compile(r'https?:\/\/[a-zA-Z0-9._\/?=&%+-]{10,300}'),\n 'domain': re.compile(r'\\b(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\\.){1,10}(?:com|net|org|io|co|xyz|top|tk|ru|cn|de|info|biz)\\b', re.I),\n 'email': re.compile(r'\\b[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}\\b'),\n 'hash_md5': re.compile(r'\\b[a-fA-F0-9]{32}\\b'),\n 'hash_sha256': re.compile(r'\\b[a-fA-F0-9]{64}\\b'),\n 'registry_run':re.compile(r'SOFTWARE\\(?:Microsoft\\Windows\\CurrentVersion\\Run|Wow6432Node)[^\"\\'\\]+', re.I),\n 'pipe': re.compile(r'\\\\\\.\\pipe\\[a-zA-Z0-9_-]{4,}'),\n 'base64_large':re.compile(r'[A-Za-z0-9+\/]{100,}={0,2}'),\n}\n\nINTERNAL_IPS = ['10.', '172.16.', '192.168.', '127.', '169.254.']\nLEGIT_DOMAINS = ['microsoft.com', 'windows.com', 'windowsupdate.com',\n 'google.com', 'akamai.com', 'cloudflare.com',\n 'amazon.com', 'amazonaws.com']\n\ndef extract_strings_from_memory(image_path: str,\n suspicious_pids: List[int]) -> str:\n \"\"\"Extract strings from suspicious process memory regions.\"\"\"\n if not suspicious_pids:\n return ''\n\n pid_args = ['--pid', ','.join(str(p) for p in suspicious_pids[:10])]\n\n cmd = [VOL3, '-f', image_path, '--renderer', 'json',\n 'windows.strings'] + pid_args\n r = subprocess.run(cmd, capture_output=True, text=True, timeout=300)\n return r.stdout if r.returncode == 0 else ''\n\ndef run_yara_scan(image_path: str) -> List[Dict]:\n \"\"\"Run Yara rules against process memory via Volatility.\"\"\"\n if not Path(YARA_RULES).exists():\n log.warning(\"Yara rules not found - skipping Yara scan\")\n return []\n\n cmd = [VOL3, '-f', image_path, '--renderer', 'json',\n 'windows.vadyarascan', '--yara-file', YARA_RULES]\n r = subprocess.run(cmd, capture_output=True, text=True, timeout=600)\n\n if r.returncode != 0:\n return []\n\n try:\n data = json.loads(r.stdout)\n rows = data.get('rows', data) if isinstance(data, dict) else data\n return rows\n except:\n return []\n\ndef extract_iocs(plugin_results: Dict, image_path: str,\n suspicious_pids: List[int]) -> Dict:\n \"\"\"Extract all IOC types from plugin results and process memory strings.\"\"\"\n iocs: Dict[str, Set] = {k: set() for k in PATTERNS}\n iocs['yara_hits'] = set()\n\n # Extract from network connections\n for conn in plugin_results.get('windows.netscan', []):\n remote = ''\n if isinstance(conn, list):\n remote = str(conn[3]) if len(conn) > 3 else ''\n elif isinstance(conn, dict):\n remote = str(conn.get('ForeignAddr', ''))\n remote_ip = remote.split(':')[0] if ':' in remote else remote\n if remote_ip and not any(remote_ip.startswith(r) for r in INTERNAL_IPS):\n iocs['ipv4'].add(remote_ip)\n\n # Extract from command lines\n for cmd_entry in plugin_results.get('windows.cmdline', []):\n cmd_text = ''\n if isinstance(cmd_entry, list) and len(cmd_entry) > 2:\n cmd_text = str(cmd_entry[2])\n elif isinstance(cmd_entry, dict):\n cmd_text = str(cmd_entry.get('Args', ''))\n\n for ioc_type, pattern in PATTERNS.items():\n for match in pattern.findall(cmd_text):\n if ioc_type == 'ipv4' and not any(match.startswith(r) for r in INTERNAL_IPS):\n iocs[ioc_type].add(match)\n elif ioc_type == 'url' and not any(d in match for d in LEGIT_DOMAINS):\n iocs[ioc_type].add(match[:200])\n elif ioc_type not in ('ipv4', 'url'):\n iocs[ioc_type].add(match[:200])\n\n # Extract from process memory strings\n if suspicious_pids:\n strings_output = extract_strings_from_memory(image_path, suspicious_pids)\n for ioc_type, pattern in PATTERNS.items():\n for match in pattern.findall(strings_output):\n if ioc_type == 'url' and not any(d in match for d in LEGIT_DOMAINS):\n iocs[ioc_type].add(match[:200])\n elif ioc_type == 'ipv4' and not any(match.startswith(r) for r in INTERNAL_IPS):\n iocs[ioc_type].add(match)\n elif ioc_type == 'pipe' and 'pipe\\' in match.lower():\n iocs[ioc_type].add(match)\n\n # Yara scan\n yara_hits = run_yara_scan(image_path)\n for hit in yara_hits:\n if isinstance(hit, list):\n rule = str(hit[2]) if len(hit) > 2 else ''\n pid = str(hit[0]) if len(hit) > 0 else ''\n proc = str(hit[1]) if len(hit) > 1 else ''\n elif isinstance(hit, dict):\n rule = str(hit.get('Rule', ''))\n pid = str(hit.get('PID', ''))\n proc = str(hit.get('Process', ''))\n else:\n continue\n if rule:\n iocs['yara_hits'].add(f\"{rule} (PID {pid} \/ {proc})\")\n log.warning(f\"YARA HIT: {rule} in PID {pid} ({proc})\")\n\n return {k: sorted(list(v)) for k, v in iocs.items()}\n<\/pre>\nPhase 5: report generation (HTML and JSON)<\/h3>\n
## File: phase5_report.py\n#!\/usr\/bin\/env python3\n\"\"\"\nPhase 5: Generate analyst-ready reports in a clean folder structure.\n\nOutput per run:\n reports\/<image>_<timestamp>\/\n report.html - full visual report with navigation, all data shown\n report.json - structured data for SIEM\/automation\n summary.txt - plain text summary for terminal\/tickets\n iocs.txt - extracted IOCs ready for blocking\n dumps\/ - any carved\/dumped files from memory\n\"\"\"\n\nimport json\nimport logging\nimport shutil\nfrom datetime import datetime\nfrom pathlib import Path\nfrom typing import Dict, List\nfrom jinja2 import Environment\n\nlog = logging.getLogger(__name__)\nREPORT_BASE = Path('\/opt\/memory-hunter\/reports')\n\n# \u2500\u2500 HTML template \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\nHTML_TEMPLATE = \"\"\"<!DOCTYPE html>\n<html lang=\"en\">\n<head>\n<meta charset=\"UTF-8\">\n<meta name=\"viewport\" content=\"width=device-width,initial-scale=1\">\n<title>Memory Analysis: {{ meta.filename }}<\/title>\n<style>\n*{box-sizing:border-box;margin:0;padding:0}\n:root{\n --bg:#0f172a;--surface:#1e293b;--surface2:#263248;--border:#334155;\n --text:#e2e8f0;--text2:#94a3b8;--text3:#64748b;\n --amber:#f59e0b;--red:#ef4444;--orange:#f97316;\n --yellow:#eab308;--green:#22c55e;--blue:#3b82f6;\n --mono:'JetBrains Mono','Fira Mono',monospace;\n}\nbody{font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',sans-serif;\n background:var(--bg);color:var(--text);font-size:13px;line-height:1.5}\n\n\/* \u2500\u2500 NAV \u2500\u2500 *\/\n.topbar{background:#0a1120;border-bottom:1px solid var(--border);\n padding:0 24px;position:sticky;top:0;z-index:100;\n display:flex;align-items:center;gap:0;height:48px}\n.topbar-brand{font-family:var(--mono);font-size:12px;color:var(--amber);\n font-weight:700;margin-right:24px;white-space:nowrap}\n.nav-links{display:flex;align-items:stretch;gap:0;overflow-x:auto;flex:1}\n.nav-link{color:var(--text2);text-decoration:none;font-size:11px;\n padding:0 14px;display:flex;align-items:center;\n border-bottom:2px solid transparent;white-space:nowrap;\n transition:all 0.15s}\n.nav-link:hover{color:var(--text);border-bottom-color:var(--amber)}\n.nav-link .badge{background:var(--surface);color:var(--text3);\n font-family:var(--mono);font-size:9px;\n padding:1px 5px;border-radius:3px;margin-left:5px}\n.nav-link.critical .badge{background:#450a0a;color:var(--red)}\n.nav-link.has-data .badge{background:#1e3a5f;color:#60a5fa}\n\n\/* \u2500\u2500 HEADER \u2500\u2500 *\/\n.header{background:linear-gradient(135deg,#0a1120 0%,#1a1f35 100%);\n padding:24px 32px;border-bottom:1px solid var(--border)}\n.header-top{display:flex;align-items:flex-start;justify-content:space-between;gap:16px}\n.header-title{font-family:var(--mono);font-size:15px;font-weight:700;\n color:var(--text)}\n.header-meta{font-family:var(--mono);font-size:10px;color:var(--text3);\n margin-top:6px;display:flex;flex-wrap:wrap;gap:16px}\n.header-meta span{color:var(--text2)}\n.risk-pill{padding:6px 18px;border-radius:20px;font-weight:800;\n font-size:13px;font-family:var(--mono);white-space:nowrap}\n.risk-CRITICAL{background:#450a0a;color:var(--red);border:1px solid var(--red)}\n.risk-HIGH {background:#431407;color:var(--orange);border:1px solid var(--orange)}\n.risk-MEDIUM {background:#422006;color:var(--yellow);border:1px solid var(--yellow)}\n.risk-LOW {background:#052e16;color:var(--green);border:1px solid var(--green)}\n\n\/* \u2500\u2500 STAT CARDS \u2500\u2500 *\/\n.stats{display:flex;gap:10px;padding:20px 32px;\n background:var(--surface);border-bottom:1px solid var(--border);\n overflow-x:auto}\n.stat{background:var(--bg);border:1px solid var(--border);border-radius:8px;\n padding:12px 18px;text-align:center;min-width:80px}\n.stat-val{font-size:22px;font-weight:800;font-family:var(--mono)}\n.stat-lbl{font-size:9px;color:var(--text3);text-transform:uppercase;\n letter-spacing:.1em;margin-top:3px}\n.c-critical{color:var(--red)}.c-high{color:var(--orange)}\n.c-medium{color:var(--yellow)}.c-low{color:var(--green)}\n.c-blue{color:var(--blue)}.c-amber{color:var(--amber)}\n\n\/* \u2500\u2500 SECTIONS \u2500\u2500 *\/\n.container{padding:0 32px 40px}\n.section{margin-top:24px;scroll-margin-top:56px}\n.section-header{display:flex;align-items:center;gap:10px;\n padding:12px 16px;background:var(--surface);\n border:1px solid var(--border);border-radius:8px 8px 0 0;\n border-bottom:none}\n.section-title{font-family:var(--mono);font-size:12px;font-weight:700;\n color:var(--amber)}\n.section-count{background:var(--bg);color:var(--text3);\n font-family:var(--mono);font-size:9px;\n padding:2px 7px;border-radius:3px;border:1px solid var(--border)}\n.section-body{background:var(--surface);border:1px solid var(--border);\n border-radius:0 0 8px 8px;overflow:hidden}\n\n\/* \u2500\u2500 FINDINGS \u2500\u2500 *\/\n.finding{padding:12px 16px;border-bottom:1px solid var(--border)}\n.finding:last-child{border-bottom:none}\n.finding-row{display:flex;align-items:flex-start;gap:10px}\n.sev{padding:2px 8px;border-radius:4px;font-size:9px;font-weight:700;\n font-family:var(--mono);white-space:nowrap;margin-top:1px}\n.sev-CRITICAL{background:#450a0a;color:var(--red);border:1px solid var(--red)}\n.sev-HIGH {background:#431407;color:var(--orange);border:1px solid var(--orange)}\n.sev-MEDIUM {background:#422006;color:var(--yellow);border:1px solid var(--yellow)}\n.sev-LOW {background:#052e16;color:var(--green);border:1px solid var(--green)}\n.finding-title{font-weight:600;font-size:12px}\n.finding-proc{color:var(--text3);font-family:var(--mono);font-size:10px;margin-left:4px}\n.finding-detail{font-family:var(--mono);font-size:10px;color:var(--text2);\n background:var(--bg);padding:6px 10px;border-radius:4px;\n margin-top:6px;white-space:pre-wrap;word-break:break-all;\n border-left:2px solid var(--border)}\n\n\/* \u2500\u2500 TABLES \u2500\u2500 *\/\n.tbl{width:100%;border-collapse:collapse}\n.tbl th{text-align:left;padding:7px 12px;font-size:9px;color:var(--text3);\n text-transform:uppercase;letter-spacing:.08em;\n background:var(--surface2);border-bottom:1px solid var(--border);\n position:sticky;top:48px}\n.tbl td{padding:7px 12px;border-bottom:1px solid #1a2336;\n font-family:var(--mono);font-size:10px;vertical-align:top}\n.tbl tr:last-child td{border-bottom:none}\n.tbl tr:hover td{background:rgba(255,255,255,0.02)}\n.tag-ext{display:inline-block;background:#450a0a;color:var(--red);\n padding:0 5px;border-radius:3px;font-size:8px;margin-left:4px}\n.tag-sus{display:inline-block;background:#422006;color:var(--orange);\n padding:0 5px;border-radius:3px;font-size:8px;margin-left:4px}\n.tag-ok {display:inline-block;background:#052e16;color:var(--green);\n padding:0 5px;border-radius:3px;font-size:8px;margin-left:4px}\n.tbl-wrap{max-height:500px;overflow-y:auto}\n\n\/* \u2500\u2500 IOC GRID \u2500\u2500 *\/\n.ioc-grid{display:grid;grid-template-columns:repeat(auto-fill,minmax(280px,1fr));\n gap:12px;padding:16px}\n.ioc-group{}\n.ioc-label{font-size:9px;font-family:var(--mono);text-transform:uppercase;\n letter-spacing:.1em;color:var(--text3);margin-bottom:6px}\n.ioc-val{font-family:var(--mono);font-size:10px;color:#60a5fa;\n background:#0c1e35;padding:3px 8px;border-radius:3px;\n margin-bottom:2px;word-break:break-all;\n border-left:2px solid #1e3a5f}\n.ioc-empty{color:var(--text3);font-size:11px;padding:16px;font-style:italic}\n\n\/* \u2500\u2500 SCROLLABLE TABLES \u2500\u2500 *\/\n.scroll-hint{font-size:9px;color:var(--text3);padding:4px 12px;\n background:var(--surface2);text-align:right;\n font-family:var(--mono)}\n\n\/* \u2500\u2500 EMPTY STATE \u2500\u2500 *\/\n.empty{padding:24px;text-align:center;color:var(--text3);font-size:11px}\n\n\/* \u2500\u2500 YARA HITS \u2500\u2500 *\/\n.yara-hit{padding:10px 16px;border-bottom:1px solid var(--border);\n font-family:var(--mono);font-size:11px;color:var(--red)}\n.yara-hit:last-child{border-bottom:none}\n\n@media(max-width:700px){\n .stats{flex-wrap:wrap}\n .container{padding:0 12px 40px}\n .header{padding:16px}\n .topbar{padding:0 12px}\n}\n<\/style>\n<\/head>\n<body>\n\n<!-- \u2500\u2500 TOP NAVIGATION \u2500\u2500 -->\n<div class=\"topbar\">\n <div class=\"topbar-brand\">\/\/ memory-hunter<\/div>\n <nav class=\"nav-links\">\n <a class=\"nav-link {{ 'critical' if risk_label == 'CRITICAL' else 'has-data' if findings else '' }}\"\n href=\"#findings\">Findings<span class=\"badge\">{{ findings|length }}<\/span><\/a>\n <a class=\"nav-link {{ 'has-data' if iocs.yara_hits else '' }}\"\n href=\"#yara\">Yara<span class=\"badge\">{{ iocs.yara_hits|length }}<\/span><\/a>\n <a class=\"nav-link {{ 'has-data' if iocs.ipv4 or iocs.url else '' }}\"\n href=\"#iocs\">IOCs<span class=\"badge\">{{ (iocs.ipv4|length) + (iocs.url|length) + (iocs.domain|length) }}<\/span><\/a>\n <a class=\"nav-link {{ 'has-data' if network else '' }}\"\n href=\"#network\">Network<span class=\"badge\">{{ network|length }}<\/span><\/a>\n <a class=\"nav-link {{ 'has-data' if processes else '' }}\"\n href=\"#processes\">Processes<span class=\"badge\">{{ processes|length }}<\/span><\/a>\n <a class=\"nav-link {{ 'has-data' if services else '' }}\"\n href=\"#services\">Services<span class=\"badge\">{{ services|length }}<\/span><\/a>\n <a class=\"nav-link {{ 'has-data' if scheduled_tasks else '' }}\"\n href=\"#tasks\">Sched Tasks<span class=\"badge\">{{ scheduled_tasks|length }}<\/span><\/a>\n <a class=\"nav-link {{ 'has-data' if modules else '' }}\"\n href=\"#modules\">Modules<span class=\"badge\">{{ modules|length }}<\/span><\/a>\n <a class=\"nav-link {{ 'has-data' if drivers else '' }}\"\n href=\"#drivers\">Drivers<span class=\"badge\">{{ drivers|length }}<\/span><\/a>\n <a class=\"nav-link {{ 'has-data' if handles else '' }}\"\n href=\"#handles\">Handles<span class=\"badge\">{{ handles|length }}<\/span><\/a>\n <a class=\"nav-link {{ 'has-data' if ie_history else '' }}\"\n href=\"#iehistory\">IE History<span class=\"badge\">{{ ie_history|length }}<\/span><\/a>\n <a class=\"nav-link {{ 'has-data' if prefetch else '' }}\"\n href=\"#prefetch\">Prefetch<span class=\"badge\">{{ prefetch|length }}<\/span><\/a>\n <a class=\"nav-link {{ 'has-data' if shimcache else '' }}\"\n href=\"#shimcache\">Shimcache<span class=\"badge\">{{ shimcache|length }}<\/span><\/a>\n <a class=\"nav-link {{ 'has-data' if mft else '' }}\"\n href=\"#mft\">MFT<span class=\"badge\">{{ mft|length }}<\/span><\/a>\n <a class=\"nav-link {{ 'has-data' if envars else '' }}\"\n href=\"#envars\">Envars<span class=\"badge\">{{ envars|length }}<\/span><\/a>\n <a class=\"nav-link {{ 'has-data' if cmdline else '' }}\"\n href=\"#cmdline\">CmdLine<span class=\"badge\">{{ cmdline|length }}<\/span><\/a>\n <\/nav>\n<\/div>\n\n<!-- \u2500\u2500 HEADER \u2500\u2500 -->\n<div class=\"header\">\n <div class=\"header-top\">\n <div>\n <div class=\"header-title\">{{ meta.filename }}<\/div>\n <div class=\"header-meta\">\n <span>SHA256: {{ meta.sha256[:16] }}...{{ meta.sha256[-8:] }}<\/span>\n <span>{{ meta.size_gb }} GB<\/span>\n <span>OS: {{ meta.os_version or 'Unknown' }}<\/span>\n {% if meta.vol2_profile %}<span>Vol2: {{ meta.vol2_profile }}<\/span>{% endif %}\n <span>{{ generated_at }}<\/span>\n <\/div>\n <\/div>\n <div class=\"risk-pill risk-{{ risk_label }}\">{{ risk_label }}<\/div>\n <\/div>\n<\/div>\n\n<!-- \u2500\u2500 STAT CARDS \u2500\u2500 -->\n<div class=\"stats\">\n <div class=\"stat\"><div class=\"stat-val c-critical\">{{ summary.critical }}<\/div><div class=\"stat-lbl\">Critical<\/div><\/div>\n <div class=\"stat\"><div class=\"stat-val c-high\">{{ summary.high }}<\/div><div class=\"stat-lbl\">High<\/div><\/div>\n <div class=\"stat\"><div class=\"stat-val c-medium\">{{ summary.medium }}<\/div><div class=\"stat-lbl\">Medium<\/div><\/div>\n <div class=\"stat\"><div class=\"stat-val c-amber\">{{ summary.yara_hits }}<\/div><div class=\"stat-lbl\">Yara Hits<\/div><\/div>\n <div class=\"stat\"><div class=\"stat-val c-blue\">{{ network|length }}<\/div><div class=\"stat-lbl\">Net Conns<\/div><\/div>\n <div class=\"stat\"><div class=\"stat-val\">{{ processes|length }}<\/div><div class=\"stat-lbl\">Processes<\/div><\/div>\n <div class=\"stat\"><div class=\"stat-val\">{{ modules|length }}<\/div><div class=\"stat-lbl\">Modules<\/div><\/div>\n <div class=\"stat\"><div class=\"stat-val\">{{ drivers|length }}<\/div><div class=\"stat-lbl\">Drivers<\/div><\/div>\n <div class=\"stat\"><div class=\"stat-val\">{{ handles|length }}<\/div><div class=\"stat-lbl\">Handles<\/div><\/div>\n <div class=\"stat\"><div class=\"stat-val c-blue\">{{ (iocs.ipv4|length)+(iocs.url|length)+(iocs.domain|length) }}<\/div><div class=\"stat-lbl\">Net IOCs<\/div><\/div>\n <div class=\"stat\"><div class=\"stat-val\">{{ risk_score }}<\/div><div class=\"stat-lbl\">Risk Score<\/div><\/div>\n<\/div>\n\n<div class=\"container\">\n\n<!-- \u2500\u2500 FINDINGS \u2500\u2500 -->\n<div class=\"section\" id=\"findings\">\n <div class=\"section-header\">\n <span class=\"section-title\">\/\/ threat findings<\/span>\n <span class=\"section-count\">{{ findings|length }}<\/span>\n <\/div>\n <div class=\"section-body\">\n {% if findings %}\n {% for f in findings %}\n <div class=\"finding\">\n <div class=\"finding-row\">\n <span class=\"sev sev-{{ f.severity }}\">{{ f.severity }}<\/span>\n <span class=\"finding-title\">{{ f.title }}<\/span>\n {% if f.process %}<span class=\"finding-proc\">{{ f.process }}{% if f.pid %} (PID {{ f.pid }}){% endif %}<\/span>{% endif %}\n <\/div>\n {% if f.detail %}<div class=\"finding-detail\">{{ f.detail }}<\/div>{% endif %}\n <\/div>\n {% endfor %}\n {% else %}\n <div class=\"empty\">No threat findings detected<\/div>\n {% endif %}\n <\/div>\n<\/div>\n\n<!-- \u2500\u2500 YARA \u2500\u2500 -->\n<div class=\"section\" id=\"yara\">\n <div class=\"section-header\">\n <span class=\"section-title\">\/\/ yara rule matches<\/span>\n <span class=\"section-count\">{{ iocs.yara_hits|length }}<\/span>\n <\/div>\n <div class=\"section-body\">\n {% if iocs.yara_hits %}\n {% for hit in iocs.yara_hits %}\n <div class=\"yara-hit\">{{ hit }}<\/div>\n {% endfor %}\n {% else %}\n <div class=\"empty\">No Yara matches<\/div>\n {% endif %}\n <\/div>\n<\/div>\n\n<!-- \u2500\u2500 IOCs \u2500\u2500 -->\n<div class=\"section\" id=\"iocs\">\n <div class=\"section-header\">\n <span class=\"section-title\">\/\/ extracted iocs<\/span>\n <span class=\"section-count\">{{ summary.ioc_count }}<\/span>\n <\/div>\n <div class=\"section-body\">\n {% set ioc_type_labels = {\n 'ipv4': 'IP Addresses', 'url': 'URLs', 'domain': 'Domains',\n 'pipe': 'Named Pipes', 'registry_run': 'Registry Run Keys',\n 'hash_sha256': 'SHA256 Hashes', 'hash_md5': 'MD5 Hashes',\n 'email': 'Email Addresses', 'base64_large': 'Base64 Data'\n } %}\n {% set has_iocs = namespace(v=false) %}\n {% for ioc_type, values in iocs.items() %}\n {% if values and ioc_type != 'yara_hits' %}{% set has_iocs.v = true %}{% endif %}\n {% endfor %}\n {% if has_iocs.v %}\n <div class=\"ioc-grid\">\n {% for ioc_type, values in iocs.items() %}\n {% if values and ioc_type != 'yara_hits' %}\n <div class=\"ioc-group\">\n <div class=\"ioc-label\">{{ ioc_type_labels.get(ioc_type, ioc_type.replace('_',' ')) }} ({{ values|length }})<\/div>\n {% for v in values %}\n <div class=\"ioc-val\">{{ v }}<\/div>\n {% endfor %}\n <\/div>\n {% endif %}\n {% endfor %}\n <\/div>\n {% else %}\n <div class=\"ioc-empty\">No IOCs extracted<\/div>\n {% endif %}\n <\/div>\n<\/div>\n\n<!-- \u2500\u2500 NETWORK CONNECTIONS \u2500\u2500 -->\n<div class=\"section\" id=\"network\">\n <div class=\"section-header\">\n <span class=\"section-title\">\/\/ network connections<\/span>\n <span class=\"section-count\">{{ network|length }}<\/span>\n <\/div>\n <div class=\"section-body\">\n {% if network %}\n <div class=\"scroll-hint\">scroll to view all rows<\/div>\n <div class=\"tbl-wrap\">\n <table class=\"tbl\">\n <tr><th>PID<\/th><th>Process<\/th><th>Local<\/th><th>Remote IP<\/th><th>Port<\/th><th>State<\/th><\/tr>\n {% for c in network %}\n {% if c is mapping %}\n {% set remote = c.get('ForeignAddr', c.get('Remote', c.get('ForeignAddress', ''))) %}\n <tr>\n <td>{{ c.get('PID', c.get('Pid', c.get('pid', ''))) }}<\/td>\n <td>{{ c.get('Owner', c.get('ImageFileName', c.get('Process', c.get('name', '')))) }}<\/td>\n <td>{{ c.get('LocalAddr', c.get('Local', '')) }}<\/td>\n <td>{{ remote }}{% if remote and not remote.startswith(('10.','192.168.','172.','127.','0.','*')) %}<span class=\"tag-ext\">EXT<\/span>{% endif %}<\/td>\n <td>{{ c.get('ForeignPort', c.get('Port', '')) }}<\/td>\n <td>{{ c.get('State', c.get('state', '')) }}<\/td>\n <\/tr>\n {% else %}\n <tr>\n <td>{{ c[6] if c|length > 6 else '' }}<\/td>\n <td>{{ c[7] if c|length > 7 else '' }}<\/td>\n <td>{{ c[1] if c|length > 1 else '' }}<\/td>\n {% set r = c[3] if c|length > 3 else '' %}\n <td>{{ r }}{% if r and not r.startswith(('10.','192.168.','172.','127.','0.','*')) %}<span class=\"tag-ext\">EXT<\/span>{% endif %}<\/td>\n <td>{{ c[4] if c|length > 4 else '' }}<\/td>\n <td>{{ c[5] if c|length > 5 else '' }}<\/td>\n <\/tr>\n {% endif %}\n {% endfor %}\n <\/table>\n <\/div>\n {% else %}<div class=\"empty\">No network connections<\/div>{% endif %}\n <\/div>\n<\/div>\n\n<!-- \u2500\u2500 PROCESS LIST \u2500\u2500 -->\n<div class=\"section\" id=\"processes\">\n <div class=\"section-header\">\n <span class=\"section-title\">\/\/ process list<\/span>\n <span class=\"section-count\">{{ processes|length }}<\/span>\n <\/div>\n <div class=\"section-body\">\n {% if processes %}\n <div class=\"tbl-wrap\">\n <table class=\"tbl\">\n <tr><th>PID<\/th><th>PPID<\/th><th>Name<\/th><th>Path \/ Command<\/th><th>Created<\/th><\/tr>\n {% for p in processes %}\n {% if p is mapping %}\n <tr>\n <td>{{ p.get('PID', p.get('Pid','')) }}<\/td>\n <td>{{ p.get('PPID', p.get('PPid','')) }}<\/td>\n <td><b>{{ p.get('ImageFileName', p.get('Name','')) }}<\/b><\/td>\n <td style=\"max-width:500px;overflow:hidden;text-overflow:ellipsis\">\n {{ p.get('Path', p.get('Exe', p.get('CommandLine', ''))) }}<\/td>\n <td>{{ p.get('CreateTime', p.get('create_time','')) }}<\/td>\n <\/tr>\n {% else %}\n <tr>\n <td>{{ p[0] if p|length > 0 else '' }}<\/td>\n <td>{{ p[1] if p|length > 1 else '' }}<\/td>\n <td><b>{{ p[2] if p|length > 2 else '' }}<\/b><\/td>\n <td style=\"max-width:500px\">{{ p[10] if p|length > 10 else p[3] if p|length > 3 else '' }}<\/td>\n <td>{{ p[8] if p|length > 8 else '' }}<\/td>\n <\/tr>\n {% endif %}\n {% endfor %}\n <\/table>\n <\/div>\n {% else %}<div class=\"empty\">No process data<\/div>{% endif %}\n <\/div>\n<\/div>\n\n<!-- \u2500\u2500 SERVICES \u2500\u2500 -->\n<div class=\"section\" id=\"services\">\n <div class=\"section-header\">\n <span class=\"section-title\">\/\/ windows services<\/span>\n <span class=\"section-count\">{{ services|length }}<\/span>\n <\/div>\n <div class=\"section-body\">\n {% if services %}\n <div class=\"tbl-wrap\">\n <table class=\"tbl\">\n <tr><th>Name<\/th><th>State<\/th><th>Type<\/th><th>Binary Path<\/th><\/tr>\n {% for s in services %}\n {% if s is mapping %}\n <tr>\n <td>{{ s.get('ServiceName', s.get('Name','')) }}<\/td>\n <td>{{ s.get('State','') }}<\/td>\n <td>{{ s.get('Type', s.get('ServiceType','')) }}<\/td>\n <td>{{ s.get('Binary', s.get('Path', s.get('ImagePath',''))) }}<\/td>\n <\/tr>\n {% else %}\n <tr>\n <td>{{ s[0] if s|length > 0 else '' }}<\/td>\n <td>{{ s[2] if s|length > 2 else '' }}<\/td>\n <td>{{ s[1] if s|length > 1 else '' }}<\/td>\n <td>{{ s[4] if s|length > 4 else '' }}<\/td>\n <\/tr>\n {% endif %}\n {% endfor %}\n <\/table>\n <\/div>\n {% else %}<div class=\"empty\">No service data<\/div>{% endif %}\n <\/div>\n<\/div>\n\n<!-- \u2500\u2500 SCHEDULED TASKS \u2500\u2500 -->\n<div class=\"section\" id=\"tasks\">\n <div class=\"section-header\">\n <span class=\"section-title\">\/\/ scheduled tasks<\/span>\n <span class=\"section-count\">{{ scheduled_tasks|length }}<\/span>\n <\/div>\n <div class=\"section-body\">\n {% if scheduled_tasks %}\n <div class=\"tbl-wrap\">\n <table class=\"tbl\">\n <tr><th>Name<\/th><th>Status<\/th><th>Command<\/th><th>Arguments<\/th><\/tr>\n {% for t in scheduled_tasks %}\n <tr>\n {% if t is mapping %}\n <td>{{ t.get('Name', t.get('TaskName', t.get('name',''))) }}<\/td>\n <td>{{ t.get('Status', t.get('Enabled', t.get('state',''))) }}<\/td>\n <td>{{ t.get('Command', t.get('Action', t.get('cmd',''))) }}<\/td>\n <td style=\"word-break:break-all\">{{ t.get('Arguments', t.get('Args', t.get('args',''))) }}<\/td>\n {% else %}\n <td colspan=\"4\" style=\"white-space:pre-wrap;font-size:10px\">{{ t }}<\/td>\n <td><\/td><td><\/td><td><\/td>\n {% endif %}\n <\/tr>\n {% endfor %}\n <\/table>\n <\/div>\n {% else %}<div class=\"empty\">No scheduled tasks<\/div>{% endif %}\n <\/div>\n<\/div>\n\n<!-- \u2500\u2500 MODULES \u2500\u2500 -->\n<div class=\"section\" id=\"modules\">\n <div class=\"section-header\">\n <span class=\"section-title\">\/\/ loaded kernel modules<\/span>\n <span class=\"section-count\">{{ modules|length }}<\/span>\n <\/div>\n <div class=\"section-body\">\n {% if modules %}\n <div class=\"tbl-wrap\">\n <table class=\"tbl\">\n <tr><th>Name<\/th><th>Base<\/th><th>Size<\/th><th>Path<\/th><\/tr>\n {% for m in modules %}\n {% if m is mapping %}\n <tr>\n <td>{{ m.get('Name', m.get('BaseDllName','')) }}<\/td>\n <td style=\"font-family:var(--mono)\">{{ '0x%x' % m.get('Base',0) if m.get('Base') else m.get('Offset','') }}<\/td>\n <td>{{ m.get('Size','') }}<\/td>\n <td>{{ m.get('Path', m.get('FullDllName','')) }}<\/td>\n <\/tr>\n {% else %}\n <tr>\n <td>{{ m[1] if m|length > 1 else '' }}<\/td>\n <td>{{ m[0] if m|length > 0 else '' }}<\/td>\n <td>{{ m[2] if m|length > 2 else '' }}<\/td>\n <td>{{ m[3] if m|length > 3 else '' }}<\/td>\n <\/tr>\n {% endif %}\n {% endfor %}\n <\/table>\n <\/div>\n {% else %}<div class=\"empty\">No module data<\/div>{% endif %}\n <\/div>\n<\/div>\n\n<!-- \u2500\u2500 DRIVERS \u2500\u2500 -->\n<div class=\"section\" id=\"drivers\">\n <div class=\"section-header\">\n <span class=\"section-title\">\/\/ driver scan<\/span>\n <span class=\"section-count\">{{ drivers|length }}<\/span>\n <\/div>\n <div class=\"section-body\">\n {% if drivers %}\n <div class=\"tbl-wrap\">\n <table class=\"tbl\">\n <tr><th>Name<\/th><th>Base\/Offset<\/th><th>Size<\/th><th>Service Key<\/th><\/tr>\n {% for d in drivers %}\n {% if d is mapping %}\n <tr>\n <td>{{ d.get('Name', d.get('DriverName','')) }}<\/td>\n <td style=\"font-family:var(--mono)\">{{ d.get('Offset', d.get('Base','')) }}<\/td>\n <td>{{ d.get('Size','') }}<\/td>\n <td>{{ d.get('ServiceKey', d.get('DriverServiceName','')) }}<\/td>\n <\/tr>\n {% else %}\n <tr>\n <td>{{ d[0] if d|length > 0 else '' }}<\/td>\n <td>{{ d[1] if d|length > 1 else '' }}<\/td>\n <td>{{ d[2] if d|length > 2 else '' }}<\/td>\n <td>{{ d[3] if d|length > 3 else '' }}<\/td>\n <\/tr>\n {% endif %}\n {% endfor %}\n <\/table>\n <\/div>\n {% else %}<div class=\"empty\">No driver data<\/div>{% endif %}\n <\/div>\n<\/div>\n\n<!-- \u2500\u2500 HANDLES \u2500\u2500 -->\n<div class=\"section\" id=\"handles\">\n <div class=\"section-header\">\n <span class=\"section-title\">\/\/ open handles<\/span>\n <span class=\"section-count\">{{ handles|length }}{% if handles|length == 200 %} (capped - see JSON for all){% endif %}<\/span>\n <\/div>\n <div class=\"section-body\">\n {% if handles %}\n <div class=\"tbl-wrap\">\n <table class=\"tbl\">\n <tr><th>PID<\/th><th>Process<\/th><th>Type<\/th><th>Handle<\/th><th>Name<\/th><\/tr>\n {% for h in handles %}\n {% if h is mapping %}\n <tr>\n <td>{{ h.get('PID', h.get('Pid','')) }}<\/td>\n <td>{{ h.get('Process','') }}<\/td>\n <td>{{ h.get('Type','') }}<\/td>\n <td>{{ h.get('HandleValue', h.get('Handle','')) }}<\/td>\n <td>{{ h.get('Name','') }}<\/td>\n <\/tr>\n {% else %}\n <tr>\n <td>{{ h[0] if h|length > 0 else '' }}<\/td>\n <td>{{ h[1] if h|length > 1 else '' }}<\/td>\n <td>{{ h[2] if h|length > 2 else '' }}<\/td>\n <td>{{ h[3] if h|length > 3 else '' }}<\/td>\n <td>{{ h[4] if h|length > 4 else '' }}<\/td>\n <\/tr>\n {% endif %}\n {% endfor %}\n <\/table>\n <\/div>\n {% else %}<div class=\"empty\">No handle data<\/div>{% endif %}\n <\/div>\n<\/div>\n\n<!-- \u2500\u2500 IE HISTORY \u2500\u2500 -->\n<div class=\"section\" id=\"iehistory\">\n <div class=\"section-header\">\n <span class=\"section-title\">\/\/ ie history (vol2)<\/span>\n <span class=\"section-count\">{{ ie_history|length }}<\/span>\n <\/div>\n <div class=\"section-body\">\n {% if ie_history %}\n <div class=\"tbl-wrap\">\n <table class=\"tbl\">\n <tr><th>Process<\/th><th>URL<\/th><th>Modified<\/th><th>Accessed<\/th><\/tr>\n {% for r in ie_history %}\n <tr>\n {% if r is mapping %}\n <td>{{ r.get('Process', r.get('process','')) }}<\/td>\n <td>{{ r.get('URL', r.get('url','')) }}<\/td>\n <td>{{ r.get('Modified','') }}<\/td>\n <td>{{ r.get('Accessed','') }}<\/td>\n {% else %}\n <td colspan=\"4\" style=\"white-space:pre-wrap\">{{ r }}<\/td>\n {% endif %}\n <\/tr>\n {% endfor %}\n <\/table>\n <\/div>\n {% else %}<div class=\"empty\">No IE history (requires Vol2 iehistory plugin)<\/div>{% endif %}\n <\/div>\n<\/div>\n\n<!-- \u2500\u2500 PREFETCH \u2500\u2500 -->\n<div class=\"section\" id=\"prefetch\">\n <div class=\"section-header\">\n <span class=\"section-title\">\/\/ prefetch (vol2)<\/span>\n <span class=\"section-count\">{{ prefetch|length }}<\/span>\n <\/div>\n <div class=\"section-body\">\n {% if prefetch %}\n <div class=\"tbl-wrap\">\n <table class=\"tbl\">\n <tr><th>Application<\/th><th>Last Run<\/th><th>Run Count<\/th><th>Path<\/th><\/tr>\n {% for r in prefetch %}\n <tr>\n {% if r is mapping %}\n <td>{{ r.get('Application', r.get('Executable','')) }}<\/td>\n <td>{{ r.get('LastRun', r.get('last_run','')) }}<\/td>\n <td>{{ r.get('RunCount', r.get('run_count','')) }}<\/td>\n <td>{{ r.get('Path','') }}<\/td>\n {% else %}\n <td colspan=\"4\" style=\"white-space:pre-wrap;font-size:10px\">{{ r }}<\/td>\n {% endif %}\n <\/tr>\n {% endfor %}\n <\/table>\n <\/div>\n {% else %}<div class=\"empty\">No prefetch data (requires Vol2 prefetchparser plugin)<\/div>{% endif %}\n <\/div>\n<\/div>\n\n<!-- \u2500\u2500 SHIMCACHE \u2500\u2500 -->\n<div class=\"section\" id=\"shimcache\">\n <div class=\"section-header\">\n <span class=\"section-title\">\/\/ shimcache (vol2)<\/span>\n <span class=\"section-count\">{{ shimcache|length }}<\/span>\n <\/div>\n <div class=\"section-body\">\n {% if shimcache %}\n <div class=\"tbl-wrap\">\n <table class=\"tbl\">\n <tr><th>Last Modified<\/th><th>Last Update<\/th><th>Path<\/th><\/tr>\n {% for r in shimcache %}\n <tr>\n {% if r is mapping %}\n <td>{{ r.get('Last Modified','') }}<\/td>\n <td>{{ r.get('Last Update','') }}<\/td>\n <td>{{ r.get('Path','') }}<\/td>\n {% else %}\n <td colspan=\"3\" style=\"white-space:pre-wrap;font-size:10px\">{{ r }}<\/td>\n {% endif %}\n <\/tr>\n {% endfor %}\n <\/table>\n <\/div>\n {% else %}<div class=\"empty\">No shimcache data (requires Vol2 shimcache plugin)<\/div>{% endif %}\n <\/div>\n<\/div>\n\n<!-- \u2500\u2500 MFT \u2500\u2500 -->\n<div class=\"section\" id=\"mft\">\n <div class=\"section-header\">\n <span class=\"section-title\">\/\/ mft entries<\/span>\n <span class=\"section-count\">{{ mft_count }}<\/span>\n <\/div>\n <div class=\"section-body\">\n <div style=\"padding:16px;font-family:var(--mono);font-size:11px;color:var(--text2)\">\n {% if mft_count > 0 %}\n MFT data ({{ mft_count }} entries) is too large to render in the browser.<br>\n Full MFT written to: <span style=\"color:var(--amber)\">mft.txt<\/span> in this report folder.<br><br>\n To search: <span style=\"color:var(--green)\">grep -i \"filename.exe\" mft.txt<\/span>\n {% else %}\n No MFT data collected.\n {% endif %}\n <\/div>\n <\/div>\n<\/div>\n\n<!-- \u2500\u2500 ENVIRONMENT VARIABLES \u2500\u2500 -->\n<div class=\"section\" id=\"envars\">\n <div class=\"section-header\">\n <span class=\"section-title\">\/\/ environment variables<\/span>\n <span class=\"section-count\">{{ envars|length }}<\/span>\n <\/div>\n <div class=\"section-body\">\n {% if envars %}\n <div class=\"tbl-wrap\">\n <table class=\"tbl\">\n <tr><th>PID<\/th><th>Process<\/th><th>Variable<\/th><th>Value<\/th><\/tr>\n {% for e in envars %}\n {% if e is mapping %}\n <tr>\n <td>{{ e.get('PID', e.get('Pid','')) }}<\/td>\n <td>{{ e.get('Process', e.get('Name','')) }}<\/td>\n <td>{{ e.get('Variable', e.get('Key','')) }}<\/td>\n <td style=\"max-width:400px;overflow:hidden;text-overflow:ellipsis\">{{ e.get('Value','') }}<\/td>\n <\/tr>\n {% else %}\n <tr>\n <td>{{ e[0] if e|length > 0 else '' }}<\/td>\n <td>{{ e[1] if e|length > 1 else '' }}<\/td>\n <td>{{ e[2] if e|length > 2 else '' }}<\/td>\n <td>{{ e[3] if e|length > 3 else '' }}<\/td>\n <\/tr>\n {% endif %}\n {% endfor %}\n <\/table>\n <\/div>\n {% else %}<div class=\"empty\">No environment variable data<\/div>{% endif %}\n <\/div>\n<\/div>\n\n<!-- \u2500\u2500 COMMAND LINES \u2500\u2500 -->\n<div class=\"section\" id=\"cmdline\">\n <div class=\"section-header\">\n <span class=\"section-title\">\/\/ command line arguments<\/span>\n <span class=\"section-count\">{{ cmdline|length }}<\/span>\n <\/div>\n <div class=\"section-body\">\n {% if cmdline %}\n <div class=\"tbl-wrap\">\n <table class=\"tbl\">\n <tr><th>PID<\/th><th>Process<\/th><th>Command Line<\/th><\/tr>\n {% for c in cmdline %}\n {% if c is mapping %}\n <tr>\n <td>{{ c.get('PID', c.get('Pid','')) }}<\/td>\n <td>{{ c.get('Process', c.get('Name','')) }}<\/td>\n <td style=\"white-space:pre-wrap;word-break:break-all\">{{ c.get('Args', c.get('CommandLine', c.get('CmdLine',''))) }}<\/td>\n <\/tr>\n {% else %}\n <tr>\n <td>{{ c[0] if c|length > 0 else '' }}<\/td>\n <td>{{ c[1] if c|length > 1 else '' }}<\/td>\n <td style=\"white-space:pre-wrap;word-break:break-all\">{{ c[2] if c|length > 2 else '' }}<\/td>\n <\/tr>\n {% endif %}\n {% endfor %}\n <\/table>\n <\/div>\n {% else %}<div class=\"empty\">No command line data<\/div>{% endif %}\n <\/div>\n<\/div>\n\n<\/div><!-- \/container -->\n<\/body>\n<\/html>\"\"\"\n\ndef _row_type(rows):\n if not rows:\n return \"empty\"\n r = rows[0]\n if hasattr(r, 'get'):\n return f\"dict ({r.__class__.__name__}): keys={list(r.keys())[:4]}\"\n try:\n return f\"list (len={len(r)})\"\n except Exception:\n return f\"unknown: {r.__class__.__name__}\"\n\ndef _get(plugin_results: Dict, *keys):\n \"\"\"Get plugin results by multiple possible key names, return first non-empty.\"\"\"\n for k in keys:\n v = plugin_results.get(k)\n if v:\n return v\n return []\n\ndef _parse_vol2_text(raw: str, min_cols: int = 2) -> List:\n \"\"\"\n Parse Vol2 text output into a list of dicts using the header row as keys.\n Falls back to returning non-empty stripped lines as strings.\n \"\"\"\n if not raw or not raw.strip():\n return []\n lines = [l for l in raw.splitlines() if l.strip()]\n if len(lines) < 2:\n return [l.strip() for l in lines if l.strip()]\n\n # Try to use first line as header\n header = lines[0]\n # Vol2 headers use spaces to separate - detect by checking for all-caps words\n import re\n cols = re.split(r'\\s{2,}', header.strip())\n if len(cols) >= min_cols and any(c.isupper() or c[0].isupper() for c in cols if c):\n result = []\n for line in lines[1:]:\n if line.startswith('-') or line.startswith('*'):\n continue\n parts = re.split(r'\\s{2,}', line.strip())\n row = {}\n for i, col in enumerate(cols):\n row[col.strip()] = parts[i].strip() if i < len(parts) else ''\n result.append(row)\n return result\n # Fallback: return lines as strings\n return [l.strip() for l in lines if l.strip() and not l.startswith('Volatility')]\n\ndef generate_reports(meta: Dict, findings: List, iocs: Dict,\n plugin_results: Dict) -> Dict:\n \"\"\"Generate all report files in a structured output folder.\"\"\"\n\n risk_score = sum(f.score for f in findings)\n risk_label = (\n 'CRITICAL' if risk_score >= 60 else\n 'HIGH' if risk_score >= 30 else\n 'MEDIUM' if risk_score >= 10 else\n 'LOW'\n )\n\n image_stem = Path(meta['filename']).stem\n timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')\n report_dir = REPORT_BASE \/ f\"{image_stem}_{timestamp}\"\n dumps_dir = report_dir \/ 'dumps'\n report_dir.mkdir(parents=True, exist_ok=True)\n dumps_dir.mkdir(exist_ok=True)\n\n findings_dicts = [f.to_dict() for f in findings]\n\n # \u2500\u2500 Move any existing dump files into dumps\/ \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n # Files written by dumpfiles or carving tools in the working directory\n import glob, os\n for pattern in ['*.dmp', '*.img', '*.dat', '*.vacb', 'file.*.img', 'file.*.vacb', '*.raw.dir']:\n for f in glob.glob(f'\/opt\/memory-hunter\/{pattern}'):\n try:\n shutil.move(f, dumps_dir \/ Path(f).name)\n log.info(f\"Moved dump file: {Path(f).name} -> dumps\/\")\n except Exception:\n pass\n\n # \u2500\u2500 Resolve all plugin data with fallbacks \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n network_rows = _get(plugin_results, 'windows.netscan', 'netscan', 'linux.netstat')\n process_rows = _get(plugin_results, 'windows.pslist', 'pslist', 'linux.pslist')\n services_raw = _get(plugin_results, 'windows.svcscan', 'svcscan')\n modules_raw = _get(plugin_results, 'windows.modules', 'modules', 'linux.lsmod')\n drivers_raw = _get(plugin_results, 'windows.driverscan', 'driverscan')\n handles_raw = _get(plugin_results, 'windows.handles', 'handles')\n cmdline_raw = _get(plugin_results, 'windows.cmdline', 'cmdline')\n mft_raw = _get(plugin_results, 'windows.mftscan', 'mftparser')\n envars_raw = _get(plugin_results, 'windows.envars', 'envars')\n tasks_raw = _get(plugin_results, 'windows.scheduled_tasks', 'scheduled_tasks')\n\n # Vol2-only plugins return text - parse to structured rows\n ie_history_raw = _get(plugin_results, 'iehistory')\n prefetch_raw = _get(plugin_results, 'prefetchparser')\n shimcache_raw = _get(plugin_results, 'shimcache')\n\n ie_history = ie_history_raw if isinstance(ie_history_raw, list) else _parse_vol2_text(str(ie_history_raw))\n prefetch = prefetch_raw if isinstance(prefetch_raw, list) else _parse_vol2_text(str(prefetch_raw))\n shimcache = shimcache_raw if isinstance(shimcache_raw, list) else _parse_vol2_text(str(shimcache_raw))\n\n log.info(f\"network:{len(network_rows)} processes:{len(process_rows)} \"\n f\"services:{len(services_raw)} modules:{len(modules_raw)} \"\n f\"drivers:{len(drivers_raw)} handles:{len(handles_raw)}\")\n log.info(f\"ie_history:{len(ie_history)} prefetch:{len(prefetch)} \"\n f\"shimcache:{len(shimcache)} mft:{len(mft_raw)}\")\n\n summary_data = {\n 'critical': sum(1 for f in findings if f.severity == 'CRITICAL'),\n 'high': sum(1 for f in findings if f.severity == 'HIGH'),\n 'medium': sum(1 for f in findings if f.severity == 'MEDIUM'),\n 'yara_hits': len(iocs.get('yara_hits', [])),\n 'ioc_count': sum(len(v) for v in iocs.values()),\n }\n\n # \u2500\u2500 report.json \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n json_data = {\n 'meta': meta,\n 'generated_at': datetime.now().isoformat(),\n 'risk_label': risk_label,\n 'risk_score': risk_score,\n 'findings': findings_dicts,\n 'iocs': iocs,\n 'summary': summary_data,\n 'network': network_rows[:1000],\n 'processes': process_rows[:1000],\n 'services': services_raw[:1000],\n 'modules': modules_raw[:500],\n 'drivers': drivers_raw[:500],\n 'scheduled_tasks':tasks_raw[:500],\n 'ie_history': ie_history[:500],\n 'prefetch': prefetch[:500],\n 'shimcache': shimcache[:1000],\n 'mft': mft_raw[:50], # capped - full data in mft.txt\n }\n\n json_path = report_dir \/ 'report.json'\n with open(json_path, 'w', encoding='utf-8') as f:\n json.dump(json_data, f, indent=2, default=str)\n\n # \u2500\u2500 report.html \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n env = Environment(autoescape=False)\n tmpl = env.from_string(HTML_TEMPLATE)\n # Cap rows passed to HTML to prevent browser freeze\n # Large datasets (14k handles, 13M MFT) are in JSON for full access\n HTML_ROW_CAPS = {\n 'network': 500,\n 'processes': 500,\n 'services': 500,\n 'modules': 300,\n 'drivers': 300,\n 'handles': 200, # 14k handles would freeze any browser\n 'cmdline': 500,\n 'mft': 50, # written to mft.txt - minimal in JSON\n 'envars': 300,\n 'scheduled_tasks':300,\n 'ie_history': 300,\n 'prefetch': 300,\n 'shimcache': 500,\n }\n\n def cap(data, key):\n limit = HTML_ROW_CAPS.get(key, 500)\n if isinstance(data, list) and len(data) > limit:\n log.info(f\"HTML cap: {key} {len(data)} -> {limit} rows (full data in JSON)\")\n return data[:limit]\n return data or []\n\n html = tmpl.render(\n meta=meta,\n generated_at=datetime.now().strftime('%Y-%m-%d %H:%M:%S UTC'),\n risk_label=risk_label,\n risk_score=risk_score,\n findings=findings_dicts,\n iocs=iocs,\n summary=summary_data,\n network=cap(network_rows, 'network'),\n processes=cap(process_rows, 'processes'),\n services=cap(services_raw, 'services'),\n modules=cap(modules_raw, 'modules'),\n drivers=cap(drivers_raw, 'drivers'),\n handles=cap(handles_raw, 'handles'),\n cmdline=cap(cmdline_raw, 'cmdline'),\n mft_count=len(mft_raw) if isinstance(mft_raw, list) else 0,\n envars=cap(envars_raw, 'envars'),\n scheduled_tasks=cap(tasks_raw, 'scheduled_tasks'),\n ie_history=cap(ie_history, 'ie_history'),\n prefetch=cap(prefetch, 'prefetch'),\n shimcache=cap(shimcache, 'shimcache'),\n )\n\n html_path = report_dir \/ 'report.html'\n with open(html_path, 'w', encoding='utf-8') as f:\n f.write(html)\n\n # \u2500\u2500 summary.txt \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n lines = [\n \"=\" * 62,\n \" MEMORY ANALYSIS SUMMARY\",\n \"=\" * 62,\n f\" Image: {meta['filename']}\",\n f\" SHA256: {meta['sha256'][:32]}...\",\n f\" Size: {meta['size_gb']} GB\",\n f\" OS: {meta.get('os_version','Unknown')}\",\n f\" Vol2 Profile: {meta.get('vol2_profile','Not detected')}\",\n f\" Analysed: {datetime.now().strftime('%Y-%m-%d %H:%M:%S UTC')}\",\n \"-\" * 62,\n f\" RISK: {risk_label} (Score: {risk_score})\",\n \"-\" * 62,\n f\" Findings: {len(findings)} ({summary_data['critical']} critical, {summary_data['high']} high, {summary_data['medium']} medium)\",\n f\" Yara hits: {summary_data['yara_hits']}\",\n f\" IOCs: {summary_data['ioc_count']}\",\n f\" Network: {len(network_rows)} connections\",\n f\" Processes: {len(process_rows)}\",\n f\" Services: {len(services_raw)}\",\n f\" Drivers: {len(drivers_raw)}\",\n f\" IE History: {len(ie_history)}\",\n f\" Prefetch: {len(prefetch)}\",\n \"-\" * 62,\n ]\n\n if findings_dicts:\n lines.append(\" TOP FINDINGS:\")\n for fd in findings_dicts[:15]:\n lines.append(f\" [{fd['severity']:8}] {fd['title']}\")\n if fd.get('process'):\n lines.append(f\" Process: {fd['process']} PID:{fd.get('pid','?')}\")\n for dl in str(fd.get('detail','')).splitlines()[:1]:\n if dl.strip():\n lines.append(f\" {dl[:78]}\")\n lines.append(\"-\" * 62)\n\n if iocs.get('yara_hits'):\n lines.append(\" YARA HITS:\")\n for hit in iocs['yara_hits']:\n lines.append(f\" -> {hit}\")\n lines.append(\"-\" * 62)\n\n if iocs.get('ipv4'):\n lines.append(\" EXTERNAL IPs:\")\n for ip in iocs['ipv4']:\n lines.append(f\" -> {ip}\")\n lines.append(\"-\" * 62)\n\n lines += [\n f\" report.html {html_path}\",\n f\" report.json {json_path}\",\n f\" iocs.txt {report_dir \/ 'iocs.txt'}\",\n f\" dumps\/ {dumps_dir}\",\n \"=\" * 62,\n ]\n\n summary_path = report_dir \/ 'summary.txt'\n summary_path.write_text('\\n'.join(lines) + '\\n', encoding='utf-8')\n\n # \u2500\u2500 iocs.txt \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n ioc_lines = [\n f\"IOCs: {meta['filename']}\",\n f\"Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S UTC')}\",\n f\"Risk: {risk_label}\",\n \"\",\n ]\n ioc_labels = {\n 'ipv4':'IP Addresses', 'url':'URLs', 'domain':'Domains',\n 'pipe':'Named Pipes', 'registry_run':'Registry Run Keys',\n 'hash_sha256':'SHA256 Hashes', 'hash_md5':'MD5 Hashes',\n 'email':'Email Addresses', 'base64_large':'Base64 Data',\n 'yara_hits':'Yara Matches',\n }\n for t, vals in iocs.items():\n if vals:\n ioc_lines += [f\"--- {ioc_labels.get(t,t.replace('_',' ').title())} ({len(vals)}) ---\"]\n ioc_lines += [str(v) for v in vals]\n ioc_lines.append(\"\")\n\n iocs_path = report_dir \/ 'iocs.txt'\n iocs_path.write_text('\\n'.join(ioc_lines) + '\\n', encoding='utf-8')\n\n # Write MFT to its own file - too large for HTML or JSON\n mft_path = report_dir \/ 'mft.txt'\n if isinstance(mft_raw, list) and mft_raw:\n with open(mft_path, 'w', encoding='utf-8', errors='replace') as f:\n f.write(f\"MFT entries from {meta['filename']}\\n\")\n f.write(f\"Total entries: {len(mft_raw)}\\n\")\n f.write(\"=\" * 80 + \"\\n\")\n for row in mft_raw:\n if isinstance(row, dict):\n f.write(\"\\t\".join(str(v) for v in row.values()) + \"\\n\")\n else:\n f.write(str(row) + \"\\n\")\n log.info(f\"MFT written to mft.txt ({len(mft_raw)} entries)\")\n elif isinstance(mft_raw, str) and mft_raw.strip():\n mft_path.write_text(mft_raw, encoding='utf-8', errors='replace')\n log.info(f\"MFT written to mft.txt\")\n\n log.info(f\"Reports saved to {report_dir}\/\")\n log.info(f\" report.html | report.json | summary.txt | iocs.txt | dumps\/\")\n\n return {\n 'report_dir': str(report_dir),\n 'html': str(html_path),\n 'json': str(json_path),\n 'summary': str(summary_path),\n 'iocs': str(iocs_path),\n 'dumps': str(dumps_dir),\n 'mft_txt': str(mft_path) if mft_raw else None,\n 'risk_label': risk_label,\n 'risk_score': risk_score,\n 'finding_count': len(findings),\n }\n<\/pre>\nThe main entry point: analyse.py<\/h3>\n
#!\/usr\/bin\/env python3\n## \/opt\/memory-hunter\/analyse.py\n## THE ONE SCRIPT TO RUN\n## Usage: python3 analyse.py \/path\/to\/memory.raw\n## Usage: python3 analyse.py \/path\/to\/memory.raw --workers 16 --no-vol2\n\nimport sys\nimport argparse\nimport logging\nimport time\nimport json\nfrom datetime import datetime\nfrom pathlib import Path\n\n# Add scripts directory to path\nsys.path.insert(0, str(Path(__file__).parent \/ 'scripts'))\n\nfrom phase1_identify import identify_image\nfrom vol_runner import collect_all_plugins\nfrom anomaly_checks import run_all_checks\nfrom phase4_iocs import extract_iocs\nfrom phase5_report import generate_reports\n\ndef setup_logging(log_file: str = None) -> None:\n handlers = [logging.StreamHandler(sys.stdout)]\n if log_file:\n handlers.append(logging.FileHandler(log_file))\n logging.basicConfig(\n level=logging.INFO,\n format='%(asctime)s [%(levelname)s] %(message)s',\n handlers=handlers\n )\n\ndef print_banner():\n print(\"\"\"\n\u2554\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2557\n\u2551 Memory Hunter - Automated Analysis \u2551\n\u2551 Volatility 2 + 3 | Windows + Linux | HTML + JSON \u2551\n\u255a\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u255d\n\"\"\")\n\ndef print_summary(report_info: dict, elapsed: float):\n risk = report_info['risk_label']\n colour = {\n 'CRITICAL': '\\033[91m', 'HIGH': '\\033[93m',\n 'MEDIUM': '\\033[33m', 'LOW': '\\033[92m'\n }.get(risk, '')\n reset = '\\033[0m'\n\n print(f\"\"\"\n{'='*60}\n ANALYSIS COMPLETE ({elapsed:.1f} seconds)\n{'='*60}\n Risk Level: {colour}{risk}{reset}\n Risk Score: {report_info['risk_score']}\n Findings: {report_info['finding_count']}\n HTML Report: {report_info['html']}\n JSON Report: {report_info['json']}\n{'='*60}\n\"\"\")\n\ndef main():\n print_banner()\n\n parser = argparse.ArgumentParser(\n description='Automated memory image analysis for threat hunters'\n )\n parser.add_argument('image', help='Path to memory image file')\n parser.add_argument('--workers', type=int, default=8,\n help='Parallel plugin workers (default: 8)')\n parser.add_argument('--no-vol2', action='store_true',\n help='Skip Volatility 2 plugins')\n parser.add_argument('--output-dir', default='\/opt\/memory-hunter\/reports',\n help='Report output directory')\n parser.add_argument('--yara-rules', default='\/opt\/memory-hunter\/yara_rules\/combined.yar',\n help='Path to compiled Yara rules')\n parser.add_argument('--log-file', help='Write log to file')\n parser.add_argument('--quiet', action='store_true',\n help='Reduce output verbosity')\n args = parser.parse_args()\n\n setup_logging(args.log_file)\n log = logging.getLogger(__name__)\n\n start_time = time.time()\n image_path = str(Path(args.image).absolute())\n\n print(f\"[*] Image: {image_path}\")\n print(f\"[*] Workers: {args.workers}\")\n print()\n\n # \u2500\u2500 Phase 1: Identify \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n print(\"[Phase 1\/5] Identifying image...\")\n meta = identify_image(image_path)\n print(f\" OS Type: {meta.get('os_type', 'unknown')}\")\n print(f\" OS Version: {meta.get('os_version', 'unknown')}\")\n print(f\" Size: {meta.get('size_gb', 0):.1f} GB\")\n print(f\" SHA256: {meta.get('sha256', '')[:32]}...\")\n print()\n\n if not meta.get('os_type'):\n print(\"[!] Could not identify OS type - check symbol tables\")\n print(\" Windows: ensure \/opt\/vol3-symbols\/windows\/ is populated\")\n print(\" Linux: ensure ISF file exists for this kernel version\")\n sys.exit(1)\n\n # \u2500\u2500 Phase 2: Collect \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n print(f\"[Phase 2\/5] Running plugin collection ({args.workers} workers)...\")\n plugin_results = collect_all_plugins(image_path, meta, args.workers)\n print(f\" Plugins completed: {len(plugin_results)}\")\n print()\n\n # \u2500\u2500 Phase 3: Detect \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n print(\"[Phase 3\/5] Running anomaly detection...\")\n findings, risk_score = run_all_checks(plugin_results)\n critical = sum(1 for f in findings if f.severity == 'CRITICAL')\n high = sum(1 for f in findings if f.severity == 'HIGH')\n print(f\" Findings: {len(findings)} total ({critical} critical, {high} high)\")\n if findings:\n print(\" Top findings:\")\n for f in findings[:5]:\n print(f\" [{f.severity}] {f.title}\")\n print()\n\n # \u2500\u2500 Phase 4: IOCs \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n print(\"[Phase 4\/5] Extracting IOCs...\")\n suspicious_pids = [f.pid for f in findings if f.pid]\n iocs = extract_iocs(plugin_results, image_path, suspicious_pids)\n ioc_count = sum(len(v) for v in iocs.values())\n print(f\" IOCs extracted: {ioc_count}\")\n if iocs.get('yara_hits'):\n print(f\" YARA hits: {len(iocs['yara_hits'])}\")\n for hit in iocs['yara_hits'][:3]:\n print(f\" -> {hit}\")\n print()\n\n # \u2500\u2500 Phase 5: Report \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\n print(\"[Phase 5\/5] Generating reports...\")\n report_info = generate_reports(meta, findings, iocs, plugin_results)\n elapsed = time.time() - start_time\n\n print_summary(report_info, elapsed)\n\n # Exit code reflects risk level for CI\/CD integration\n exit_codes = {'CRITICAL': 3, 'HIGH': 2, 'MEDIUM': 1, 'LOW': 0}\n sys.exit(exit_codes.get(report_info['risk_label'], 0))\n\nif __name__ == '__main__':\n main()<\/pre>\nMaking it truly one command to run<\/h3>\n
## Install the script system-wide\nsudo ln -sf \/opt\/memory-hunter\/analyse.py \/usr\/local\/bin\/memory-hunt\nsudo chmod +x \/opt\/memory-hunter\/analyse.py\n\n## Now you can run from anywhere:\nmemory-hunt \/path\/to\/image.raw\n\n## With options:\nmemory-hunt \/path\/to\/image.raw --workers 16 --log-file \/tmp\/analysis.log\n\n## In a pipeline (exit code reflects risk):\nmemory-hunt suspicious.raw && echo \"CLEAN\" || echo \"THREATS FOUND\"\n\n## Process multiple images in parallel\nls \/srv\/memory\/landing\/*.raw | \\\n parallel -j 4 memory-hunt {} --log-file \/srv\/memory\/logs\/{\/.}.log\n\n## Quick check with reduced scope (faster for initial triage)\nmemory-hunt image.raw --workers 4 --no-vol2<\/pre>\nLinux image support and ISF generation<\/h3>\n
## Linux memory analysis requires ISF (Intermediate Symbol Format) files\n## These must be generated for each specific kernel version being analysed\n\n## Method 1: Generate ISF from a running system (same kernel as the image)\n## Install dwarf2json\nwget https:\/\/github.com\/volatilityfoundation\/dwarf2json\/releases\/latest\/download\/dwarf2json-linux-amd64\nchmod +x dwarf2json-linux-amd64\n\n## Generate ISF from the running kernel (on the target or identical system)\nsudo .\/dwarf2json-linux-amd64 linux \\\n --elf \/usr\/lib\/debug\/boot\/vmlinux-$(uname -r) \\\n > \/opt\/vol3-symbols\/linux\/$(uname -r).json\n\n## Method 2: Generate from vmlinux debug symbols package\n## On Ubuntu\/Debian:\nsudo apt install linux-image-$(uname -r)-dbgsym 2>\/dev\/null || \\\n sudo apt install linux-image-$(uname -r)-dbg\nsudo .\/dwarf2json-linux-amd64 linux \\\n --elf \/usr\/lib\/debug\/boot\/vmlinux-$(uname -r) \\\n --system-map \/boot\/System.map-$(uname -r) \\\n > \/opt\/vol3-symbols\/linux\/$(uname -r).json\n\n## Verify Volatility can use the ISF\nvol -f linux_memory.lime linux.pslist\n\n## Automating ISF generation for a fleet\n## Run this on each unique kernel version in your environment\npython3 << 'EOF'\nimport subprocess, os, sys\nfrom pathlib import Path\n\nISF_DIR = Path('\/opt\/vol3-symbols\/linux')\nISF_DIR.mkdir(parents=True, exist_ok=True)\n\nkernel_version = subprocess.run(['uname', '-r'],\n capture_output=True, text=True).stdout.strip()\nisf_path = ISF_DIR \/ f"{kernel_version}.json"\n\nif isf_path.exists():\n print(f"ISF already exists for {kernel_version}")\n sys.exit(0)\n\n# Try to find vmlinux debug symbols\nvmlinux_paths = [\n f'\/usr\/lib\/debug\/boot\/vmlinux-{kernel_version}',\n f'\/usr\/lib\/debug\/lib\/modules\/{kernel_version}\/vmlinux',\n f'\/boot\/vmlinux-{kernel_version}',\n]\n\nvmlinux = next((p for p in vmlinux_paths if os.path.exists(p)), None)\nif not vmlinux:\n print(f"No debug symbols found for {kernel_version}")\n print(f"Install: apt install linux-image-{kernel_version}-dbgsym")\n sys.exit(1)\n\nresult = subprocess.run([\n '\/opt\/dwarf2json-linux-amd64', 'linux',\n '--elf', vmlinux,\n], capture_output=True, timeout=300)\n\nif result.returncode == 0:\n isf_path.write_bytes(result.stdout)\n print(f"ISF generated: {isf_path} ({len(result.stdout)\/\/1024}KB)")\nelse:\n print(f"dwarf2json failed: {result.stderr}")\n sys.exit(1)\nEOF<\/pre>\nTroubleshooting the automation pipeline<\/h3>\n
## Common issues and their fixes\n\n## Issue 1: \"Unsatisfied requirement\" errors from Volatility 3\n## This means symbol tables are missing or mislinked\nsource \/opt\/vol3-env\/bin\/activate\npython3 -c \"\nimport volatility3.symbols as sym\nimport os\nsym_path = os.path.dirname(sym.__file__)\nwin_path = os.path.join(sym_path, 'windows')\nprint(f'Symbol path: {sym_path}')\nprint(f'Windows syms: {os.path.isdir(win_path)}')\nif os.path.isdir(win_path):\n files = os.listdir(win_path)\n print(f'Files: {len(files)} (sample: {files[:2]})')\n\"\n\n## Fix: re-link symbols\nSITE=$(python3 -c \"import site; print(site.getsitepackages()[0])\")\nln -sf \/opt\/vol3-symbols\/windows $SITE\/volatility3\/symbols\/windows\n\n## Issue 2: Plugin times out on large images\n## Increase timeout values in vol_runner.py or limit to core plugins only\n## The --workers flag does not help here - timeouts are per-plugin\n\n## Issue 3: Yara scan returns no results despite known malware in image\n## Check compiled ruleset is not corrupted\npython3 -c \"\nimport yara\ntry:\n rules = yara.load('\/opt\/memory-hunter\/yara_rules\/combined.yar')\n print('Ruleset loaded OK')\nexcept Exception as e:\n print(f'Error: {e} - recompile the ruleset')\n\"\n\n## Recompile from source rules:\npython3 -c \"\nimport yara, glob\nrule_files = {}\nfor f in glob.glob('\/opt\/memory-hunter\/yara_rules\/rules\/*.yar'):\n rule_files[f.split('\/')[-1].replace('.yar','')] = f\ncombined = yara.compile(filepaths=rule_files)\ncombined.save('\/opt\/memory-hunter\/yara_rules\/combined.yar')\nprint(f'Compiled {len(rule_files)} rule files')\n\"\n\n## Issue 4: HTML report renders but shows no data\n## The template assumes Vol3 list format [pid, ppid, name, ...path at index 10]\n## Different Vol3 versions may change column ordering\n## Debug by checking raw plugin output:\nvol -f image.raw --renderer json windows.pslist | \\\n python3 -c \"import json,sys; d=json.load(sys.stdin); print(d['columns'])\"\n\n## Issue 5: analyse.py runs but risk score is always 0\n## Check anomaly_checks.py is receiving data by printing plugin result sizes:\npython3 -c \"\nimport sys; sys.path.insert(0,'\/opt\/memory-hunter\/scripts')\nfrom vol_runner import collect_all_plugins\nfrom phase1_identify import identify_image\nmeta = identify_image(sys.argv[1])\nresults = collect_all_plugins(sys.argv[1], meta, 4)\nfor name, data in sorted(results.items()):\n count = len(data) if isinstance(data, list) else len(data.splitlines())\n print(f'{name}: {count} records')\n\" \/path\/to\/image.raw<\/pre>\nIntegrating with the fleet pipeline<\/h3>\n
The single-image automation script integrates cleanly with the fleet collection pipeline described in the companion post. When the image watcher detects a new image it can call the analyse.py script directly instead of the Celery task chain, which is useful for simpler deployments that do not need the full distributed pipeline.<\/p>\n
## Simple integration with the image watcher\n## Replace the validate_image.delay() call with a direct script invocation\n\n## In image_watcher.py, replace:\n## validate_image.delay(image_id)\n\n## With:\nimport subprocess\nsubprocess.Popen([\n '\/opt\/vol3-env\/bin\/python3',\n '\/opt\/memory-hunter\/analyse.py',\n str(path),\n '--workers', '8',\n '--log-file', f'\/srv\/memory\/logs\/{hostname}_{timestamp}.log',\n '--output-dir', f'\/srv\/memory\/reports\/{hostname}',\n])\n\n## The exit code from analyse.py maps to risk level:\n## 0 = LOW (no concerning findings)\n## 1 = MEDIUM\n## 2 = HIGH\n## 3 = CRITICAL\n## Use this in automation to trigger different response actions\n\n## Example: auto-isolate a host if CRITICAL findings\nresult = subprocess.run([\n '\/opt\/vol3-env\/bin\/python3',\n '\/opt\/memory-hunter\/analyse.py',\n image_path,\n], capture_output=True)\n\nif result.returncode == 3:\n log.warning(f\"CRITICAL findings in {hostname} - triggering isolation workflow\")\n # Call your EDR\/firewall API to isolate the host\n isolate_host(hostname)\nelif result.returncode == 2:\n log.warning(f\"HIGH findings in {hostname} - notifying SOC\")\n notify_soc(hostname)<\/pre>\nThe HTML report template in full<\/h3>\n
The report template referenced in phase5_report.py is a Jinja2 template that lives at
\/opt\/memory-hunter\/templates\/report.html.j2<\/code>. The template code was included inline in the phase5_report.py listing above as theHTML_TEMPLATE<\/code> string. To use it as a standalone file instead, replace the inline string with a file load:<\/p>\n## In phase5_report.py, replace the HTML_TEMPLATE string with:\nfrom jinja2 import Environment, FileSystemLoader\n\nenv = Environment(loader=FileSystemLoader('\/opt\/memory-hunter\/templates'))\ntmpl = env.get_template('report.html.j2')\nhtml = tmpl.render(...)<\/pre>\nSave the template content from the
HTML_TEMPLATE<\/code> variable in phase5_report.py to\/opt\/memory-hunter\/templates\/report.html.j2<\/code>. The template uses standard Jinja2 syntax throughout:{{ variable }}<\/code> for output,{% for item in list %}<\/code> for loops,{% if condition %}<\/code> for conditionals. No additional template dependencies are needed beyond Jinja2 itself.<\/p>\nYara rules: complete content for all four rule files<\/h3>\n
These are the four rule files that belong in
\/opt\/memory-hunter\/yara_rules\/rules\/<\/code>. Each focuses on a different threat category relevant to Windows endpoint memory analysis.<\/p>\n## File: cobalt_strike.yar\n\/\/ Cobalt Strike beacon detection rules\n\nrule CobaltStrike_Beacon_Config_Decoded {\n meta:\n description = \"Detects decoded Cobalt Strike beacon configuration in process memory\"\n author = \"justruss\"\n date = \"2026-05-24\"\n confidence = \"high\"\n strings:\n $cfg_header = { 00 01 00 01 00 00 00 ?? 00 02 00 01 }\n $uri_check = \"\/updates\/check\" ascii wide\n $uri_submit = \"\/submit.php\" ascii wide\n $uri_cdn = \"\/CDN\/\" ascii wide\n $sleep_mask = { C7 44 24 ?? 01 00 00 00 EB ?? }\n $ref_loader = \"ReflectiveLoader\" ascii fullword\n $pipe_msse = \"\\\\.\\pipe\\MSSE-\" ascii wide\n $pipe_postex = \"\\\\.\\pipe\\postex_\" ascii wide\n $pipe_status = \"\\\\.\\pipe\\status_\" ascii wide\n $watermark = { 00 27 00 01 }\n condition:\n ($ref_loader or $sleep_mask) and\n 1 of ($uri_check, $uri_submit, $uri_cdn, $cfg_header, $watermark)\n or 2 of ($pipe_msse, $pipe_postex, $pipe_status)\n}\n\nrule CobaltStrike_Shellcode_Stager {\n meta:\n description = \"Detects Cobalt Strike shellcode stager in memory\"\n confidence = \"medium\"\n strings:\n $stager_x64 = { FC 48 83 E4 F0 E8 C0 00 00 00 }\n $stager_x86 = { FC E8 82 00 00 00 60 89 E5 }\n condition:\n any of them\n}\n\nrule CobaltStrike_MalleableC2_Indicators {\n meta:\n description = \"Detects indicators of Cobalt Strike Malleable C2 profiles\"\n confidence = \"low\"\n strings:\n $ref_loader = \"ReflectiveLoader\" ascii fullword\n $sleep_mask = { C7 44 24 ?? 01 00 00 00 EB ?? }\n $amz_host = \"s3.amazonaws.com\" ascii wide\n $o365 = \"outlook.office365.com\" ascii wide\n $ua_excel = \"Microsoft Excel\" ascii wide\n $ua_teams = \"Teams\/1.\" ascii wide\n condition:\n ($ref_loader or $sleep_mask) and\n 1 of ($amz_host, $o365, $ua_excel, $ua_teams)\n}\n<\/pre>\n## File: meterpreter.yar\n\/\/ Meterpreter detection rules\n\nrule Meterpreter_Reflective_DLL_x64 {\n meta:\n description = \"Detects Meterpreter x64 reflective DLL loaded in process memory\"\n author = \"justruss\"\n date = \"2026-05-24\"\n confidence = \"high\"\n strings:\n $mz = { 4D 5A }\n $ref_loader = \"ReflectiveLoader\" ascii fullword\n $stdapi = \"stdapi_\" ascii\n $priv = \"priv_elevate\" ascii\n $incognito = \"incognito_\" ascii\n $kiwi = \"kiwi_cmd\" ascii\n $transport = \"METERPRETER_TRANSPORT_\" ascii\n $pivot = \"pivot_\" ascii\n $session_chan = \"MeterpreterSession\" ascii wide nocase\n condition:\n $mz at 0 and $ref_loader and\n 2 of ($stdapi, $priv, $incognito, $kiwi, $transport, $pivot, $session_chan)\n}\n\nrule Meterpreter_Shellcode_Reverse_TCP {\n meta:\n description = \"Detects Meterpreter reverse TCP shellcode in memory\"\n confidence = \"high\"\n strings:\n $rev_tcp_x64 = { 49 BE ?? ?? ?? ?? ?? ?? ?? ?? 41 FF E6 }\n $lib_resolve = { 48 31 C9 48 81 EC D0 00 00 00 }\n condition:\n any of them\n}\n\nrule Meterpreter_Python_Stage {\n meta:\n description = \"Detects Python Meterpreter stage in memory\"\n confidence = \"medium\"\n strings:\n $py_met1 = \"met_api\" ascii\n $py_met2 = \"meterpreter.core\" ascii\n $py_met3 = \"MeterpreterSession\" ascii\n $py_met4 = \"from metasploit\" ascii nocase\n condition:\n 2 of them\n}\n<\/pre>\n## File: credential_tools.yar\n\/\/ Credential access tool detection rules\n\nrule Mimikatz_In_Memory {\n meta:\n description = \"Detects Mimikatz and variants loaded in process memory\"\n author = \"justruss\"\n date = \"2026-05-24\"\n confidence = \"high\"\n strings:\n $sekurlsa = \"sekurlsa::\" ascii wide nocase\n $lsadump = \"lsadump::\" ascii wide nocase\n $kerberos = \"kerberos::\" ascii wide nocase\n $crypto = \"crypto::\" ascii wide nocase\n $dpapi = \"dpapi::\" ascii wide nocase\n $mimikatz = \"mimikatz\" ascii wide nocase\n $priv_debug = \"privilege::debug\" ascii wide nocase\n $logonpw = \"logonPasswords\" ascii wide\n $wdigest = \"wdigest.dll\" ascii wide\n $lsasrv = \"lsasrv.dll\" ascii wide\n $ntlm_hash = \"NTLM hash\" ascii wide nocase\n $aes256_key = \"AES256 HMAC\" ascii wide\n condition:\n 2 of ($sekurlsa, $lsadump, $kerberos, $crypto, $dpapi, $mimikatz, $priv_debug)\n or ($wdigest and $lsasrv and 1 of ($sekurlsa, $lsadump, $logonpw))\n or ($ntlm_hash and $aes256_key and $wdigest)\n}\n\nrule Rubeus_Kerberos_Toolkit {\n meta:\n description = \"Detects Rubeus .NET Kerberos attack toolkit in memory\"\n confidence = \"high\"\n strings:\n $rubeus_id = \"Rubeus\" ascii wide\n $asktgt = \"asktgt\" ascii wide nocase\n $kerberoast = \"kerberoast\" ascii wide nocase\n $asreproast = \"asreproast\" ascii wide nocase\n $s4u = \" s4u \" ascii wide nocase\n $ptt = \"ptt\" ascii wide\n $harvest = \"harvest\" ascii wide nocase\n $monitor = \"monitor\" ascii wide nocase\n $dump_cmd = \"dump\" ascii wide\n $dotnet = { 4D 5A 90 00 03 00 00 00 04 00 00 00 FF FF 00 00 }\n condition:\n $rubeus_id and $dotnet and\n 2 of ($asktgt, $kerberoast, $asreproast, $s4u, $ptt, $harvest, $monitor, $dump_cmd)\n}\n\nrule SharpHound_BloodHound_Collector {\n meta:\n description = \"Detects SharpHound\/BloodHound AD enumeration tool in memory\"\n confidence = \"high\"\n strings:\n $sh1 = \"SharpHound\" ascii wide nocase\n $sh2 = \"BloodHound\" ascii wide nocase\n $sh3 = \"Invoke-BloodHound\" ascii wide nocase\n $ldap1 = \"GetAllDomainTrusts\" ascii wide\n $ldap2 = \"GetDomainControllers\" ascii wide\n $ldap3 = \"GetDomainComputers\" ascii wide\n $ldap4 = \"LdapSearcher\" ascii wide\n $zip = \"BloodHound.zip\" ascii wide\n $json1 = \"computers.json\" ascii wide\n $json2 = \"users.json\" ascii wide\n $json3 = \"groups.json\" ascii wide\n condition:\n 1 of ($sh1, $sh2, $sh3)\n or 3 of ($ldap1, $ldap2, $ldap3, $ldap4, $zip, $json1, $json2, $json3)\n}\n\nrule NanoDump_LSASS_Dumper {\n meta:\n description = \"Detects NanoDump or similar LSASS dumping tools in memory\"\n confidence = \"high\"\n strings:\n $nano1 = \"nanodump\" ascii wide nocase\n $nano2 = \"NanoDump\" ascii wide\n $syscall_pattern = { 4C 8B D1 B8 ?? 00 00 00 0F 05 C3 }\n $minidump = \"MiniDumpWriteDump\" ascii wide\n $lsass_name = \"lsass.exe\" ascii wide nocase\n condition:\n $nano1 or $nano2\n or ($syscall_pattern and $lsass_name)\n or ($minidump and $lsass_name and $syscall_pattern)\n}\n\nrule Seatbelt_Recon_Tool {\n meta:\n description = \"Detects Seatbelt post-exploitation recon tool in memory\"\n confidence = \"high\"\n strings:\n $sb1 = \"Seatbelt\" ascii wide\n $sb2 = \"WindowsCredentialFiles\" ascii wide\n $sb3 = \"DpapiMasterKeys\" ascii wide\n $sb4 = \"RDPSavedConnections\" ascii wide\n $sb5 = \"NetworkProfiles\" ascii wide\n $sb6 = \"TokenPrivileges\" ascii wide\n $sb7 = \"PowerShellHistory\" ascii wide\n condition:\n $sb1 or 4 of ($sb2, $sb3, $sb4, $sb5, $sb6, $sb7)\n}\n<\/pre>\n## File: generic_suspicious.yar\n\/\/ Generic suspicious pattern detection rules\n\nrule RWX_PE_In_Anonymous_Memory {\n meta:\n description = \"PE file in executable anonymous memory - possible reflective loading\"\n author = \"justruss\"\n date = \"2026-05-24\"\n confidence = \"medium\"\n strings:\n $mz_header = { 4D 5A 90 00 }\n $pe_sig = { 50 45 00 00 }\n condition:\n $mz_header at 0 and $pe_sig\n}\n\nrule Shellcode_Common_x64_Preambles {\n meta:\n description = \"Common x64 shellcode entry patterns in executable memory\"\n confidence = \"medium\"\n strings:\n $preamble1 = { FC 48 83 E4 F0 E8 }\n $getpc1 = { E8 00 00 00 00 59 }\n $getpc2 = { E8 00 00 00 00 5B }\n $peb_walk = { 64 48 8B 04 25 60 00 00 00 }\n condition:\n any of them\n}\n\nrule Suspicious_Named_Pipe {\n meta:\n description = \"Named pipe patterns associated with common C2 frameworks\"\n confidence = \"high\"\n strings:\n $cs_msse = \"\\\\.\\pipe\\MSSE-\" ascii wide\n $cs_postex = \"\\\\.\\pipe\\postex_\" ascii wide\n $cs_msagent = \"\\\\.\\pipe\\msagent_\" ascii wide\n $cs_status = \"\\\\.\\pipe\\status_\" ascii wide\n $msf_pipe = \"\\\\.\\pipe\\metsrv\" ascii wide\n $empire_pipe = \"\\\\.\\pipe\\empire\" ascii wide nocase\n condition:\n any of them\n}\n\nrule AMSI_Bypass_Patterns {\n meta:\n description = \"Detects common AMSI bypass technique byte patterns in memory\"\n confidence = \"high\"\n strings:\n $amsi_patch1 = { B8 57 00 07 80 C3 }\n $amsi_str = \"amsi.dll\" ascii wide nocase\n $amsi_func = \"AmsiScanBuffer\" ascii wide\n condition:\n $amsi_patch1 and ($amsi_str or $amsi_func)\n}\n\nrule ETW_Tamper_Patterns {\n meta:\n description = \"Detects ETW patching techniques in process memory\"\n confidence = \"high\"\n strings:\n $etw_patch = { C2 14 00 }\n $etw_func = \"EtwEventWrite\" ascii wide\n $etw_disable = \"EtwEventUnregister\" ascii wide\n condition:\n $etw_patch and ($etw_func or $etw_disable)\n}\n\nrule PowerShell_Download_Cradle {\n meta:\n description = \"Detects decoded PowerShell download cradles in process memory\"\n confidence = \"medium\"\n strings:\n $dl_string = \"DownloadString\" ascii wide nocase\n $dl_file = \"DownloadFile\" ascii wide nocase\n $webclient = \"Net.WebClient\" ascii wide nocase\n $iex = \"IEX\" ascii wide\n $invoke_exp = \"Invoke-Expression\" ascii wide nocase\n $ref_load = \"[Reflection.Assembly]::Load\" ascii wide\n condition:\n 2 of them\n}\n<\/pre>\nCompiling the combined Yara ruleset<\/h3>\n
After placing all four files in the rules directory, compile them into a single binary ruleset that Volatility’s vadyarascan plugin can load. The compiled format is faster to load than re-parsing individual text files on every scan.<\/p>\n
## Compile all rules into combined.yar\nsource \/opt\/vol3-env\/bin\/activate\n\npython3 < {OUTPUT}\")\nprint(f\"Ruleset size: {Path(OUTPUT).stat().st_size \/ 1024:.1f} KB\")\nEOF\n\n## Verify the compiled ruleset loads correctly\npython3 -c \"\nimport yara\nrules = yara.load('\/opt\/memory-hunter\/yara_rules\/combined.yar')\nprint('Ruleset loaded OK')\nprint(f'Rules available for memory scanning')\n\"\n\n## Test against a known-clean binary to check false positive rate\npython3 -c \"\nimport yara\nrules = yara.load('\/opt\/memory-hunter\/yara_rules\/combined.yar')\nimport os\ntest_files = ['\/bin\/ls', '\/bin\/cat', '\/usr\/bin\/python3']\nfor f in test_files:\n if os.path.exists(f):\n matches = rules.match(f)\n if matches:\n print(f'FP WARNING: {f} matched {[m.rule for m in matches]}')\n else:\n print(f'Clean: {f}')\n\"<\/pre>\nAdding your own Yara rules<\/h3>\n
The four files above are a starting point covering the most commonly encountered C2 frameworks and credential theft tools. As you encounter new malware families or build rules from your own analysis, add new
.yar<\/code> files to the rules directory and recompile. A few practical notes on writing rules that work well in a memory scanning context.<\/p>\nAvoid filesize conditions entirely since Yara scanning process memory regions does not have a meaningful file size. Avoid conditions that depend on PE structure offsets like
pe.entry_point<\/code> unless you are certain the region you are scanning is a complete PE and not a fragment or raw shellcode. String conditions that usefullword<\/code> are more reliable in memory than substring matches because memory contains a lot of incidental short strings. For strings that appear in both legitimate software and malware (like “cmd.exe” or “powershell”), always pair them with at least one other more unique indicator before the rule can match. A condition that requires three or more strings is almost always more reliable than one requiring a single string, even if the single string seems highly distinctive on the initial sample.<\/p>\nThe
clean_samples<\/code> directory at\/opt\/memory-hunter\/yara_rules\/clean_samples\/<\/code> is intended for known-clean Windows executables that you run new rules against before deploying them. A useful set to maintain there: a clean copy of common system binaries (ntdll.dll, kernel32.dll, powershell.exe), clean copies of legitimate admin tools (PsExec, Process Explorer), and a small set of benign .NET assemblies. Running every new rule against these before adding it to the compiled ruleset catches false positives before they generate noise in production scans.<\/p>\nThe complete repository structure<\/h3>\n
## Clone-ready repository layout\n## All scripts from this post organised for immediate use\n\n\/opt\/memory-hunter\/\n\u251c\u2500\u2500 analyse.py # Main entry point - run this\n\u251c\u2500\u2500 setup_volatility_dual.sh # One-shot environment setup\n\u251c\u2500\u2500 requirements.txt # Python dependencies\n\u251c\u2500\u2500 README.md # Quick start guide\n\u2502\n\u251c\u2500\u2500 scripts\/\n\u2502 \u251c\u2500\u2500 phase1_identify.py # OS detection and image metadata\n\u2502 \u251c\u2500\u2500 vol_runner.py # Vol2\/Vol3 abstraction + parallel collection\n\u2502 \u251c\u2500\u2500 anomaly_checks.py # Detection logic and Finding class\n\u2502 \u251c\u2500\u2500 phase4_iocs.py # IOC extraction and Yara scanning\n\u2502 \u2514\u2500\u2500 phase5_report.py # HTML and JSON report generation\n\u2502\n\u251c\u2500\u2500 templates\/\n\u2502 \u2514\u2500\u2500 report.html.j2 # Jinja2 HTML report template\n\u2502\n\u251c\u2500\u2500 yara_rules\/\n\u2502 \u251c\u2500\u2500 rules\/ # Individual .yar files (add yours here)\n\u2502 \u2502 \u251c\u2500\u2500 cobalt_strike.yar\n\u2502 \u2502 \u251c\u2500\u2500 meterpreter.yar\n\u2502 \u2502 \u251c\u2500\u2500 credential_tools.yar\n\u2502 \u2502 \u2514\u2500\u2500 generic_suspicious.yar\n\u2502 \u251c\u2500\u2500 combined.yar # Compiled ruleset (auto-generated)\n\u2502 \u2514\u2500\u2500 clean_samples\/ # Known-clean files for FP testing\n\u2502\n\u251c\u2500\u2500 reports\/ # Analysis output (gitignored)\n\u2514\u2500\u2500 logs\/ # Run logs (gitignored)\n\n## requirements.txt\nvolatility3\nyara-python\npefile\ncapstone\npython-magic\nrequests\njinja2\ntqdm\ntabulate\ncolorama\n\n## Quick start\ngit clone https:\/\/github.com\/yourrepo\/memory-hunter \/opt\/memory-hunter\ncd \/opt\/memory-hunter\nbash setup_volatility_dual.sh\nln -sf \/opt\/memory-hunter\/analyse.py \/usr\/local\/bin\/memory-hunt\n\n## Compile Yara rules\npython3 -c \"\nimport yara, glob\nrule_files = {f.split('\/')[-1].replace('.yar',''): f\n for f in glob.glob('\/opt\/memory-hunter\/yara_rules\/rules\/*.yar')}\nyara.compile(filepaths=rule_files).save('\/opt\/memory-hunter\/yara_rules\/combined.yar')\nprint(f'Compiled {len(rule_files)} rule files')\n\"\n\n## Run your first analysis\nmemory-hunt \/path\/to\/memory.raw<\/pre>\nThe pipeline produces two output files per image: a structured JSON file suitable for SIEM ingestion, scripted comparison across multiple images, or feeding into a correlation pipeline, and an HTML report that opens in any browser with colour-coded severity, IOC tables, full process list, and network connections. The HTML report is designed to be shared with stakeholders who need to understand the findings without running any tools themselves.<\/p>\n
The most important design decision in the whole pipeline is the exit code contract. By mapping risk levels to exit codes (0 for clean, 3 for critical), the script integrates cleanly into any automation that knows how to act on a process exit code. Shell scripts, CI\/CD pipelines, the image watcher daemon, and orchestration frameworks all speak exit codes natively. A threat hunter who wants to process fifty images and immediately see which ones need attention can run a single parallel command and act on the results without reading any output until something non-zero comes back.<\/p>\n","protected":false},"excerpt":{"rendered":"
A complete guide to automating single memory image analysis with Volatility 2 and 3. Covers when to use each version, parallel plugin collection, anomaly detection logic, IOC extraction, Yara scanning, and a full pipeline that produces an HTML report and structured JSON from a single command. Supports both Windows and Linux images.<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[4,13],"tags":[],"class_list":["post-366","post","type-post","status-publish","format-standard","hentry","category-dfir","category-threat-hunting"],"_links":{"self":[{"href":"https:\/\/justruss.tech\/index.php\/wp-json\/wp\/v2\/posts\/366","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/justruss.tech\/index.php\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/justruss.tech\/index.php\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/justruss.tech\/index.php\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/justruss.tech\/index.php\/wp-json\/wp\/v2\/comments?post=366"}],"version-history":[{"count":11,"href":"https:\/\/justruss.tech\/index.php\/wp-json\/wp\/v2\/posts\/366\/revisions"}],"predecessor-version":[{"id":377,"href":"https:\/\/justruss.tech\/index.php\/wp-json\/wp\/v2\/posts\/366\/revisions\/377"}],"wp:attachment":[{"href":"https:\/\/justruss.tech\/index.php\/wp-json\/wp\/v2\/media?parent=366"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/justruss.tech\/index.php\/wp-json\/wp\/v2\/categories?post=366"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/justruss.tech\/index.php\/wp-json\/wp\/v2\/tags?post=366"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}