Was macht Aufbau einer IOCs-Anreicherungspipeline mit OpenCTI?
Übersicht
OpenCTI ist eine Open-Source-Plattform für das Management von Cyber-Bedrohungsinformationen, die auf STIX 2.1 als nativem Datenmodell basiert. Diese Fähigkeit beschreibt den Aufbau einer automatisierten Pipeline zur Anreicherung von Indikatoren (IOCs) unter Verwendung des Connector-Ökosystems von OpenCTI, um Indikatoren mit Kontext aus VirusTotal, Shodan, AbuseIPDB, GreyNoise und anderen Quellen anzureichern. Die Pipeline reichert neu erfasste Indikatoren automatisch an, korreliert sie mit bekannten Bedrohungsakteuren und Kampagnen und bewertet sie für die Priorisierung durch Analysten.
Einsatzbereiche
- Beim Bereitstellen oder Konfigurieren von Funktionen zum Aufbau einer IOCs-Anreicherungspipeline mit OpenCTI in Ihrer Umgebung
- Beim Einrichten von Sicherheitskontrollen gemäß Compliance-Anforderungen
- Beim Aufbau oder der Verbesserung der Sicherheitsarchitektur für diesen Bereich
- Bei der Durchführung von Sicherheitsbewertungen, die diese Implementierung erfordern
Voraussetzungen
- Docker und Docker Compose für die OpenCTI-Bereitstellung
- Python 3.9+ mit der Bibliothek
pycti - API-Schlüssel für Anreicherungsdienste: VirusTotal, Shodan, AbuseIPDB, GreyNoise
- Verständnis des STIX 2.1-Datenmodells und seiner Beziehungen
- ElasticSearch oder OpenSearch für das OpenCTI-Backend
- RabbitMQ oder Redis für die Connector-Kommunikation
Schlüsselkonzepte
OpenCTI-Architektur
OpenCTI verwendet ein GraphQL-API-Frontend, das auf ElasticSearch zur Speicherung und Redis/RabbitMQ für die Connector-Kommunikation basiert. Daten werden nativ als STIX 2.1-Objekte mit Beziehungen gespeichert. Connectors werden kategorisiert in: Externer Import (Feed-Erfassung), Interner Import (Dateiparsing), Interne Anreicherung (Kontexterweiterung) und Stream (Echtzeit-Export).
Modell des Anreicherungs-Connectors
Interne Anreicherungs-Connectors werden automatisch ausgelöst, wenn neue Observablen erstellt werden, oder manuell von Analysten. Jeder Connector empfängt STIX-Objekte, fragt externe Dienste ab und gibt STIX 2.1-Bundles zurück, die die ursprüngliche Observable mit zusätzlichem Kontext, Labels und Beziehungen anreichern.
Vertrauensbewertung
OpenCTI verwendet eine Vertrauensskala von 0-100 für Indikatoren. Anreicherungs-Connectors können Vertrauenswerte basierend auf externer Validierung aktualisieren: VirusTotal-Erkennungsraten, Shodan-Expositionsdaten, AbuseIPDB-Meldezahlen und GreyNoise-Klassifikationsergebnisse.
Arbeitsablauf
Schritt 1: OpenCTI mit Docker Compose bereitstellen
# docker-compose.yml (wichtige Dienste)
version: '3'
services:
opencti:
image: opencti/platform:6.4.4
environment:
- APP__PORT=8080
- [email protected]
- APP__ADMIN__PASSWORD=ChangeMeNow
- APP__ADMIN__TOKEN=your-admin-token-uuid
- ELASTICSEARCH__URL=http://elasticsearch:9200
- MINIO__ENDPOINT=minio
- RABBITMQ__HOSTNAME=rabbitmq
ports:
- "8080:8080"
depends_on:
- elasticsearch
- minio
- rabbitmq
- redis
connector-virustotal:
image: opencti/connector-virustotal:6.4.4
environment:
- OPENCTI_URL=http://opencti:8080
- OPENCTI_TOKEN=your-admin-token-uuid
- CONNECTOR_ID=connector-virustotal-id
- CONNECTOR_NAME=VirusTotal
- CONNECTOR_SCOPE=StixFile,Artifact,IPv4-Addr,Domain-Name,Url
- CONNECTOR_AUTO=true
- VIRUSTOTAL_TOKEN=your-vt-api-key
- VIRUSTOTAL_MAX_TLP=TLP:AMBER
connector-shodan:
image: opencti/connector-shodan:6.4.4
environment:
- OPENCTI_URL=http://opencti:8080
- OPENCTI_TOKEN=your-admin-token-uuid
- CONNECTOR_ID=connector-shodan-id
- CONNECTOR_NAME=Shodan
- CONNECTOR_SCOPE=IPv4-Addr
- CONNECTOR_AUTO=true
- SHODAN_TOKEN=your-shodan-api-key
- SHODAN_MAX_TLP=TLP:AMBER
connector-abuseipdb:
image: opencti/connector-abuseipdb:6.4.4
environment:
- OPENCTI_URL=http://opencti:8080
- OPENCTI_TOKEN=your-admin-token-uuid
- CONNECTOR_ID=connector-abuseipdb-id
- CONNECTOR_NAME=AbuseIPDB
- CONNECTOR_SCOPE=IPv4-Addr
- CONNECTOR_AUTO=true
- ABUSEIPDB_API_KEY=your-abuseipdb-key
Schritt 2: Benutzerdefinierten Anreicherungs-Connector erstellen
import os
from pycti import OpenCTIConnectorHelper, get_config_variable
from stix2 import (
Bundle, Indicator, Note, Relationship,
IPv4Address, DomainName
)
import requests
class CustomEnrichmentConnector:
def __init__(self):
config = {
"opencti": {
"url": os.environ.get("OPENCTI_URL"),
"token": os.environ.get("OPENCTI_TOKEN"),
},
"connector": {
"id": os.environ.get("CONNECTOR_ID"),
"name": "CustomEnrichment",
"scope": "IPv4-Addr,Domain-Name,Url",
"auto": True,
"type": "INTERNAL_ENRICHMENT",
},
}
self.helper = OpenCTIConnectorHelper(config)
self.helper.listen(self._process_message)
def _process_message(self, data):
entity_id = data["entity_id"]
stix_object = self.helper.api.stix_cyber_observable.read(id=entity_id)
if not stix_object:
return "Observable nicht gefunden"
observable_type = stix_object["entity_type"]
observable_value = stix_object.get("value", "")
enrichment_results = []
if observable_type == "IPv4-Addr":
enrichment_results = self._enrich_ip(observable_value, entity_id)
elif observable_type == "Domain-Name":
enrichment_results = self._enrich_domain(observable_value, entity_id)
if enrichment_results:
bundle = Bundle(objects=enrichment_results, allow_custom=True)
self.helper.send_stix2_bundle(bundle.serialize())
return "Anreicherung abgeschlossen"
def _enrich_ip(self, ip_address, entity_id):
"""IP-Adresse mit GreyNoise-, AbuseIPDB-Kontext anreichern."""
objects = []
# GreyNoise Community API
try:
gn_response = requests.get(
f"https://api.greynoise.io/v3/community/{ip_address}",
headers={"key": os.environ.get("GREYNOISE_API_KEY")},
timeout=30,
)
if gn_response.status_code == 200:
gn_data = gn_response.json()
classification = gn_data.get("classification", "unknown")
noise = gn_data.get("noise", False)
riot = gn_data.get("riot", False)
note_content = (
f"## GreyNoise-Anreicherung\n"
f"- Klassifikation: {classification}\n"
f"- Internet-Rauschen: {noise}\n"
f"- RIOT (Legitimer Dienst): {riot}\n"
f"- Name: {gn_data.get('name', 'N/A')}\n"
f"- Zuletzt gesehen: {gn_data.get('last_seen', 'N/A')}"
)
note = Note(
content=note_content,
object_refs=[entity_id],
abstract=f"GreyNoise: {classification}",
allow_custom=True,
)
objects.append(note)
# Labels basierend auf Klassifikation hinzufügen
if classification == "malicious":
self.helper.api.stix_cyber_observable.add_label(
id=entity_id, label_name="greynoise:schadhaft"
)
elif riot:
self.helper.api.stix_cyber_observable.add_label(
id=entity_id, label_name="greynoise:legitimer-dienst"
)
except Exception as e:
self.helper.log_error(f"GreyNoise-Anreicherung fehlgeschlagen: {e}")
return objects
def _enrich_domain(self, domain, entity_id):
"""Domain mit WHOIS- und DNS-Kontext anreichern."""
objects = []
try:
# SecurityTrails API für Domain-Anreicherung verwenden
st_response = requests.get(
f"https://api.securitytrails.com/v1/domain/{domain}",
headers={"APIKEY": os.environ.get("SECURITYTRAILS_API_KEY")},
timeout=30,
)
if st_response.status_code == 200:
st_data = st_response.json()
current_dns = st_data.get("current_dns", {})
a_records = [
r.get("ip") for r in current_dns.get("a", {}).get("values", [])
]
note_content = (
f"## SecurityTrails-Anreicherung\n"
f"- A-Einträge: {', '.join(a_records)}\n"
f"- Alexa-Rang: {st_data.get('alexa_rank', 'N/A')}\n"
f"- Hostname: {st_data.get('hostname', 'N/A')}"
)
note = Note(
content=note_content,
object_refs=[entity_id],
abstract=f"SecurityTrails: {domain}",
allow_custom=True,
)
objects.append(note)
except Exception as e:
self.helper.log_error(f"SecurityTrails-Anreicherung fehlgeschlagen: {e}")
return objects
if __name__ == "__main__":
connector = CustomEnrichmentConnector()