f3nris
086150b200
A tenant whose basic detection rules only notify — email, Slack — creates no
investigation and no alert any API can read: /investigations stays empty and the
Alert Triage API is MDR-gated. The rule's LEQL query is the only artefact left,
so replay it on a schedule and turn every matching line into an alert.
query_log and query_log_set gain:
- an ingest block (results_path events, dedup_key event_id, watermark start_time);
- event_id = log_id + the line's sequence number, the line's own address in Log
Search. It has to be stable across runs because the scheduler deliberately
re-reads the last minute of the previous window on every tick, and anything
derived from the fetch time would make that overlap a second incident. A hash
of (log_id, timestamp, message) covers a source that does not number entries;
- start_time read as ISO-8601, UNIX seconds or milliseconds alike. The API only
takes milliseconds, but which one the scheduler sends is an operator setting on
the rule — the feed must not fetch the wrong window because of a dropdown;
- a bounded page walk (limit, default 1000): unattended every few minutes, a
query matching more than intended would otherwise hold the run open long
enough for the next tick to find it in flight and skip itself;
- detection_name, stamped on every event. The log line does not know which rule
it is being replayed for, and without it an incident is titled by its own first
characters where the source console shows an alert name.
Plus mappers/query_log{,_set}.yaml: the parsed event lives in `message` as a
JSON-encoded string, so account, addresses, asset and process come out through
$parseJson. A raw unparsed line is not JSON — those entries resolve to nothing
and are skipped, and custom:rapid7_log_line still carries the line verbatim.
Severity is left unmapped: a log line has no priority of its own, so the alert
rule's default applies.
Known limits, by construction: this endpoint rejects calculations, so a rule with
a threshold (N occurrences in M minutes) cannot be reproduced — only pattern
rules are 1:1 — and an inactivity rule never can, since it fires on the absence
of events. The query also becomes a copy: editing the rule in InsightIDR leaves
the feed reading the old one.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UjtRtCzw6FEajDDHJ3mTQf
97 lines
8.1 KiB
YAML
97 lines
8.1 KiB
YAML
name: "Rapid7 InsightIDR LEQL events → OCSF"
|
|
description: |
|
|
Maps one Log Search event — a single log line returned by a LEQL query — to OCSF.
|
|
Serves both query_log and query_log_set (results_path = events on either): the event
|
|
objects are identical, only the endpoint differs. This copy is the one registered for
|
|
query_log — a single log rather than a log set.
|
|
|
|
This mapper exists for the tenants that cannot ingest anything else. A basic detection
|
|
rule whose action is notification only (email, Slack) writes nothing the platform can
|
|
read back: no investigation, and the Alert Triage API is MDR-gated. What does survive
|
|
is the rule's own LEQL query, so the feed replays it on a schedule and every matching
|
|
line becomes an alert. Rapid7 is then no longer the detector — the query is — which is
|
|
why the rule's name has to be carried in by hand (detection_name on the command) and
|
|
why a rule with a threshold or an inactivity condition cannot be reproduced this way.
|
|
|
|
An event is {timestamp, sequence_number, log_id, message} plus the event_id the script
|
|
builds for dedup. Everything an analyst actually needs is inside `message`, a
|
|
JSON-encoded string for every parsed log set, which is why nearly every entry below
|
|
goes through $parseJson. A raw, unparsed log line is not JSON: those entries resolve to
|
|
nothing and are skipped, and custom:rapid7_log_line still carries the line verbatim.
|
|
|
|
Severity is deliberately unmapped. A log line carries no priority of its own, so the
|
|
incident takes the alert rule's default — set it on the rule, per detection.
|
|
field_mappings:
|
|
# The rule name when the feed was told it, the line itself otherwise. A feed left
|
|
# without detection_name still produces readable incidents; it just reads like a log.
|
|
title: "jsonata: detection_name ? detection_name : ('Rapid7 Log Search — ' & $substring($string(message), 0, 80))"
|
|
source: "jsonata: 'Rapid7 InsightIDR Log Search'"
|
|
# Always populated, JSON or not — the one field an analyst can always fall back on.
|
|
"custom:rapid7_log_line": "message"
|
|
"custom:rapid7_log_id": "log_id"
|
|
# Key/value rows of the parsed event, for a table custom field. A single object
|
|
# becomes key/value pairs, which is exactly the shape wanted here.
|
|
"custom:rapid7_event": "jsonata: $toTable($parseJson(message))"
|
|
ocsf:
|
|
# ── Finding ───────────────────────────────────────────────────────
|
|
# event_id is log_id + the line's sequence number in Log Search: its address, stable
|
|
# across the one-minute overlap the scheduler replays on every run.
|
|
- { source_path: "event_id", ocsf_field: "finding_info.uid" }
|
|
- { source_path: "event_id", ocsf_field: "metadata.uid" }
|
|
- { source_path: "detection_name", ocsf_field: "finding_info.title" }
|
|
- { source_path: "detection_name", ocsf_field: "finding_info.analytic.name" }
|
|
- { source_path: "$fromMillis(timestamp)", ocsf_field: "finding_info.created_time" }
|
|
# ── Detection time: the MTTD anchor ───────────────────────────────
|
|
# The log's own timestamp, not the poll's. A feed that ran late must not make the
|
|
# detection look late with it.
|
|
- { source_path: "$fromMillis(timestamp)", ocsf_field: "time" }
|
|
# ── Product identity ──────────────────────────────────────────────
|
|
- { source_path: "'Rapid7 InsightIDR'", ocsf_field: "metadata.product.name" }
|
|
- { source_path: "'Rapid7'", ocsf_field: "metadata.product.vendor_name" }
|
|
- { source_path: "log_id", ocsf_field: "metadata.log_source" }
|
|
# ── The parsed event ──────────────────────────────────────────────
|
|
# Whole thing first, so a log set nobody anticipated still reaches the analyst and the
|
|
# AI analysis intact. The typed entries below then pull out what has to be indexed as
|
|
# an entity: a value under `detection.` is only ever matched by the IOC regex, so an
|
|
# account left there would never be indexed as a user.
|
|
- { source_path: "$parseJson(message)", ocsf_field: "detection" }
|
|
# ── The acting account ────────────────────────────────────────────
|
|
# Log sets are not consistent about the spelling; weakest first, strongest last.
|
|
- { source_path: "$parseJson(message).user", ocsf_field: "actor.user.name" }
|
|
- { source_path: "$parseJson(message).username", ocsf_field: "actor.user.name" }
|
|
- { source_path: "$parseJson(message).account", ocsf_field: "actor.user.name" }
|
|
- { source_path: "$parseJson(message).source_account", ocsf_field: "actor.user.name" }
|
|
# actor.user is who acted, user is who it was done TO. Keeping them apart is what
|
|
# lets an analyst read a lateral move in one line.
|
|
- { source_path: "$parseJson(message).destination_account", ocsf_field: "user.name" }
|
|
# ── The affected host ─────────────────────────────────────────────
|
|
# device.hostname and src_endpoint.hostname both index the host as an entity; the
|
|
# duplication is what the other mappers here do and what the pivot graph reads.
|
|
- { source_path: "$parseJson(message).hostname", ocsf_field: "device.hostname" }
|
|
- { source_path: "$parseJson(message).asset", ocsf_field: "device.hostname" }
|
|
- { source_path: "$parseJson(message).source_asset", ocsf_field: "device.hostname" }
|
|
- { source_path: "$parseJson(message).hostname", ocsf_field: "src_endpoint.hostname" }
|
|
- { source_path: "$parseJson(message).asset", ocsf_field: "src_endpoint.hostname" }
|
|
- { source_path: "$parseJson(message).source_asset", ocsf_field: "src_endpoint.hostname" }
|
|
- { source_path: "$parseJson(message).destination_asset", ocsf_field: "dst_endpoint.hostname" }
|
|
- { source_path: "$parseJson(message).destination_asset_address", ocsf_field: "dst_endpoint.hostname" }
|
|
# ── Network ───────────────────────────────────────────────────────
|
|
- { source_path: "$parseJson(message).source_ip", ocsf_field: "src_endpoint.ip" }
|
|
- { source_path: "$parseJson(message).source_address", ocsf_field: "src_endpoint.ip" }
|
|
- { source_path: "$parseJson(message).destination_ip", ocsf_field: "dst_endpoint.ip" }
|
|
- { source_path: "$parseJson(message).destination_address", ocsf_field: "dst_endpoint.ip" }
|
|
- { source_path: "$parseJson(message).destination_port", ocsf_field: "dst_endpoint.port" }
|
|
- { source_path: "$parseJson(message).geoip_country_name", ocsf_field: "src_endpoint.location.country" }
|
|
# ── The process, when the log set has one ─────────────────────────
|
|
- { source_path: "$parseJson(message).process.name", ocsf_field: "process.name" }
|
|
- { source_path: "$parseJson(message).process.name", ocsf_field: "process.file.name" }
|
|
- { source_path: "$parseJson(message).process.cmd_line", ocsf_field: "process.cmd_line" }
|
|
- { source_path: "$parseJson(message).process.exe_path", ocsf_field: "process.file.path" }
|
|
- { source_path: "$parseJson(message).process.pid", ocsf_field: "process.pid" }
|
|
- { source_path: "$parseJson(message).parent_process.name", ocsf_field: "actor.process.parent_process.name" }
|
|
- { source_path: "$parseJson(message).parent_process.cmd_line", ocsf_field: "actor.process.parent_process.cmd_line" }
|
|
- { source_path: "$parseJson(message).parent_process.exe_path", ocsf_field: "actor.process.parent_process.file.path" }
|
|
# ── What the source made of it ────────────────────────────────────
|
|
- { source_path: "$parseJson(message).service", ocsf_field: "activity_name" }
|
|
- { source_path: "$parseJson(message).result", ocsf_field: "status_detail" }
|