performing-automated-malware-analysis-with-cape部署和操作 CAPEv2 沙箱，进行自动化恶意软件分析，具备行为监控、载荷提取、配置解析和反规避能力。
Independent GitHub snapshot · Automated classification · Original documentation governs
CAPE（Config And Payload Extraction，配置与载荷提取）是一款从 Cuckoo 派生的开源恶意软件沙箱，可自动化执行行为分析、载荷转储和配置提取。CAPEv2 具备用于行为插桩的 API 钩挂功能，可捕获执行过程中创建/修改/删除的文件，以 PCAP 格式记录网络流量，并包含 70+ 个针对 Emotet、TrickBot、Cobalt Strike、AsyncRAT 和 Rhadamanthys 等家族的自定义配置提取器（cape-parsers）。签名系统包含 1000+ 个行为签名，可检测规避技术、持久化、凭据窃取和勒索软件行为。CAPE 的调试器支持通过在 YARA 签名中组合调试器操作来动态绕过反规避机制。推荐部署方式：Ubuntu LTS 宿主机 + Windows 10 21H2 客户机虚拟机。
#!/usr/bin/env python3
"""用于自动化恶意软件提交和分析的 CAPE 沙箱 API 客户端。"""
import requests
import json
import time
import sys
from pathlib import Path
class CAPEClient:
def __init__(self, base_url="http://localhost:8000", api_token=None):
self.base_url = base_url.rstrip("/")
self.headers = {}
if api_token:
self.headers["Authorization"] = f"Token {api_token}"
def submit_file(self, filepath, options=None):
"""提交文件进行分析。"""
url = f"{self.base_url}/apiv2/tasks/create/file/"
files = {"file": open(filepath, "rb")}
data = options or {}
data.setdefault("timeout", 120)
data.setdefault("enforce_timeout", False)
resp = requests.post(url, files=files, data=data, headers=self.headers)
resp.raise_for_status()
result = resp.json()
task_id = result.get("data", {}).get("task_ids", [None])[0]
print(f"[+] 已提交 {filepath} -> 任务 ID：{task_id}")
return task_id
def get_status(self, task_id):
"""检查任务分析状态。"""
url = f"{self.base_url}/apiv2/tasks/status/{task_id}/"
resp = requests.get(url, headers=self.headers)
return resp.json().get("data", "unknown")
def wait_for_completion(self, task_id, poll_interval=15, max_wait=600):
"""等待分析完成。"""
elapsed = 0
while elapsed < max_wait:
status = self.get_status(task_id)
if status == "reported":
print(f"[+] 任务 {task_id} 已完成")
return True
time.sleep(poll_interval)
elapsed += poll_interval
print(f" 等待中...（{elapsed}s，状态：{status}）")
return False
def get_report(self, task_id):
"""获取完整分析报告。"""
url = f"{self.base_url}/apiv2/tasks/get/report/{task_id}/"
resp = requests.get(url, headers=self.headers)
return resp.json()
def get_config(self, task_id):
"""获取提取的恶意软件配置。"""
report = self.get_report(task_id)
configs = report.get("CAPE", {}).get("configs", [])
return configs
def get_dropped_files(self, task_id):
"""列出分析期间投放的文件。"""
report = self.get_report(task_id)
return report.get("dropped", [])
def get_network_iocs(self, task_id):
"""从分析结果中提取网络 IoC。"""
report = self.get_report(task_id)
network = report.get("network", {})
iocs = {
"dns": [d.get("request") for d in network.get("dns", [])],
"http": [h.get("uri") for h in network.get("http", [])],
"tcp": [f"{h.get('dst')}:{h.get('dport')}"
for h in network.get("tcp", [])],
}
return iocs
def analyze_sample(self, filepath):
"""完整的自动化分析流水线。"""
task_id = self.submit_file(filepath)
if not task_id:
return None
if self.wait_for_completion(task_id):
report = {
"task_id": task_id,
"config": self.get_config(task_id),
"network_iocs": self.get_network_iocs(task_id),
"dropped_files": len(self.get_dropped_files(task_id)),
}
return report
return None
if __name__ == "__main__":
if len(sys.argv) < 2:
print(f"用法：{sys.argv[0]} <malware_sample> [cape_url]")
sys.exit(1)
url = sys.argv[2] if len(sys.argv) > 2 else "http://localhost:8000"
client = CAPEClient(url)
result = client.analyze_sample(sys.argv[1])
if result:
print(json.dumps(result, indent=2))