Aufbau einer Pipeline zum Defangen und Teilen von IOCs

Was macht Aufbau einer Pipeline zum Defangen und Teilen von IOCs?

Übersicht

Beim Defangen von IOCs werden potenziell schädliche Indikatoren (URLs, IP-Adressen, Domains, E-Mail-Adressen) so modifiziert, dass versehentliche Klicks oder Ausführungen verhindert werden, während die Lesbarkeit für Analyse und Weitergabe erhalten bleibt. Diese Fähigkeit beschreibt den Aufbau einer automatisierten Pipeline, die rohe IOCs aus mehreren Quellen aufnimmt, normalisiert und dedupliziert, sie für die sichere menschliche Nutzung entschärft (defangt), in das STIX 2.1-Format für die maschinelle Verarbeitung konvertiert und über TAXII-Server, MISP-Instanzen sowie E-Mail-Berichte verteilt.

Wann verwenden

• Wenn Sie Fähigkeiten zum Aufbau einer Pipeline zum Defangen und Teilen von IOCs in Ihrer Umgebung bereitstellen oder konfigurieren
• Wenn Sie Sicherheitskontrollen einrichten, die auf Compliance-Anforderungen abgestimmt sind
• Wenn Sie die Sicherheitsarchitektur für diesen Bereich aufbauen oder verbessern
• Wenn Sie Sicherheitsbewertungen durchführen, die diese Implementierung erfordern

Voraussetzungen

• Python 3.9+ mit den Bibliotheken defang, ioc-fanger, stix2, requests, validators
• MISP-Instanz oder TAXII-Server für die automatisierte Weitergabe
• Verständnis von IOC-Typen: IPv4/IPv6, Domains, URLs, E-Mail-Adressen, Dateihashes
• Vertrautheit mit STIX 2.1 Indicator Patterns und TLP-Markierungsdefinitionen
• Zugriff auf Threat-Intelligence-Feeds für die IOC-Aufnahme

Schlüsselkonzepte

Standards zum Defangen von IOCs

Beim Defangen werden aktive Protokoll- und Domainkomponenten ersetzt, um die Ausführung zu verhindern: http:// wird zu hxxp://, https:// wird zu hxxps://, Punkte in Domains/IPs werden zu [.], @ in E-Mails wird zu [@]. Dies ist entscheidend für die Weitergabe von IOCs in Berichten, E-Mails, Slack-Kanälen und Paste-Sites, wo automatische Verlinkung Netzwerkverbindungen zu schädlicher Infrastruktur auslösen könnte.

IOC-Normalisierung

Rohe IOCs aus verschiedenen Quellen liegen in inkonsistenten Formaten vor. Die Normalisierung umfasst die Umwandlung in Kleinbuchstaben, das Entfernen von abschließenden Schrägstrichen und Leerzeichen, das Extrahieren von Domains aus URLs, das Auflösen von URL-Kodierungen, die Validierung der Formatkorrektheit und die Deduplizierung über Quellen hinweg.

STIX 2.1 Indicator Patterns

STIX Patterns drücken IOCs in einem standardisierten Format aus: [ipv4-addr:value = '203.0.113.1'], [domain-name:value = 'bösartig.example.com'], [url:value = 'http://evil.com/payload'], [file:hashes.'SHA-256' = 'abc123...']. Jeder Indikator enthält valid_from, indicator_types, confidence und optionale TLP-Markierungen.

Arbeitsablauf

Schritt 1: IOC-Extraktion und -Normalisierung aufbauen

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

class IOCExtractor:
    """Extrahiert und normalisiert IOCs aus Text."""

    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):
        """Extrahiert alle IOC-Typen aus freiem Text."""
        # Zuerst bereits entschärfte Indikatoren wiederherstellen
        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)

        # Entferne Domains, die Teil von URLs sind
        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"[+] {total} eindeutige IOCs aus Text extrahiert")
        return {k: sorted(v) for k, v in iocs.items()}

    def _refang(self, text):
        """Wandelt entschärfte Indikatoren zurück in die aktive Form."""
        text = text.replace("hxxp://", "http://").replace("hxxps://", "https://")
        text = text.replace("[.]", ".").replace("[@]", "@")
        text = text.replace("[://]", "://").replace("(.)", ".")
        return text

    def _normalize(self, value, ioc_type):
        """Normalisiert einen IOC-Wert."""
        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):
        """Prüft, ob IOC in der Whitelist ist."""
        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 = """
Malware C2: hxxps://evil-domain[.]com/beacon
Drops payload from 192.168.1.100 and contacts 10[.]0[.]0[.]1
SHA256: 275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f
Phishing email from attacker[@]phishing-domain[.]com
"""
iocs = extractor.extract_from_text(sample_text)

Schritt 2: Defanging-Engine

class IOCDefanger:
    """Entschärft IOCs für die sichere Weitergabe in Berichten und Kommunikation."""

    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):
        """Entschärft alle IOCs in einem Dictionary."""
        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  # Hashes müssen nicht entschärft werden
        return defanged

    def generate_sharing_report(self, iocs, defanged, report_name="IOC Report"):
        """Erzeugt einen menschenlesbaren Bericht mit entschärften IOCs."""
        report = f"# {report_name}\n"
        report += f"Generiert: {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, "Malware Campaign IOCs")
print(report)

Schritt 3: In das STIX 2.1-Format konvertieren

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

class STIXConverter:
    """Konvertiert rohe IOCs in STIX 2.1 Indicator-Objekte."""

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

    def iocs_to_stix(self, iocs, tlp="green", confidence=75):
        """Konvertiert IOC-Dictionary in ein STIX 2.1 Bundle."""
        stix_objects = []
        marking = self.TLP_MAP.get(tlp, TLP_GREEN)

        for ip in iocs.get("ipv4", []):
            stix_objects.append(Indicator(
                name=f"Bösartige IP: {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"Bösartige Domain: {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"Bösartige URL: {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"Bösartiger Dateihash: {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"[+] STIX-Bundle mit {len(stix_objects)} Indikatoren erstellt")
        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))

Schritt 4: Über MISP und TAXII verteilen

import requests
import json

class IOCDistributor:
    """Verteilt IOCs über verschiedene Kanäle."""

    def push_to_misp(self, iocs, misp_url, misp_key, event_info):
        """Überträgt IOCs als neues Event an MISP."""
        headers = {
            "Authorization": misp_key,
            "Content-Type": "application/json",
            "Accept": "application/json",
        }
        event = {
            "Event": {
                "info": event_info,
                "distribution": "1",  # Nur diese Community
                "threat_level_id": "2",  # Mittel
                "analysis": "2",  # Abgeschlossen
                "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": "Network activity" if ioc_type in ("ipv4", "domain", "url") else "Payload delivery",
                        "to_ids": True,
                    })

        resp = requests.post(
            f"{misp_url}/events",
            headers=headers,
            json=event,
            verify=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true",  # Setzen Sie SKIP_TLS_VERIFY=true für selbstsignierte Zertifikate in Laborumgebungen
        )
        if resp.status_code == 200:
            event_id = resp.json().get("Event", {}).get("id", "")
            print(f"[+] MISP-Event erstellt: {event_id}")
            return event_id
        else:
            print(f"[-] MISP-Fehler: {resp.status_code} - {resp.text[:200]}")
            return None

    def push_to_taxii(self, stix_bundle, taxii_url, collection_id, username, password):
        """Überträgt STIX-Bundle an eine TAXII 2.1-Collection."""
        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 veröffentlicht, Status: {response.status}")
        return response

distributor = IOCDistributor()
distributor.push_to_misp(
    iocs,
    misp_url="https://misp.organization.com",
    misp_key="YOUR_MISP_API_KEY",
    event_info="Malware Campaign IOCs - 2025",
)

Validierungskriterien

• IOCs werden korrekt aus freiem Text extrahiert, inkl. Wiederherstellung entschärfter IOCs
• Defanging erzeugt sichere, nicht anklickbare Indikatoren
• STIX 2.1-Bundle enthält gültige Indikator- Patterns
• IOCs werden erfolgreich an MISP und TAXII verteilt
• Deduplizierung verhindert doppelte Indikatoren
• Whitelisting verhindert Fehlalarme bei bekannten sicheren Domains

Referenzen

• CISA: Cybersecurity Automation and Threat Intelligence Sharing
• Defang Python Library
• ioc-fanger
• STIX 2.1 Indicator Specification
• Grokipedia: Defanging
• Hunt.io: Best IOC Feeds