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

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)

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)

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)