计划执行与检查节点
executing-plans
按既定实施计划完成工作,保持过程检查和反馈。
performing-malware-ioc-extraction【效率与协作】恶意软件 IOC(失陷指标)提取是指通过分析恶意软件,识别可操作的失陷指标,包括文件哈希、网络指标(C2 域名、IP 地址、URL)、注册表修改、互斥体名称、嵌入字符串和行为产物。
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
恶意软件 IOC(失陷指标,Indicator of Compromise)提取是指通过分析恶意软件,识别可操作的失陷指标,包括文件哈希、网络指标(C2 域名、IP 地址、URL)、注册表修改、互斥体名称、嵌入字符串和行为产物。本技能涵盖:使用 PE 解析和字符串提取进行静态分析、通过沙箱引爆进行动态分析、使用 YARA 等工具进行自动化 IOC 提取,以及将结果格式化为 STIX 2.1 指标以供共享。
pefile、yara-python、oletools、stix2 库YARA 是一种用于识别和分类恶意软件的模式匹配工具。规则由字符串(文本、十六进制、正则表达式)和定义匹配逻辑的条件组成。规则可以检测恶意软件家族、加壳程序、漏洞利用工具包和特定活动工具。
import pefile
import hashlib
import os
def analyze_pe(filepath):
"""通过静态分析从 PE 文件中提取 IOC。"""
iocs = {"hashes": {}, "pe_info": {}, "strings": [], "imports": []}
# 计算文件哈希
with open(filepath, "rb") as f:
data = f.read()
iocs["hashes"]["md5"] = hashlib.md5(data).hexdigest()
iocs["hashes"]["sha1"] = hashlib.sha1(data).hexdigest()
iocs["hashes"]["sha256"] = hashlib.sha256(data).hexdigest()
iocs["hashes"]["file_size"] = len(data)
# 解析 PE 头部
try:
pe = pefile.PE(filepath)
iocs["hashes"]["imphash"] = pe.get_imphash()
iocs["pe_info"]["compilation_time"] = str(pe.FILE_HEADER.TimeDateStamp)
iocs["pe_info"]["machine_type"] = hex(pe.FILE_HEADER.Machine)
iocs["pe_info"]["subsystem"] = pe.OPTIONAL_HEADER.Subsystem
# 提取节信息
iocs["pe_info"]["sections"] = []
for section in pe.sections:
iocs["pe_info"]["sections"].append({
"name": section.Name.decode("utf-8", errors="ignore").strip("\x00"),
"virtual_size": section.Misc_VirtualSize,
"raw_size": section.SizeOfRawData,
"entropy": section.get_entropy(),
"md5": section.get_hash_md5(),
})
# 提取导入表
if hasattr(pe, "DIRECTORY_ENTRY_IMPORT"):
for entry in pe.DIRECTORY_ENTRY_IMPORT:
dll_name = entry.dll.decode("utf-8", errors="ignore")
functions = [
imp.name.decode("utf-8", errors="ignore")
for imp in entry.imports
if imp.name
]
iocs["imports"].append({"dll": dll_name, "functions": functions})
# 检查可疑特征
iocs["pe_info"]["is_dll"] = pe.is_dll()
iocs["pe_info"]["is_driver"] = pe.is_driver()
iocs["pe_info"]["is_exe"] = pe.is_exe()
# 版本信息
if hasattr(pe, "VS_VERSIONINFO"):
for entry in pe.FileInfo:
for st in entry:
for item in st.entries.items():
key = item[0].decode("utf-8", errors="ignore")
val = item[1].decode("utf-8", errors="ignore")
iocs["pe_info"][f"version_{key}"] = val
pe.close()
except pefile.PEFormatError as e:
iocs["pe_info"]["error"] = str(e)
return iocs
import re
def extract_ioc_strings(filepath):
"""从二进制文件中提取与 IOC 相关的字符串。"""
patterns = {
"ipv4": re.compile(
r"\b(?:(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}"
r"(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\b"
),
"domain": re.compile(
r"\b(?:[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+"
r"(?:com|net|org|io|ru|cn|tk|xyz|top|info|biz|cc|ws|pw)\b"
),
"url": re.compile(
r"https?://[^\s\"'<>]{5,200}"
),
"email": re.compile(
r"\b[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}\b"
),
"registry": re.compile(
r"(?:HKEY_[A-Z_]+|HKLM|HKCU|HKU|HKCR|HKCC)"
r"\\[\\a-zA-Z0-9_ .{}-]+"
),
"filepath_windows": re.compile(
r"[A-Z]:\\(?:[^\\/:*?\"<>|\r\n]+\\)*[^\\/:*?\"<>|\r\n]+"
),
"mutex": re.compile(
r"(?:Global\\|Local\\)[a-zA-Z0-9_\-{}.]{4,}"
),
"useragent": re.compile(
r"Mozilla/[45]\.0[^\"']{10,200}"
),
"bitcoin": re.compile(
r"\b[13][a-km-zA-HJ-NP-Z1-9]{25,34}\b"
),
"pdb_path": re.compile(
r"[A-Z]:\\[^\"]{5,200}\.pdb"
),
}
with open(filepath, "rb") as f:
data = f.read()
# 提取 ASCII 字符串(最小长度 4)
ascii_strings = re.findall(rb"[\x20-\x7e]{4,}", data)
# 提取 Unicode 字符串
unicode_strings = re.findall(
rb"(?:[\x20-\x7e]\x00){4,}", data
)
all_strings = [s.decode("ascii", errors="ignore") for s in ascii_strings]
all_strings += [
s.decode("utf-16-le", errors="ignore") for s in unicode_strings
]
extracted = {category: set() for category in patterns}
for string in all_strings:
for category, pattern in patterns.items():
matches = pattern.findall(string)
for match in matches:
extracted[category].add(match)
# 将集合转换为排序后的列表
return {k: sorted(v) for k, v in extracted.items() if v}
import yara
def scan_with_yara(filepath, rules_path):
"""使用 YARA 规则扫描文件进行恶意软件分类。"""
rules = yara.compile(filepath=rules_path)
matches = rules.match(filepath)
results = []
for match in matches:
result = {
"rule": match.rule,
"namespace": match.namespace,
"tags": match.tags,
"meta": match.meta,
"strings": [],
}
for offset, identifier, data in match.strings:
result["strings"].append({
"offset": hex(offset),
"identifier": identifier,
"data": data.hex() if len(data) < 100 else data[:100].hex() + "...",
})
results.append(result)
return results
# 常见恶意软件指标的示例 YARA 规则
SAMPLE_YARA_RULE = """
rule Suspicious_Network_Indicators {
meta:
description = "检测可疑的网络相关字符串"
author = "CTI Analyst"
severity = "medium"
strings:
$ua1 = "Mozilla/5.0" ascii
$cmd1 = "cmd.exe /c" ascii nocase
$ps1 = "powershell" ascii nocase
$wget = "wget" ascii nocase
$curl = "curl" ascii nocase
$b64 = "base64" ascii nocase
$reg1 = "HKLM\\SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Run" ascii nocase
condition:
uint16(0) == 0x5A4D and
(2 of ($ua1, $cmd1, $ps1, $wget, $curl, $b64)) or $reg1
}
rule Packed_Binary {
meta:
description = "检测可能加壳的二进制文件"
author = "CTI Analyst"
condition:
uint16(0) == 0x5A4D and
for any section in pe.sections : (
section.entropy >= 7.0
)
}
"""
from stix2 import (
Bundle, Indicator, Malware, Relationship,
File as STIXFile, DomainName, IPv4Address,
ObservedData,
)
from datetime import datetime
def create_stix_bundle(pe_iocs, string_iocs, yara_results, sample_name):
"""从提取的 IOC 创建 STIX 2.1 包。"""
objects = []
# 创建恶意软件 SDO
malware = Malware(
name=sample_name,
is_family=False,
malware_types=["unknown"],
description=f"已分析的恶意软件样本: {pe_iocs['hashes']['sha256']}",
allow_custom=True,
)
objects.append(malware)
# 文件哈希指标
sha256 = pe_iocs["hashes"]["sha256"]
hash_indicator = Indicator(
name=f"恶意软件哈希: {sha256[:16]}...",
pattern=f"[file:hashes.'SHA-256' = '{sha256}']",
pattern_type="stix",
valid_from=datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
indicator_types=["malicious-activity"],
allow_custom=True,
)
objects.append(hash_indicator)
objects.append(Relationship(
relationship_type="indicates",
source_ref=hash_indicator.id,
target_ref=malware.id,
))
# 字符串中的网络指标
for ip in string_iocs.get("ipv4", []):
if not ip.startswith(("10.", "172.", "192.168.", "127.")):
ip_indicator = Indicator(
name=f"C2 IP: {ip}",
pattern=f"[ipv4-addr:value = '{ip}']",
pattern_type="stix",
valid_from=datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
indicator_types=["malicious-activity"],
allow_custom=True,
)
objects.append(ip_indicator)
objects.append(Relationship(
relationship_type="indicates",
source_ref=ip_indicator.id,
target_ref=malware.id,
))
for domain in string_iocs.get("domain", []):
domain_indicator = Indicator(
name=f"C2 域名: {domain}",
pattern=f"[domain-name:value = '{domain}']",
pattern_type="stix",
valid_from=datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
indicator_types=["malicious-activity"],
allow_custom=True,
)
objects.append(domain_indicator)
objects.append(Relationship(
relationship_type="indicates",
source_ref=domain_indicator.id,
target_ref=malware.id,
))
bundle = Bundle(objects=objects, allow_custom=True)
return bundle