From 6b76270cc6997bf55d4b090fb19279474d63eb7c Mon Sep 17 00:00:00 2001 From: Guillaume Bourgeois Date: Fri, 16 Oct 2026 11:23:03 +0000 Subject: [PATCH] feat(webhook-relay): read the alerts vendors push to riposte-relay (1.0.0) Riposte accepts no inbound connection, so a vendor that can only push (Rapid7 InsightIDR Essential basic detection rules) posts to the relay, and an alert rule on this integration reads one relay source back over outbound HTTPS. fetch_messages takes the rule's watermark as `since` and walks every page within the run, since the next watermark is this run's instant. A window holding more than `limit` fails the run instead of truncating, so the watermark stays put. relay_status and test_connection tell a 404 from the proxy (this Riposte's address is not listed) from a refused token. The generic mapper keeps the raw body, the sender's address and the checks the delivery passed (custom:webhook_verified). Tested against a local relay: 2503 messages over three pages, a millisecond watermark, the limit, unknown source and bad token. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Xa1QDbsPTYiVREv2Cvs3Zn --- .../incident-types/webhook-relay-message.yaml | 3 + integrations/webhook-relay/manifest.yaml | 60 ++++++++++++++++ .../webhook-relay/mappers/fetch_messages.yaml | 35 ++++++++++ .../webhook-relay/scripts/fetch_messages.py | 68 +++++++++++++++++++ .../webhook-relay/scripts/relay_status.py | 22 ++++++ .../webhook-relay/scripts/test_connection.py | 42 ++++++++++++ 6 files changed, 230 insertions(+) create mode 100644 integrations/webhook-relay/incident-types/webhook-relay-message.yaml create mode 100644 integrations/webhook-relay/manifest.yaml create mode 100644 integrations/webhook-relay/mappers/fetch_messages.yaml create mode 100644 integrations/webhook-relay/scripts/fetch_messages.py create mode 100644 integrations/webhook-relay/scripts/relay_status.py create mode 100644 integrations/webhook-relay/scripts/test_connection.py diff --git a/integrations/webhook-relay/incident-types/webhook-relay-message.yaml b/integrations/webhook-relay/incident-types/webhook-relay-message.yaml new file mode 100644 index 0000000..9bed383 --- /dev/null +++ b/integrations/webhook-relay/incident-types/webhook-relay-message.yaml @@ -0,0 +1,3 @@ +name: "Webhook Relay Message" +color: "#0288d1" +icon: "alert" diff --git a/integrations/webhook-relay/manifest.yaml b/integrations/webhook-relay/manifest.yaml new file mode 100644 index 0000000..ae4b826 --- /dev/null +++ b/integrations/webhook-relay/manifest.yaml @@ -0,0 +1,60 @@ +id: webhook_relay +name: Webhook Relay +version: 1.0.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.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: + properties: + base_url: + type: string + description: "The relay's public address, e.g. https://hermod.acknowledge.tools" + read_token: + type: string + description: "The relay's consumer token. The relay holds only its SHA-256 (RELAY_CONSUMER_TOKEN_SHA256); this Riposte must also call from an address in RELAY_CONSUMER_ALLOW_IPS." + x-soar-sensitive: true + required: + - base_url + - read_token + +commands: + - id: fetch_messages + name: webhook-relay-fetch-messages + description: "Messages a relay source received since a watermark, oldest first, every page followed. Ingestion: results path = messages, one alert per message, dedup on the relay's message id. Each message carries source, received_at, remote_ip, headers, verified (the checks it passed: path_token, header_token, hmac, ip), body (or body_b64) and json when the body parses. Reading consumes nothing." + risk: read + inputs_schema: + properties: + source: + type: string + description: "Relay source id, e.g. r7-acme — the mailbox to read. The client this alert rule is bound to must be the one this source was made for." + 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, the relay's own per-source cap). A window holding more fails the run rather than dropping the rest: the watermark stays put and the next run reads it again." + required: [source] + outputs_schema: { properties: {} } + ingest: + results_path: messages + dedup_key: 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." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } + + - id: test_connection + name: webhook-relay-test-connection + description: "Verify the relay answers this Riposte and the read token (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/webhook-relay/mappers/fetch_messages.yaml b/integrations/webhook-relay/mappers/fetch_messages.yaml new file mode 100644 index 0000000..31efcea --- /dev/null +++ b/integrations/webhook-relay/mappers/fetch_messages.yaml @@ -0,0 +1,35 @@ +name: "Webhook Relay message (generic)" +description: | + Maps one relayed message — whatever the vendor sent — to an alert, without knowing + the vendor. It is the mapper a new source starts with: the alert exists, the raw + payload is readable in Riposte, and a vendor mapper is then written from it + (clone this one, or pick the vendor's own when the marketplace ships it). + + A message is {id, source, received_at, remote_ip, content_type, headers, verified, + body | body_b64, json}. `json` is the body already decoded when it is JSON, so a + vendor field reads json.. + + The title tries the fields alert payloads most often name themselves with, and + falls back to the source. Severity is left to the alert rule's default: a generic + payload has no scale this mapper could trust. + + custom:webhook_verified is the trust level: the checks the delivery passed at the + relay (path_token, header_token, hmac, ip). An alert authenticated by a URL alone + is plausible, not proven — declare the field and keep destructive automation off + such alerts until a playbook has confirmed them against the vendor. +field_mappings: + title: "jsonata: $exists(json.title) ? $string(json.title) : $exists(json.name) ? $string(json.name) : $exists(json.alert.name) ? $string(json.alert.name) : $exists(json.rule.name) ? $string(json.rule.name) : $exists(json.detection_name) ? $string(json.detection_name) : ('Webhook · ' & source)" + source: "jsonata: 'Webhook relay · ' & source" + description: "jsonata: body ? $substring(body, 0, 8000) : ('Binary body, ' & $string($length(body_b64)) & ' base64 characters')" + "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" + # Key/value rows of the decoded payload's top level, for a table field. + "custom:webhook_payload": "jsonata: json ? $toTable(json) : body" +ocsf: + - { source_path: "id", ocsf_field: "finding_info.uid" } + - { source_path: "id", ocsf_field: "metadata.uid" } + - { source_path: "received_at", ocsf_field: "time" } + - { source_path: "source", ocsf_field: "metadata.product.name" } diff --git a/integrations/webhook-relay/scripts/fetch_messages.py b/integrations/webhook-relay/scripts/fetch_messages.py new file mode 100644 index 0000000..44356f4 --- /dev/null +++ b/integrations/webhook-relay/scripts/fetch_messages.py @@ -0,0 +1,68 @@ +import 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"{}") + + +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")) + print(json.dumps({"messages": messages, "count": len(messages), "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) diff --git a/integrations/webhook-relay/scripts/relay_status.py b/integrations/webhook-relay/scripts/relay_status.py new file mode 100644 index 0000000..566815b --- /dev/null +++ b/integrations/webhook-relay/scripts/relay_status.py @@ -0,0 +1,22 @@ +import json, os, sys, urllib.request, urllib.error + + +def run(): + 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") + req = urllib.request.Request(base + "/relay/v1/status", method="GET", headers={ + "Authorization": "Bearer " + str(s.get("read_token") or "").strip(), "Accept": "application/json"}) + with urllib.request.urlopen(req, timeout=30) as r: + print(r.read().decode("utf-8")) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/webhook-relay/scripts/test_connection.py b/integrations/webhook-relay/scripts/test_connection.py new file mode 100644 index 0000000..cdaf40f --- /dev/null +++ b/integrations/webhook-relay/scripts/test_connection.py @@ -0,0 +1,42 @@ +import json, os, sys, urllib.request, urllib.error + + +def run(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("base_url") or "").strip().rstrip("/") + if not base.startswith("https://"): + print(json.dumps({"ok": False, "error": "base_url must be the relay's https:// address."})) + sys.exit(1) + # /healthz first: it answers anyone, so a failure there is the network or + # the address, not the credential. + try: + urllib.request.urlopen(base + "/healthz", timeout=15).read() + except Exception as e: + print(json.dumps({"ok": False, "error": "The relay does not answer at %s/healthz: %s" % (base, e)})) + sys.exit(1) + req = urllib.request.Request(base + "/relay/v1/status", method="GET", headers={ + "Authorization": "Bearer " + str(s.get("read_token") or "").strip(), "Accept": "application/json"}) + try: + with urllib.request.urlopen(req, timeout=15) as r: + status = json.loads(r.read() or b"{}") + except urllib.error.HTTPError as e: + if e.code == 404: + err = ("The relay answers, but not to this Riposte: its proxy hides the reading API from addresses it " + "does not list. Add this Riposte's egress address to the Caddyfile and to " + "RELAY_CONSUMER_ALLOW_IPS — and check it does not leave through a proxy that changes it.") + elif e.code == 401: + err = ("The relay refused the read token, or this Riposte's address is not in " + "RELAY_CONSUMER_ALLOW_IPS. The relay's log line 'consumer refused' says which.") + else: + err = "HTTP %d from the relay." % e.code + print(json.dumps({"ok": False, "error": err})) + sys.exit(1) + sources = sorted((status.get("sources") or {}).keys()) + print(json.dumps({"ok": True, "sources": sources})) + + +try: + run() +except Exception as e: + print(json.dumps({"ok": False, "error": str(e)})) + sys.exit(1)