Pipeline automatisé de soumission et d'analyse de malwares

Construction d'un Pipeline Automatisé de Soumission de Malwares

Quand l'utiliser

Utilisez cette compétence lorsque :

• Les équipes SOC font face à un volume élevé d'alertes de fichiers suspects nécessitant une analyse en sandbox
• La soumission manuelle au sandbox crée des goulots d'étranglement dans le workflow de tri des alertes
• Les outils de sécurité des endpoints et des emails mettent en quarantaine des fichiers nécessitant une détermination automatisée de verdict
• La réponse aux incidents nécessite une identification rapide des familles de malwares et l'extraction d'IOCs

N'utilisez pas cette compétence pour analyser des échantillons de malwares en direct dans des environnements de production — utilisez toujours une infrastructure sandbox isolée.

Prérequis

• Environnement sandbox : Cuckoo Sandbox, Joe Sandbox, Any.Run ou VMRay
• Clé API VirusTotal (Enterprise pour la soumission, gratuite pour la recherche)
• Accès à l'API MalwareBazaar pour la recherche de malwares connus
• Mécanisme de collecte de fichiers : API de quarantaine EDR, export de passerelle email, capture réseau
• Python 3.8+ avec les bibliothèques requests, vt-py, pefile
• Réseau d'analyse isolé sans connectivité de production

Workflow

Étape 1 : Construire le Pipeline de Collecte de Fichiers

Collectez les fichiers suspects depuis plusieurs sources :

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):
        """Récupérer les fichiers mis en quarantaine depuis CrowdStrike Falcon"""
        headers = {"Authorization": f"Bearer {api_token}"}

        # Obtenir les événements de quarantaine récents
        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:
            # Télécharger le fichier en quarantaine
            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):
        """Récupérer les pièces jointes depuis la quarantaine de la passerelle email"""
        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 "inconnu"

                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):
        """Calculer les empreintes MD5, SHA1, SHA256 d'un fichier"""
        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)
        }

Étape 2 : Pré-trier avec des Recherches d'Empreintes

Vérifiez si le fichier est déjà connu avant de le soumettre au 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):
        """Rechercher l'empreinte dans VirusTotal"""
        try:
            file_obj = self.vt_client.get_object(f"/files/{sha256}")
            stats = file_obj.last_analysis_stats
            return {
                "found": True,
                "malicious": stats.get("malicious", 0),
                "suspicious": stats.get("suspicious", 0),
                "undetected": stats.get("undetected", 0),
                "total": sum(stats.values()),
                "threat_label": getattr(file_obj, "popular_threat_classification", {}).get(
                    "suggested_threat_label", "Inconnu"
                ),
                "type": getattr(file_obj, "type_description", "Inconnu")
            }
        except vt.APIError:
            return {"found": False}

    def check_malwarebazaar(self, sha256):
        """Rechercher l'empreinte dans MalwareBazaar"""
        response = requests.post(
            self.mb_api_url,
            data={"query": "get_info", "hash": sha256}
        )
        data = response.json()
        if data["query_status"] == "ok":
            entry = data["data"][0]
            return {
                "found": True,
                "signature": entry.get("signature", "Inconnu"),
                "tags": entry.get("tags", []),
                "file_type": entry.get("file_type", "Inconnu"),
                "first_seen": entry.get("first_seen", "Inconnu")
            }
        return {"found": False}

    def pre_screen(self, sha256):
        """Exécuter toutes les vérifications de pré-tri"""
        vt_result = self.check_virustotal(sha256)
        mb_result = self.check_malwarebazaar(sha256)

        verdict = "INCONNU"
        if vt_result["found"] and vt_result.get("malicious", 0) > 10:
            verdict = "CONNU_MALVEILLANT"
        elif vt_result["found"] and vt_result.get("malicious", 0) == 0:
            verdict = "PROBABLEMENT_SAIN"

        return {
            "sha256": sha256,
            "virustotal": vt_result,
            "malwarebazaar": mb_result,
            "pre_screen_verdict": verdict,
            "needs_sandbox": verdict == "INCONNU"
        }

    def close(self):
        self.vt_client.close()

Étape 3 : Soumettre au Sandbox pour l'Analyse Dynamique

Soumission à Cuckoo Sandbox :

class SandboxSubmitter:
    def __init__(self, cuckoo_url="http://cuckoo.internal:8090"):
        self.cuckoo_url = cuckoo_url

    def submit_to_cuckoo(self, filepath, timeout=300):
        """Soumettre un fichier à Cuckoo Sandbox"""
        with open(filepath, "rb") as f:
            response = requests.post(
                f"{self.cuckoo_url}/tasks/create/file",
                files={"file": f},
                data={
                    "timeout": timeout,
                    "options": "procmemdump=yes,route=none",
                    "priority": 2,
                    "machine": "win10_x64"
                }
            )
        task_id = response.json()["task_id"]
        return task_id

    def wait_for_analysis(self, task_id, poll_interval=30, max_wait=600):
        """Attendre la fin de l'analyse sandbox"""
        import time
        elapsed = 0
        while elapsed < max_wait:
            response = requests.get(f"{self.cuckoo_url}/tasks/view/{task_id}")
            status = response.json()["task"]["status"]
            if status == "reported":
                return self.get_report(task_id)
            elif status == "failed_analysis":
                return {"error": "Échec de l'analyse"}
            time.sleep(poll_interval)
            elapsed += poll_interval
        return {"error": "L'analyse a expiré"}

    def get_report(self, task_id):
        """Récupérer le rapport d'analyse"""
        response = requests.get(f"{self.cuckoo_url}/tasks/report/{task_id}")
        report = response.json()

        # Extraire les indicateurs clés
        return {
            "task_id": task_id,
            "score": report.get("info", {}).get("score", 0),
            "signatures": [
                {"name": s["name"], "severity": s["severity"], "description": s["description"]}
                for s in report.get("signatures", [])
            ],
            "network": {
                "dns": [d["request"] for d in report.get("network", {}).get("dns", [])],
                "http": [
                    {"url": h["uri"], "method": h["method"]}
                    for h in report.get("network", {}).get("http", [])
                ],
                "hosts": report.get("network", {}).get("hosts", [])
            },
            "dropped_files": [
                {"name": f["name"], "sha256": f["sha256"], "size": f["size"]}
                for f in report.get("dropped", [])
            ],
            "processes": [
                {"name": p["process_name"], "pid": p["pid"], "command_line": p.get("command_line", "")}
                for p in report.get("behavior", {}).get("processes", [])
            ],
            "registry_keys": [
                k for k in report.get("behavior", {}).get("summary", {}).get("regkey_written", [])
            ]
        }

    def submit_to_joesandbox(self, filepath, joe_api_key, joe_url="https://jbxcloud.joesecurity.org/api"):
        """Soumettre à Joe Sandbox Cloud"""
        with open(filepath, "rb") as f:
            response = requests.post(
                f"{joe_url}/v2/submission/new",
                headers={"Authorization": f"Bearer {joe_api_key}"},
                files={"sample": f},
                data={
                    "systems": "w10_64",
                    "internet-access": False,
                    "report-cache": True
                }
            )
        return response.json()["data"]["webid"]

Étape 4 : Extraire les IOCs et Générer le Verdict

class VerdictGenerator:
    def __init__(self):
        self.malicious_threshold = 7  # Seuil de score Cuckoo

    def generate_verdict(self, pre_screen, sandbox_report):
        """Combiner les résultats du pré-tri et du sandbox pour le verdict final"""
        iocs = {
            "ips": [],
            "domains": [],
            "urls": [],
            "hashes": [],
            "registry_keys": [],
            "files_dropped": []
        }

        # Extraire les IOCs du rapport sandbox
        if sandbox_report:
            iocs["domains"] = sandbox_report.get("network", {}).get("dns", [])
            iocs["ips"] = sandbox_report.get("network", {}).get("hosts", [])
            iocs["urls"] = [
                h["url"] for h in sandbox_report.get("network", {}).get("http", [])
            ]
            iocs["hashes"] = [
                f["sha256"] for f in sandbox_report.get("dropped_files", [])
            ]
            iocs["registry_keys"] = sandbox_report.get("registry_keys", [])[:10]
            iocs["files_dropped"] = sandbox_report.get("dropped_files", [])

        # Déterminer le verdict
        vt_malicious = pre_screen.get("virustotal", {}).get("malicious", 0)
        sandbox_score = sandbox_report.get("score", 0) if sandbox_report else 0
        sig_count = len(sandbox_report.get("signatures", [])) if sandbox_report else 0

        combined_score = (vt_malicious * 2) + (sandbox_score * 10) + (sig_count * 5)

        if combined_score >= 100:
            verdict = "MALVEILLANT"
            confidence = "ÉLEVÉE"
        elif combined_score >= 50:
            verdict = "SUSPECT"
            confidence = "MOYENNE"
        elif combined_score >= 20:
            verdict = "POTENTIELLEMENT_INDÉSIRABLE"
            confidence = "FAIBLE"
        else:
            verdict = "SAIN"
            confidence = "ÉLEVÉE"

        return {
            "verdict": verdict,
            "confidence": confidence,
            "combined_score": combined_score,
            "iocs": iocs,
            "vt_detections": vt_malicious,
            "sandbox_score": sandbox_score,
            "signatures": sandbox_report.get("signatures", []) if sandbox_report else []
        }

Étape 5 : Pousser les Résultats vers le SIEM

def push_to_splunk(verdict_result, splunk_url, splunk_token):
    """Envoyer le verdict d'analyse de malware à Splunk HEC"""
    import json

    event = {
        "sourcetype": "malware_analysis",
        "source": "malware_pipeline",
        "event": {
            "sha256": verdict_result["sha256"],
            "verdict": verdict_result["verdict"],
            "confidence": verdict_result["confidence"],
            "score": verdict_result["combined_score"],
            "vt_detections": verdict_result["vt_detections"],
            "sandbox_score": verdict_result["sandbox_score"],
            "malware_family": verdict_result.get("threat_label", "Inconnu"),
            "iocs": verdict_result["iocs"],
            "signatures": [s["name"] for s in verdict_result["signatures"]]
        }
    }

    response = requests.post(
        f"{splunk_url}/services/collector/event",
        headers={
            "Authorization": f"Splunk {splunk_token}",
            "Content-Type": "application/json"
        },
        json=event,
        verify=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true",  # Définir SKIP_TLS_VERIFY=true pour les certificats auto-signés dans les environnements de laboratoire
    )
    return response.status_code == 200

def push_iocs_to_blocklist(iocs, firewall_api):
    """Pousser les IOCs extraits vers l'infrastructure de blocage"""
    for ip in iocs.get("ips", []):
        requests.post(
            f"{firewall_api}/block",
            json={"type": "ip", "value": ip, "action": "block", "source": "malware_pipeline"}
        )
    for domain in iocs.get("domains", []):
        requests.post(
            f"{firewall_api}/block",
            json={"type": "domain", "value": domain, "action": "sinkhole", "source": "malware_pipeline"}
        )

Étape 6 : Orchestrer le Pipeline Complet

def run_full_pipeline():
    """Exécuter le pipeline complet de collecte, pré-tri, sandbox et génération de verdict"""
    # Configuration
    EDR_API_URL = os.environ["EDR_API_URL"]
    EDR_API_TOKEN = os.environ["EDR_API_TOKEN"]
    VT_API_KEY = os.environ["VT_API_KEY"]
    SPLUNK_URL = os.environ["SPLUNK_URL"]
    SPLUNK_TOKEN = os.environ["SPLUNK_TOKEN"]

    # Initialiser les composants
    collector = MalwareCollector()
    prescreener = MalwarePreScreener(VT_API_KEY)
    submitter = SandboxSubmitter()
    verdict_gen = VerdictGenerator()

    # Collecter depuis les sources EDR et email
    all_files = []
    for file_info in collector.collect_from_edr(EDR_API_URL, EDR_API_TOKEN):
        all_files.append(file_info)
    for file_info in collector.collect_from_email_gateway("/var/quarantine/email"):
        all_files.append(file_info)

    # Traiter chaque fichier
    for file_info in all_files:
        try:
            # Étape 1 : Calculer les empreintes
            hashes = collector.compute_hashes(file_info["path"])
            sha256 = hashes["sha256"]

            # Étape 2 : Pré-trier
            prescreen_result = prescreener.pre_screen(sha256)

            if prescreen_result["needs_sandbox"]:
                # Étape 3 : Soumettre au sandbox
                task_id = submitter.submit_to_cuckoo(file_info["path"])
                sandbox_report = submitter.wait_for_analysis(task_id)

                # Étape 4 : Générer le verdict
                verdict = verdict_gen.generate_verdict(prescreen_result, sandbox_report)

                # Ajouter les métadonnées
                verdict["sha256"] = sha256
                verdict["source"] = file_info["source"]
                verdict["threat_label"] = prescreen_result.get("virustotal", {}).get("threat_label", "Inconnu")

                # Étape 5 : Pousser vers le SIEM
                push_to_splunk(verdict, SPLUNK_URL, SPLUNK_TOKEN)

                # Bloquer les IOCs si le verdict est malveillant
                if verdict["verdict"] == "MALVEILLANT":
                    push_iocs_to_blocklist(
                        verdict["iocs"],
                        os.environ.get("FIREWALL_API", "http://firewall.local/api")
                    )

        except Exception as e:
            print(f"Erreur lors du traitement de {file_info['sha256']}: {e}")
            continue

    # Nettoyer
    prescreener.close()

Considérations de Sécurité

• Always sandbox in isolation : Les sandbox pour malwares doivent être sur un réseau isolé sans accès à la production. Utilisez des VLANs dédiés ou des machines physiques déconnectées.
• Stockage des échantillons : Chiffrez les échantillons de malwares au repos. Utilisez des permissions restrictives (0600) sur les fichiers d'échantillons.
• Rotation des clés API : Faites pivoter régulièrement les clés API pour VirusTotal, MalwareBazaar et les sandbox.
• Protection des données : Anonymisez ou pseudonymisez les métadonnées (expéditeurs, objets) avant de les stocker dans le SIEM.
• Contrôle de version : Utilisez cette compétence comme base et versionnez-la dans un dépôt Git privé pour suivre les modifications et les audits.

Options de Sandbox Alternatives

Plateforme	Type	Méthode API	Cas d'usage
Cuckoo	Open-source	REST	Environnements locaux, entièrement personnalisable
Any.Run	Cloud	REST API	Analyse interactive, partage public/privé
Joe Sandbox	Cloud/On-prem	REST	Classification avancée, détection YARA, familles de malwares
VMRay	On-prem	REST	Analyse basée sur la rétro-ingénierie, faible taux de faux positifs