From 5085d966286c6480a5f2e2b83b2d267d6722a176 Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Mon, 19 Oct 2026 07:56:45 +0000 Subject: [PATCH] feat(webhook-relay): map Rapid7 InsightIDR basic rule alerts (1.1.0) fetch_rapid7_alerts opens the alert InsightIDR Essential posts as a JSON string in a single payload field (JSON or form body) and returns one flat item per relayed message; the correlation id is the dedup key. Own mapper and alert type, written from a real delivery. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Xa1QDbsPTYiVREv2Cvs3Zn --- .../rapid7-insightidr-basic-rule.yaml | 3 + integrations/webhook-relay/manifest.yaml | 26 +++- .../mappers/fetch_rapid7_alerts.yaml | 59 ++++++++ .../scripts/fetch_rapid7_alerts.py | 138 ++++++++++++++++++ 4 files changed, 224 insertions(+), 2 deletions(-) create mode 100644 integrations/webhook-relay/incident-types/rapid7-insightidr-basic-rule.yaml create mode 100644 integrations/webhook-relay/mappers/fetch_rapid7_alerts.yaml create mode 100644 integrations/webhook-relay/scripts/fetch_rapid7_alerts.py diff --git a/integrations/webhook-relay/incident-types/rapid7-insightidr-basic-rule.yaml b/integrations/webhook-relay/incident-types/rapid7-insightidr-basic-rule.yaml new file mode 100644 index 0000000..fc1d17b --- /dev/null +++ b/integrations/webhook-relay/incident-types/rapid7-insightidr-basic-rule.yaml @@ -0,0 +1,3 @@ +name: "Rapid7 InsightIDR Basic Rule" +color: "#e8590c" +icon: "alert" diff --git a/integrations/webhook-relay/manifest.yaml b/integrations/webhook-relay/manifest.yaml index 2fd6e73..50a6b4d 100644 --- a/integrations/webhook-relay/manifest.yaml +++ b/integrations/webhook-relay/manifest.yaml @@ -1,8 +1,8 @@ id: webhook_relay name: Webhook Relay -version: 1.0.1 +version: 1.1.0 description: "Reads the alerts that vendors pushed to riposte-relay, the public mailbox for alert webhooks (riposte/relay/). Riposte accepts no inbound connection, so a source whose only output is a webhook — Rapid7 InsightIDR Essential basic detection rules, for one — posts to the relay, and an alert rule on this integration reads one relay source back over outbound HTTPS. The relay knows no vendor: the rule chooses the client, the alert type and the mapper. A read consumes nothing; the rule's watermark is the read's `since`, so after a failover the standby reads the same mailbox from its own watermark." -changelog: "1.0.1 — The generic mapper fills the OCSF fields that have a place for a relayed message: metadata.logged_time (receipt), metadata.log_provider and log_name (relay, source), metadata.labels (the checks passed, relay:hmac…), raw_data (the payload). The sender address goes to unmapped.relay.sender_ip, not src_endpoint.ip: it is the vendor's cloud, and as an entity it would tie every alert of that vendor together. metadata.product.name no longer holds the source id. | 1.0.0 — Initial release: fetch_messages (ingestion: one alert per relayed message, `since` watermark, page walk), relay_status, test_connection, and a generic mapper that keeps the raw body and the checks the message passed." +changelog: "1.1.0 — New fetch_rapid7_alerts command, with its own mapper and a Rapid7 InsightIDR Basic Rule alert type, written from a real InsightIDR Essential delivery. A basic rule posts its alert as a JSON string in a single `payload` field; the command opens it (as a JSON body or a form body) and returns one flat item per message: rule name, correlation id, event time, log set, log, host name and log line. The correlation id is the dedup key, so a redelivery of one firing makes no second alert. The payload carries no severity, user or address: those are in the log line, which the rule may withhold — the alert then says so. | 1.0.1 — The generic mapper fills the OCSF fields that have a place for a relayed message: metadata.logged_time (receipt), metadata.log_provider and log_name (relay, source), metadata.labels (the checks passed, relay:hmac…), raw_data (the payload). The sender address goes to unmapped.relay.sender_ip, not src_endpoint.ip: it is the vendor's cloud, and as an entity it would tie every alert of that vendor together. metadata.product.name no longer holds the source id. | 1.0.0 — Initial release: fetch_messages (ingestion: one alert per relayed message, `since` watermark, page walk), relay_status, test_connection, and a generic mapper that keeps the raw body and the checks the message passed." category: ingestion config_schema: @@ -41,6 +41,28 @@ commands: dedup_key: id incremental_field: since + - id: fetch_rapid7_alerts + name: webhook-relay-fetch-rapid7-alerts + description: "fetch_messages for a source fed by Rapid7 InsightIDR Essential basic detection rules (notification Webhook). Each message's `payload` — the alert as a JSON string — is opened into one flat item: alert_id (Rapid7's correlation_id, the dedup key), rule, event_time, log_set (Rapid7's host.name), log_name, hostname, log_line and log_event (when the line is JSON), log_line_withheld, plus the relay's id, source, received_at, remote_ip, verified and body. A message that does not decode still yields an item, with decode_error and alert_id relay:. Results path = alerts." + risk: read + inputs_schema: + properties: + source: + type: string + description: "Relay source id fed by the Rapid7 rule, e.g. r7-acme." + since: + type: string + description: "Lower bound on received_at: ISO-8601, UNIX seconds or milliseconds. Incremental fetch watermark." + limit: + type: number + description: "Most messages one run may gather (default 10000). A window holding more fails the run rather than dropping the rest." + required: [source] + outputs_schema: { properties: {} } + ingest: + results_path: alerts + dedup_key: alert_id + incremental_field: since + - id: relay_status name: webhook-relay-status description: "Per relay source: messages waiting, oldest and newest, accepted and rejected counts by reason (ip, path_token, header_token, hmac, size, rate, full…), last accepted, last read. Counters restart with the relay. A rising rejected count on a source is a misconfigured vendor — or someone trying." diff --git a/integrations/webhook-relay/mappers/fetch_rapid7_alerts.yaml b/integrations/webhook-relay/mappers/fetch_rapid7_alerts.yaml new file mode 100644 index 0000000..2155e68 --- /dev/null +++ b/integrations/webhook-relay/mappers/fetch_rapid7_alerts.yaml @@ -0,0 +1,59 @@ +name: "Rapid7 InsightIDR basic rule (via Webhook Relay)" +description: | + Maps one Rapid7 InsightIDR Essential basic detection rule alert, received by the + relay's Webhook notification, to an alert. fetch_rapid7_alerts has already opened + the payload (a JSON string in a `payload` field): each item carries rule, + correlation_id, event_time, log_set, log_name, hostname, log_line (log_event when + the line is JSON) and the relay's own fields (id, source, received_at, verified). + + The payload has no severity and names no user or source address of its own: those + live in the log line, which Rapid7 leaves out unless the rule includes it. When it + is left out, log_line_withheld is true and the description says so — turn on the + log line in the rule's notification, or the alert only says which rule fired. + + `host.name` in Rapid7's payload is the log set (Asset Authentication), not a + machine; it goes to unmapped.rapid7.log_set. device.hostname takes host.hostname, + which basic rules often leave null. +field_mappings: + title: "jsonata: rule ? rule : ('Rapid7 · ' & source)" + source: "jsonata: 'Rapid7 InsightIDR'" + description: "jsonata: decode_error ? ('Undecodable Rapid7 delivery (' & decode_error & '):\n' & $substring(body, 0, 8000)) : ('Rapid7 InsightIDR basic rule: ' & rule & '\nLog set: ' & (log_set ? log_set : '—') & ' · log: ' & (log_name ? log_name : '—') & (hostname ? '\nHost: ' & hostname : '') & (event_time ? '\nEvent time: ' & event_time : '') & '\n\n' & (log_line_withheld ? 'Log line withheld by the rule: enable it in the rule notification to see the triggering event.' : (log_line ? 'Log line:\n' & $substring(log_line, 0, 8000) : '')))" + "custom:webhook_source": "source" + "custom:webhook_verified": "jsonata: $join(verified, ',')" + "custom:webhook_message_id": "id" + "custom:webhook_received_at": "received_at" + "custom:webhook_sender_ip": "remote_ip" + "custom:webhook_payload": "jsonata: log_event ? $toTable(log_event) : rapid7 ? $toTable({'rule': rule, 'log_set': log_set, 'log': log_name, 'hostname': hostname, 'event_time': event_time, 'region': region, 'correlation_id': correlation_id}) : body" +ocsf: + # ── Finding ─────────────────────────────────────────────────────── + # Rapid7's correlation id (falls back to relay: for an undecodable + # delivery) — also the command's dedup key. + - { source_path: "alert_id", ocsf_field: "finding_info.uid" } + - { source_path: "rule", ocsf_field: "finding_info.title" } + - { source_path: "rule", ocsf_field: "finding_info.analytic.name" } + - { source_path: "'Rule'", ocsf_field: "finding_info.analytic.type" } + - { source_path: "event_time", ocsf_field: "finding_info.created_time" } + # The time of the log event that fired the rule; receipt stays in logged_time. + - { source_path: "event_time ? event_time : received_at", ocsf_field: "time" } + - { source_path: "received_at", ocsf_field: "metadata.logged_time" } + - { source_path: "id", ocsf_field: "metadata.uid" } + - { source_path: "correlation_id", ocsf_field: "metadata.correlation_uid" } + # ── Product ─────────────────────────────────────────────────────── + - { source_path: "'InsightIDR'", ocsf_field: "metadata.product.name" } + - { source_path: "'Rapid7'", ocsf_field: "metadata.product.vendor_name" } + - { source_path: "'riposte-relay'", ocsf_field: "metadata.log_provider" } + - { source_path: "source", ocsf_field: "metadata.log_name" } + - { source_path: "[verified.('relay:' & $)]", ocsf_field: "metadata.labels" } + - { source_path: "body", ocsf_field: "raw_data" } + # ── Asset ───────────────────────────────────────────────────────── + - { source_path: "hostname", ocsf_field: "device.hostname" } + # ── Rapid7 specifics ────────────────────────────────────────────── + - { source_path: "log_set", ocsf_field: "unmapped.rapid7.log_set" } + - { source_path: "log_name", ocsf_field: "unmapped.rapid7.log_name" } + - { source_path: "log_key", ocsf_field: "unmapped.rapid7.log_key" } + - { source_path: "region", ocsf_field: "unmapped.rapid7.region" } + - { source_path: "event_sequence", ocsf_field: "unmapped.rapid7.event_sequence" } + - { source_path: "log_line", ocsf_field: "unmapped.rapid7.log_line" } + - { source_path: "log_line_withheld", ocsf_field: "unmapped.rapid7.log_line_withheld" } + # The vendor's cloud, never an entity (see the generic mapper). + - { source_path: "remote_ip", ocsf_field: "unmapped.relay.sender_ip" } diff --git a/integrations/webhook-relay/scripts/fetch_rapid7_alerts.py b/integrations/webhook-relay/scripts/fetch_rapid7_alerts.py new file mode 100644 index 0000000..41affe4 --- /dev/null +++ b/integrations/webhook-relay/scripts/fetch_rapid7_alerts.py @@ -0,0 +1,138 @@ +import datetime, json, os, sys, urllib.request, urllib.parse, urllib.error + +PAGE = 1000 # the relay's largest page +DEFAULT_LIMIT = 10000 # the relay's default max_pending per source + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("base_url") or "").strip().rstrip("/") + if not base.startswith("https://"): + raise ValueError("base_url must be the relay's https:// address") + return base, {"Authorization": "Bearer " + str(s.get("read_token") or "").strip(), "Accept": "application/json"} + + +def _get(path, params): + base, headers = _cfg() + url = base + path + "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v not in (None, "")}) + req = urllib.request.Request(url, headers=headers, method="GET") + with urllib.request.urlopen(req, timeout=60) as r: + return json.loads(r.read() or b"{}") + + +# InsightIDR basic detection rules post a single field, `payload`, holding the +# alert as a JSON-encoded string — as a JSON body {"payload": "..."} or as a form +# body payload=... depending on the delivery. Both are opened here, in Python: +# event.s is a 64-bit sequence that a float-based JSON reader would round. +OPTED_OUT = "The log line that triggered this alert is not included" + + +def _payload(m): + j = m.get("json") + if isinstance(j, dict) and "payload" in j: + p = j["payload"] + return json.loads(p) if isinstance(p, str) else p + body = m.get("body") or "" + if isinstance(j, dict): + return j + form = urllib.parse.parse_qs(body) + if "payload" in form: + return json.loads(form["payload"][0]) + raise ValueError("no Rapid7 payload in this message") + + +def _iso_ms(t): + try: + return datetime.datetime.fromtimestamp(int(t) / 1000, datetime.timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") + except (TypeError, ValueError, OverflowError, OSError): + return None + + +def rapid7(m): + out = {k: m.get(k) for k in ("id", "source", "received_at", "remote_ip", "content_type", "verified", "body")} + # A message the relay could not tie to a Rapid7 alert still becomes an + # alert: dropping it would hide a delivery nobody would then look for. + out["alert_id"] = "relay:" + str(m.get("id")) + try: + r = _payload(m) + except Exception as e: + out["decode_error"] = str(e) + return out + if not isinstance(r, dict): + out["decode_error"] = "payload is not an object" + return out + alert, event, log, host = (r.get(k) if isinstance(r.get(k), dict) else {} for k in ("alert", "event", "log", "host")) + if r.get("correlation_id"): + # Rapid7's id for this firing: a redelivery carries it again, a new + # relay message id does not, so it is the dedup key. + out["alert_id"] = str(r["correlation_id"]) + out["correlation_id"] = r.get("correlation_id") + out["rule"] = alert.get("name") + out["region"] = r.get("region") + out["event_time"] = _iso_ms(event.get("t")) + out["event_sequence"] = str(event["s"]) if event.get("s") is not None else None + line = event.get("m") + out["log_line_withheld"] = isinstance(line, str) and line.startswith(OPTED_OUT) + out["log_line"] = None if out["log_line_withheld"] else line + if isinstance(line, str) and not out["log_line_withheld"]: + try: + out["log_event"] = json.loads(line) + except ValueError: + pass + out["log_name"] = log.get("name") + out["log_key"] = log.get("key") + # In a basic rule's payload `host.name` is the log set (Asset Authentication), + # not a machine; host.hostname is the machine, when Rapid7 knows one. + out["log_set"] = host.get("name") + out["hostname"] = host.get("hostname") + out["rapid7"] = r + return out + + +def run(): + inp = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + source = str(inp.get("source") or "").strip() + if not source: + raise ValueError("source is required: the relay source id, e.g. r7-acme") + try: + limit = int(inp.get("limit") or DEFAULT_LIMIT) + except (TypeError, ValueError): + limit = DEFAULT_LIMIT + since = str(inp.get("since") or "").strip() + + # Every page is read within this one run. The next run's watermark is the + # instant of this one, so a page left for later would never be asked for. + messages, after = [], "" + while True: + page = _get("/relay/v1/messages", {"source": source, "since": since, "after": after, "limit": PAGE}) + messages.extend(page.get("messages") or []) + if not page.get("more"): + break + after = page.get("next") or "" + if not after: + break + if len(messages) >= limit: + # Failing keeps the watermark where it is; returning a truncated + # list would move it past what was never read. + raise RuntimeError( + "more than %d messages wait on source %s since %s: raise this command's limit " + "(nothing was ingested; the next run reads the same window)" % (limit, source, since or "the beginning")) + alerts = [rapid7(m) for m in messages] + print(json.dumps({"alerts": alerts, "count": len(alerts), "source": source, "since": since})) + + +try: + run() +except urllib.error.HTTPError as e: + detail = e.read().decode("utf-8", "replace") + out = {"error": "HTTP " + str(e.code), "detail": detail} + if e.code == 401: + out["hint"] = "The relay refused the read token, or this Riposte's address is not in RELAY_CONSUMER_ALLOW_IPS." + elif e.code == 404: + out["hint"] = ("Unknown source on this relay, or this Riposte's address is not allowed by the proxy in front " + "of it (it answers 404 to readers it does not know). Run test_connection.") + print(json.dumps(out)) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1)