Sophos SIEM: probar HTTP Ingest y parser con un ejemplo

Este ejemplo conecta una fuente JSON sintética con HTTP Ingest, un Custom Parser para el esquema Auth y una búsqueda específica. Muestra así el recorrido entre las guías individuales de HTTP Ingest y Custom Parsers. El ejemplo solo comunica logins ficticios de laboratorio; no ejecuta ningún inicio de sesión real ni comprueba credenciales.

Preparar el laboratorio y definir el éxito

Se necesita un tenant de prueba con subscription Next-Gen SIEM, un rol Admin o Super Admin autorizado y un equipo aprobado con Python 3 y acceso HTTPS al endpoint Ingest asignado. La planificación de SIEM explica los límites de licencia y conservación. Un acceso XDR existente no basta por sí solo para transporte personalizado y Custom Parsers.

El formato acordado para el laboratorio es deliberadamente estricto: un mensaje JSON por POST, el formato de campos mostrado abajo y la fuente fija avanet-lab-source. La ayuda HTTP describe POST por HTTPS con token Bearer, pero no establece allí un body de respuesta, código de estado o envoltorio de payload generales para todas las fuentes. Antes de enviar, comprobar el endpoint realmente proporcionado y sus instrucciones vigentes. Si la implementación concreta exige otra estructura del body, adaptarla primero según su documentación; la plantilla no es una especificación universal de API Ingest.

Antes de empezar se documentan tenant, responsable, nombre de integración, nombre del parser y hora de prueba. El resultado esperado son tres evidencias separadas:

1. La fuente de laboratorio autorizada ha enviado exactamente el mensaje previsto.
2. El parser asigna correctamente los valores ficticios de origen al esquema Auth.
3. El mensaje puede encontrarse en el tenant correcto con hora, fuente e identificador de prueba adecuados.

Un éxito HTTP, un conector verde o un script sintácticamente válido no bastan para superar esta validación. No se espera una Detection ni tratamiento MDR de este evento de laboratorio.

Formato de prueba y plantilla completa del parser

La fuente de laboratorio utiliza esta única línea 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 es una dirección de documentación. pilot-user no es un usuario del tenant. La session_id identifica esta ejecución ficticia de laboratorio, no una sesión real. En el envío posterior, el script genera la hora UTC actual y un identificador nuevo; la hora fija del ejemplo solo sirve para el editor del parser.

En Security Operations > Integrations > Custom Parsers > Add Custom Parser, crear un parser nuevo llamado AVANET_LAB_JSON_AUTH. Seleccionar Standalone, formato origen JSON y esquema destino Auth. No sobrescribir ni ampliar parsers globales existentes. La siguiente plantilla completa describe exclusivamente el formato de laboratorio mostrado:

!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

La Confirm String está limitada deliberadamente a esta serialización compacta exacta con orden fijo de campos. Un JSON formateado con sangría u otro orden no coincide. Para una fuente real se diseña de nuevo el patrón de reconocimiento a partir de mensajes originales representativos; buscar de forma amplia únicamente la palabra «logon» no es un sustituto equivalente.

En este laboratorio, los campos Auth utilizados y sus valores enum fijos se asignan de la siguiente manera:

Los números son enums del esquema, no niveles de gravedad elegidos libremente. La plantilla temporal Go de DATETIME() conserva el indicador UTC o un offset de zona ISO 8601. En el editor se contrastan valor bruto y hora normalizada en microsegundos. La plantilla no establece por sí misma tenantId$ ni resourceId$; no se sustituyen por identificadores inventados de tenant o evento.

Comprobar en el editor sin activar

Pegar la línea de ejemplo en Sample Message. Abrir Parse Message y después Source Destination Mapping. Comparar esquema, usuario, dirección, hora, resultado y extractor path con la tabla. Una advertencia sobre otro parser debe aclararse antes de activar.

Después, guardar varios mensajes individuales como contrapruebas:

• Cambiar result a failure: se esperan action=8 y action_result=20.
• Cambiar product a other-lab: el mensaje no debe asignarse a este parser.
• Cambiar result a unknown o quitar user: el mensaje no debe normalizarse como login correcto; la Confirm String estricta debe excluirlo.
• Introducir una hora no válida: un error o la ausencia de un valor temporal correcto impiden la aprobación. No aceptar la hora de importación como un éxito aparente.

Run Test comprueba el esquema y los Extractor Paths de todos los samples introducidos. Sophos recomienda distintos samples y 12–15 ejecuciones de prueba; durante ellas, comprobar la ruta del parser realmente utilizada. Parse Message proporciona Parsed Source Message y Source Destination Mapping para la comprobación individual. Una prueba no guarda el parser; Save y Deploy son pasos separados. Guardar no demuestra una prueba. Mantener inicialmente desactivado el parser nuevo. En parsers ya activos, los cambios se aplicarían inmediatamente al guardar; por eso este laboratorio utiliza un parser nuevo propio. El runbook general de parsers explica activación y recuperación.

Preparar un único envío HTTP

Tras la autorización en el tenant de prueba, crear una integración separada AVANET-LAB-HTTP en Security Operations > Integrations > Marketplace > Custom > HTTP Ingest > Set Up. Guardar inmediatamente Integration Key y URL en el Secret Store aprobado; Sophos solo las muestra una vez. Todavía no cambiar ninguna fuente existente.

El ejemplo Python siguiente envía exactamente un mensaje generado por él mismo. URL y clave se solicitan sin mostrarlas, no se almacenan como argumentos de línea de comandos, variables de shell ni parte del código fuente. La comprobación normal del certificado TLS permanece activada; no se siguen redirecciones HTTP para impedir que la clave Bearer llegue a otro destino. El script no incluye reintentos automáticos deliberadamente, ya que podrían producir eventos duplicados sin un comportamiento confirmado.

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

Guardar el código en un archivo local de laboratorio sin secretos e iniciarlo con Python 3. La URL debe proceder exactamente de la integración de laboratorio autorizada; el código no comprueba la propiedad de un host HTTPS arbitrario. No copiar clave ni URL en grabaciones públicas de consola o tickets. El script no genera datos de clientes, contraseñas ni sesiones productivas.

Incluso tras una respuesta HTTP todavía no se afirma una ingesta correcta. Ante un timeout no está claro si el servidor ya aceptó el mensaje. Por tanto, antes de volver a enviarlo, buscar el identificador de prueba anotado. Ante 401/403, aclarar credenciales asignadas y estado de autorización sin ampliar permisos por suposición. Ante 4xx por el formato del body, revisar la especificación concreta de Ingest en lugar de probar envoltorios JSON al azar.

Buscar el mensaje bruto y el evento normalizado

Primero comprobar el transporte por separado. En Data Lake Search > Query Editor, buscar el identificador mostrado dentro de una ventana temporal estrecha:

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

Sustituir el valor del ejemplo por el identificador realmente mostrado. En Original Data, producto, identificador de fuente, usuario, hora y resultado deben corresponder a la línea enviada. El sensor_id del evento genérico puede venir determinado por la ruta de incorporación. Por ello, el nombre visible de una integración no es automáticamente el identificador de búsqueda correcto. Solo adoptar el valor realmente presente si se desea filtrar adicionalmente por sensor_id.

Tras superar las pruebas del editor, activar conscientemente el parser nuevo de laboratorio y enviar un mensaje de prueba nuevo. No se presupone que los datos brutos ya incorporados se normalicen retrospectivamente. El evento Auth esperado se busca mediante los campos de laboratorio definidos:

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

También aquí se sustituye el identificador. En el resultado, comprobar Normalized Data: source_user_name=pilot-user, source_address=192.0.2.25, hora correspondiente al origen, action=1, auth_category=10 y action_result=10, o los nombres de enum que muestre la interfaz. El extractor path debe pertenecer al parser de laboratorio. Para aceptar el piloto, deben cumplirse todos los criterios de esta lista.

Para la prueba controlada de fallo, cambiar únicamente result a failure en el generador propio, generar otro identificador y enviar exactamente una vez. El esquema debe mostrar ahora action=8 y action_result=20. Una normalización adicional correcta para unknown o un producto ajeno sería, en cambio, motivo para detenerse.

Acotar errores y retirar el laboratorio

Sin mensaje bruto: Comprobar hora de envío e identificador de prueba, respuesta del transporte, tenant correcto, ventana temporal y fuente. No interpretar la ausencia de resultados como prueba única de fallo del transporte; la búsqueda genérica bruta requiere parámetros avanzados y el acceso a original_data tiene un límite temporal.

Mensaje bruto presente, sin resultado Auth: Comprobar estado del parser, serialización exacta, Confirm String, esquema y ruta del parser. No ampliar parsers productivos para capturar de cualquier manera el evento de laboratorio.

Resultado Auth con resultado u hora incorrectos: Comparar origen y mapeo por separado. Desactivar el parser y corregir la versión guardada del script; no completar valores con valores predeterminados inventados.

Después de validar, detener el emisor propio y desactivar únicamente el parser nuevo de laboratorio mediante Disable. Así termina su normalización adicional; los datos genéricos y eventos ya guardados no se consideran eliminados. Retirar la integración de laboratorio únicamente después de aclarar cómo sustituir o revocar su clave y de recibir autorización. La retirada no debe modificar ninguna integración compartida ni otra fuente de datos.

El mantenimiento posterior sigue el procedimiento SIEM: documentar juntos formato de origen y versión del parser, volver a validar tras cambios y planificar por separado conservación y tratamiento de alertas.