Webhooks

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

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

// express.raw, not express.json — verification needs the raw bytes.
app.post('/maat', express.raw({ type: 'application/json' }), (req, res) => {
    const timestamp = req.header('X-Signature-Timestamp') ?? '';

    // 1. Reject a stale timestamp before comparing anything.
    if (!timestamp || Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) {
        return res.status(401).end();
    }

    // 2. HMAC over "v1:" + timestamp + ":" + the raw body.
    const expected = createHmac('sha256', secret)
        .update(`v1:${timestamp}:`)
        .update(req.body) // a Buffer — the bytes as delivered
        .digest('hex');

    // 3. Constant-time compare — against either signature during a rotation.
    const matches = (candidate) =>
        typeof candidate === 'string' &&
        candidate.length === expected.length &&
        timingSafeEqual(Buffer.from(candidate), Buffer.from(expected));

    const verified =
        matches(req.header('X-Signature')) || matches(req.header('X-Signature-Previous'));
    if (!verified) return res.status(401).end();

    // 4. Only now parse.
    const event = JSON.parse(req.body.toString('utf8'));

    // 5. event_id is stable across our retries — use it as your idempotency key.
    //    A duplicate you have already processed should be answered 200 and ignored.

    if (event.event_type === 'endpoint.challenge') {
        return res.status(200).json({ challenge: event.data.challenge });
    }

    res.status(200).end();
});

app.listen(8443);

import hashlib
import hmac
import os
import time

from flask import Flask, jsonify, request

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


@app.post("/maat")
def maat_webhook():
    raw_body = request.get_data()  # the raw bytes, before any JSON parsing
    timestamp = request.headers.get("X-Signature-Timestamp", "")

    # 1. Reject a stale timestamp before comparing anything.
    if not timestamp or abs(time.time() - int(timestamp)) > 300:
        return "", 401

    # 2. HMAC over b"v1:" + timestamp + b":" + the raw body.
    expected = hmac.new(
        SECRET, b"v1:" + timestamp.encode() + b":" + raw_body, hashlib.sha256
    ).hexdigest()

    # 3. Constant-time compare — against either signature during a rotation.
    candidates = [
        request.headers.get("X-Signature", ""),
        request.headers.get("X-Signature-Previous", ""),
    ]
    if not any(sig and hmac.compare_digest(sig, expected) for sig in candidates):
        return "", 401

    # 4. Only now parse.
    event = request.get_json(force=True)

    # 5. event_id is stable across retries — use it as your idempotency key.

    if event["event_type"] == "endpoint.challenge":
        return jsonify(challenge=event["data"]["challenge"])

    return "", 200