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" }
  # ── Event time (OCSF `time`) ───────────────────────────────────────
  # Riposte no longer reads it to date detection (MTTD retired, migration 0256):
  # it is the alert's own record of when the source saw the event.
  # The log's own timestamp, not the poll's.
  - { 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" }