analyse-de-la-configuration-dun-beacon-cobalt-strike

Analyse de la configuration d'un beacon Cobalt Strike

Vue d'ensemble

Cobalt Strike est un outil commercial de simulation adverse largement utilisé par les acteurs malveillants pour des opérations de post-exploitation. Les charges utiles Beacon contiennent des données de configuration intégrées qui révèlent les adresses des serveurs C2, les protocoles de communication, les intervalles de sommeil, les valeurs de jitter, les paramètres du profil C2 malleable, les identifiants Watermark et les clés de chiffrement. L'extraction de cette configuration à partir de fichiers PE, de code shell ou de dumps mémoire est essentielle pour les intervenants en cybersécurité afin de cartographier l'infrastructure des attaquants et d'attribuer des campagnes. La configuration du beacon est encodée par XOR à l'aide d'un seul octet (0x69 pour la version 3, 0x2e pour la version 4) et stockée dans un format Type-Longueur-Valeur (TLV) dans la section .data.

Quand utiliser

• Lors de l'investigation d'incidents de sécurité nécessitant l'analyse de la configuration d'un beacon Cobalt Strike
• Lors de la création de règles de détection ou de requêtes de chasse aux menaces pour ce domaine
• Lorsque les analystes SOC ont besoin de procédures structurées pour ce type d'analyse
• Lors de la validation de la couverture de surveillance de sécurité pour les techniques d'attaque associées

Prérequis

• Python 3.9+ avec dissect.cobaltstrike, pefile, yara-python
• Le parseur SentinelOne CobaltStrikeParser (parse_beacon_config.py)
• Éditeur hexadécimal (010 Editor, HxD) pour inspection manuelle
• Compréhension du format des fichiers PE et du codage XOR
• Outils d'acquisition de dumps mémoire (Volatility3, WinDbg)
• Outils d'analyse réseau (Wireshark) pour la corrélation du trafic C2

Concepts clés

Structure de la configuration du beacon

Les beacons Cobalt Strike stockent leur configuration sous la forme d'un bloc d'entrées TLV (Type-Longueur-Valeur) dans la section .data du PE. Les beacons sans stagère effectuent un XOR sur l'ensemble du code du beacon avec une clé de 4 octets. Le bloc de configuration lui-même utilise une clé XOR d'un seul octet. Chaque entrée TLV contient un identifiant de type sur 2 octets (par exemple, 0x0001 pour BeaconType, 0x0008 pour C2Server), une longueur sur 2 octets et des données de longueur variable.

Profils C2 Malleable

La configuration du beacon encode le profil C2 malleable qui dicte les transformations des requêtes/réponses HTTP, y compris les chemins URI, les en-têtes, le codage des métadonnées (Base64, NetBIOS) et les transformations de données. L'analyse de ces paramètres révèle comment le beacon déguise son trafic pour se fondre dans le trafic web légitime.

Identification du Watermark et de la licence

Chaque licence Cobalt Strike intègre un watermark unique (entier de 4 octets) dans les beacons générés. L'extraction du watermark peut lier plusieurs beacons au même opérateur ou à une licence fissurée. Les bases de données de watermarks connues, maintenues par les fournisseurs de renseignement sur les menaces, associent les watermarks à des acteurs malveillants spécifiques ou à des clés de licence divulguées.

Flux de travail

Étape 1 : Extraire la configuration avec CobaltStrikeParser

#!/usr/bin/env python3
"""Extraire la configuration du beacon Cobalt Strike à partir d'un PE ou d'un dump mémoire."""
import sys
import json

# Utilisation du CobaltStrikeParser de SentinelOne
# pip install dissect.cobaltstrike
from dissect.cobaltstrike.beacon import BeaconConfig

def extract_beacon_config(filepath):
    """Analyser la configuration du beacon à partir d'un fichier."""
    configs = list(BeaconConfig.from_path(filepath))

    if not configs:
        print(f"[-] Aucune configuration de beacon trouvée dans {filepath}")
        return None

    for i, config in enumerate(configs):
        print(f"\n[+] Configuration du beacon #{i+1}")
        print(f"{'='*60}")

        settings = config.as_dict()

        # Champs critiques pour la réponse aux incidents
        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):
    """Extraire les indicateurs C2 exploitables de la configuration du beacon."""
    indicators = {
        "c2_domains": [],
        "c2_ips": [],
        "c2_urls": [],
        "user_agent": "",
        "named_pipes": [],
        "spawn_processes": [],
        "watermark": "",
    }

    if not config:
        return indicators

    # Extraire les domaines 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)

    # Extraire le user agent
    indicators["user_agent"] = str(config.get("SETTING_USERAGENT", ""))

    # Extraire les pipes nommés
    pipe = config.get("SETTING_PIPENAME", "")
    if pipe:
        indicators["named_pipes"].append(str(pipe))

    # Extraire les processus spawn-to
    for arch in ["SETTING_SPAWNTO_X86", "SETTING_SPAWNTO_X64"]:
        proc = config.get(arch, "")
        if proc:
            indicators["spawn_processes"].append(str(proc))

    # Extraire le watermark
    indicators["watermark"] = str(config.get("SETTING_WATERMARK", ""))

    return indicators


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

    config = extract_beacon_config(sys.argv[1])
    if config:
        indicators = extract_c2_indicators(config)
        print(f"\n[+] Indicateurs C2 extraits :")
        print(json.dumps(indicators, indent=2))

Étape 2 : Déchiffrement manuel XOR de la configuration du beacon

import struct

def find_and_decrypt_config(data):
    """Localiser et déchiffrer manuellement la configuration du beacon."""
    # Cobalt Strike 4.x utilise 0x2e comme clé XOR
    xor_keys = [0x2e, 0x69]  # v4, v3

    for xor_key in xor_keys:
        # Rechercher les octets magiques de la configuration après XOR
        # La configuration commence par 0x0001 (BeaconType) XORé avec la clé
        magic = bytes([0x00 ^ xor_key, 0x01 ^ xor_key,
                       0x00 ^ xor_key, 0x02 ^ xor_key])

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

        print(f"[+] Configuration trouvée au décalage 0x{offset:x} (clé XOR : 0x{xor_key:02x})")

        # Déchiffrer le bloc de configuration (généralement 4096 octets)
        config_size = 4096
        encrypted = data[offset:offset + config_size]
        decrypted = bytes([b ^ xor_key for b in encrypted])

        # Analyser les entrées TLV
        entries = parse_tlv(decrypted)
        return entries

    return None


def parse_tlv(data):
    """Analyser les entrées de configuration de type Type-Longueur-Valeur."""
    entries = {}
    offset = 0

    # Correspondance des types de champs 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:  # Court
            value = struct.unpack(">H", value_data[:2])[0]
        elif entry_len_type == 2:  # Entier
            value = struct.unpack(">I", value_data[:4])[0]
        elif entry_len_type == 3:  # Chaîne/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

Étape 3 : Règle YARA pour la détection du beacon

import yara

cobalt_strike_rule = """
rule CobaltStrike_Beacon_Config {
    meta:
        description = "Détecte la configuration du beacon Cobalt Strike"
        author = "Équipe d'analyse de logiciels malveillants"
        date = "2025-01-01"

    strings:
        // Marqueur de configuration XORé pour CS 4.x (clé 0x2e)
        $config_v4 = { 2e 2f 2e 2c }

        // Marqueur de configuration XORé pour CS 3.x (clé 0x69)
        $config_v3 = { 69 68 69 6b }

        // Chaînes courantes du beacon
        $str_pipe = "\\\\.\\pipe\\" ascii wide
        $str_beacon = "beacon" ascii nocase
        $str_sleeptime = "sleeptime" ascii nocase

        // Motif du chargeur réflexif
        $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):
    """Analyser un fichier avec des règles YARA pour les beacons Cobalt Strike."""
    rules = yara.compile(source=cobalt_strike_rule)
    matches = rules.match(filepath)

    for match in matches:
        print(f"[+] Correspondance YARA : {match.rule}")
        for string_match in match.strings:
            offset = string_match.instances[0].offset
            print(f"    Chaîne : {string_match.identifier} au décalage 0x{offset:x}")

    return matches

Étape 4 : Corrélation du trafic réseau

from dissect.cobaltstrike.c2 import HttpC2Config

def analyze_c2_profile(beacon_config):
    """Analyser le profil C2 malleable à partir de la configuration du beacon."""
    print("\n[+] Analyse du profil C2 Malleable")
    print("=" * 60)

    # Configuration HTTP GET
    get_verb = beacon_config.get("SETTING_C2_VERB_GET", "GET")
    get_uri = beacon_config.get("SETTING_C2_REQUEST", "")
    print(f"\n  Requête HTTP GET :")
    print(f"    Verbe : {get_verb}")
    print(f"    URI : {get_uri}")

    # Configuration HTTP POST
    post_verb = beacon_config.get("SETTING_C2_VERB_POST", "POST")
    post_uri = beacon_config.get("SETTING_C2_POSTREQ", "")
    print(f"\n  Requête HTTP POST :")
    print(f"    Verbe : {post_verb}")
    print(f"    URI : {post_uri}")

    # User Agent
    ua = beacon_config.get("SETTING_USERAGENT", "")
    print(f"\n  User-Agent : {ua}")

    # En-tête Host
    host = beacon_config.get("SETTING_C2_HOSTHEADER", "")
    print(f"  En-tête Host : {host}")

    # Sommeil et jitter pour le modèle de trafic
    sleep_ms = beacon_config.get("SETTING_SLEEPTIME", 60000)
    jitter = beacon_config.get("SETTING_JITTER", 0)
    print(f"\n  Temps de sommeil : {sleep_ms}ms")
    print(f"  Jitter : {jitter}%")

    # Générer des signatures Suricata/Snort
    print(f"\n[+] Signatures réseau suggérées :")
    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;)')

Critères de validation

• Configuration du beacon extraite avec succès d'un fichier PE ou d'un dump mémoire
• Domaines/IP des serveurs C2 correctement identifiés avec le port et le protocole
• Paramètres du profil C2 malleable décodés montrant les transformations HTTP
• Valeur du watermark extraite pour la corrélation d'attribution
• Temps de sommeil et valeurs de jitter correspondant aux intervalles de beacon réseau observés
• Règles YARA détectant le beacon dans les échantillons compressés et non compressés
• Signatures réseau générées à partir du profil C2 extrait

Références

• SentinelOne CobaltStrikeParser
• Bibliothèque dissect.cobaltstrike
• Analyse de la configuration des beacons par SentinelLabs
• Extraction de la configuration et staging de Cobalt Strike
• MITRE ATT&CK - Cobalt Strike S0154