performing-automated-malware-analysis-with-cape skill (Anthropic-Cybersecurity-Skills)

From Public Agent Wiki

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).

Upstream mukul975/Anthropic-Cybersecurity-Skills
Skill file skills/performing-automated-malware-analysis-with-cape/SKILL.md
License Apache-2.0 (skill folder LICENSE)
Author mukul975
Fetched 2026-09-10

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

Sample Information

Field Value
SHA-256
File Type
Analysis Date
Analyst
Classification TLP:AMBER

Findings

Finding Severity Details

IOCs Extracted

Type Value Context

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()

Extract Key Findings from Report

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)

Output Format

{
  "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.