construction-d-un-pipeline-de-defanging-et-partage-d-ioc

Construction d'un pipeline de défanging et de partage d'IOC

Aperçu

Le défanging d'IOC modifie les indicateurs potentiellement malveillants (URLs, adresses IP, domaines, adresses email) pour éviter les clics ou exécutions accidentels, tout en préservant leur lisibilité pour l'analyse et le partage. Cette compétence couvre la construction d'un pipeline automatisé qui ingère des IOC brutes provenant de multiples sources, les normalise et les déduplique, applique un défanging pour une consommation humaine sécurisée, les convertit au format STIX 2.1 pour une consommation machine, et les distribue via des serveurs TAXII, des instances MISP et des rapports email.

Quand l'utiliser

• Lors du déploiement ou de la configuration de capacités de pipeline de défanging et de partage d'IOC dans votre environnement
• Lors de l'établissement de contrôles de sécurité alignés sur les exigences de conformité
• Lors de la construction ou de l'amélioration d'une architecture de sécurité pour ce domaine
• Lors de la réalisation d'évaluations de sécurité nécessitant cette implémentation

Prérequis

• Python 3.9+ avec les bibliothèques defang, ioc-fanger, stix2, requests, validators
• Instance MISP ou serveur TAXII pour le partage automatisé
• Compréhension des types d'IOC : IPv4/IPv6, domaines, URLs, adresses email, hachages de fichiers
• Familiarité avec les motifs d'indicateurs STIX 2.1 et les définitions de marquage TLP
• Accès à des flux de renseignement sur les menaces pour l'ingestion d'IOC

Concepts clés

Normes de défanging d'IOC

Le défanging remplace les composants actifs de protocole et de domaine pour empêcher toute exécution : http:// devient hxxp://, https:// devient hxxps://, les points dans les domaines/IP deviennent [.], @ dans les emails devient [@]. C'est essentiel pour partager des IOC dans des rapports, emails, canaux Slack et sites de partage où l'auto-liaison pourrait déclencher des connexions réseau vers une infrastructure malveillante.

Normalisation des IOC

Les IOC brutes provenant de différentes sources arrivent dans des formats inconsistants. La normalisation consiste à convertir en minuscules, supprimer les slashes et espaces de fin, extraire les domaines des URLs, résoudre l'encodage URL, valider la correction du format et dédupliquer entre les sources.

Motifs d'indicateurs STIX 2.1

Les motifs STIX expriment les IOC dans un format standardisé : [ipv4-addr:value = '203.0.113.1'], [domain-name:value = 'malicious.example.com'], [url:value = 'http://evil.com/payload'], [file:hashes.'SHA-256' = 'abc123...']. Chaque indicateur inclut valid_from, indicator_types, confidence et des marquages TLP optionnels.

Workflow

Étape 1 : Construire l'extraction et la normalisation des IOC

import re
import hashlib
from urllib.parse import urlparse, unquote
from datetime import datetime

class IOCExtractor:
    """Extraire et normaliser les IOC à partir de texte."""

    PATTERNS = {
        "ipv4": r'\b(?:(?:25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)\.){3}(?:25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)\b',
        "domain": r'\b(?:[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,}\b',
        "url": r'https?://[^\s<>"{}|\\^`\[\]]+',
        "email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
        "md5": r'\b[a-fA-F0-9]{32}\b',
        "sha1": r'\b[a-fA-F0-9]{40}\b',
        "sha256": r'\b[a-fA-F0-9]{64}\b',
    }

    WHITELIST_DOMAINS = {
        "google.com", "microsoft.com", "amazon.com", "github.com",
        "cloudflare.com", "akamai.com", "example.com",
    }

    def extract_from_text(self, text):
        """Extraire tous les types d'IOC d'un texte libre."""
        # Désactiver le défanging des indicateurs déjà défangés en premier
        text = self._refang(text)
        iocs = {"ipv4": set(), "domain": set(), "url": set(),
                "email": set(), "md5": set(), "sha1": set(), "sha256": set()}

        for ioc_type, pattern in self.PATTERNS.items():
            matches = re.findall(pattern, text)
            for match in matches:
                normalized = self._normalize(match, ioc_type)
                if normalized and not self._is_whitelisted(normalized, ioc_type):
                    iocs[ioc_type].add(normalized)

        # Supprimer les domaines qui font partie d'URLs
        url_domains = set()
        for url in iocs["url"]:
            parsed = urlparse(url)
            url_domains.add(parsed.netloc)
        iocs["domain"] -= url_domains

        total = sum(len(v) for v in iocs.values())
        print(f"[+] Extrait {total} IOC uniques du texte")
        return {k: sorted(v) for k, v in iocs.items()}

    def _refang(self, text):
        """Convertir les indicateurs défangés en forme active."""
        text = text.replace("hxxp://", "http://").replace("hxxps://", "https://")
        text = text.replace("[.]", ".").replace("[@]", "@")
        text = text.replace("[:]", ":").replace("[://]", "://").replace("(.)", ".")
        return text

    def _normalize(self, value, ioc_type):
        """Normaliser une valeur d'IOC."""
        value = value.strip().lower()
        if ioc_type == "url":
            value = unquote(value).rstrip("/")
        elif ioc_type == "domain":
            value = value.rstrip(".")
        return value

    def _is_whitelisted(self, value, ioc_type):
        """Vérifier si l'IOC est dans la liste blanche."""
        if ioc_type == "domain":
            return value in self.WHITELIST_DOMAINS
        if ioc_type == "url":
            parsed = urlparse(value)
            return parsed.netloc in self.WHITELIST_DOMAINS
        return False

extractor = IOCExtractor()
sample_text = """
C2 du malware : hxxps://evil-domain[.]com/beacon
Livre la charge utile depuis 192.168.1.100 et contacte 10[.]0[.]0[.]1
SHA256 : 275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f
Email de phishing depuis attacker[@]phishing-domain[.]com
"""
iocs = extractor.extract_from_text(sample_text)

Étape 2 : Moteur de défanging

class IOCDefanger:
    """Défanger les IOC pour un partage sécurisé dans les rapports et communications."""

    def defang_url(self, url):
        return url.replace("http://", "hxxp://").replace("https://", "hxxps://").replace(".", "[.]")

    def defang_domain(self, domain):
        return domain.replace(".", "[.]")

    def defang_ip(self, ip):
        return ip.replace(".", "[.]")

    def defang_email(self, email):
        return email.replace("@", "[@]").replace(".", "[.]")

    def defang_all(self, iocs):
        """Défanger toutes les IOC d'un dictionnaire."""
        defanged = {}
        for ioc_type, values in iocs.items():
            if ioc_type == "url":
                defanged[ioc_type] = [self.defang_url(v) for v in values]
            elif ioc_type == "domain":
                defanged[ioc_type] = [self.defang_domain(v) for v in values]
            elif ioc_type == "ipv4":
                defanged[ioc_type] = [self.defang_ip(v) for v in values]
            elif ioc_type == "email":
                defanged[ioc_type] = [self.defang_email(v) for v in values]
            else:
                defanged[ioc_type] = values  # Les hachages n'ont pas besoin de défanging
        return defanged

    def generate_sharing_report(self, iocs, defanged, report_name="Rapport IOC"):
        """Générer un rapport IOC défangé lisible par un humain."""
        report = f"# {report_name}\n"
        report += f"Généré le : {datetime.now().isoformat()}\n\n"

        for ioc_type in ["url", "domain", "ipv4", "email", "sha256", "sha1", "md5"]:
            values = defanged.get(ioc_type, [])
            if values:
                report += f"## {ioc_type.upper()} ({len(values)})\n"
                for v in values:
                    report += f"- `{v}`\n"
                report += "\n"

        return report

defanger = IOCDefanger()
defanged = defanger.defang_all(iocs)
report = defanger.generate_sharing_report(iocs, defanged, "IOC de campagne de malware")
print(report)

Étape 3 : Conversion au format STIX 2.1

from stix2 import Indicator, Bundle, TLP_WHITE, TLP_GREEN, TLP_AMBER
from datetime import datetime

class STIXConverter:
    """Convertir les IOC brutes en objets Indicator STIX 2.1."""

    TLP_MAP = {"white": TLP_WHITE, "green": TLP_GREEN, "amber": TLP_AMBER}

    def iocs_to_stix(self, iocs, tlp="green", confidence=75):
        """Convertir un dictionnaire d'IOC en bundle STIX 2.1."""
        stix_objects = []
        marking = self.TLP_MAP.get(tlp, TLP_GREEN)

        for ip in iocs.get("ipv4", []):
            stix_objects.append(Indicator(
                name=f"IP malveillante : {ip}",
                pattern=f"[ipv4-addr:value = '{ip}']",
                pattern_type="stix",
                valid_from=datetime.now(),
                indicator_types=["malicious-activity"],
                confidence=confidence,
                object_marking_refs=[marking],
            ))

        for domain in iocs.get("domain", []):
            stix_objects.append(Indicator(
                name=f"Domaine malveillant : {domain}",
                pattern=f"[domain-name:value = '{domain}']",
                pattern_type="stix",
                valid_from=datetime.now(),
                indicator_types=["malicious-activity"],
                confidence=confidence,
                object_marking_refs=[marking],
            ))

        for url in iocs.get("url", []):
            escaped = url.replace("'", "\\'")
            stix_objects.append(Indicator(
                name=f"URL malveillante : {url[:60]}",
                pattern=f"[url:value = '{escaped}']",
                pattern_type="stix",
                valid_from=datetime.now(),
                indicator_types=["malicious-activity"],
                confidence=confidence,
                object_marking_refs=[marking],
            ))

        for sha256 in iocs.get("sha256", []):
            stix_objects.append(Indicator(
                name=f"Hachage de fichier malveillant : {sha256[:16]}...",
                pattern=f"[file:hashes.'SHA-256' = '{sha256}']",
                pattern_type="stix",
                valid_from=datetime.now(),
                indicator_types=["malicious-activity"],
                confidence=confidence,
                object_marking_refs=[marking],
            ))

        bundle = Bundle(objects=stix_objects)
        print(f"[+] Bundle STIX créé avec {len(stix_objects)} indicateurs")
        return bundle

converter = STIXConverter()
stix_bundle = converter.iocs_to_stix(iocs, tlp="amber", confidence=80)
with open("iocs_stix_bundle.json", "w") as f:
    f.write(stix_bundle.serialize(pretty=True))

Étape 4 : Distribution via MISP et TAXII

import requests
import json

class IOCDistributor:
    """Distribuer les IOC via différents canaux."""

    def push_to_misp(self, iocs, misp_url, misp_key, event_info):
        """Pousser les IOC vers MISP sous forme d'un nouvel événement."""
        headers = {
            "Authorization": misp_key,
            "Content-Type": "application/json",
            "Accept": "application/json",
        }
        event = {
            "Event": {
                "info": event_info,
                "distribution": "1",  # Cette communauté uniquement
                "threat_level_id": "2",  # Moyen
                "analysis": "2",  # Terminé
                "Attribute": [],
            }
        }

        type_mapping = {
            "ipv4": "ip-dst",
            "domain": "domain",
            "url": "url",
            "email": "email-src",
            "md5": "md5",
            "sha1": "sha1",
            "sha256": "sha256",
        }

        for ioc_type, values in iocs.items():
            misp_type = type_mapping.get(ioc_type)
            if misp_type:
                for value in values:
                    event["Event"]["Attribute"].append({
                        "type": misp_type,
                        "value": value,
                        "category": "Activité réseau" if ioc_type in ("ipv4", "domain", "url") else "Livraison de charge utile",
                        "to_ids": True,
                    })

        resp = requests.post(
            f"{misp_url}/events",
            headers=headers,
            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
        )
        if resp.status_code == 200:
            event_id = resp.json().get("Event", {}).get("id", "")
            print(f"[+] Événement MISP créé : {event_id}")
            return event_id
        else:
            print(f"[-] Erreur MISP : {resp.status_code} - {resp.text[:200]}")
            return None

    def push_to_taxii(self, stix_bundle, taxii_url, collection_id, username, password):
        """Pousser un bundle STIX vers une collection TAXII 2.1."""
        from taxii2client.v21 import Collection
        collection = Collection(
            f"{taxii_url}/collections/{collection_id}/",
            user=username, password=password,
        )
        response = collection.add_objects(stix_bundle.serialize())
        print(f"[+] TAXII : Bundle publié, statut : {response.status}")
        return response

distributor = IOCDistributor()
distributor.push_to_misp(
    iocs,
    misp_url="https://misp.organization.com",
    misp_key="VOTRE_CLE_API_MISP",
    event_info="IOC de campagne de malware - 2025",
)

Critères de validation

• Les IOC sont extraites correctement d'un texte libre avec prise en charge du refanging
• Le défanging produit des indicateurs sécurisés, non cliquables
• Le bundle STIX 2.1 contient des motifs d'indicateurs valides
• Les IOC sont distribuées avec succès vers MISP et TAXII
• La déduplication empêche les indicateurs en double
• La liste blanche évite les faux positifs sur les domaines légitimes connus

Références

• CISA : Automatisation de la cybersécurité et partage de renseignements sur les menaces
• Bibliothèque Python Defang
• ioc-fanger
• Spécification des indicateurs STIX 2.1
• Grokipedia : Défanging
• Hunt.io : Meilleurs flux IOC