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)