Construindo um Pipeline Automatizado de Submissão de Malware

Quando Usar

Use esta funcionalidade quando:

• Equipes de SOC enfrentam um alto volume de alertas de arquivos suspeitos que exigem análise em sandbox.
• A submissão manual em sandbox cria gargalos no fluxo de trabalho de triagem de alertas.
• Ferramentas de segurança de endpoint e e-mail quarentenam arquivos que necessitam de determinação automatizada de veredicto.
• A resposta a incidentes requer identificação rápida da família de malware e extração de IOCs.

Não use para analisar amostras de malware ativas em ambientes de produção — sempre utilize infraestrutura de sandbox isolada.

Pré-requisitos

• Ambiente de sandbox: Cuckoo Sandbox, Joe Sandbox, Any.Run ou VMRay.
• Chave de API do VirusTotal (Enterprise para submissão, gratuita para consulta).
• Acesso à API do MalwareBazaar para consulta de malware conhecido.
• Mecanismo de coleta de arquivos: API de quarentena do EDR, exportação de gateway de e-mail, captura de rede.
• Python 3.8+ com as bibliotecas requests, vt-py, pefile.
• Rede de análise isolada, sem conectividade com a produção.

Fluxo de Trabalho

Etapa 1: Construir o Pipeline de Coleta de Arquivos

Colete arquivos suspeitos de várias fontes:

import requests
import hashlib
import os
from pathlib import Path
from datetime import datetime

class MalwareCollector:
    def __init__(self, quarantine_dir="/opt/malware_quarantine"):
        self.quarantine_dir = Path(quarantine_dir)
        self.quarantine_dir.mkdir(exist_ok=True)

    def collect_from_edr(self, edr_api_url, api_token):
        """Pull quarantined files from CrowdStrike Falcon"""
        headers = {"Authorization": f"Bearer {api_token}"}

        # Get recent quarantine events
        response = requests.get(
            f"{edr_api_url}/quarantine/queries/quarantined-files/v1",
            headers=headers,
            params={"filter": "state:'quarantined'", "limit": 50}
        )
        file_ids = response.json()["resources"]

        for file_id in file_ids:
            # Download quarantined file
            dl_response = requests.get(
                f"{edr_api_url}/quarantine/entities/quarantined-files/v1",
                headers=headers,
                params={"ids": file_id}
            )
            file_data = dl_response.content
            sha256 = hashlib.sha256(file_data).hexdigest()

            filepath = self.quarantine_dir / f"{sha256}.sample"
            filepath.write_bytes(file_data)
            yield {"sha256": sha256, "path": str(filepath), "source": "edr"}

    def collect_from_email_gateway(self, smtp_quarantine_path):
        """Pull attachments from email gateway quarantine"""
        import email
        from email import policy

        for eml_file in Path(smtp_quarantine_path).glob("*.eml"):
            msg = email.message_from_binary_file(
                eml_file.open("rb"), policy=policy.default
            )
            for attachment in msg.iter_attachments():
                content = attachment.get_content()
                if isinstance(content, str):
                    content = content.encode()
                sha256 = hashlib.sha256(content).hexdigest()
                filename = attachment.get_filename() or "unknown"

                filepath = self.quarantine_dir / f"{sha256}.sample"
                filepath.write_bytes(content)
                yield {
                    "sha256": sha256,
                    "path": str(filepath),
                    "source": "email",
                    "original_filename": filename,
                    "sender": msg["From"],
                    "subject": msg["Subject"]
                }

    def compute_hashes(self, filepath):
        """Calculate MD5, SHA1, SHA256 for a file"""
        with open(filepath, "rb") as f:
            content = f.read()
        return {
            "md5": hashlib.md5(content).hexdigest(),
            "sha1": hashlib.sha1(content).hexdigest(),
            "sha256": hashlib.sha256(content).hexdigest(),
            "size": len(content)
        }

Etapa 2: Pré-triagem com Consultas de Hash

Verifique se o arquivo já é conhecido antes da submissão para sandbox:

import vt

class MalwarePreScreener:
    def __init__(self, vt_api_key, mb_api_url="https://mb-api.abuse.ch/api/v1/"):
        self.vt_client = vt.Client(vt_api_key)
        self.mb_api_url = mb_api_url

    def check_virustotal(self, sha256):
        """Consulta hash no VirusTotal"""
        try:
            file_obj = self.vt_client.get_object(f"/files/{sha256}")
            stats = file_obj.last_analysis_stats
            return {
                "found": True,

…