What it does. Deploy and operate the CAPEv2 malware sandbox (a Cuckoo derivative) to run samples in a Part of mukul975/Anthropic-Cybersecurity-Skills (817 security skills) (mukul975/Anthropic-Cybersecurity-Skills).
Install
npx skills add mukul975/Anthropic-Cybersecurity-Skills --skill performing-automated-malware-analysis-with-cape, or copy the skill folder into ~/.claude/skills/performing-automated-malware-analysis-with-cape/.
- Raw file:
curl -sL https://raw.githubusercontent.com/mukul975/Anthropic-Cybersecurity-Skills/HEAD/skills/performing-automated-malware-analysis-with-cape/SKILL.md
SKILL.md (verbatim)
name: performing-automated-malware-analysis-with-cape
description: Deploy and operate the CAPEv2 malware sandbox (a Cuckoo derivative) to run samples in a
monitored Windows guest VM, capturing behavioral signatures, dropped files, PCAP network traffic,
and family-specific configuration extraction (e.g. Emotet, TrickBot, Cobalt Strike) via
cape-parsers. Use when a suspicious file or payload needs automated dynamic analysis, anti-evasion
debugger tricks, or config/payload extraction.
domain: cybersecurity
subdomain: malware-analysis
tags:
- cape
- sandbox
- automated-analysis
- malware-analysis
- behavioral-analysis
- payload-extraction
- cuckoo
version: '1.0'
author: mahipal
license: Apache-2.0
nist_csf:
- DE.AE-02
- RS.AN-03
- ID.RA-01
- DE.CM-01
mitre_attack:
- T1027
- T1055
- T1140
- T1497
- T1070
Performing Automated Malware Analysis with CAPE
Overview
CAPE (Config And Payload Extraction) is an open-source malware sandbox derived from Cuckoo that automates behavioral analysis, payload dumping, and configuration extraction. CAPEv2 features API hooking for behavioral instrumentation, captures files created/modified/deleted during execution, records network traffic in PCAP format, and includes 70+ custom configuration extractors (cape-parsers) for families like Emotet, TrickBot, Cobalt Strike, AsyncRAT, and Rhadamanthys. The signature system includes 1000+ behavioral signatures detecting evasion techniques, persistence, credential theft, and ransomware behavior. CAPE's debugger enables dynamic anti-evasion bypasses combining debugger actions within YARA signatures. Recommended deployment: Ubuntu LTS host with Windows 10 21H2 guest VM.
When to Use
- When conducting security assessments that involve performing automated malware analysis with cape
- When following incident response procedures for related security events
- When performing scheduled security testing or auditing activities
- When validating security controls through hands-on testing
Prerequisites
- Ubuntu 22.04 LTS server (8+ CPU cores, 32GB+ RAM, 500GB+ SSD)
- KVM/QEMU virtualization support
- Windows 10 21H2 guest image
- Python 3.9+ with CAPEv2 dependencies
- Network configuration for isolated analysis network
Workflow
Step 1: Submit and Analyze Samples via API
#!/usr/bin/env python3
"""CAPE sandbox API client for automated malware submission and analysis."""
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):
"""Submit a file for analysis."""
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"[+] Submitted {filepath} -> Task ID: {task_id}")
return task_id
def get_status(self, task_id):
"""Check task analysis status."""
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):
"""Wait for analysis to complete."""
elapsed = 0
while elapsed < max_wait:
status = self.get_status(task_id)
if status == "reported":
print(f"[+] Task {task_id} completed")
return True
time.sleep(poll_interval)
elapsed += poll_interval
print(f" Waiting... ({elapsed}s, status: {status})")
return False
def get_report(self, task_id):
"""Retrieve full analysis report."""
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):
"""Get extracted malware configuration."""
report = self.get_report(task_id)
configs = report.get("CAPE", {}).get("configs", [])
return configs
def get_dropped_files(self, task_id):
"""List files dropped during analysis."""
report = self.get_report(task_id)
return report.get("dropped", [])
def get_network_iocs(self, task_id):
"""Extract network IOCs from analysis."""
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):
"""Full automated analysis pipeline."""
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"Usage: {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))
Validation Criteria
- Samples submitted and analyzed within configured timeout
- Behavioral signatures triggered for known malware families
- Malware configurations extracted by cape-parsers
- Network traffic captured and IOCs extracted
- Dropped files and payloads collected for further analysis
- Anti-evasion bypasses effective against sandbox-aware malware
References
Other files in this skill
assets/template.md (verbatim)
Analysis Report Template - performing-automated-malware-analysis-with-cape
| Field |
Value |
| SHA-256 |
|
| File Type |
|
| Analysis Date |
|
| Analyst |
|
| Classification |
TLP:AMBER |
Findings
Recommendations
references/api-reference.md (verbatim)
API Reference: CAPE Sandbox Automated Malware Analysis
Libraries Used
| Library |
Purpose |
requests |
HTTP client for CAPE REST API v2 |
json |
Parse analysis reports and task metadata |
os |
Read CAPE_URL and CAPE_API_KEY environment variables |
time |
Poll task status until analysis completes |
Installation
pip install requests
Authentication
import requests
import os
CAPE_URL = os.environ.get("CAPE_URL", "http://cape.example.com:8000")
CAPE_KEY = os.environ.get("CAPE_API_KEY", "")
headers = {"Authorization": f"Token {CAPE_KEY}"} if CAPE_KEY else {}
REST API v2 Endpoints
| Method |
Endpoint |
Description |
| POST |
/apiv2/tasks/create/file/ |
Submit a file for analysis |
| POST |
/apiv2/tasks/create/url/ |
Submit a URL for analysis |
| GET |
/apiv2/tasks/list/ |
List all analysis tasks |
| GET |
/apiv2/tasks/view/{id}/ |
Get task status and metadata |
| GET |
/apiv2/tasks/report/{id}/ |
Get full analysis report |
| GET |
/apiv2/tasks/report/{id}/lite/ |
Get lightweight report |
| DELETE |
/apiv2/tasks/delete/{id}/ |
Delete a task and its data |
| GET |
/apiv2/tasks/screenshots/{id}/ |
Get analysis screenshots |
| GET |
/apiv2/tasks/procmemory/{id}/ |
Get process memory dumps |
| GET |
/apiv2/files/view/sha256/{hash}/ |
Look up file by SHA-256 |
| GET |
/apiv2/files/get/{sha256}/ |
Download the sample binary |
| GET |
/apiv2/pcap/get/{id}/ |
Download PCAP network capture |
| GET |
/apiv2/machines/list/ |
List analysis VMs |
| GET |
/apiv2/cuckoo/status/ |
Server status and version |
Core Operations
Submit a File for Analysis
def submit_file(file_path, timeout_mins=5, machine=None):
files = {"file": open(file_path, "rb")}
data = {
"timeout": timeout_mins * 60,
"enforce_timeout": True,
"options": "procmemdump=yes,import_reconstruction=yes",
}
if machine:
data["machine"] = machine
resp = requests.post(
f"{CAPE_URL}/apiv2/tasks/create/file/",
files=files,
data=data,
headers=headers,
timeout=60,
)
resp.raise_for_status()
result = resp.json()
return result["data"]["task_ids"][0]
Submit a URL for Analysis
def submit_url(url, timeout_mins=3):
resp = requests.post(
f"{CAPE_URL}/apiv2/tasks/create/url/",
data={
"url": url,
"timeout": timeout_mins * 60,
"options": "procmemdump=yes",
},
headers=headers,
timeout=30,
)
resp.raise_for_status()
return resp.json()["data"]["task_ids"][0]
Poll Task Until Complete
import time
def wait_for_task(task_id, poll_interval=30, max_wait=600):
elapsed = 0
while elapsed < max_wait:
resp = requests.get(
f"{CAPE_URL}/apiv2/tasks/view/{task_id}/",
headers=headers,
timeout=30,
)
status = resp.json()["data"]["status"]
if status == "reported":
return True
if status in ("failed_analysis", "failed_processing"):
raise RuntimeError(f"Task {task_id} failed: {status}")
time.sleep(poll_interval)
elapsed += poll_interval
raise TimeoutError(f"Task {task_id} did not complete within {max_wait}s")
Retrieve Analysis Report
def get_report(task_id, lite=False):
endpoint = "lite" if lite else ""
resp = requests.get(
f"{CAPE_URL}/apiv2/tasks/report/{task_id}/{endpoint}",
headers=headers,
timeout=120,
)
resp.raise_for_status()
return resp.json()
def extract_findings(report):
info = report.get("info", {})
findings = {
"score": info.get("score", 0),
"duration": info.get("duration", 0),
"signatures": [],
"network_iocs": {"domains": [], "ips": [], "urls": []},
"dropped_files": [],
"yara_matches": [],
}
# Behavioral signatures
for sig in report.get("signatures", []):
findings["signatures"].append({
"name": sig["name"],
"severity": sig["severity"],
"description": sig["description"],
})
# Network IOCs
network = report.get("network", {})
findings["network_iocs"]["domains"] = [
d["domain"] for d in network.get("domains", [])
]
findings["network_iocs"]["ips"] = [
h["ip"] for h in network.get("hosts", [])
]
# YARA matches
for target_yara in report.get("target", {}).get("file", {}).get("yara", []):
findings["yara_matches"].append(target_yara["name"])
return findings
Download Network PCAP
def download_pcap(task_id, output_path):
resp = requests.get(
f"{CAPE_URL}/apiv2/pcap/get/{task_id}/",
headers=headers,
timeout=60,
)
resp.raise_for_status()
with open(output_path, "wb") as f:
f.write(resp.content)
{
"info": {
"id": 42,
"score": 8.5,
"duration": 120,
"machine": {"name": "win10-01", "label": "win10-01"},
"started": "2025-01-15T10:30:00",
"ended": "2025-01-15T10:32:00"
},
"signatures": [
{"name": "ransomware_bcdedit", "severity": 5, "description": "Modifies boot configuration"},
{"name": "creates_exe", "severity": 3, "description": "Creates executable files on disk"}
],
"network": {
"hosts": [{"ip": "198.51.100.42", "country": "US"}],
"domains": [{"domain": "c2.evil.example.com", "ip": "198.51.100.42"}]
},
"target": {
"file": {
"name": "sample.exe",
"size": 245760,
"sha256": "a1b2c3d4e5f6..."
}
}
}
references/standards.md (verbatim)
Standards Reference - performing-automated-malware-analysis-with-cape
Applicable Standards
- MITRE ATT&CK Framework
- NIST SP 800-83 Guide to Malware Incident Prevention
- NIST SP 800-86 Guide to Integrating Forensic Techniques
See SKILL.md for specific technique mappings.
references/workflows.md (verbatim)
Analysis Workflows - performing-automated-malware-analysis-with-cape
Primary Workflow
[Sample Collection] --> [Static Analysis] --> [Dynamic Analysis] --> [IOC Extraction]
|
v
[Report Generation]
See SKILL.md for detailed step-by-step procedures.
Back to mukul975/Anthropic-Cybersecurity-Skills (817 security skills) or Agent skills.