Alert Slack on real changes

import hashlib, hmac, json, os, time

import requests
from flask import Flask, abort, request

ONYM_SECRET = os.environ["ONYM_WEBHOOK_SECRET"]  # whsec_...
SLACK_URL = os.environ["SLACK_WEBHOOK_URL"]      # https://hooks.slack.com/...
app = Flask(__name__)
seen: set[str] = set()


def verify(raw: bytes, header: str) -> bool:
    try:
        parts = dict(p.split("=", 1) for p in header.split(","))
        ts, sig = int(parts["t"]), parts["v1"]
    except (KeyError, ValueError):
        return False
    expected = hmac.new(ONYM_SECRET.encode(), f"{ts}.".encode() + raw, hashlib.sha256).hexdigest()
    return abs(time.time() - ts) <= 300 and hmac.compare_digest(expected, sig)


@app.post("/hooks/onym")
def onym():
    if not verify(request.get_data(), request.headers.get("X-Onym-Signature", "")):
        abort(400)
    event = request.get_json()
    if event["id"] in seen or event["event"] != "change.detected":
        return "", 204
    seen.add(event["id"])
    d = event["data"]
    bad = [c for c in d["events"] if c["tone"] == "bad"]  # only the bad news
    if bad:
        lines = "\n".join(f"• *{c['title']}*: {c['detail']}" for c in bad)
        requests.post(SLACK_URL, json={"text": f":warning: {d['domain']['brand']} on {d['run_date']}\n{lines}\n"
                                               f"<{d['domain']['dashboard']}/changes|Open in Onym>"}, timeout=5)
    return "", 204

import crypto from "node:crypto";
import express from "express";

const { ONYM_WEBHOOK_SECRET, SLACK_WEBHOOK_URL } = process.env;
const app = express();
const seen = new Set();

function verify(raw, header) {
  const p = Object.fromEntries(header.split(",").map((x) => x.split("=")));
  if (!p.t || !p.v1 || Math.abs(Date.now() / 1000 - Number(p.t)) > 300) return false;
  const expected = crypto.createHmac("sha256", ONYM_WEBHOOK_SECRET).update(`${p.t}.`).update(raw).digest("hex");
  return expected.length === p.v1.length && crypto.timingSafeEqual(Buffer.from(expected), Buffer.from(p.v1));
}

app.post("/hooks/onym", express.raw({ type: "application/json" }), async (req, res) => {
  if (!verify(req.body, req.get("X-Onym-Signature") || "")) return res.sendStatus(400);
  const event = JSON.parse(req.body);
  res.sendStatus(204); // answer fast, then work
  if (seen.has(event.id) || event.event !== "change.detected") return;
  seen.add(event.id);
  const d = event.data;
  const bad = d.events.filter((c) => c.tone === "bad");
  if (!bad.length) return;
  const lines = bad.map((c) => `• *${c.title}*: ${c.detail}`).join("\n");
  await fetch(SLACK_WEBHOOK_URL, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ text: `:warning: ${d.domain.brand} on ${d.run_date}\n${lines}\n<${d.domain.dashboard}/changes|Open in Onym>` }),
  });
});

app.listen(3000);