Verify the signature

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

const app = express();
const SECRET = process.env.TALLWATCH_WEBHOOK_SECRET;
const TOLERANCE_SEC = 300;

// Capture the raw body BEFORE express.json() parses it.
app.use(
  "/webhooks/tallwatch",
  express.raw({ type: "application/json" }),
  (req, res) => {
    const header = req.header("X-Tallwatch-Signature") ?? "";
    const parts = Object.fromEntries(
      header.split(",").map((kv) => kv.split("=")),
    );
    const { t, v1 = "" } = parts;

    // Sign over the exact bytes: "t=<ts>." followed by the raw body.
    const signed = Buffer.concat([Buffer.from(`t=${t}.`), req.body]);
    const expected = crypto
      .createHmac("sha256", SECRET)
      .update(signed)
      .digest("hex");

    const a = Buffer.from(v1);
    const b = Buffer.from(expected);
    if (a.length !== b.length || !crypto.timingSafeEqual(a, b)) {
      return res.status(401).send("invalid signature");
    }

    // Reject stale timestamps to blunt replay attacks.
    if (Math.abs(Date.now() / 1000 - Number(t)) > TOLERANCE_SEC) {
      return res.status(401).send("stale timestamp");
    }

    const payload = JSON.parse(req.body.toString("utf8"));
    // ... process the payload ...
    res.status(204).end();
  },
);

package main

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"io"
	"net/http"
	"os"
	"strings"
)

var secret = []byte(os.Getenv("TALLWATCH_WEBHOOK_SECRET"))

func handleWebhook(w http.ResponseWriter, r *http.Request) {
	body, err := io.ReadAll(r.Body)
	if err != nil {
		http.Error(w, "read failed", http.StatusBadRequest)
		return
	}

	var t, v1 string
	for _, part := range strings.Split(r.Header.Get("X-Tallwatch-Signature"), ",") {
		kv := strings.SplitN(part, "=", 2)
		if len(kv) != 2 {
			continue
		}
		switch kv[0] {
		case "t":
			t = kv[1]
		case "v1":
			v1 = kv[1]
		}
	}

	mac := hmac.New(sha256.New, secret)
	mac.Write([]byte("t=" + t + "."))
	mac.Write(body)
	expected := hex.EncodeToString(mac.Sum(nil))

	if !hmac.Equal([]byte(v1), []byte(expected)) {
		http.Error(w, "invalid signature", http.StatusUnauthorized)
		return
	}

	// ... process the body ...
	w.WriteHeader(http.StatusNoContent)
}

func main() {
	http.HandleFunc("/webhooks/tallwatch", handleWebhook)
	http.ListenAndServe(":8080", nil)
}

import hmac
import hashlib
import os

from flask import Flask, request, abort

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


@app.route("/webhooks/tallwatch", methods=["POST"])
def handle_webhook():
    raw_body = request.get_data()
    header = request.headers.get("X-Tallwatch-Signature", "")
    parts = dict(p.split("=", 1) for p in header.split(",") if "=" in p)
    t = parts.get("t", "")
    v1 = parts.get("v1", "")

    signed = b"t=" + t.encode() + b"." + raw_body
    expected = hmac.new(SECRET, signed, hashlib.sha256).hexdigest()

    if not hmac.compare_digest(v1, expected):
        abort(401, "invalid signature")

    payload = request.get_json(force=True)
    # ... process the payload ...
    return "", 204

require "sinatra"
require "openssl"
require "json"

SECRET = ENV.fetch("TALLWATCH_WEBHOOK_SECRET")

post "/webhooks/tallwatch" do
  body = request.body.read
  header = request.env["HTTP_X_TALLWATCH_SIGNATURE"].to_s
  parts = header.split(",").map { |kv| kv.split("=", 2) }.to_h
  t = parts["t"].to_s
  v1 = parts["v1"].to_s

  signed = "t=#{t}.#{body}"
  expected = OpenSSL::HMAC.hexdigest("SHA256", SECRET, signed)

  halt 401, "invalid signature" unless Rack::Utils.secure_compare(v1, expected)

  payload = JSON.parse(body)
  # ... process the payload ...
  status 204
end