Sophos SIEM: przykład kontroli HTTP Ingest i parsera

Ten przykład łączy syntetyczne źródło JSON z HTTP Ingest, Custom Parserem dla schematu Auth i ukierunkowanym wyszukiwaniem. Pokazuje więc drogę między osobnymi instrukcjami HTTP Ingest i Custom Parsers. Przykład zgłasza jedynie fikcyjne logowania laboratoryjne; nie wykonuje prawdziwego logowania użytkownika ani nie sprawdza danych dostępowych.

Przygotowanie laboratorium i zdefiniowanie powodzenia

Potrzebny jest tenant testowy z subskrypcją Next-Gen SIEM, autoryzowana rola Admin lub Super Admin oraz zatwierdzony komputer z Python 3 i dostępem HTTPS do przydzielonego endpointu Ingest. Planowanie SIEM wyjaśnia granice licencji i retencji. Sam dostęp XDR nie wystarcza do transportu niestandardowego i Custom Parsers.

Kontrakt formatu laboratoryjnego jest celowo wąski: jedna wiadomość JSON na POST, format pól pokazany poniżej i stałe źródło avanet-lab-source. Pomoc HTTP opisuje HTTPS POST z tokenem Bearer, ale nie definiuje tam wspólnego response body, kodu statusu ani otoczki payloadu dla wszystkich źródeł. Przed testem wysyłki należy sprawdzić faktycznie udostępniony endpoint i jego aktualne wymagania. Jeśli konkretne wdrożenie wymaga innej struktury body, najpierw dostosować ją na podstawie właściwej dokumentacji; szablon nie jest uniwersalną specyfikacją API Ingest.

Przed rozpoczęciem dokumentuje się tenant, osobę odpowiedzialną, nazwę integracji, nazwę parsera i czas testu. Oczekiwany wynik obejmuje trzy osobne dowody:

1. Zatwierdzone źródło laboratoryjne wysłało dokładnie zamierzoną wiadomość.
2. Parser prawidłowo mapuje fikcyjne wartości źródłowe na schemat Auth.
3. Wiadomość jest możliwa do znalezienia we właściwym tenancie z odpowiednim czasem, źródłem i identyfikatorem testu.

Sam sukces HTTP, zielony stan konektora lub poprawna składnia skryptu nie zaliczają odbioru. Nie oczekuje się ani wykrycia, ani obsługi tego zdarzenia laboratoryjnego przez MDR.

Format testowy i pełny szablon parsera

Źródło laboratoryjne używa następującego pojedynczego wiersza JSON:

{"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 jest adresem dokumentacyjnym. pilot-user nie jest użytkownikiem tenanta. session_id identyfikuje ten fikcyjny przebieg laboratoryjny, nie sesję prawdziwego logowania. W późniejszym teście wysyłki skrypt generuje aktualny czas UTC i nowy identyfikator; stały czas przykładowy służy wyłącznie edytorowi parsera.

W Security Operations > Integrations > Custom Parsers > Add Custom Parser utworzyć nowy parser AVANET_LAB_JSON_AUTH. Wybrać Standalone, format źródłowy JSON i schemat docelowy Auth. Nie nadpisywać ani nie rozszerzać istniejących parserów globalnych. Poniższy pełny szablon opisuje wyłącznie pokazany format laboratoryjny:

!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

Confirm String jest celowo ograniczona do dokładnie tej zwartej serializacji ze stałą kolejnością pól. JSON z wcięciami lub inna kolejność nie pasują. Dla rzeczywistego źródła należy ponownie opracować wzorzec rozpoznawania na podstawie reprezentatywnych oryginalnych wiadomości; szerokie wyszukiwanie samego słowa „logon” nie jest równoważnym zamiennikiem.

W tym laboratorium używane pola Auth i ich stałe wartości enum są mapowane następująco:

Liczby są wartościami enum schematu, nie dowolnie wybranymi poziomami ważności. Wzorzec czasu Go w DATETIME() zachowuje oznaczenie UTC albo offset strefy ISO 8601. W edytorze porównuje się wartość surową ze znormalizowanym czasem w mikrosekundach. Szablon sam nie ustawia tenantId$ ani resourceId$; tych wartości nie wolno zastępować wymyślonymi identyfikatorami tenanta lub zdarzenia.

Kontrola w edytorze parsera bez włączania

Wkleić pojedynczy wiersz przykładowy do Sample Message. Otworzyć Parse Message, a następnie Source Destination Mapping. Porównać schemat, użytkownika, adres, czas, wynik i extractor path z tabelą. Ostrzeżenie o innym parserze musi zostać wyjaśnione przed aktywacją.

Następnie dodać kilka pojedynczych wiadomości jako próby kontrolne:

• Zmienić result na failure: oczekiwane są action=8 i action_result=20.
• Zmienić product na other-lab: wiadomość nie może zostać przypisana do tego parsera.
• Zmienić result na unknown lub usunąć user: wiadomość nie może zostać znormalizowana jako udane logowanie; wąska Confirm String powinna ją wykluczyć.
• Wstawić nieprawidłowy czas: błąd lub brak prawidłowego czasu wstrzymuje zatwierdzenie. Nie uznawać czasu importu za pozorny sukces.

Run Test sprawdza schemat i Extractor Paths wszystkich zapisanych próbek. Sophos zaleca różne próbki i 12–15 przebiegów testowych; należy przy tym kontrolować faktycznie używaną ścieżkę parsera. Parse Message udostępnia Parsed Source Message i Source Destination Mapping do kontroli pojedynczej próbki. Test nie zapisuje parsera; Save i Deploy to osobne kroki. Zapisanie nie jest dowodem testu. Nowy parser pozostawić początkowo wyłączony. Zmiany w już aktywnych parserach działałyby natychmiast po zapisaniu; dlatego to laboratorium używa własnego nowego parsera. Ogólny runbook parserów wyjaśnia aktywację i drogę powrotu.

Przygotowanie pojedynczej wysyłki HTTP

Po zatwierdzeniu w tenancie testowym utworzyć osobną integrację AVANET-LAB-HTTP pod Security Operations > Integrations > Marketplace > Custom > HTTP Ingest > Set Up. Od razu zapisać Integration Key i URL w zatwierdzonym magazynie sekretów; Sophos pokazuje je tylko raz. Nie przełączać jeszcze istniejącego źródła.

Poniższy przykład Python wysyła dokładnie jedną samodzielnie wygenerowaną wiadomość. URL i klucz są pobierane z ukrytym echem, nie są zapisywane jako argumenty polecenia, zmienna powłoki ani część kodu źródłowego. Standardowa weryfikacja certyfikatu TLS pozostaje włączona; przekierowania HTTP nie są śledzone, aby klucz Bearer nie trafił do innego celu. Skrypt celowo nie ponawia automatycznie, ponieważ bez potwierdzonego zachowania mogłoby to tworzyć zduplikowane zdarzenia.

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.")

Zapisać kod w lokalnym pliku laboratoryjnym bez wartości sekretów i uruchomić Pythonem 3. URL musi pochodzić dokładnie z zatwierdzonej integracji laboratoryjnej; kod nie weryfikuje własności dowolnego hosta HTTPS. Nie umieszczać klucza i URL w publicznych nagraniach konsoli ani zgłoszeniach. Skrypt nie generuje danych klientów, haseł ani produkcyjnych sesji.

Nawet po odpowiedzi HTTP nie potwierdza się jeszcze udanego przyjęcia danych. Po timeout nie wiadomo, czy serwer przyjął już wiadomość. Przed kolejną transmisją należy więc szukać zapisanego identyfikatora testu. Przy 401/403 wyjaśnić przydzielone dane dostępowe i stan zatwierdzenia, nie rozszerzając uprawnień na próbę. Przy 4xx dotyczącym formatu body sprawdzić konkretną specyfikację Ingest, zamiast losowo próbować nowych otoczek JSON.

Wyszukanie wiadomości surowej i zdarzenia znormalizowanego

Najpierw osobno sprawdzić transport. W Data Lake Search > Query Editor wyszukać wyświetlony identyfikator w wąskim oknie czasowym:

FROM generic WHERE original_data CONTAINS 'avanet-lab-20261029T101500Z' earliest=-1h

Zastąpić przykładową wartość faktycznie wyświetlonym identyfikatorem. W Original Data produkt, identyfikator źródła, użytkownik, czas i wynik muszą odpowiadać wysłanemu wierszowi. sensor_id w zdarzeniu generic może wynikać ze ścieżki pobierania. Nazwa wyświetlana integracji nie jest więc automatycznie właściwym identyfikatorem wyszukiwania. Faktycznie istniejącą wartość przejąć dopiero wtedy, gdy ma być użyty dodatkowy filtr sensor_id.

Po zaliczonych testach edytora świadomie aktywować nowy parser laboratoryjny i wysłać nową wiadomość testową. Nie zakłada się przy tym ponownej normalizacji wstecz już pobranych danych surowych. Oczekiwane zdarzenie Auth wyszukuje się przy użyciu zdefiniowanych pól laboratoryjnych:

FROM auth WHERE sensor_id = 'avanet-lab-source' AND session_id = 'avanet-lab-20261029T101500Z' earliest=-1h

Także tutaj zastąpić identyfikator. W trafieniu sprawdzić Normalized Data: source_user_name=pilot-user, source_address=192.0.2.25, czas zgodny ze źródłem, action=1, auth_category=10 i action_result=10 albo ich nazwy enum wyświetlane w interfejsie. extractor path musi należeć do parsera laboratoryjnego. Ta lista określa kryteria odbioru.

Dla kontrolowanej próby niepowodzenia we własnym generatorze zmienić tylko result na failure, wygenerować nowy identyfikator i znów wysłać dokładnie raz. Schemat powinien teraz pokazywać action=8 i action_result=20. Dodatkowy udany wynik normalizacji dla unknown lub obcego identyfikatora produktu byłby natomiast powodem przerwania.

Lokalizacja błędów i wycofanie laboratorium

Brak wiadomości surowej: Sprawdzić czas wysyłki i identyfikator, odpowiedź transportu, właściwy tenant, okno czasowe i źródło. Brak trafień nie jest samodzielnym dowodem awarii transportu; wyszukiwanie danych surowych generic wymaga zaawansowanych parametrów, a dostęp do original_data jest ograniczony czasowo.

Wiadomość surowa istnieje, brak trafienia Auth: Sprawdzić stan parsera, dokładną serializację, Confirm String, schemat i ścieżkę parsera. Nie rozszerzać parserów produkcyjnych tylko po to, aby jakoś przechwycić zdarzenie laboratoryjne.

Trafienie Auth z błędnym wynikiem lub czasem: Osobno porównać źródło i mapowanie. Wyłączyć parser i poprawić zachowaną wersję skryptu; nie uzupełniać wymyślonymi wartościami domyślnymi.

Po odbiorze zakończyć własny nadajnik i wyłączyć przez Disable tylko nowo utworzony parser laboratoryjny. Kończy to jego dodatkową normalizację; dane generic i zapisane wcześniej zdarzenia nie są traktowane jako usunięte. Integrację laboratoryjną usunąć dopiero po wyjaśnieniu ścieżki wymiany/unieważnienia jej klucza i po zatwierdzeniu. Wycofanie nie może zmienić współdzielonej integracji ani innego źródła danych.

Dalsza pielęgnacja odbywa się według procedury SIEM: dokumentować wspólnie format źródła i wersję parsera, po zmianach powtórzyć odbiór oraz osobno planować retencję i obsługę alarmów.