performing-automated-malware-analysis-with-cape

CAPE を使用した自動化マルウェア解析の実行

概要

CAPE（Config And Payload Extraction）は Cuckoo から派生したオープンソースのマルウェアサンドボックスで、挙動解析、ペイロードのダンプ、設定情報の抽出を自動化します。CAPEv2 は挙動計測のための API フッキング機能を備え、実行中に作成・変更・削除されたファイルを取得し、ネットワークトラフィックを PCAP 形式で記録します。また、Emotet、TrickBot、Cobalt Strike、AsyncRAT、Rhadamanthys などのファミリー向けに、70 以上のカスタム設定抽出器（cape-parsers）を含みます。シグネチャシステムには 1000 以上の挙動シグネチャがあり、回避技術、永続化、認証情報窃取、ランサムウェアの挙動を検出します。CAPE のデバッガーは、YARA シグネチャ内でデバッガーのアクションを組み合わせることで、動的なアンチ回避の迂回を可能にします。推奨構成は、Ubuntu LTS ホストと Windows 10 21H2 ゲスト VM です。

使用するタイミング

• マルウェアの自動解析を CAPE で行うセキュリティアセスメントを実施するとき
• 関連するセキュリティイベントに対してインシデント対応手順に従うとき
• 定期的なセキュリティテストや監査活動を実施するとき
• 実機によるテストを通じてセキュリティ統制を検証するとき

前提条件

• Ubuntu 22.04 LTS サーバー（CPU 8 コア以上、RAM 32GB 以上、SSD 500GB 以上）
• KVM/QEMU による仮想化対応
• Windows 10 21H2 ゲストイメージ
• CAPEv2 の依存関係を含む Python 3.9 以上
• 隔離された解析ネットワーク用のネットワーク構成

ワークフロー

ステップ 1: API 経由でサンプルを送信して解析する

#!/usr/bin/env python3
"""CAPE sandbox API client for automated malware submission and analysis."""
import requests
import json
import time
import sys
from pathlib import Path


class CAPEClient:
    def __init__(self, base_url="http://localhost:8000", api_token=None):
        self.base_url = base_url.rstrip("/")
        self.headers = {}
        if api_token:
            self.headers["Authorization"] = f"Token {api_token}"

    def submit_file(self, filepath, options=None):
        """Submit a file for analysis."""
        url = f"{self.base_url}/apiv2/tasks/create/file/"
        files = {"file": open(filepath, "rb")}
        data = options or {}
        data.setdefault("timeout", 120)
        data.setdefault("enforce_timeout", False)

        resp = requests.post(url, files=files, data=data, headers=self.headers)
        resp.raise_for_status()
        result = resp.json()
        task_id = result.get("data", {}).get("task_ids", [None])[0]
        print(f"[+] Submitted {filepath} -> Task ID: {task_id}")
        return task_id

    def get_status(self, task_id):
        """Check task analysis status."""
        url = f"{self.base_url}/apiv2/tasks/status/{task_id}/"
        resp = requests.get(url, headers=self.headers)
        return resp.json().get("data", "unknown")

    def wait_for_completion(self, task_id, poll_interval=15, max_wait=600):
        """Wait for analysis to complete."""
        elapsed = 0
        while elapsed < max_wait:
            status = self.get_status(task_id)
            if status == "reported":
                print(f"[+] Task {task_id} completed")
                return True
            time.sleep(poll_interval)
            elapsed += poll_interval
            print(f"  Waiting... ({elapsed}s, status: {status})")
        return False

    def get_report(self, task_id):
        """Retrieve full analysis report."""
        url = f"{self.base_url}/apiv2/tasks/get/report/{task_id}/"
        resp = requests.get(url, headers=self.headers)
        return resp.json()

    def get_config(self, task_id):
        """Get extracted malware configuration."""
        report = self.get_report(task_id)
        configs = report.get("CAPE", {}).get("configs", [])
        return configs

    def get_dropped_files(self, task_id):
        """List files dropped during analysis."""
        report = self.get_report(task_id)
        return report.get("dropped", [])

    def get_network_iocs(self, task_id):
        """Extract network IOCs from analysis."""
        report = self.get_report(task_id)
        network = report.get("network", {})
        iocs = {
            "dns": [d.get("request") for d in network.get("dns", [])],
            "http": [h.get("uri") for h in network.get("http", [])],
            "tcp": [f"{h.get('dst')}:{h.get('dport')}"
                    for h in network.get("tcp", [])],
        }
        return iocs

    def analyze_sample(self, filepath):
        """Full automated analysis pipeline."""
        task_id = self.submit_file(filepath)
        if not task_id:
            return None

        if self.wait_for_completion(task_id):
            report = {
                "task_id": task_id,
                "config": self.get_config(task_id),
                "network_iocs": self.get_network_iocs(task_id),
                "dropped_files": len(self.get_dropped_files(task_id)),
            }
            return report
        return None


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

    url = sys.argv[2] if len(sys.argv) > 2 else "http://localhost:8000"
    client = CAPEClient(url)
    result = client.analyze_sample(sys.argv[1])
    if result:
        print(json.dumps(result, indent=2))

検証基準

• 設定されたタイムアウト内にサンプルが送信・解析されること
• 既知のマルウェアファミリーに対して挙動シグネチャが発火すること
• cape-parsers によりマルウェアの設定情報が抽出されること
• ネットワークトラフィックが取得され、IOC が抽出されること
• ドロップされたファイルとペイロードが、さらなる解析のために収集されること
• サンドボックスを認識するマルウェアに対して、アンチ回避の迂回が有効であること

参考資料

• CAPEv2 GitHub
• CAPE Sandbox Documentation
• Automating Malware Analysis with CAPE
• Installing CAPEv2 on Ubuntu