analyzing-cobalt-strike-beacon-configuration
Compare original and translation side by side
🇺🇸
Original
English🇨🇳
Translation
ChineseAnalyzing Cobalt Strike Beacon Configuration
分析Cobalt Strike Beacon配置
Overview
概述
Cobalt Strike is a commercial adversary simulation tool widely abused by threat actors for post-exploitation operations. Beacon payloads contain embedded configuration data that reveals C2 server addresses, communication protocols, sleep intervals, jitter values, malleable C2 profile settings, watermark identifiers, and encryption keys. Extracting this configuration from PE files, shellcode, or memory dumps is critical for incident responders to map attacker infrastructure and attribute campaigns. The beacon configuration is XOR-encoded using a single byte (0x69 for version 3, 0x2e for version 4) and stored in a Type-Length-Value (TLV) format within the .data section.
Cobalt Strike是一款被威胁行为者广泛滥用的商业对手仿真工具，用于攻击后操作。Beacon载荷包含嵌入式配置数据，这些数据会暴露C2服务器地址、通信协议、休眠间隔、抖动值、可变形C2配置文件设置、水印标识符以及加密密钥。从PE文件、shellcode或内存转储中提取这些配置对于事件响应人员绘制攻击者基础设施图谱并关联攻击活动至关重要。Beacon配置使用单字节进行XOR加密（版本3为0x69，版本4为0x2e），并以Type-Length-Value（TLV）格式存储在.data节中。
When to Use
适用场景
- When investigating security incidents that require analyzing cobalt strike beacon configuration
- When building detection rules or threat hunting queries for this domain
- When SOC analysts need structured procedures for this analysis type
- When validating security monitoring coverage for related attack techniques
- 当调查需要分析Cobalt Strike Beacon配置的安全事件时
- 当构建该领域的检测规则或威胁狩猎查询时
- 当SOC分析师需要此类分析的结构化流程时
- 当验证相关攻击技术的安全监控覆盖范围时
Prerequisites
前提条件
- Python 3.9+ with ,
dissect.cobaltstrike,pefileyara-python - SentinelOne CobaltStrikeParser ()
parse_beacon_config.py - Hex editor (010 Editor, HxD) for manual inspection
- Understanding of PE file format and XOR encoding
- Memory dump acquisition tools (Volatility3, WinDbg)
- Network analysis tools (Wireshark) for C2 traffic correlation
- 安装Python 3.9+及、
dissect.cobaltstrike、pefile库yara-python - SentinelOne CobaltStrikeParser（）
parse_beacon_config.py - 用于手动检查的十六进制编辑器（010 Editor、HxD）
- 了解PE文件格式和XOR加密
- 内存转储获取工具（Volatility3、WinDbg）
- 用于C2流量关联的网络分析工具（Wireshark）
Key Concepts
核心概念
Beacon Configuration Structure
Beacon配置结构
Cobalt Strike beacons store their configuration as a blob of TLV (Type-Length-Value) entries within the .data section of the PE. Stageless beacons XOR the entire beacon code with a 4-byte key. The configuration blob itself uses a single-byte XOR key. Each TLV entry contains a 2-byte type identifier (e.g., 0x0001 for BeaconType, 0x0008 for C2Server), a 2-byte length, and variable-length data.
Cobalt Strike Beacon将其配置存储为TLV（Type-Length-Value）条目组成的二进制块，位于PE文件的.data节中。无阶段Beacon使用4字节密钥对整个Beacon代码进行XOR加密。配置块本身则使用单字节XOR密钥。每个TLV条目包含2字节类型标识符（例如0x0001代表BeaconType，0x0008代表C2Server）、2字节长度以及可变长度的数据。
Malleable C2 Profiles
可变形C2配置文件
The beacon configuration encodes the malleable C2 profile that dictates HTTP request/response transformations, including URI paths, headers, metadata encoding (Base64, NetBIOS), and data transforms. Analyzing these settings reveals how the beacon disguises its traffic to blend with legitimate web traffic.
Beacon配置中编码了可变形C2配置文件，该文件规定了HTTP请求/响应转换规则，包括URI路径、请求头、元数据编码（Base64、NetBIOS）以及数据转换方式。分析这些设置可以揭示Beacon如何伪装其流量以融入合法网络流量。
Watermark and License Identification
水印与许可证识别
Each Cobalt Strike license embeds a unique watermark (4-byte integer) into generated beacons. Extracting the watermark can link multiple beacons to the same operator or cracked license. Known watermark databases maintained by threat intelligence providers map watermarks to specific threat actors or leaked license keys.
每个Cobalt Strike许可证都会在生成的Beacon中嵌入唯一的水印（4字节整数）。提取水印可以将多个Beacon关联到同一操作者或破解的许可证。威胁情报提供商维护的已知水印数据库可将水印映射到特定威胁行为者或泄露的许可证密钥。
Workflow
工作流程
Step 1: Extract Configuration with CobaltStrikeParser
步骤1：使用CobaltStrikeParser提取配置
python
#!/usr/bin/env python3
"""Extract Cobalt Strike beacon config from PE or memory dump."""
import sys
import jsonpython
#!/usr/bin/env python3
"""Extract Cobalt Strike beacon config from PE or memory dump."""
import sys
import jsonUsing SentinelOne's CobaltStrikeParser
Using SentinelOne's CobaltStrikeParser
pip install dissect.cobaltstrike
pip install dissect.cobaltstrike
from dissect.cobaltstrike.beacon import BeaconConfig
def extract_beacon_config(filepath):
"""Parse beacon configuration from file."""
configs = list(BeaconConfig.from_path(filepath))
if not configs:
print(f"[-] No beacon configuration found in {filepath}")
return None
for i, config in enumerate(configs):
print(f"\n[+] Beacon Configuration #{i+1}")
print(f"{'='*60}")
settings = config.as_dict()
# Critical fields for incident response
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 Nonedef extract_c2_indicators(config):
"""Extract actionable C2 indicators from beacon config."""
indicators = {
"c2_domains": [],
"c2_ips": [],
"c2_urls": [],
"user_agent": "",
"named_pipes": [],
"spawn_processes": [],
"watermark": "",
}
if not config:
return indicators
# Extract C2 domains
domains = config.get("SETTING_DOMAINS", "")
if domains:
for domain in str(domains).split(","):
domain = domain.strip().rstrip("/")
if domain:
indicators["c2_domains"].append(domain)
# Extract user agent
indicators["user_agent"] = str(config.get("SETTING_USERAGENT", ""))
# Extract named pipes
pipe = config.get("SETTING_PIPENAME", "")
if pipe:
indicators["named_pipes"].append(str(pipe))
# Extract spawn-to processes
for arch in ["SETTING_SPAWNTO_X86", "SETTING_SPAWNTO_X64"]:
proc = config.get(arch, "")
if proc:
indicators["spawn_processes"].append(str(proc))
# Extract watermark
indicators["watermark"] = str(config.get("SETTING_WATERMARK", ""))
return indicatorsif name == "main":
if len(sys.argv) < 2:
print(f"Usage: {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[+] Extracted C2 Indicators:")
print(json.dumps(indicators, indent=2))undefinedfrom dissect.cobaltstrike.beacon import BeaconConfig
def extract_beacon_config(filepath):
"""Parse beacon configuration from file."""
configs = list(BeaconConfig.from_path(filepath))
if not configs:
print(f"[-] No beacon configuration found in {filepath}")
return None
for i, config in enumerate(configs):
print(f"\n[+] Beacon Configuration #{i+1}")
print(f"{'='*60}")
settings = config.as_dict()
# Critical fields for incident response
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 Nonedef extract_c2_indicators(config):
"""Extract actionable C2 indicators from beacon config."""
indicators = {
"c2_domains": [],
"c2_ips": [],
"c2_urls": [],
"user_agent": "",
"named_pipes": [],
"spawn_processes": [],
"watermark": "",
}
if not config:
return indicators
# Extract C2 domains
domains = config.get("SETTING_DOMAINS", "")
if domains:
for domain in str(domains).split(","):
domain = domain.strip().rstrip("/")
if domain:
indicators["c2_domains"].append(domain)
# Extract user agent
indicators["user_agent"] = str(config.get("SETTING_USERAGENT", ""))
# Extract named pipes
pipe = config.get("SETTING_PIPENAME", "")
if pipe:
indicators["named_pipes"].append(str(pipe))
# Extract spawn-to processes
for arch in ["SETTING_SPAWNTO_X86", "SETTING_SPAWNTO_X64"]:
proc = config.get(arch, "")
if proc:
indicators["spawn_processes"].append(str(proc))
# Extract watermark
indicators["watermark"] = str(config.get("SETTING_WATERMARK", ""))
return indicatorsif name == "main":
if len(sys.argv) < 2:
print(f"Usage: {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[+] Extracted C2 Indicators:")
print(json.dumps(indicators, indent=2))undefinedStep 2: Manual XOR Decryption of Beacon Config
步骤2：手动XOR解密Beacon配置
python
import struct
def find_and_decrypt_config(data):
"""Manually locate and decrypt beacon configuration."""
# Cobalt Strike 4.x uses 0x2e as XOR key
xor_keys = [0x2e, 0x69] # v4, v3
for xor_key in xor_keys:
# Search for the config magic bytes after XOR
# Config starts with 0x0001 (BeaconType) XOR'd with key
magic = bytes([0x00 ^ xor_key, 0x01 ^ xor_key,
0x00 ^ xor_key, 0x02 ^ xor_key])
offset = data.find(magic)
if offset == -1:
continue
print(f"[+] Found config at offset 0x{offset:x} (XOR key: 0x{xor_key:02x})")
# Decrypt the config blob (typically 4096 bytes)
config_size = 4096
encrypted = data[offset:offset + config_size]
decrypted = bytes([b ^ xor_key for b in encrypted])
# Parse TLV entries
entries = parse_tlv(decrypted)
return entries
return None
def parse_tlv(data):
"""Parse Type-Length-Value configuration entries."""
entries = {}
offset = 0
# TLV field type mapping
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 entriespython
import struct
def find_and_decrypt_config(data):
"""Manually locate and decrypt beacon configuration."""
# Cobalt Strike 4.x uses 0x2e as XOR key
xor_keys = [0x2e, 0x69] # v4, v3
for xor_key in xor_keys:
# Search for the config magic bytes after XOR
# Config starts with 0x0001 (BeaconType) XOR'd with key
magic = bytes([0x00 ^ xor_key, 0x01 ^ xor_key,
0x00 ^ xor_key, 0x02 ^ xor_key])
offset = data.find(magic)
if offset == -1:
continue
print(f"[+] Found config at offset 0x{offset:x} (XOR key: 0x{xor_key:02x})")
# Decrypt the config blob (typically 4096 bytes)
config_size = 4096
encrypted = data[offset:offset + config_size]
decrypted = bytes([b ^ xor_key for b in encrypted])
# Parse TLV entries
entries = parse_tlv(decrypted)
return entries
return None
def parse_tlv(data):
"""Parse Type-Length-Value configuration entries."""
entries = {}
offset = 0
# TLV field type mapping
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 entriesStep 3: YARA Rule for Beacon Detection
步骤3：用于Beacon检测的YARA规则
python
import yara
cobalt_strike_rule = """
rule CobaltStrike_Beacon_Config {
meta:
description = "Detects Cobalt Strike beacon configuration"
author = "Malware Analysis Team"
date = "2025-01-01"
strings:
// XOR'd config marker for CS 4.x (key 0x2e)
$config_v4 = { 2e 2f 2e 2c }
// XOR'd config marker for CS 3.x (key 0x69)
$config_v3 = { 69 68 69 6b }
// Common beacon strings
$str_pipe = "\\\\.\\pipe\\" ascii wide
$str_beacon = "beacon" ascii nocase
$str_sleeptime = "sleeptime" ascii nocase
// Reflective loader pattern
$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):
"""Scan file with YARA rules for Cobalt Strike beacons."""
rules = yara.compile(source=cobalt_strike_rule)
matches = rules.match(filepath)
for match in matches:
print(f"[+] YARA Match: {match.rule}")
for string_match in match.strings:
offset = string_match.instances[0].offset
print(f" String: {string_match.identifier} at offset 0x{offset:x}")
return matchespython
import yara
cobalt_strike_rule = """
rule CobaltStrike_Beacon_Config {
meta:
description = "Detects Cobalt Strike beacon configuration"
author = "Malware Analysis Team"
date = "2025-01-01"
strings:
// XOR'd config marker for CS 4.x (key 0x2e)
$config_v4 = { 2e 2f 2e 2c }
// XOR'd config marker for CS 3.x (key 0x69)
$config_v3 = { 69 68 69 6b }
// Common beacon strings
$str_pipe = "\\\\.\\pipe\\" ascii wide
$str_beacon = "beacon" ascii nocase
$str_sleeptime = "sleeptime" ascii nocase
// Reflective loader pattern
$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):
"""Scan file with YARA rules for Cobalt Strike beacons."""
rules = yara.compile(source=cobalt_strike_rule)
matches = rules.match(filepath)
for match in matches:
print(f"[+] YARA Match: {match.rule}")
for string_match in match.strings:
offset = string_match.instances[0].offset
print(f" String: {string_match.identifier} at offset 0x{offset:x}")
return matchesStep 4: Network Traffic Correlation
步骤4：网络流量关联
python
from dissect.cobaltstrike.c2 import HttpC2Config
def analyze_c2_profile(beacon_config):
"""Analyze malleable C2 profile from beacon configuration."""
print("\n[+] Malleable C2 Profile Analysis")
print("=" * 60)
# HTTP GET configuration
get_verb = beacon_config.get("SETTING_C2_VERB_GET", "GET")
get_uri = beacon_config.get("SETTING_C2_REQUEST", "")
print(f"\n HTTP GET Request:")
print(f" Verb: {get_verb}")
print(f" URI: {get_uri}")
# HTTP POST configuration
post_verb = beacon_config.get("SETTING_C2_VERB_POST", "POST")
post_uri = beacon_config.get("SETTING_C2_POSTREQ", "")
print(f"\n HTTP POST Request:")
print(f" Verb: {post_verb}")
print(f" URI: {post_uri}")
# User Agent
ua = beacon_config.get("SETTING_USERAGENT", "")
print(f"\n User-Agent: {ua}")
# Host header
host = beacon_config.get("SETTING_C2_HOSTHEADER", "")
print(f" Host Header: {host}")
# Sleep and jitter for traffic pattern
sleep_ms = beacon_config.get("SETTING_SLEEPTIME", 60000)
jitter = beacon_config.get("SETTING_JITTER", 0)
print(f"\n Sleep Time: {sleep_ms}ms")
print(f" Jitter: {jitter}%")
# Generate Suricata/Snort signatures
print(f"\n[+] Suggested Network Signatures:")
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;)')python
from dissect.cobaltstrike.c2 import HttpC2Config
def analyze_c2_profile(beacon_config):
"""Analyze malleable C2 profile from beacon configuration."""
print("\n[+] Malleable C2 Profile Analysis")
print("=" * 60)
# HTTP GET configuration
get_verb = beacon_config.get("SETTING_C2_VERB_GET", "GET")
get_uri = beacon_config.get("SETTING_C2_REQUEST", "")
print(f"\n HTTP GET Request:")
print(f" Verb: {get_verb}")
print(f" URI: {get_uri}")
# HTTP POST configuration
post_verb = beacon_config.get("SETTING_C2_VERB_POST", "POST")
post_uri = beacon_config.get("SETTING_C2_POSTREQ", "")
print(f"\n HTTP POST Request:")
print(f" Verb: {post_verb}")
print(f" URI: {post_uri}")
# User Agent
ua = beacon_config.get("SETTING_USERAGENT", "")
print(f"\n User-Agent: {ua}")
# Host header
host = beacon_config.get("SETTING_C2_HOSTHEADER", "")
print(f" Host Header: {host}")
# Sleep and jitter for traffic pattern
sleep_ms = beacon_config.get("SETTING_SLEEPTIME", 60000)
jitter = beacon_config.get("SETTING_JITTER", 0)
print(f"\n Sleep Time: {sleep_ms}ms")
print(f" Jitter: {jitter}%")
# Generate Suricata/Snort signatures
print(f"\n[+] Suggested Network Signatures:")
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;)')Validation Criteria
验证标准
- Beacon configuration successfully extracted from PE file or memory dump
- C2 server domains/IPs correctly identified with port and protocol
- Malleable C2 profile parameters decoded showing HTTP transforms
- Watermark value extracted for attribution correlation
- Sleep time and jitter values match observed network beacon intervals
- YARA rules detect beacon in both packed and unpacked samples
- Network signatures generated from extracted C2 profile
- 成功从PE文件或内存转储中提取Beacon配置
- 正确识别C2服务器域名/IP以及端口和协议
- 解码出可变形C2配置文件参数，展示HTTP转换规则
- 提取水印值用于关联攻击归属
- 提取的休眠时间和抖动值与观察到的网络Beacon间隔匹配
- YARA规则可检测打包和解包样本中的Beacon
- 根据提取的C2配置生成网络签名