Sophos SIEM: verify HTTP Ingest and a parser with an example

This example connects a synthetic JSON source to HTTP Ingest, a Custom Parser for the Auth schema and a targeted search. It bridges the individual guides to HTTP Ingest and Custom Parsers. It reports only fictional lab logins; it performs no real user login and checks no credentials.

Prepare the lab and define success

The requirements are a test tenant with a Next-Gen SIEM subscription, an authorised Admin or Super Admin role, and an approved computer with Python 3 and HTTPS access to the assigned ingest endpoint. SIEM planning explains licensing and retention boundaries. Existing XDR access alone is insufficient for custom transport and Custom Parsers.

The lab contract is deliberately narrow: one JSON message per POST, the field format below and the fixed source avanet-lab-source. The HTTP help describes HTTPS POST with a Bearer token but does not establish a general response body, status code or payload envelope for every source. Check the endpoint actually supplied and its current requirements before the send test. If the specific deployment requires a different body structure, adapt it against the corresponding documentation first; this template is not a universal ingest API specification.

Before starting, record the tenant, owner, integration name, parser name and test time. The expected result comprises three separate pieces of evidence:

1. The approved lab source sent precisely the intended message.
2. The parser correctly maps the fictional source values to the Auth schema.
3. The message is searchable in the correct tenant with the matching time, source and test identifier.

HTTP success, a green connector or a syntactically valid script alone does not pass this acceptance. Neither a detection nor MDR handling of this lab event is expected.

Test format and complete parser template

The lab source uses this single JSON line:

{"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 is a documentation address. pilot-user is not a tenant user. The session_id identifies this fictional lab run, not a real login session. The later send-test script generates a current UTC time and new identifier; the fixed example time is only for the parser editor.

Under Security Operations > Integrations > Custom Parsers > Add Custom Parser, create a new parser named AVANET_LAB_JSON_AUTH. Select Standalone, source format JSON and destination schema Auth. Neither overwrite nor expand existing global parsers. The following complete template describes only the lab format shown:

!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

The Confirm String deliberately matches only this precise compact serialisation with fixed field order. Pretty-printed JSON or a different order will not match. For a real source, redesign this recognition pattern against representative original messages; a broad search merely for the word “logon” is not an equivalent substitute.

For this lab, the Auth fields used and their fixed enum values are mapped as follows:

These numbers are schema enums, not arbitrary severity values. The Go time-format template in DATETIME() retains the UTC designator or an ISO 8601 zone offset. Compare the raw value and normalised microsecond time in the editor. The template does not set tenantId$ or resourceId$ itself; do not substitute invented tenant or event identifiers for them.

Check in the parser editor without enabling

Paste the single example line into Sample Message. Open Parse Message, then Source Destination Mapping. Compare schema, user, address, time, outcome and extractor path with the table. Resolve any warning about another parser before activation.

Then add several individual messages as negative checks:

• Change result to failure: expect action=8 and action_result=20.
• Change product to other-lab: the message must not be assigned to this parser.
• Change result to unknown or remove user: the message must not normalise as a successful login; the narrow Confirm String should exclude it.
• Insert an invalid time: an error or missing correct timestamp stops approval. Do not accept ingestion time as apparent success.

Run Test checks the schema and Extractor Paths for all supplied samples. Sophos recommends varied samples and 12–15 test runs; check the parser path actually used during these runs. Parse Message provides the Parsed Source Message and Source Destination Mapping for individual inspection. A test does not save the parser; Save and Deploy are separate steps. Saving is not proof of a test. Initially leave a new parser disabled. Changes to an already active parser would take effect immediately on saving; this lab therefore uses a separate new parser. The general parser runbook explains activation and rollback.

Prepare a single HTTP send step

After approval in the test tenant, create a separate AVANET-LAB-HTTP integration under Security Operations > Integrations > Marketplace > Custom > HTTP Ingest > Set Up. Immediately capture the Integration Key and URL in the approved secret store; Sophos displays them only once. Do not switch an existing source yet.

The Python example below sends precisely one self-generated message. URL and key are entered without display, rather than stored as command-line arguments, shell variables or source-code literals. Normal TLS certificate verification remains enabled; HTTP redirects are not followed so the Bearer key cannot reach another destination. The script deliberately has no automatic retries because they could create duplicate events without confirmed behaviour.

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

Save the code in a local lab file without secret values and start it with Python 3. The URL must come directly from the approved lab integration; the code does not verify ownership of an arbitrary HTTPS host. Do not copy the key or URL into public console recordings or tickets. The script produces no customer, password or production-session data.

Even after an HTTP response, successful ingestion is not claimed yet. A timeout leaves it unclear whether the server has already accepted the message. Search using the recorded test identifier before sending again. For 401/403, clarify the assigned credentials and approval state without speculatively expanding permissions. For a 4xx response caused by body format, check the precise ingest requirements rather than trying arbitrary JSON envelopes.

Search for the raw message and normalised event

Check transport separately first. Under Data Lake Search > Query Editor, search for the printed test identifier within a narrow time window:

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

Replace the example value with the identifier actually printed. In Original Data, the product, source identifier, user, time and outcome must match the sent line. The generic event’s sensor_id may be determined by the ingestion path. An integration display name is therefore not automatically the correct search identifier. Use the value actually present if an additional sensor_id filter is needed.

After the editor tests pass, deliberately enable the new lab parser and send a new test message. Do not assume that existing raw data will be retrospectively normalised again. Search for the expected Auth event using the defined lab fields:

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

Replace the identifier here too. In the result, check Normalized Data: source_user_name=pilot-user, source_address=192.0.2.25, the source-matching time, action=1, auth_category=10 and action_result=10, or their enum names displayed in the UI. The extractor path must belong to the lab parser. Use these criteria to assess acceptance.

For the controlled failed attempt, change only result to failure in the generator, create a new test identifier and send precisely once again. The schema should now show action=8 and action_result=20. Successful normalisation for unknown or a different product identifier would instead be a reason to stop.

Isolate faults and clean up the lab

No raw message: Check send time and test identifier, transport response, correct tenant, time window and source. Do not treat the absence of search results alone as proof of a transport failure; generic raw search needs advanced parameters and access to original_data is time-limited.

Raw message present, no Auth result: Check parser state, exact serialisation, Confirm String, schema and parser path. Do not broaden production parsers simply to capture the lab event somehow.

Auth result with the wrong outcome or time: Compare source and mapping separately. Disable the parser and correct the saved script baseline; do not fill gaps with invented default values.

After acceptance, stop the custom sender and disable only the newly created lab parser using Disable. This ends its additional normalisation; generic data and previously stored events are not considered deleted. Remove the lab integration only after its key-replacement/revocation path is clear and removal is approved. Cleanup must not alter a shared integration or another source.

Further maintenance follows the SIEM workflow: document source format and parser version together, repeat acceptance after changes, and plan retention and alert handling separately.