Verify webhooks

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

export function verifyMachinemailSignature(secret: string, header: string | null, rawBody: string, toleranceSeconds = 300): boolean {
  if (!header) return false;
  const parts = new Map(
    header.split(',').map((p) => {
      const i = p.indexOf('=');
      return [p.slice(0, i).trim(), p.slice(i + 1).trim()] as const;
    }),
  );
  const t = Number(parts.get('t'));
  const v1 = parts.get('v1');
  if (!Number.isFinite(t) || !v1) return false;
  if (Math.abs(Date.now() / 1000 - t) > toleranceSeconds) return false;
  const expected = createHmac('sha256', secret).update(`${t}.${rawBody}`).digest('hex');
  const a = Buffer.from(expected, 'hex');
  const b = Buffer.from(v1, 'hex');
  return a.length === b.length && timingSafeEqual(a, b);
}

// Bun (or any runtime with the Fetch API Request/Response).
import { verifyMachinemailSignature } from './verify';

const secret = process.env.MACHINEMAIL_WEBHOOK_SECRET!;
const seen = new Set<string>(); // use your database in production

Bun.serve({
  port: 3000,
  async fetch(req) {
    if (req.method !== 'POST') return new Response('Method not allowed', { status: 405 });
    const raw = await req.text(); // raw body, before JSON.parse
    if (!verifyMachinemailSignature(secret, req.headers.get('x-machinemail-signature'), raw)) {
      return new Response('Invalid signature', { status: 401 });
    }
    const eventId = req.headers.get('x-machinemail-event-id')!;
    if (seen.has(eventId)) return new Response('Duplicate', { status: 200 });
    seen.add(eventId);

    const event = JSON.parse(raw);
    if (event.event_type === 'message.received') {
      // Queue the work and return quickly.
      console.log('New message', event.data.message.message_id, event.data.message.trust?.level);
    }
    return new Response('ok');
  },
});

import hashlib, hmac, time


def verify_machinemail_signature(secret: str, header: str | None, raw_body: bytes, tolerance: int = 300) -> bool:
    if not header:
        return False
    parts = dict(p.strip().split("=", 1) for p in header.split(",") if "=" in p)
    try:
        t = int(parts["t"])
        v1 = parts["v1"]
    except (KeyError, ValueError):
        return False
    if abs(time.time() - t) > tolerance:
        return False
    expected = hmac.new(secret.encode(), f"{t}.".encode() + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, v1)

# Flask
import json, os
from flask import Flask, request
from verify import verify_machinemail_signature

app = Flask(__name__)
SECRET = os.environ["MACHINEMAIL_WEBHOOK_SECRET"]
seen = set()  # use your database in production


@app.post("/hooks/machinemail")
def machinemail_hook():
    raw = request.get_data()  # raw bytes, before parsing
    if not verify_machinemail_signature(SECRET, request.headers.get("X-MachineMail-Signature"), raw):
        return "Invalid signature", 401
    event_id = request.headers["X-MachineMail-Event-Id"]
    if event_id in seen:
        return "Duplicate", 200
    seen.add(event_id)

    event = json.loads(raw)
    if event["event_type"] == "message.received":
        msg = event["data"]["message"]
        print("New message", msg["message_id"], (msg.get("trust") or {}).get("level"))
    return "ok", 200

package main

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"io"
	"log"
	"math"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"
)

func verifyMachinemailSignature(secret, header string, body []byte, tolerance time.Duration) bool {
	var t int64
	var v1 string
	for _, part := range strings.Split(header, ",") {
		k, v, ok := strings.Cut(strings.TrimSpace(part), "=")
		if !ok {
			continue
		}
		switch k {
		case "t":
			t, _ = strconv.ParseInt(v, 10, 64)
		case "v1":
			v1 = v
		}
	}
	if t == 0 || v1 == "" {
		return false
	}
	if math.Abs(float64(time.Now().Unix()-t)) > tolerance.Seconds() {
		return false
	}
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte(strconv.FormatInt(t, 10) + "."))
	mac.Write(body)
	expected := mac.Sum(nil)
	got, err := hex.DecodeString(v1)
	if err != nil {
		return false
	}
	return hmac.Equal(expected, got)
}

func main() {
	secret := os.Getenv("MACHINEMAIL_WEBHOOK_SECRET")
	http.HandleFunc("/hooks/machinemail", func(w http.ResponseWriter, r *http.Request) {
		body, err := io.ReadAll(io.LimitReader(r.Body, 50<<20))
		if err != nil {
			http.Error(w, "bad body", http.StatusBadRequest)
			return
		}
		if !verifyMachinemailSignature(secret, r.Header.Get("X-MachineMail-Signature"), body, 300*time.Second) {
			http.Error(w, "invalid signature", http.StatusUnauthorized)
			return
		}
		var event struct {
			EventID   string          `json:"event_id"`
			EventType string          `json:"event_type"`
			Data      json.RawMessage `json:"data"`
		}
		if err := json.Unmarshal(body, &event); err != nil {
			http.Error(w, "bad json", http.StatusBadRequest)
			return
		}
		log.Printf("%s %s", event.EventType, event.EventID)
		w.WriteHeader(http.StatusOK)
	})
	log.Fatal(http.ListenAndServe(":3000", nil))
}