performing-memory-forensics-with-volatility3-plugins

Volatility3 プラグインによるメモリフォレンジックの実施

概要

Volatility3（v2.26.0以降、2025年5月の機能同等リリース）は、メモリフォレンジックの標準フレームワークであり、非推奨となったVolatility2を置き換えるものです。Windows、Linux、macOSのRAMダンプを解析し、ディスクベースのフォレンジックでは明らかにできない、悪意のあるプロセス、コードインジェクション、ルートキット、認証情報の収集、ネットワーク接続を検出します。主なプラグインは、windows.malfind（インジェクションを示すRWXメモリ領域の検出）、windows.psscan（隠しプロセスの発見）、windows.dlllist（読み込まれたモジュールの列挙）、windows.netscan（アクティブなネットワーク接続）、windows.handles（開いているファイルおよびレジストリのハンドル）です。2024 Plugin Contestでは、メモリからEvent Tracing for Windowsのデータを抽出するETW Scanが紹介されました。

使用するタイミング

• volatility3 プラグインを用いたメモリフォレンジックを伴うセキュリティ評価を実施する場合
• 関連するセキュリティイベントに対してインシデント対応手順に従う場合
• 定期的なセキュリティテストまたは監査活動を実施する場合
• ハンズオンテストを通じてセキュリティ管理策を検証する場合

前提条件

• volatility3 フレームワークがインストールされたPython 3.9以降
• メモリダンプファイル（.raw、.dmp、.vmem、.lime）
• Windowsシンボルテーブル（ISFファイル、自動ダウンロード）
• Windowsのプロセスメモリアーキテクチャに関する理解
• メモリ内パターンスキャンのためのYARA連携

ワークフロー

ステップ 1: マルウェア検出のためのプロセス解析

#!/usr/bin/env python3
"""Volatility3-based memory forensics automation for malware analysis."""
import subprocess
import json
import sys
import os


class Vol3Analyzer:
    """Automate Volatility3 plugin execution for malware analysis."""

    def __init__(self, dump_path, vol3_path="vol"):
        self.dump_path = dump_path
        self.vol3 = vol3_path
        self.results = {}

    def run_plugin(self, plugin, extra_args=None):
        """Execute a Volatility3 plugin and capture output."""
        cmd = [
            self.vol3, "-f", self.dump_path,
            "-r", "json", plugin,
        ]
        if extra_args:
            cmd.extend(extra_args)

        try:
            result = subprocess.run(
                cmd, capture_output=True, text=True, timeout=300
            )
            if result.returncode == 0:
                return json.loads(result.stdout)
        except (subprocess.TimeoutExpired, json.JSONDecodeError) as e:
            print(f"  [!] {plugin} failed: {e}")
        return None

    def detect_process_injection(self):
        """Use malfind to detect injected code regions."""
        print("[+] Running windows.malfind (code injection detection)")
        results = self.run_plugin("windows.malfind")

        injected = []
        if results:
            for entry in results:
                injected.append({
                    "pid": entry.get("PID"),
                    "process": entry.get("Process"),
                    "address": entry.get("Start VPN"),
                    "protection": entry.get("Protection"),
                    "hexdump": entry.get("Hexdump", "")[:200],
                })
                print(f"  [!] Injection in PID {entry.get('PID')} "
                      f"({entry.get('Process')}) at {entry.get('Start VPN')}")

        self.results["injected_processes"] = injected
        return injected

    def find_hidden_processes(self):
        """Compare pslist vs psscan to find hidden processes."""
        print("[+] Running process comparison (pslist vs psscan)")

        pslist = self.run_plugin("windows.pslist")
        psscan = self.run_plugin("windows.psscan")

        if not pslist or not psscan:
            return []

        list_pids = {e.get("PID") for e in pslist}
        scan_pids = {e.get("PID") for e in psscan}

        hidden = scan_pids - list_pids
        if hidden:
            print(f"  [!] {len(hidden)} hidden processes found!")
            for entry in psscan:
                if entry.get("PID") in hidden:
                    print(f"    PID {entry['PID']}: {entry.get('ImageFileName')}")

        self.results["hidden_processes"] = list(hidden)
        return list(hidden)

    def analyze_network(self):
        """Extract active network connections."""
        print("[+] Running windows.netscan")
        results = self.run_plugin("windows.netscan")

        connections = []
        if results:
            for entry in results:
                conn = {
                    "pid": entry.get("PID"),
                    "process": entry.get("Owner"),
                    "local": f"{entry.get('LocalAddr')}:{entry.get('LocalPort')}",
                    "remote": f"{entry.get('ForeignAddr')}:{entry.get('ForeignPort')}",
                    "state": entry.get("State"),
                    "protocol": entry.get("Proto"),
                }
                connections.append(conn)

        self.results["network_connections"] = connections
        return connections

    def extract_dlls(self, pid=None):
        """List loaded DLLs per process."""
        print(f"[+] Running windows.dlllist{f' (PID {pid})' if pid else ''}")
        args = ["--pid", str(pid)] if pid else None
        results = self.run_plugin("windows.dlllist", args)

        dlls = []
        if results:
            for entry in results:
                dlls.append({
                    "pid": entry.get("PID"),
                    "process": entry.get("Process"),
                    "base": entry.get("Base"),
                    "name": entry.get("Name"),
                    "path": entry.get("Path"),
                    "size": entry.get("Size"),
                })

        self.results["loaded_dlls"] = dlls
        return dlls

    def scan_with_yara(self, rules_path):
        """Scan memory with YARA rules."""
        print(f"[+] Running windows.yarascan with {rules_path}")
        results = self.run_plugin(
            "windows.yarascan",
            ["--yara-file", rules_path]
        )

        matches = []
        if results:
            for entry in results:
                matches.append({
                    "rule": entry.get("Rule"),
                    "pid": entry.get("PID"),
                    "process": entry.get("Process"),
                    "offset": entry.get("Offset"),
                })

        self.results["yara_matches"] = matches
        return matches

    def full_triage(self):
        """Run full malware-focused memory triage."""
        print(f"[*] Full memory triage: {self.dump_path}")
        print("=" * 60)

        self.detect_process_injection()
        self.find_hidden_processes()
        self.analyze_network()

        return self.results


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <memory_dump>")
        sys.exit(1)

    analyzer = Vol3Analyzer(sys.argv[1])
    results = analyzer.full_triage()
    print(json.dumps(results, indent=2, default=str))

検証基準

• メモリダンプが正しいOSプロファイルで正常に解析されている
• malfindでRWX領域を伴うインジェクションされたプロセスが検出されている
• pslistとpsscanの比較により隠しプロセスが特定されている
• ネットワーク接続からC2通信の宛先が判明している
• メモリ内の既知のマルウェアシグネチャにYARAルールが一致している
• lsassプロセスのメモリから認証情報の痕跡が抽出されている

参考資料

• Volatility Foundation
• Volatility3 GitHub
• 2024 Volatility Plugin Contest
• Memory Forensics with Volatility 3
• MITRE ATT&CK T1055 - Process Injection