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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Xa1QDbsPTYiVREv2Cvs3Zn

integrations/webhook-relay/incident-types/rapid7-insightidr-basic-rule.yaml

@@ -0,0 +1,3 @@
+name: "Rapid7 InsightIDR Basic Rule"
+color: "#e8590c"
+icon: "alert"

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:<message id>. 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."

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:<message id> 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" }

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)