Webhooks

import hashlib, hmac, os, time
from flask import Flask, request, abort

app = Flask(__name__)
SECRET = os.environ["LEAKCHECK_WEBHOOK_SECRET"].encode()

@app.post("/leakcheck")
def leakcheck():
    ts = request.headers.get("X-LeakCheck-Timestamp", "")
    sig = request.headers.get("X-LeakCheck-Signature", "")
    if not ts.isdigit() or abs(time.time() - int(ts)) > 300:
        abort(400)
    expected = "sha256=" + hmac.new(SECRET, ts.encode() + b"." + request.get_data(), hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, sig):
        abort(401)
    alert = request.get_json()
    print(alert["monitor"]["value"], [s["name"] for s in alert["sources"]])
    return "", 204

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

const app = express();
const secret = process.env.LEAKCHECK_WEBHOOK_SECRET;

// Keep the raw body: the signature covers the bytes as sent.
app.post("/leakcheck", express.raw({ type: "application/json" }), (req, res) => {
  const ts = req.get("X-LeakCheck-Timestamp") ?? "";
  const sig = req.get("X-LeakCheck-Signature") ?? "";
  if (!/^\d+$/.test(ts) || Math.abs(Date.now() / 1000 - Number(ts)) > 300) return res.sendStatus(400);

  const expected = "sha256=" + createHmac("sha256", secret).update(`${ts}.`).update(req.body).digest("hex");
  if (expected.length !== sig.length || !timingSafeEqual(Buffer.from(expected), Buffer.from(sig))) return res.sendStatus(401);

  const alert = JSON.parse(req.body);
  console.log(alert.monitor.value, alert.sources.map((s) => s.name));
  res.sendStatus(204);
});

<?php
$secret = getenv('LEAKCHECK_WEBHOOK_SECRET');
$body = file_get_contents('php://input');
$ts = $_SERVER['HTTP_X_LEAKCHECK_TIMESTAMP'] ?? '';
$sig = $_SERVER['HTTP_X_LEAKCHECK_SIGNATURE'] ?? '';

if (!ctype_digit($ts) || abs(time() - (int) $ts) > 300) {
    http_response_code(400); exit;
}
$expected = 'sha256=' . hash_hmac('sha256', $ts . '.' . $body, $secret);
if (!hash_equals($expected, $sig)) {
    http_response_code(401); exit;
}

$alert = json_decode($body, true);
error_log($alert['monitor']['value'] . ': ' . implode(', ', array_column($alert['sources'], 'name')));
http_response_code(204);