Webhooks

import { createHmac, timingSafeEqual } from "node:crypto";

// rawBody is the request body as bytes, exactly as received.
export function verify(rawBody, signatureHeader, secret) {
  const parts = Object.fromEntries(signatureHeader.split(",").map((kv) => kv.split("=")));
  if (!parts.t || !parts.v1) return false;

  const fresh = Math.abs(Date.now() / 1000 - Number(parts.t)) < 300;
  const expected = createHmac("sha256", secret).update(`${parts.t}.${rawBody}`).digest("hex");
  const a = Buffer.from(expected), b = Buffer.from(parts.v1);
  return fresh && a.length === b.length && timingSafeEqual(a, b);
}

// Express: keep the raw body, verify, then parse.
app.post("/murusai", express.raw({ type: "application/json" }), (req, res) => {
  if (!verify(req.body, req.header("x-murusai-signature") ?? "", process.env.MURUSAI_WEBHOOK_SECRET)) {
    return res.status(401).end();
  }
  const alert = JSON.parse(req.body);
  res.status(200).end();   // answer first; do the work after
  handle(alert);
});

import hmac, hashlib, time

def verify(raw_body: bytes, signature_header: str, secret: str) -> bool:
    parts = dict(kv.split("=", 1) for kv in signature_header.split(","))
    if "t" not in parts or "v1" not in parts:
        return False
    fresh = abs(time.time() - int(parts["t"])) < 300
    expected = hmac.new(secret.encode(), f"{parts['t']}.".encode() + raw_body, hashlib.sha256).hexdigest()
    return fresh and hmac.compare_digest(expected, parts["v1"])

# Flask
@app.post("/murusai")
def murusai():
    if not verify(request.get_data(), request.headers.get("X-Murusai-Signature", ""), WEBHOOK_SECRET):
        return "", 401
    alert = request.get_json()
    handle_later(alert)   # queue it; answer now
    return "", 200