Realizando Análise Automatizada de Malware com CAPE

Visão Geral

CAPE (Config And Payload Extraction) é uma sandbox de malware de código aberto derivada do Cuckoo que automatiza a análise comportamental, extração de payloads e extração de configurações. O CAPEv2 apresenta API hooking para instrumentação comportamental, captura arquivos criados/modificados/excluídos durante a execução, grava tráfego de rede no formato PCAP e inclui mais de 70 extratores de configuração personalizados (cape-parsers) para famílias como Emotet, TrickBot, Cobalt Strike, AsyncRAT e Rhadamanthys. O sistema de assinaturas inclui mais de 1000 assinaturas comportamentais que detectam técnicas de evasão, persistência, roubo de credenciais e comportamento de ransomware. O debugger do CAPE permite contornar anti-evasões dinâmicas combinando ações do debugger dentro de assinaturas YARA. Implantação recomendada: host Ubuntu LTS com VM guest Windows 10 21H2.

Quando Usar

• Ao conduzir avaliações de segurança que envolvam a realização de análise automatizada de malware com CAPE
• Ao seguir procedimentos de resposta a incidentes para eventos de segurança relacionados
• Ao realizar testes de segurança ou atividades de auditoria agendadas
• Ao validar controles de segurança através de testes práticos

Pré-requisitos

• Servidor Ubuntu 22.04 LTS (8+ núcleos de CPU, 32GB+ RAM, 500GB+ SSD)
• Suporte de virtualização KVM/QEMU
• Imagem guest Windows 10 21H2
• Python 3.9+ com dependências do CAPEv2
• Configuração de rede para rede de análise isolada

Fluxo de Trabalho

Etapa 1: Enviar e Analisar Amostras via API

#!/usr/bin/env python3
"""Cliente API da sandbox CAPE para submissão e análise automatizada de malware."""
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):
        """Submete um arquivo para análise."""
        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"[+] Submetido {filepath} -> ID da Tarefa: {task_id}")
        return task_id

    def get_status(self, task_id):
        """Verifica o status da análise da tarefa."""
        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):
        """Espera a análise ser concluída."""
        elapsed = 0
        while elapsed < max_wait:
            status = self.get_status(task_id)
            if status == "reported":
                print(f"[+] Tarefa {task_id} concluída")
                return True
            time.sleep(poll_interval)
            elapsed += poll_interval
            print(f"  Aguardando... ({elapsed}s, status: {status})")
        return False

    def get_report(self, task_id):
        """Recupera o relatório completo da análise."""
        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):
        """Obtém a configuração extraída do malware."""
        report = self.get_report(task_id)
        configs = report.get("CAPE", {}).get("configs", [])
        return configs

    def get_dropped_files(self, task_id):
        """Lista os arquivos descartados durante a análise."""
        report = self.get_report(task_id)
        return report.get("dropped", [])

    def get_network_iocs(self, task_id):
        """Extrai IOCs de rede da análise."""
        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):
        """Pipeline completo de análise automatizada."""