rapid7-insightidr 1.12.0: ingest LEQL query results as alerts

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

README.md

@@ -26,8 +26,12 @@ and walks the whole tree looking for files named exactly **`manifest.yaml`**.
 integrations/
 └── <integration-id>/
     ├── manifest.yaml          # required — the integration definition
-    └── scripts/               # optional — only for script-based commands
-        └── <command-id>.py
+    ├── scripts/               # optional — only for script-based commands
+    │   └── <command-id>.py
+    ├── mappers/               # optional — bundled mapper templates, one per command
+    │   └── <command-id>.yaml
+    └── incident-types/        # optional — bundled incident types
+        └── <anything>.yaml
 ```
 
 ---
@@ -132,6 +136,72 @@ Two rules the catalog applies consistently, because they are not obvious:
 
 ---
 
+## Bundled mappers (`mappers/<command-id>.yaml`)
+
+A mapper turns one raw element of a command's output into an incident: the
+fields Riposte shows and routes on, and an OCSF projection for entity
+indexing. Bundled mappers are registered as **read-only library templates**
+on install and **replaced on every update** — an operator's clones are never
+touched, which also means a clone never receives your fixes. Ship correct
+templates; let operators clone only to diverge on purpose.
+
+```yaml
+name: "My Product Investigations → OCSF"   # required — shown in the mapper library
+description: |
+  Free text. Say what one element looks like and what the mapper reads.
+
+# Incident fields. Target: title, severity, source, description, custom:<name>.
+field_mappings:
+  title: "title"                              # dotted path into the element
+  severity: "priority"                        # values run through toSeverity (CRITICAL→5 … LOW→2)
+  "custom:vendor_status": "status"            # lands in the declared custom field <name>
+
+# OCSF projection. source_path is full JSONata; ocsf_field is the destination.
+ocsf:
+  - { source_path: "rrn",          ocsf_field: "finding_info.uid" }
+  - { source_path: "created_time", ocsf_field: "finding_info.created_time" }
+```
+
+### Expressions: dotted paths and `jsonata:`
+
+A `field_mappings` expression is a dotted path (with an optional
+`cond ? then : else` ternary). Prefixed with **`jsonata:`**, it is evaluated
+by the full JSONata engine instead — the same one `ocsf.source_path` always
+uses — which can index arrays, filter, and reshape:
+
+```yaml
+field_mappings:
+  "custom:evidence":  "jsonata: $toTable(alerts.evidences.fields[])"
+  "custom:alert_name": "jsonata: $parseJson(alerts[0].evidences[0].data).source_json.AlertDisplayName"
+  severity:            "jsonata: priority = 'HIGH' ? 4 : 2"
+```
+
+Two Riposte extensions exist alongside the standard JSONata functions:
+
+| Function | What it does |
+|----------|--------------|
+| `$parseJson(str)` | Opens a JSON-encoded string — the original log line most SIEMs bury inside a JSON string field. |
+| `$toTable(x [, cols])` | Turns an array of objects into rows for a `table` custom field (a single object becomes key/value rows). `cols` keeps only the named columns. |
+
+Mind JSONata's singleton semantics: a path that finds **one** element unwraps
+it from its array. End the path with `[]` (`alerts.evidences.fields[]`) so a
+one-alert investigation still produces table rows instead of key/value pairs.
+
+**Convention: reshape in the mapper, not in the script.** A command script
+should return what the API gives, plus at most the flattening the API forces
+on it. Building display structures in Python (pre-joined rows, markdown
+summaries) welds them to the integration version — as expressions in the
+mapper they are visible, testable against the rule's captured samples
+(editor → Test), and an operator can adapt them without a package release.
+Pre-1.11 integrations that ship pre-built fields keep working; don't extend
+the pattern.
+
+An expression that fails costs its own field and never the alert — silently,
+by design. The mapper editor's **Test** button is where those errors surface,
+named by field, evaluated against the rule's last captured samples.
+
+---
+
 ## Adding this catalog to Riposte
 
 In Riposte → **Integrations → Marketplace → Add source**:
@@ -152,9 +222,11 @@ Then **Sync**. Discovered integrations appear in the marketplace, ready to insta
 
 1. Create `integrations/<id>/manifest.yaml` (one directory per integration).
 2. Add `scripts/<command-id>.py` only for script-based commands.
-3. Bump `version` (semver) on every change — Riposte tracks versions per source.
-4. Validate the YAML parses and `id`/`name` are set.
-5. Open a merge request.
+3. Add `mappers/<command-id>.yaml` for every ingestion command — reshape with
+   `jsonata:` expressions there rather than in the script (see above).
+4. Bump `version` (semver) on every change — Riposte tracks versions per source.
+5. Validate the YAML parses and `id`/`name` are set.
+6. Open a merge request.
 
 See [`templates/manifest.example.yaml`](templates/manifest.example.yaml) for a
 fully-commented starting point (that file is intentionally **not** ingested).

integrations/rapid7-insightidr/manifest.yaml

@@ -1,8 +1,8 @@
 id: rapid7_insightidr
 name: Rapid7 InsightIDR
-version: 1.11.1
+version: 1.12.0
 description: "Rapid7 InsightIDR (REST API) — investigation ingestion and full lifecycle (list/get/search/create/update/assign/set-status/bulk-close), investigation alerts and Rapid7 product alerts, custom threat indicators (add/replace), log management and LEQL log/log-set queries with downloads, and user directory search."
-changelog: "1.11.1 — Strip the whitespace out of the comma-separated filter params (statuses, priorities, sources, tags) before the URL. \"HIGH, CRITICAL\" as typed sent \" CRITICAL\" to the API, which rejects the whole call — so a priority filter written naturally fetched nothing. | 1.11.0 —Build one evidence row per alert evidence into custom:rapid7_evidence, for the table field: sortable and filterable where the Markdown summary only reads. | 1.10.0 — Render every detection key as Markdown into custom:rapid7_summary, so a layout can show what the OCSF catalogue has no field for. Map both ends of an account action, read source_json top-level scalars, drop placeholder values, and force evidences to stay an array. | 1.9.2 — Read the evidence data in both of its shapes. A parsed source states its event flat, with no source_data/custom_data wrapper, and the parser returned {} for all of them. | 1.9.1 — get_alert_process_trees: explain the 404, which the API returns with an empty body for three different situations. | 1.9.0 — Fetch each alert process tree and flatten it into ctx.process_tree, the shape the layout Process Tree block reads. Gated on the event type by default, since an authentication alert never has one. | 1.8.0 — Fetch each alert evidence at ingestion and parse it: the log line behind the alert lives in a JSON-encoded string no mapper can reach, and it is where the source and destination addresses, the account and the device actually are. | 1.7.0 — Map source/destination assets and accounts to src_endpoint, dst_endpoint, actor.user and user. A connection rule names both ends and neither is \"the host\", so an RDP-to-DC alert produced no entity at all and nothing a layout could show. | 1.6.0 — Surface region_failures: the Alert Triage API answers HTTP 200 with a partial result when it cannot reach a region, and that list was being dropped. Also report the regions seen in the RRNs against the configured one. | 1.5.0 — Flatten the whole of rule_keys_of_interest into the context. It is an open vocabulary — a process rule and an authentication rule share no keys — so the hardcoded list only covered rules already seen and left every other alert empty. | 1.4.1 — Read the alert RRN from `rrn` before `id` and drop anything not RRN-shaped: the investigation alert list also returns bare UUIDs, and one of those rejects the entire /alerts/ops/rrns batch (HTTP 400), losing every alert in the run. | 1.4.0 — Add get_investigations_full: the investigation with its alerts inlined, so it stays the unit that becomes an incident. Drop-in for list_investigations (results_path data, dedup rrn). | 1.3.0 — Add search_investigation_alerts: ingest only the alerts that belong to an investigation, with their full detection content, in 1+N+1 calls. Filtering search_alerts on investigation_rrn does not work — the term is accepted and silently matches nothing. | 1.2.1 — search_alerts: report a malformed terms_json with the input named and echoed, instead of json's bare \"Expecting value: line 1 column 1\". | 1.2.0 — Add the search_alerts OCSF mapper: rule_keys_of_interest carries the host, the process, its command line, its parent and the file hashes, so an alert-based feed indexes entities where the investigation feed indexed none. Validated against a live ABA alert. | 1.1.1 — list_alert_fields: default path_depth to 1 and expose the search filter. The V2 fields endpoint refuses an unbounded walk of the tree (\"root path with path depth greater than 1 is not supported for V2 without a search filter\"), so the bare call returned HTTP 400. | 1.1.0 — Add the Alert Triage API (/idr/at): search alerts, get an alert, its evidence, its actors and its process trees, plus the tenant's alert field catalogue. These carry the detection content the investigation endpoints do not — MITRE T-codes, matched rule keys, assets and accounts, command lines. MDR-gated. | 1.0.0 — Initial release: investigation ingestion (list_investigations) with an exhaustive OCSF mapper, 18 commands across investigations, alerts, custom threats, logs/LEQL search and users. API v1/v2 selectable per instance and per command."
+changelog: "1.12.0 — Make the LEQL queries ingestible: query_log and query_log_set gain an ingest block, a stable event_id for dedup, a bounded page walk and a detection_name stamp, and they now read start_time as ISO-8601, UNIX seconds or milliseconds alike. A tenant whose basic detection rules only notify (email, Slack) creates no investigation and no alert an API can read, so replaying the rule's own query on a schedule is the only way its detections reach a SOAR. | 1.11.1 — Strip the whitespace out of the comma-separated filter params (statuses, priorities, sources, tags) before the URL. \"HIGH, CRITICAL\" as typed sent \" CRITICAL\" to the API, which rejects the whole call — so a priority filter written naturally fetched nothing. | 1.11.0 —Build one evidence row per alert evidence into custom:rapid7_evidence, for the table field: sortable and filterable where the Markdown summary only reads. | 1.10.0 — Render every detection key as Markdown into custom:rapid7_summary, so a layout can show what the OCSF catalogue has no field for. Map both ends of an account action, read source_json top-level scalars, drop placeholder values, and force evidences to stay an array. | 1.9.2 — Read the evidence data in both of its shapes. A parsed source states its event flat, with no source_data/custom_data wrapper, and the parser returned {} for all of them. | 1.9.1 — get_alert_process_trees: explain the 404, which the API returns with an empty body for three different situations. | 1.9.0 — Fetch each alert process tree and flatten it into ctx.process_tree, the shape the layout Process Tree block reads. Gated on the event type by default, since an authentication alert never has one. | 1.8.0 — Fetch each alert evidence at ingestion and parse it: the log line behind the alert lives in a JSON-encoded string no mapper can reach, and it is where the source and destination addresses, the account and the device actually are. | 1.7.0 — Map source/destination assets and accounts to src_endpoint, dst_endpoint, actor.user and user. A connection rule names both ends and neither is \"the host\", so an RDP-to-DC alert produced no entity at all and nothing a layout could show. | 1.6.0 — Surface region_failures: the Alert Triage API answers HTTP 200 with a partial result when it cannot reach a region, and that list was being dropped. Also report the regions seen in the RRNs against the configured one. | 1.5.0 — Flatten the whole of rule_keys_of_interest into the context. It is an open vocabulary — a process rule and an authentication rule share no keys — so the hardcoded list only covered rules already seen and left every other alert empty. | 1.4.1 — Read the alert RRN from `rrn` before `id` and drop anything not RRN-shaped: the investigation alert list also returns bare UUIDs, and one of those rejects the entire /alerts/ops/rrns batch (HTTP 400), losing every alert in the run. | 1.4.0 — Add get_investigations_full: the investigation with its alerts inlined, so it stays the unit that becomes an incident. Drop-in for list_investigations (results_path data, dedup rrn). | 1.3.0 — Add search_investigation_alerts: ingest only the alerts that belong to an investigation, with their full detection content, in 1+N+1 calls. Filtering search_alerts on investigation_rrn does not work — the term is accepted and silently matches nothing. | 1.2.1 — search_alerts: report a malformed terms_json with the input named and echoed, instead of json's bare \"Expecting value: line 1 column 1\". | 1.2.0 — Add the search_alerts OCSF mapper: rule_keys_of_interest carries the host, the process, its command line, its parent and the file hashes, so an alert-based feed indexes entities where the investigation feed indexed none. Validated against a live ABA alert. | 1.1.1 — list_alert_fields: default path_depth to 1 and expose the search filter. The V2 fields endpoint refuses an unbounded walk of the tree (\"root path with path depth greater than 1 is not supported for V2 without a search filter\"), so the bare call returned HTTP 400. | 1.1.0 — Add the Alert Triage API (/idr/at): search alerts, get an alert, its evidence, its actors and its process trees, plus the tenant's alert field catalogue. These carry the detection content the investigation endpoints do not — MITRE T-codes, matched rule keys, assets and accounts, command lines. MDR-gated. | 1.0.0 — Initial release: investigation ingestion (list_investigations) with an exhaustive OCSF mapper, 18 commands across investigations, alerts, custom threats, logs/LEQL search and users. API v1/v2 selectable per instance and per command."
 category: siem
 
 # Per-instance configuration. The Insight platform region selects the API host
@@ -251,35 +251,47 @@ commands:
 
   - id: query_log
     name: rapid7-insight-idr-query-log
-    description: "Run a LEQL query against a single log, following pagination links to gather all events."
+    description: "Run a LEQL query against a single log, following pagination links to gather all events. Usable as an ingestion feed: point it at the LEQL of a basic detection rule and every matching line becomes an alert (results path = events, one incident per log entry)."
     risk: read
     inputs_schema:
       properties:
         log_id: { type: string, description: "Log key to query" }
-        query: { type: string, description: "Valid LEQL query (no calculations)" }
+        query: { type: string, description: "Valid LEQL query (no calculations: groupby/calculate are rejected by this endpoint)" }
         time_range: { type: string, description: "Relative time range (e.g. 1 week, 1 day). When set, start_time/end_time are not needed." }
-        start_time: { type: string, description: "Lower bound as a UNIX timestamp in milliseconds" }
-        end_time: { type: string, description: "Upper bound as a UNIX timestamp in milliseconds" }
+        start_time: { type: string, description: "Lower bound: UNIX milliseconds, UNIX seconds or ISO-8601 — all three are accepted. Incremental fetch watermark." }
+        end_time: { type: string, description: "Upper bound: UNIX milliseconds, UNIX seconds or ISO-8601" }
         logs_per_page: { type: string, description: "Maximum log entries per page (default 50)" }
         sequence_number: { type: string, description: "Earliest sequence number of a log entry to start from" }
+        limit: { type: number, description: "Maximum events to gather across pages (default 1000). Bounds an unattended run against a query that matches more than it was meant to." }
+        detection_name: { type: string, description: "Name of the detection rule this query stands for. Stamped on every event and used as the incident title — a log line does not know which rule it was replayed for." }
       required: [log_id, query]
     outputs_schema: { properties: {} }
+    ingest:
+      results_path: events
+      dedup_key: event_id
+      incremental_field: start_time
 
   - id: query_log_set
     name: rapid7-insight-idr-query-log-set
-    description: "Run a LEQL query against a log set, following pagination links to gather all events."
+    description: "Run a LEQL query against a log set, following pagination links to gather all events. Usable as an ingestion feed: point it at the LEQL of a basic detection rule and every matching line becomes an alert (results path = events, one incident per log entry)."
     risk: read
     inputs_schema:
       properties:
         log_set_id: { type: string, description: "Log set ID to query" }
-        query: { type: string, description: "Valid LEQL query (no calculations)" }
+        query: { type: string, description: "Valid LEQL query (no calculations: groupby/calculate are rejected by this endpoint)" }
         time_range: { type: string, description: "Relative time range (e.g. 1 week, 1 day). When set, start_time/end_time are not needed." }
-        start_time: { type: string, description: "Lower bound as a UNIX timestamp in milliseconds" }
-        end_time: { type: string, description: "Upper bound as a UNIX timestamp in milliseconds" }
+        start_time: { type: string, description: "Lower bound: UNIX milliseconds, UNIX seconds or ISO-8601 — all three are accepted. Incremental fetch watermark." }
+        end_time: { type: string, description: "Upper bound: UNIX milliseconds, UNIX seconds or ISO-8601" }
         logs_per_page: { type: string, description: "Maximum log entries per page (default 50)" }
         sequence_number: { type: string, description: "Earliest sequence number of a log entry to start from" }
+        limit: { type: number, description: "Maximum events to gather across pages (default 1000). Bounds an unattended run against a query that matches more than it was meant to." }
+        detection_name: { type: string, description: "Name of the detection rule this query stands for. Stamped on every event and used as the incident title — a log line does not know which rule it was replayed for." }
       required: [log_set_id, query]
     outputs_schema: { properties: {} }
+    ingest:
+      results_path: events
+      dedup_key: event_id
+      incremental_field: start_time
 
   - id: download_logs
     name: rapid7-insight-idr-download-logs

integrations/rapid7-insightidr/mappers/query_log.yaml

@@ -0,0 +1,96 @@
+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" }

integrations/rapid7-insightidr/mappers/query_log_set.yaml

@@ -0,0 +1,95 @@
+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 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" }

integrations/rapid7-insightidr/scripts/query_log.py

@@ -1,7 +1,11 @@
-import json, os, sys, time, urllib.request, urllib.parse, urllib.error
+import datetime, hashlib, json, os, sys, time, urllib.request, urllib.parse, urllib.error
 
 _UNITS = {"second": 1, "minute": 60, "hour": 3600, "day": 86400, "week": 604800, "month": 2592000, "year": 31536000}
 
+# Beyond this a single value can only be milliseconds: 1e11 seconds is the year 5138,
+# and 1e11 milliseconds was 1973. Nothing a feed can legitimately ask for sits between.
+_MS_FLOOR = 100000000000
+
 
 def _cfg():
     s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
@@ -30,6 +34,27 @@ def _rel_ms_range(text):
     return now_ms - num * _UNITS[unit] * 1000, now_ms
 
 
+def _to_ms(v):
+    """Epoch milliseconds, whatever the caller wrote.
+
+    As an ingestion feed this command receives start_time from the scheduler's
+    high-water mark, and the mark's format is an operator setting on the rule
+    (iso8601, unix, unix_ms). The API only takes milliseconds. Reading all three
+    here is what keeps the feed from silently fetching the wrong window — or
+    nothing at all — because of a dropdown three screens away.
+    """
+    if v in (None, ""):
+        return None
+    s = str(v).strip()
+    if s.isdigit():
+        n = int(s)
+        return n if n >= _MS_FLOOR else n * 1000
+    try:
+        return int(datetime.datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
+    except Exception:
+        return None
+
+
 def request(method, url, params=None):
     base, headers = _cfg()
     full = url if url.startswith("http") else base + url.lstrip("/")
@@ -43,26 +68,73 @@ def request(method, url, params=None):
         return json.loads(raw) if raw else {}
 
 
-def _collect(first):
-    # Follow the query's pagination links until no more pages remain.
+def _collect(first, limit):
+    """Walk the pagination links, stopping at `limit`.
+
+    The walk is bounded because this runs unattended every few minutes: a query
+    that matches more than it was meant to would otherwise hold the run open for
+    as many pages as the window contains, and the next tick would find the rule
+    still in flight and skip itself.
+    """
     events = []
     queue = [first]
-    while queue:
+    while queue and len(events) < limit:
         page = queue.pop(0)
         events.extend(page.get("events", []) or [])
         for link in page.get("links", []) or []:
             href = link.get("href")
             if href:
                 queue.append(request("GET", href))
-    return events
+    return events[:limit]
+
+
+def _identify(events, fallback_log_id, detection_name):
+    """Give every event a stable id, and the name of the rule it stands for.
+
+    dedup_key needs a field that is identical across runs for the same log line.
+    The scheduler deliberately re-reads the last minute of the previous window on
+    every run (so an event written while a poll was in flight is not lost), which
+    means the same line WILL come back — anything derived from the fetch time, or
+    from the position in the page, would make it a second incident every time.
+    log_id + sequence_number is the line's own address in Log Search; the hash is
+    the fallback for a source that does not number its entries.
+
+    detection_name is stamped here because the log line has no idea which basic
+    detection rule it is being replayed for. Without it an incident is titled by
+    its own first characters, and an analyst reads a log line where the source
+    console shows an alert name.
+    """
+    out = []
+    for e in events:
+        if not isinstance(e, dict):
+            continue
+        log_id = e.get("log_id") or fallback_log_id or ""
+        seq = e.get("sequence_number")
+        if seq in (None, ""):
+            digest = hashlib.sha1(
+                ("%s|%s|%s" % (log_id, e.get("timestamp"), e.get("message"))).encode("utf-8", "replace")
+            ).hexdigest()
+            e["event_id"] = "%s:%s" % (log_id, digest)
+        else:
+            e["event_id"] = "%s:%s" % (log_id, seq)
+        if not e.get("log_id"):
+            e["log_id"] = log_id
+        if detection_name:
+            e["detection_name"] = detection_name
+        out.append(e)
+    return out
 
 
 def run():
     inp = _inputs()
-    start_time = inp.get("start_time")
-    end_time = inp.get("end_time")
+    start_time = _to_ms(inp.get("start_time"))
+    end_time = _to_ms(inp.get("end_time"))
     if inp.get("time_range"):
         start_time, end_time = _rel_ms_range(inp.get("time_range"))
+    try:
+        limit = int(inp.get("limit") or 1000)
+    except (TypeError, ValueError):
+        limit = 1000
     params = {
         "query": inp.get("query"),
         "from": start_time,
@@ -72,7 +144,8 @@ def run():
     }
     log_id = inp.get("log_id", "")
     first = request("GET", "log_search/query/logs/" + urllib.parse.quote(log_id, safe=""), params=params)
-    print(json.dumps({"events": _collect(first)}))
+    events = _identify(_collect(first, limit), log_id, inp.get("detection_name"))
+    print(json.dumps({"events": events, "count": len(events), "from": start_time, "to": end_time}))
 
 
 try:

integrations/rapid7-insightidr/scripts/query_log_set.py

@@ -1,7 +1,11 @@
-import json, os, sys, time, urllib.request, urllib.parse, urllib.error
+import datetime, hashlib, json, os, sys, time, urllib.request, urllib.parse, urllib.error
 
 _UNITS = {"second": 1, "minute": 60, "hour": 3600, "day": 86400, "week": 604800, "month": 2592000, "year": 31536000}
 
+# Beyond this a single value can only be milliseconds: 1e11 seconds is the year 5138,
+# and 1e11 milliseconds was 1973. Nothing a feed can legitimately ask for sits between.
+_MS_FLOOR = 100000000000
+
 
 def _cfg():
     s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
@@ -30,6 +34,27 @@ def _rel_ms_range(text):
     return now_ms - num * _UNITS[unit] * 1000, now_ms
 
 
+def _to_ms(v):
+    """Epoch milliseconds, whatever the caller wrote.
+
+    As an ingestion feed this command receives start_time from the scheduler's
+    high-water mark, and the mark's format is an operator setting on the rule
+    (iso8601, unix, unix_ms). The API only takes milliseconds. Reading all three
+    here is what keeps the feed from silently fetching the wrong window — or
+    nothing at all — because of a dropdown three screens away.
+    """
+    if v in (None, ""):
+        return None
+    s = str(v).strip()
+    if s.isdigit():
+        n = int(s)
+        return n if n >= _MS_FLOOR else n * 1000
+    try:
+        return int(datetime.datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
+    except Exception:
+        return None
+
+
 def request(method, url, params=None):
     base, headers = _cfg()
     full = url if url.startswith("http") else base + url.lstrip("/")
@@ -43,25 +68,73 @@ def request(method, url, params=None):
         return json.loads(raw) if raw else {}
 
 
-def _collect(first):
+def _collect(first, limit):
+    """Walk the pagination links, stopping at `limit`.
+
+    The walk is bounded because this runs unattended every few minutes: a query
+    that matches more than it was meant to would otherwise hold the run open for
+    as many pages as the window contains, and the next tick would find the rule
+    still in flight and skip itself.
+    """
     events = []
     queue = [first]
-    while queue:
+    while queue and len(events) < limit:
         page = queue.pop(0)
         events.extend(page.get("events", []) or [])
         for link in page.get("links", []) or []:
             href = link.get("href")
             if href:
                 queue.append(request("GET", href))
-    return events
+    return events[:limit]
+
+
+def _identify(events, fallback_log_id, detection_name):
+    """Give every event a stable id, and the name of the rule it stands for.
+
+    dedup_key needs a field that is identical across runs for the same log line.
+    The scheduler deliberately re-reads the last minute of the previous window on
+    every run (so an event written while a poll was in flight is not lost), which
+    means the same line WILL come back — anything derived from the fetch time, or
+    from the position in the page, would make it a second incident every time.
+    log_id + sequence_number is the line's own address in Log Search; the hash is
+    the fallback for a source that does not number its entries.
+
+    detection_name is stamped here because the log line has no idea which basic
+    detection rule it is being replayed for. Without it an incident is titled by
+    its own first characters, and an analyst reads a log line where the source
+    console shows an alert name.
+    """
+    out = []
+    for e in events:
+        if not isinstance(e, dict):
+            continue
+        log_id = e.get("log_id") or fallback_log_id or ""
+        seq = e.get("sequence_number")
+        if seq in (None, ""):
+            digest = hashlib.sha1(
+                ("%s|%s|%s" % (log_id, e.get("timestamp"), e.get("message"))).encode("utf-8", "replace")
+            ).hexdigest()
+            e["event_id"] = "%s:%s" % (log_id, digest)
+        else:
+            e["event_id"] = "%s:%s" % (log_id, seq)
+        if not e.get("log_id"):
+            e["log_id"] = log_id
+        if detection_name:
+            e["detection_name"] = detection_name
+        out.append(e)
+    return out
 
 
 def run():
     inp = _inputs()
-    start_time = inp.get("start_time")
-    end_time = inp.get("end_time")
+    start_time = _to_ms(inp.get("start_time"))
+    end_time = _to_ms(inp.get("end_time"))
     if inp.get("time_range"):
         start_time, end_time = _rel_ms_range(inp.get("time_range"))
+    try:
+        limit = int(inp.get("limit") or 1000)
+    except (TypeError, ValueError):
+        limit = 1000
     params = {
         "query": inp.get("query"),
         "from": start_time,
@@ -71,7 +144,9 @@ def run():
     }
     log_set_id = inp.get("log_set_id", "")
     first = request("GET", "log_search/query/logsets/" + urllib.parse.quote(log_set_id, safe=""), params=params)
-    print(json.dumps({"events": _collect(first)}))
+    # No fallback log id: a log set spans several logs and every event names its own.
+    events = _identify(_collect(first, limit), "", inp.get("detection_name"))
+    print(json.dumps({"events": events, "count": len(events), "from": start_time, "to": end_time}))
 
 
 try: