Sophos SIEM: HTTP Ingest und Parser an einem Beispiel prüfen
Dieses Beispiel verbindet eine synthetische JSON-Quelle mit HTTP Ingest, einem Custom Parser für das Auth-Schema und einer gezielten Suche. Es zeigt damit den Weg zwischen den Einzelanleitungen zu HTTP Ingest und Custom Parsern. Das Beispiel meldet nur fiktive Labor-Anmeldungen; es führt keine echte Benutzeranmeldung aus und prüft keine Zugangsdaten.
Labor vorbereiten und Erfolg definieren
Benötigt werden ein Testtenant mit Next-Gen-SIEM-Subscription, eine autorisierte Admin- oder Super-Admin-Rolle und ein freigegebener Rechner mit Python 3 und HTTPS-Zugriff zum zugeteilten Ingest-Endpunkt. Die SIEM-Planung erklärt Lizenz- und Aufbewahrungsgrenzen. Ein vorhandener XDR-Zugang allein genügt für Custom-Transport und Custom Parser nicht.
Der Laborvertrag ist absichtlich eng: eine JSON-Meldung pro POST, das unten gezeigte Feldformat und die feste Quelle avanet-lab-source. Die HTTP-Hilfe beschreibt HTTPS-POST mit Bearer-Token, legt dort aber keinen allgemeinen Response-Body, Statuscode oder Payload-Umschlag für sämtliche Quellen fest. Vor dem Sendetest den tatsächlich bereitgestellten Endpunkt und dessen aktuelle Vorgaben prüfen. Falls die konkrete Bereitstellung einen anderen Body-Aufbau verlangt, diesen zuerst anhand der zugehörigen Dokumentation anpassen; die Vorlage ist keine universelle Ingest-API-Spezifikation.
Vor dem Start werden Tenant, verantwortliche Person, Integrationsname, Parsername und Testzeit dokumentiert. Das erwartete Ergebnis sind drei getrennte Nachweise:
- Die freigegebene Laborquelle hat genau die beabsichtigte Meldung gesendet.
- Der Parser ordnet die fiktiven Quellwerte dem Auth-Schema korrekt zu.
- Die Meldung ist im richtigen Tenant mit passender Zeit, Quelle und Testkennung auffindbar.
Ein HTTP-Erfolg, grüner Connector oder syntaktisch gültiges Skript allein besteht diese Abnahme nicht. Es wird weder eine Detection noch eine MDR-Bearbeitung dieses Laborereignisses erwartet.
Testformat und vollständige Parser-Vorlage
Die Laborquelle verwendet diese einzelne JSON-Zeile:
{"product":"avanet-siem-lab","schema_version":1,"event_type":"logon","sensor_id":"avanet-lab-source","user":"pilot-user","src_ip":"192.0.2.25","session_id":"avanet-lab-20261029T101500Z","time":"2026-10-29T10:15:00Z","result":"success"}
192.0.2.25 ist eine Dokumentationsadresse. pilot-user ist kein Tenant-Benutzer. Die session_id ist eine Kennung dieses fiktiven Labordurchlaufs, nicht die Sitzungs-ID einer realen Anmeldung. Beim späteren Sendetest erzeugt das Skript eine aktuelle UTC-Zeit und eine neue Kennung; die feste Beispielzeit dient nur dem Parsereditor.
Unter Security Operations > Integrations > Custom Parsers > Add Custom Parser einen neuen Parser namens AVANET_LAB_JSON_AUTH anlegen. Standalone, Quellformat JSON und Zielschema Auth auswählen. Vorhandene globale Parser weder überschreiben noch erweitern. Die folgende vollständige Vorlage beschreibt ausschliesslich das gezeigte Laborformat:
!SCHEMA=scwx.auth
!CONFIRMWITH=PATTERN
!CONFIRMSTRING=^\{"product":"avanet-siem-lab","schema_version":1,"event_type":"logon","sensor_id":"avanet-lab-source","user":"[^"]+","src_ip":"[^"]+","session_id":"[^"]+","time":"[^"]+","result":"(success|failure)"\}$
!TRIMALLOFF
!SAMPLE={"product":"avanet-siem-lab","schema_version":1,"event_type":"logon","sensor_id":"avanet-lab-source","user":"pilot-user","src_ip":"192.0.2.25","session_id":"avanet-lab-20261029T101500Z","time":"2026-10-29T10:15:00Z","result":"success"}
json = JSON(originalData$)
sensorType$ = "AVANET_SIEM_LAB"
sensorId$ = json["$.sensor_id"]
sourceUserName$ = json["$.user"]
sourceAddress$ = json["$.src_ip"]
sessionId$ = json["$.session_id"]
authSystem$ = json["$.product"]
eventTimeUsec$ = DATETIME(json["$.time"], "2006-01-02T15:04:05Z07:00")
authCategory$ = 10
action$ = IF json["$.result"] == "failure" THEN 8 ELSE 1
actionResult$ = IF json["$.result"] == "success" THEN 10 ELSE 20
Die Confirm String ist bewusst auf genau diese kompakte Serialisierung mit fester Feldreihenfolge begrenzt. Ein hübsch eingerücktes JSON oder eine andere Reihenfolge passt nicht. Für eine reale Quelle wird dieses Erkennungsmuster anhand repräsentativer Originalmeldungen neu entworfen; eine breite Suche nur nach dem Wort „logon“ wäre kein gleichwertiger Ersatz.
Die verwendeten Auth-Felder und ihre festen Enum-Werte werden für dieses Labor wie folgt zugeordnet:
| Rohwert | Parserfeld | Erwartete Normalisierung |
|---|---|---|
sensor_id | sensorId$ | sensor_id |
user | sourceUserName$ | source_user_name |
src_ip | sourceAddress$ | source_address |
session_id | sessionId$ | session_id |
UTC-time | eventTimeUsec$ | event_time_usec |
| Labor-Anmeldung | authCategory$ = 10 | ACCOUNT_LOGON |
success | action$ = 1, actionResult$ = 10 | LOGON, SUCCESS |
failure | action$ = 8, actionResult$ = 20 | FAILURE, FAILED |
Die Zahlen sind Schema-Enums, keine frei gewählten Schweregrade. Die Go-Zeitformatvorlage in DATETIME() erhält das UTC-Zeichen beziehungsweise einen ISO-8601-Zonenoffset. Rohwert und normalisierte Mikrosekundenzeit werden im Editor gegeneinander geprüft. Die Vorlage setzt weder tenantId$ noch resourceId$ selbst; diese Werte dürfen nicht durch frei erfundene Tenant- oder Ereigniskennungen ersetzt werden.
Ohne Einschalten im Parsereditor prüfen
Die einzelne Beispielzeile in Sample Message einfügen. Parse Message und danach Source Destination Mapping öffnen. Schema, Benutzer, Adresse, Zeit, Ergebnis und extractor path gegen die Tabelle vergleichen. Eine Warnung über einen anderen Parser muss vor der Aktivierung geklärt werden.
Danach mehrere Einzelmeldungen als Gegenproben hinterlegen:
resultauffailureändern: erwartet werdenaction=8undaction_result=20.productaufother-labändern: die Meldung darf diesem Parser nicht zugeordnet werden.resultaufunknownändern oderuserentfernen: die Meldung darf nicht als erfolgreiche Anmeldung normalisiert werden; die enge Confirm String soll sie ausschliessen.- Einen ungültigen Zeitwert einsetzen: ein Fehler oder fehlender korrekter Zeitwert stoppt die Freigabe. Nicht die Importzeit als scheinbaren Erfolg akzeptieren.
Run Test prüft Schema und Extractor Paths aller hinterlegten Samples. Sophos empfiehlt verschiedene Samples und 12–15 Testläufe; dabei den tatsächlich verwendeten Parserpfad kontrollieren. Parse Message liefert die Parsed Source Message und das Source Destination Mapping für die Einzelprüfung. Ein Test speichert den Parser nicht; Save und Deploy sind getrennte Schritte. Das Speichern ist kein Testnachweis. Einen neuen Parser zunächst ausgeschaltet lassen. Bei bereits aktiven Parsern würden Änderungen beim Speichern sofort wirken; deshalb verwendet dieses Labor einen eigenen neuen Parser. Das allgemeine Parser-Runbook erklärt Aktivierung und Rückweg.
Einen einzelnen HTTP-Sendeschritt vorbereiten
Nach Freigabe im Testtenant unter Security Operations > Integrations > Marketplace > Custom > HTTP Ingest > Set Up eine separate Integration AVANET-LAB-HTTP erstellen. Integration Key und URL sofort in den freigegebenen Secret-Store übernehmen; Sophos zeigt sie nur einmal. Noch keine bestehende Quelle umstellen.
Das folgende Python-Beispiel sendet genau eine selbst erzeugte Meldung. URL und Schlüssel werden verdeckt abgefragt, nicht als Kommandozeilenargument, Shell-Variable oder Bestandteil des Quelltexts gespeichert. Die normale TLS-Zertifikatsprüfung bleibt eingeschaltet; HTTP-Weiterleitungen werden nicht verfolgt, damit der Bearer-Schlüssel nicht an ein anderes Ziel gelangt. Das Skript enthält bewusst keine automatischen Wiederholungen, weil diese ohne bestätigtes Verhalten doppelte Ereignisse erzeugen könnten.
import datetime
import getpass
import json
import urllib.error
import urllib.parse
import urllib.request
import uuid
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None
url = getpass.getpass("Integration URL: ").strip()
key = getpass.getpass("Integration Key: ")
parsed = urllib.parse.urlsplit(url)
if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password or parsed.fragment:
raise SystemExit("Nur die freigegebene HTTPS-URL ohne eingebettete Zugangsdaten verwenden.")
now = datetime.datetime.now(datetime.timezone.utc)
time_text = now.strftime("%Y-%m-%dT%H:%M:%SZ")
run_id = "avanet-lab-" + now.strftime("%Y%m%dT%H%M%SZ") + "-" + uuid.uuid4().hex[:8]
event = {
"product": "avanet-siem-lab",
"schema_version": 1,
"event_type": "logon",
"sensor_id": "avanet-lab-source",
"user": "pilot-user",
"src_ip": "192.0.2.25",
"session_id": run_id,
"time": time_text,
"result": "success",
}
request = urllib.request.Request(
url,
data=json.dumps(event, separators=(",", ":")).encode("utf-8"),
headers={"Authorization": "Bearer " + key, "Content-Type": "application/json"},
method="POST",
)
print("Testkennung:", run_id, "UTC:", time_text)
try:
with urllib.request.build_opener(NoRedirect()).open(request, timeout=15) as response:
print("HTTP-Status:", response.status)
except urllib.error.HTTPError as error:
print("HTTP-Fehler:", error.code, "— nicht automatisch erneut senden.")
except urllib.error.URLError:
print("Verbindungs- oder TLS-Fehler — URL, Vertrauenskette und Netzwerkweg prüfen.")
Den Code in einer lokalen Labor-Datei ohne Secret-Werte speichern und mit Python 3 starten. Die URL muss exakt aus der freigegebenen Laborintegration stammen; der Code prüft keinen Besitznachweis für einen beliebigen HTTPS-Host. Schlüssel und URL nicht in öffentliche Konsolenaufzeichnungen oder Tickets übernehmen. Das Skript erzeugt keine Kunden-, Passwort- oder produktiven Sitzungsdaten.
Auch nach einer HTTP-Antwort wird noch kein erfolgreicher Ingest behauptet. Bei einem Timeout ist unklar, ob der Server die Meldung bereits angenommen hat. Vor einer erneuten Übertragung deshalb anhand der notierten Testkennung suchen. Bei 401/403 die zugeteilten Zugangsdaten und den Freigabestatus klären, ohne Rechte auf Verdacht zu erweitern. Bei 4xx wegen des Body-Formats die konkrete Ingest-Vorgabe prüfen, statt wahllos neue JSON-Umschläge zu probieren.
Rohmeldung und normalisiertes Ereignis suchen
Zuerst den Transport getrennt prüfen. Unter Data Lake Search > Query Editor in einem engen Zeitfenster nach der ausgegebenen Testkennung suchen:
FROM generic WHERE original_data CONTAINS 'avanet-lab-20261029T101500Z' earliest=-1h
Den Beispielwert durch die tatsächlich ausgegebene Kennung ersetzen. In Original Data müssen Produkt, Quellkennung, Benutzer, Zeit und Ergebnis der gesendeten Zeile entsprechen. sensor_id im generischen Ereignis kann vom Aufnahmeweg bestimmt sein. Ein Integrationsanzeigename ist deshalb nicht automatisch die richtige Suchkennung. Erst den tatsächlich vorhandenen Wert übernehmen, falls zusätzlich nach sensor_id gefiltert werden soll.
Nach bestandenen Editortests den neuen Laborparser bewusst aktivieren und eine neue Testmeldung senden. Die bereits aufgenommenen Rohdaten werden dadurch nicht als nachträglich neu normalisiert vorausgesetzt. Das erwartete Auth-Ereignis wird mit den definierten Laborfeldern gesucht:
FROM auth WHERE sensor_id = 'avanet-lab-source' AND session_id = 'avanet-lab-20261029T101500Z' earliest=-1h
Auch hier die Kennung ersetzen. Im Treffer die Normalized Data prüfen: source_user_name=pilot-user, source_address=192.0.2.25, zur Quelle passende Zeit, action=1, auth_category=10 und action_result=10 beziehungsweise deren im UI angezeigte Enum-Namen. Der extractor path muss zum Laborparser gehören. Diese Werte bilden die Soll-Abnahme.
Für den kontrollierten Fehlversuch im eigenen Generator nur result auf failure ändern, eine neue Testkennung erzeugen und erneut genau einmal senden. Das Schema soll nun action=8 und action_result=20 zeigen. Ein zusätzliches erfolgreiches Normalisierungsergebnis für unknown oder eine fremde Produktkennung wäre dagegen ein Abbruchgrund.
Fehler eingrenzen und Labor zurückbauen
Keine Rohmeldung: Sendezeit und Testkennung, Transportantwort, richtigen Tenant, Zeitfenster und Quelle prüfen. Keine Suchtreffer als alleinigen Beweis eines Transportausfalls behandeln; generische Rohsuche benötigt die erweiterten Parameter und der Zugriff auf original_data ist zeitlich begrenzt.
Rohmeldung vorhanden, kein Auth-Treffer: Parserstatus, exakte Serialisierung, Confirm String, Schema und Parserpfad prüfen. Keine Produktionsparser breiter fassen, um das Laborereignis irgendwie zu erfassen.
Auth-Treffer mit falschem Ergebnis oder Zeitwert: Quelle und Mapping einzeln vergleichen. Parser deaktivieren und den gesicherten Skriptstand korrigieren; nicht mit erfundenen Standardwerten auffüllen.
Nach der Abnahme den eigenen Sender beenden und nur den neu angelegten Laborparser über Disable ausschalten. Damit endet dessen zusätzliche Normalisierung; generische Daten und bereits gespeicherte Ereignisse werden nicht als gelöscht betrachtet. Die Laborintegration erst nach geklärtem Ersatz-/Widerrufsweg für ihren Schlüssel und nach Freigabe entfernen. Der Rückbau darf keine gemeinsam genutzte Integration oder andere Datenquelle verändern.
Die weitere Pflege folgt dem SIEM-Arbeitsablauf: Quellformat und Parserversion zusammen dokumentieren, nach Änderungen erneut abnehmen und Aufbewahrung sowie Alarmbearbeitung getrennt planen.