diff --git a/integrations/rapid7-insightidr/manifest.yaml b/integrations/rapid7-insightidr/manifest.yaml index 77b821f..98473a1 100644 --- a/integrations/rapid7-insightidr/manifest.yaml +++ b/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 diff --git a/integrations/rapid7-insightidr/mappers/query_log.yaml b/integrations/rapid7-insightidr/mappers/query_log.yaml new file mode 100644 index 0000000..7ec652e --- /dev/null +++ b/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" } diff --git a/integrations/rapid7-insightidr/mappers/query_log_set.yaml b/integrations/rapid7-insightidr/mappers/query_log_set.yaml new file mode 100644 index 0000000..08a4301 --- /dev/null +++ b/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" } diff --git a/integrations/rapid7-insightidr/scripts/query_log.py b/integrations/rapid7-insightidr/scripts/query_log.py index 96d2de3..d520593 100644 --- a/integrations/rapid7-insightidr/scripts/query_log.py +++ b/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: diff --git a/integrations/rapid7-insightidr/scripts/query_log_set.py b/integrations/rapid7-insightidr/scripts/query_log_set.py index 7e7d2da..8ef9586 100644 --- a/integrations/rapid7-insightidr/scripts/query_log_set.py +++ b/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: