分析 Cobalt Strike Beacon 設定 是做什麼的？
概述
Cobalt Strike 是一款商業化的對手模擬工具，廣泛被威脅行為者濫用於入侵後階段的作業。Beacon 酬載中內嵌的組態資料會揭示 C2 伺服器位址、通訊協定、睡眠間隔、抖動值、可塑 C2 設定檔設定、浮水印識別碼和加密金鑰。從 PE 檔案、Shellcode 或記憶體傾印中萃取此組態，對於事件應變人員對抗攻擊者基礎設施以及歸因攻擊活動至關重要。Beacon 組態使用單一位元組 (版本 3 為 0x69，版本 4 為 0x2e) 進行 XOR 編碼，並以類型-長度-值 (TLV) 格式儲存在 PE 的 .data 節區中。
使用時機
- 調查需要分析 Cobalt Strike beacon 設定的安全事件時
- 為此領域建置偵測規則或威脅獵殺查詢時
- 安全營運中心分析師需要此分析類型的結構化程序時
- 驗證相關攻擊技術的安全監控覆蓋範圍時
先決條件
- Python 3.9+ 且安裝
dissect.cobaltstrike、pefile、yara-python - SentinelOne CobaltStrikeParser (
parse_beacon_config.py) - 十六進位編輯器 (010 Editor, HxD) 用於手動檢查
- 了解 PE 檔案格式和 XOR 編碼
- 記憶體傾印擷取工具 (Volatility3, WinDbg)
- 網路分析工具 (Wireshark) 用於 C2 流量關聯
關鍵概念
Beacon 設定結構
Cobalt Strike beacon 會將其組態儲存為 PE 檔案 .data 節區中的 TLV (類型-長度-值) 項目區塊。無階段 (Stageless) 的 beacon 會使用 4 位元組金鑰對整個 beacon 程式碼進行 XOR 操作。組態區塊本身則使用單位元組 XOR 金鑰。每個 TLV 項目包含一個 2 位元組的類型識別碼 (例如，0x0001 為 BeaconType，0x0008 為 C2Server)、一個 2 位元組的長度以及可變長度的資料。
可塑 C2 設定檔
Beacon 組態中會編碼可塑 C2 設定檔，此設定檔規定了 HTTP 請求/回應的轉換方式，包含 URI 路徑、標頭、元資料編碼 (Base64、NetBIOS) 和資料轉換。分析這些設定可以揭示 beacon 如何偽裝其流量以融入合法網路流量。
浮水印與授權識別
每個 Cobalt Strike 授權證都會在產生的 beacon 中嵌入一個唯一的浮水印 (4 位元組整數)。萃取浮水印可以將多個 beacon 連結到同一個操作者或破解的授權證。威脅情報提供者維護的已知浮水印資料庫，可將浮水印對應到特定的威脅行為者或外洩的授權金鑰。
工作流程
步驟 1：使用 CobaltStrikeParser 萃取設定
#!/usr/bin/env python3
"""從 PE 或記憶體傾印中萃取 Cobalt Strike beacon 設定。"""
import sys
import json
# 使用 SentinelOne 的 CobaltStrikeParser
# pip install dissect.cobaltstrike
from dissect.cobaltstrike.beacon import BeaconConfig
def extract_beacon_config(filepath):
"""從檔案中解析 beacon 設定。"""
configs = list(BeaconConfig.from_path(filepath))
if not configs:
print(f"[-] 在 {filepath} 中未找到 beacon 設定")
return None
for i, config in enumerate(configs):
print(f"\n[+] Beacon 設定 #{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):
"""從 beacon 設定中萃取消極的 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)
# 萃取 User-Agent
indicators["user_agent"] = str(config.get("SETTING_USERAGENT", ""))
# 萃取具名管道
pipe = config.get("SETTING_PIPENAME", "")
if pipe:
indicators["named_pipes"].append(str(pipe))
# 萃取 spawn-to 程序
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 解密 Beacon 設定
import struct
def find_and_decrypt_config(data):
"""手動定位並解密 beacon 設定。"""
# Cobalt Strike 4.x 使用 0x2e 作為 XOR 金鑰
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：用於 Beacon 偵測的 YARA 規則
import yara
cobalt_strike_rule = """
rule CobaltStrike_Beacon_Config {
meta:
description = "偵測 Cobalt Strike beacon 設定"
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 }
// 常見的 beacon 字串
$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 beacon。"""
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):
"""從 beacon 設定中分析可塑 C2 設定檔。"""
print("\n[+] 可塑 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}")
# User-Agent
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 檔案或記憶體傾印成功萃取出 Beacon 設定
- 正確識別 C2 伺服器域名/IP，包含連接埠和協定
- 解碼可塑 C2 設定檔參數，顯示 HTTP 轉換
- 萃取浮水印值以進行歸因關聯
- 睡眠時間和抖動值與觀察到的網路 Beacon 間隔相符
- YARA 規則能偵測到已打包和未打包的樣本
- 從萃取的 C2 設定檔產生網路簽章