Cobalt Strikeビーコン設定の解析

Cobalt Strikeビーコン設定の解析 は何をしますか？

概要

Cobalt Strikeは、脅威アクターによって権限昇格後の活動に広く悪用されている商用侵入シミュレーションツールです。ビーコンペイロードには、C2サーバーアドレス、通信プロトコル、スリープ間隔、ジッター値、Malleable C2プロファイル設定、ウォーターマーク識別子、暗号化キーなどの設定データが埋め込まれています。PEファイル、シェルコード、またはメモリダンプからこの設定を抽出することは、インシデント対応担当者が攻撃者のインフラをマッピングし、キャンペーンを属性評価するために重要です。ビーコン設定は、1バイトのXORキー（バージョン3では0x69、バージョン4では0x2e）でXORエンコードされ、.dataセクション内にType-Length-Value（TLV）形式で保存されています。

使用するタイミング

• Cobalt Strikeビーコン設定の分析を必要とするセキュリティインシデントを調査する場合
• このドメインの検出ルールや脅威ハンティングクエリを作成する場合
• SOCアナリストがこの分析タイプの構造化された手順を必要とする場合
• 関連する攻撃手法に対するセキュリティ監視カバレッジを検証する場合

前提条件

• Python 3.9+とdissect.cobaltstrike、pefile、yara-python
• SentinelOne CobaltStrikeParser（parse_beacon_config.py）
• 16進エディタ（010 Editor、HxD）手動検査用
• PEファイル形式とXORエンコードの理解
• メモリダンプ取得ツール（Volatility3、WinDbg）
• C2トラフィックの相関分析のためのネットワーク分析ツール（Wireshark）

主要な概念

ビーコン設定の構造

Cobalt Strikeビーコンは、PEの.dataセクション内にTLV（Type-Length-Value）エントリのブロブとして設定を保存します。ステージレスビーコンは、ビーコンコード全体を4バイトのキーでXORします。設定ブロブ自体は単一バイトのXORキーを使用します。各TLVエントリには、2バイトのタイプ識別子（例：0x0001はBeaconType、0x0008はC2Server）、2バイトの長さ、および可変長データが含まれます。

Malleable C2プロファイル

ビーコン設定は、URIパス、ヘッダー、メタデータエンコード（Base64、NetBIOS）、データ変換などを含むHTTPリクエスト/レスポンス変換を決定するMalleable C2プロファイルをエンコードします。これらの設定を分析することで、ビーコンが正規のWebトラフィックに紛れるためにトラフィックを偽装する方法が明らかになります。

ウォーターマークとライセンス識別

各Cobalt Strikeライセンスは、生成されたビーコンに一意のウォーターマーク（4バイト整数）を埋め込みます。ウォーターマークを抽出することで、複数のビーコンを同じオペレーターやクラックされたライセンスにリンクできます。脅威インテリジェンスプロバイダーが維持する既知のウォーターマークデータベースは、ウォーターマークを特定の脅威アクターや漏洩したライセンスキーにマッピングします。

ワークフロー

ステップ1: CobaltStrikeParserを使用した設定の抽出

#!/usr/bin/env python3
"""PEまたはメモリダンプからCobalt Strikeビーコン設定を抽出します"""
import sys
import json

# SentinelOneのCobaltStrikeParserを使用
# pip install dissect.cobaltstrike
from dissect.cobaltstrike.beacon import BeaconConfig

def extract_beacon_config(filepath):
    """ファイルからビーコン設定を解析します"""
    configs = list(BeaconConfig.from_path(filepath))

    if not configs:
        print(f"[-] {filepath} にビーコン設定が見つかりません")
        return None

    for i, config in enumerate(configs):
        print(f"\n[+] ビーコン設定 #{i+1}")
        print(f"{'='*60}")

        settings = config.as_dict()

        # インシデント対応に重要なフィールド
        critical_fields = [
            "SETTING_C2_REQUEST",
            "SETTING_C2_RECOVER",
            "SETTING_PUBKEY",
            "SETTING_DOMAINS",
            "SETTING_BEACONTYPE",
            "SETTING_PORT",
            "SETTING_SLEEPTIME",
            "SETTING_JITTER",
            "SETTING_MAXGET",
            "SETTING_SPAWNTO_X86",
            "SETTING_SPAWNTO_X64",
            "SETTING_PIPENAME",
            "SETTING_WATERMARK",
            "SETTING_C2_VERB_GET",
            "SETTING_C2_VERB_POST",
            "SETTING_USERAGENT",
            "SETTING_PROTOCOL",
        ]

        for field in critical_fields:
            value = settings.get(field, "N/A")
            print(f"  {field}: {value}")

        return settings

    return None


def extract_c2_indicators(config):
    """ビーコン設定からアクション可能なC2インジケーターを抽出します"""
    indicators = {
        "c2_domains": [],
        "c2_ips": [],
        "c2_urls": [],
        "user_agent": "",
        "named_pipes": [],
        "spawn_processes": [],
        "watermark": "",
    }

    if not config:
        return indicators

    # C2ドメインを抽出
    domains = config.get("SETTING_DOMAINS", "")
    if domains:
        for domain in str(domains).split(","):
            domain = domain.strip().rstrip("/")
            if domain:
                indicators["c2_domains"].append(domain)

    # ユーザーエージェントを抽出
    indicators["user_agent"] = str(config.get("SETTING_USERAGENT", ""))

    # 名前付きパイプを抽出
    pipe = config.get("SETTING_PIPENAME", "")
    if pipe:
        indicators["named_pipes"].append(str(pipe))

    # スポーン先プロセスを抽出
    for arch in ["SETTING_SPAWNTO_X86", "SETTING_SPAWNTO_X64"]:
        proc = config.get(arch, "")
        if proc:
            indicators["spawn_processes"].append(str(proc))

    # ウォーターマークを抽出
    indicators["watermark"] = str(config.get("SETTING_WATERMARK", ""))

    return indicators


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"使用方法: {sys.argv[0]} <beacon_file_or_dump>")
        sys.exit(1)

    config = extract_beacon_config(sys.argv[1])
    if config:
        indicators = extract_c2_indicators(config)
        print(f"\n[+] 抽出されたC2インジケーター:")
        print(json.dumps(indicators, indent=2))

ステップ2: ビーコン設定の手動XOR復号

import struct

def find_and_decrypt_config(data):
    """手動でビーコン設定を特定し復号します"""
    # Cobalt Strike 4.x は XOR キーとして 0x2e を使用
    xor_keys = [0x2e, 0x69]  # v4, v3

    for xor_key in xor_keys:
        # XOR後の設定マジックバイトを検索
        # 設定は 0x0001 (BeaconType) をキーでXORしたものから始まる
        magic = bytes([0x00 ^ xor_key, 0x01 ^ xor_key,
                       0x00 ^ xor_key, 0x02 ^ xor_key])

        offset = data.find(magic)
        if offset == -1:
            continue

        print(f"[+] オフセット 0x{offset:x} で設定を発見 (XORキー: 0x{xor_key:02x})")

        # 設定ブロブを復号（通常4096バイト）
        config_size = 4096
        encrypted = data[offset:offset + config_size]
        decrypted = bytes([b ^ xor_key for b in encrypted])

        # TLVエントリを解析
        entries = parse_tlv(decrypted)
        return entries

    return None


def parse_tlv(data):
    """Type-Length-Value 設定エントリを解析します"""
    entries = {}
    offset = 0

    # TLV フィールドタイプマッピング
    field_names = {
        0x0001: "BeaconType",
        0x0002: "Port",
        0x0003: "SleepTime",
        0x0004: "MaxGetSize",
        0x0005: "Jitter",
        0x0006: "MaxDNS",
        0x0007: "Deprecated_PublicKey",
        0x0008: "C2Server",
        0x0009: "UserAgent",
        0x000a: "PostURI",
        0x000b: "Malleable_C2_Instructions",
        0x000c: "Deprecated_HttpGet_Metadata",
        0x000d: "SpawnTo_x86",
        0x000e: "SpawnTo_x64",
        0x000f: "CryptoScheme",
        0x001a: "Watermark",
        0x001d: "C2_HostHeader",
        0x0024: "PipeName",
        0x0025: "Year",
        0x0026: "Month",
        0x0027: "Day",
        0x0036: "ProxyHostname",
    }

    while offset + 6 <= len(data):
        entry_type = struct.unpack(">H", data[offset:offset+2])[0]
        entry_len_type = struct.unpack(">H", data[offset+2:offset+4])[0]
        entry_len = struct.unpack(">H", data[offset+4:offset+6])[0]

        if entry_type == 0:
            break

        value_start = offset + 6
        value_end = value_start + entry_len
        value_data = data[value_start:value_end]

        field_name = field_names.get(entry_type, f"Unknown_0x{entry_type:04x}")

        if entry_len_type == 1:  # Short
            value = struct.unpack(">H", value_data[:2])[0]
        elif entry_len_type == 2:  # Int
            value = struct.unpack(">I", value_data[:4])[0]
        elif entry_len_type == 3:  # String/Blob
            value = value_data.rstrip(b'\x00').decode('utf-8', errors='replace')
        else:
            value = value_data.hex()

        entries[field_name] = value
        print(f"  {field_name}: {value}")

        offset = value_end

    return entries

ステップ3: ビーコン検出のYARAルール

import yara

cobalt_strike_rule = """
rule CobaltStrike_Beacon_Config {
    meta:
        description = "Cobalt Strike ビーコン設定を検出"
        author = "マルウェア解析チーム"
        date = "2025-01-01"

    strings:
        // CS 4.x の XOR 後の設定マーカー (キー 0x2e)
        $config_v4 = { 2e 2f 2e 2c }

        // CS 3.x の XOR 後の設定マーカー (キー 0x69)
        $config_v3 = { 69 68 69 6b }

        // 一般的なビーコン文字列
        $str_pipe = "\\\\.\\pipe\\" ascii wide
        $str_beacon = "beacon" ascii nocase
        $str_sleeptime = "sleeptime" ascii nocase

        // リフレクティブローダーパターン
        $reflective = { 4D 5A 41 52 55 48 89 E5 }

    condition:
        ($config_v4 or $config_v3) or
        (2 of ($str_*) and $reflective)
}
"""

def scan_for_beacons(filepath):
    """YARAルールでファイルをスキャンしCobalt Strikeビーコンを検出します"""
    rules = yara.compile(source=cobalt_strike_rule)
    matches = rules.match(filepath)

    for match in matches:
        print(f"[+] YARA一致: {match.rule}")
        for string_match in match.strings:
            offset = string_match.instances[0].offset
            print(f"    文字列: {string_match.identifier} (オフセット 0x{offset:x})")

    return matches

ステップ4: ネットワークトラフィックの相関分析

from dissect.cobaltstrike.c2 import HttpC2Config

def analyze_c2_profile(beacon_config):
    """ビーコン設定からMalleable C2プロファイルを分析します"""
    print("\n[+] Malleable C2 プロファイル分析")
    print("=" * 60)

    # HTTP GET設定
    get_verb = beacon_config.get("SETTING_C2_VERB_GET", "GET")
    get_uri = beacon_config.get("SETTING_C2_REQUEST", "")
    print(f"\n  HTTP GETリクエスト:")
    print(f"    メソッド: {get_verb}")
    print(f"    URI: {get_uri}")

    # HTTP POST設定
    post_verb = beacon_config.get("SETTING_C2_VERB_POST", "POST")
    post_uri = beacon_config.get("SETTING_C2_POSTREQ", "")
    print(f"\n  HTTP POSTリクエスト:")
    print(f"    メソッド: {post_verb}")
    print(f"    URI: {post_uri}")

    # ユーザーエージェント
    ua = beacon_config.get("SETTING_USERAGENT", "")
    print(f"\n  User-Agent: {ua}")

    # Hostヘッダー
    host = beacon_config.get("SETTING_C2_HOSTHEADER", "")
    print(f"  Host ヘッダー: {host}")

    # スリープ間隔とジッター（トラフィックパターン用）
    sleep_ms = beacon_config.get("SETTING_SLEEPTIME", 60000)
    jitter = beacon_config.get("SETTING_JITTER", 0)
    print(f"\n  スリープ時間: {sleep_ms}ms")
    print(f"  ジッター: {jitter}%")

    # Suricata/Snortシグネチャの生成
    print(f"\n[+] 推奨ネットワークシグネチャ:")
    if ua:
        print(f'  alert http any any -> any any (msg:"CS Beacon UA"; '
              f'content:"{ua}"; http_user_agent; sid:1000001; rev:1;)')
    if get_uri:
        print(f'  alert http any any -> any any (msg:"CS Beacon URI"; '
              f'content:"{get_uri}"; http_uri; sid:1000002; rev:1;)')

検証基準

• PEファイルまたはメモリダンプからビーコン設定が正常に抽出されること
• C2サーバーのドメイン/IPがポートとプロトコルとともに正しく識別されること
• HTTP変換を示すMalleable C2プロファイルパラメーターがデコードされること
• 属性評価の相関のためにウォーターマーク値が抽出されること
• スリープ時間とジッター値が観測されたネットワークビーコン間隔と一致すること
• YARAルールがパックされたサンプルとアンパックされたサンプルの両方でビーコンを検出すること
• 抽出されたC2プロファイルからネットワークシグネチャが生成されること

参考文献

• SentinelOne CobaltStrikeParser
• dissect.cobaltstrike Library
• SentinelLabs Beacon Configuration Analysis
• Cobalt Strike Staging and Config Extraction
• MITRE ATT&CK - Cobalt Strike S0154