07 - Pipelines und Routen

Revisionshistorie

Inhaltsverzeichnis

1. Pipeline-Konzept in Cribl
2. Die 5 Pipelines im Detail
   1. pipeline_syslog_enrichment
   2. pipeline_apache_clf
   3. pipeline_docker_json
   4. pipeline_security_auth
   5. pipeline_generic_passthrough
3. Route-Tabelle
4. Design-Entscheidung: Security via S2S, Web/IoT via HEC
5. Testing von Pipelines

1. Pipeline-Konzept in Cribl

In Cribl Stream ist eine Pipeline eine geordnete Kette von Verarbeitungsschritten (Functions), die auf eingehende Events angewendet werden. Jede Function kann Events transformieren, anreichern, filtern oder verwerfen.

Kernkomponenten

Der Datenfluss ist: Quelle → Route (Filter) → Pipeline (Functions) → Destination

2. Die 5 Pipelines im Detail

2.1 pipeline_syslog_enrichment

Datei: configs/stream/pipelines/pipeline-syslog-enrichment.json
Zweck: Anreicherung und Parsing von Syslog-formatierten Events (z.B. Apache Error Logs, Systemd Journal)

Regex-Pattern:

/^(?<syslog_timestamp>\w{3}\s+\d{1,2}\s+\d{2}:\d{2}:\d{2})\s+
  (?<syslog_host>\S+)\s+
  (?<syslog_program>[^\[:\s]+)
  (?:\[(?<syslog_pid>\d+)\])?:?\s+
  (?<message>.*)/

Beispiel-Input:

Feb 10 14:23:01 nuc sshd[12345]: Connection from 192.168.1.100 port 52341

Extrahierte Felder:

2.2 pipeline_apache_clf

Datei: configs/stream/pipelines/pipeline-apache-clf.json
Zweck: Parsing des Apache Combined Log Format (CLF) mit vollstaendiger Feldextraktion

Beispiel-Input:

192.168.1.50 - admin [10/Feb/2026:14:30:15 +0100] "GET /dashboard HTTP/1.1" 200 4523 "https://portal.example.de/" "Mozilla/5.0"

Extrahierte Felder:

2.3 pipeline_docker_json

Datei: configs/stream/pipelines/pipeline-docker-json.json
Zweck: Parsing von Docker-JSON-Logs mit Container-ID-Extraktion und Log-Level-Erkennung

Beispiel-Input (Docker JSON):

{"log":"2026-02-10T14:30:15.123Z INFO Starting Home Assistant Core...\n","stream":"stdout","time":"2026-02-10T14:30:15.123456789Z"}

Extrahierte Felder:

2.4 pipeline_security_auth

Datei: configs/stream/pipelines/pipeline-security-auth.json
Zweck: Sicherheitsorientiertes Parsing von SSH-Authentifizierungs-Events mit MITRE ATT&CK-Tagging

MITRE ATT&CK-Anreicherung

Bei erkannten fehlgeschlagenen SSH-Logins werden folgende MITRE-Felder automatisch hinzugefuegt:

Beispiel-Input (fehlgeschlagener Login):

Feb 10 14:45:22 nuc sshd[9876]: Failed password for invalid user admin from 10.0.0.50 port 44312 ssh2

Resultierende Felder:

{
  "auth_host": "nuc",
  "auth_program": "sshd",
  "auth_pid": "9876",
  "failed_user": "admin",
  "src_ip": "10.0.0.50",
  "src_port": "44312",
  "auth_action": "failed_login",
  "security_tag": "authentication_failure",
  "mitre_technique_id": "T1110",
  "mitre_technique_name": "Brute Force",
  "mitre_tactic": "Credential Access",
  "cribl_pipeline": "pipeline_security_auth"
}

2.5 pipeline_generic_passthrough

Datei: configs/stream/pipelines/pipeline-generic-passthrough.json
Zweck: Minimale Verarbeitung – fuegt nur Herkunftsmetadaten hinzu, ohne das Event zu veraendern

Diese Pipeline wird fuer Quellen verwendet, die entweder bereits in einem gut strukturierten Format vorliegen (z.B. Samba-Logs) oder fuer die im PoC kein spezifisches Parsing definiert wurde. Sie dient als Auffang-Pipeline fuer die Default-Route.

3. Route-Tabelle

Die Routen sind in configs/stream/pipelines/routes.json definiert und werden in der angegebenen Reihenfolge (Prioritaet) ausgewertet. Die erste passende Route gewinnt ("final": true).

Filter-Logik: __inputId

Die Filter basieren auf dem internen Feld __inputId, das Cribl automatisch fuer jedes Event setzt. Das Format ist Quellentyp:Quellen-ID:Protokoll:

• journal_files:in_journal_local – Systemd-Journal-Quelle
• syslog:in_syslog:tcp – Syslog via TCP
• syslog:in_syslog:udp – Syslog via UDP

4. Design-Entscheidung: Security via S2S, Web/IoT via HEC

Die Zuordnung der Datenquellen zu den Destinations folgt einem bewussten Sicherheits- und Architekturkonzept:

Sicherheitsbegruendung

• Trennung der Datenstroeme: Security-Daten und Anwendungsdaten nutzen unterschiedliche Transportwege, was die Auswirkung eines Ausfalls oder einer Kompromittierung begrenzt (Blast Radius).
• Priorisierung: Bei einem HEC-Ausfall sind Security-Daten nicht betroffen (und umgekehrt).
• Compliance: Die Trennung entspricht dem Prinzip der Aufgabentrennung (Segregation of Duties) gemaess MaRisk AT 4.3.1.

5. Testing von Pipelines

5.1 Sample Data (Testdaten)

Fuer jede Pipeline sollten repraesentative Testdaten vorbereitet werden. Die Testdaten koennen in der Cribl-UI unter Pipelines → [Pipeline] → Preview eingefuegt werden.

Beispiel-Testdaten je Pipeline

# pipeline_syslog_enrichment
Feb 10 14:23:01 nuc sshd[12345]: Connection from 192.168.1.100 port 52341
Feb 10 14:23:05 nuc systemd[1]: Started Session 42 of user root.

# pipeline_apache_clf
192.168.1.50 - admin [10/Feb/2026:14:30:15 +0100] "GET /api/v1/status HTTP/1.1" 200 1234 "-" "curl/7.68.0"
10.0.0.1 - - [10/Feb/2026:14:30:16 +0100] "POST /login HTTP/1.1" 401 89 "https://app.example.de/login" "Mozilla/5.0"

# pipeline_docker_json
{"log":"2026-02-10T14:30:15.123Z INFO Starting service...\n","stream":"stdout","time":"2026-02-10T14:30:15.123456789Z"}

# pipeline_security_auth
Feb 10 14:45:22 nuc sshd[9876]: Failed password for invalid user admin from 10.0.0.50 port 44312 ssh2
Feb 10 14:45:30 nuc sshd[9877]: Accepted publickey for mpauli from 192.168.1.1 port 55123 ssh2

# pipeline_generic_passthrough
[2026/02/10 14:30:00, 0] smbd: server started

5.2 Preview-Funktion

Die Preview-Funktion in der Cribl-UI ermoeglicht es, Testdaten durch eine Pipeline zu schicken und das Ergebnis sofort zu sehen:

1. Navigieren zu Processing → Pipelines → [Pipeline auswaehlen]
2. Rechts auf Preview klicken
3. Testdaten einfuegen oder eine Datei hochladen
4. Die Ausgabe zeigt die extrahierten Felder und Transformationen
5. Jede Function kann einzeln aktiviert/deaktiviert werden, um Schritte zu isolieren

5.3 Verifikations-Checkliste

• Jede Pipeline hat mindestens 3 repraesentative Testdatensaetze
• Alle erwarteten Felder werden korrekt extrahiert (Preview pruefen)
• Timestamps werden korrekt normalisiert
• Numerische Felder (status, bytes) haben den korrekten Typ
• MITRE ATT&CK-Tags werden bei fehlgeschlagenen Logins korrekt gesetzt
• Die Metadaten-Felder (cribl_host, cribl_pipeline) sind in jedem Output vorhanden
• Die Route-Tabelle leitet jede Quelle an die korrekte Pipeline und Destination
• Die Default-Route faengt nicht zuvor gematchte Events auf
• End-to-End-Test: Event von Quelle bis Splunk-Index verifiziert