Webhooks

curl -X POST "https://api.enform.co/api/v1/webhooks" \
  -H "Authorization: Bearer $ENFORM_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"formId":"c3a1f7e2-1b4d-4e8a-9c2f-6d5e4b3a2f10","url":"https://example.com/enform/webhook"}'

const res = await fetch("https://api.enform.co/api/v1/webhooks", {
  method: "POST",
  headers: {
    Authorization: "Bearer " + process.env.ENFORM_API_KEY,
    "Content-Type": "application/json",
  },
  body: JSON.stringify({"formId":"c3a1f7e2-1b4d-4e8a-9c2f-6d5e4b3a2f10","url":"https://example.com/enform/webhook"}),
});
const json = await res.json();

import os, requests

res = requests.post(
    "https://api.enform.co/api/v1/webhooks",
    headers={"Authorization": f"Bearer {os.environ['ENFORM_API_KEY']}"},
    json={"formId":"c3a1f7e2-1b4d-4e8a-9c2f-6d5e4b3a2f10","url":"https://example.com/enform/webhook"},
)
data = res.json()

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

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

app.post("/enform/webhook", express.raw({ type: "application/json" }), (req, res) => {
  const id = req.header("webhook-id");
  const timestamp = req.header("webhook-timestamp");
  const body = req.body.toString("utf8");

  const expected = createHmac("sha256", secret).update(`${id}.${timestamp}.${body}`).digest("base64");
  const given = (req.header("webhook-signature") ?? "").split(" ").map((part) => part.split(",")[1] ?? "");
  const matches = given.some(
    (sig) => sig.length === expected.length && timingSafeEqual(Buffer.from(sig), Buffer.from(expected)),
  );
  const fresh = Math.abs(Date.now() / 1000 - Number(timestamp)) < 300;
  if (!matches || !fresh) return res.status(400).end();

  const event = JSON.parse(body);
  console.log(event.data.answers);
  res.status(200).end();
});

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

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

@app.post("/enform/webhook")
def enform_webhook():
    msg_id = request.headers["webhook-id"]
    timestamp = request.headers["webhook-timestamp"]
    body = request.get_data(as_text=True)

    signed = f"{msg_id}.{timestamp}.{body}".encode()
    expected = base64.b64encode(hmac.new(SECRET.encode(), signed, hashlib.sha256).digest()).decode()
    given = [part.split(",", 1)[-1] for part in request.headers.get("webhook-signature", "").split()]
    matches = any(hmac.compare_digest(expected, sig) for sig in given)
    fresh = abs(time.time() - int(timestamp)) < 300
    if not matches or not fresh:
        return "", 400

    event = request.get_json()
    print(event["data"]["answers"])
    return "", 200