웹훅 받기

Next.js 라우트 핸들러, Express, Cloudflare Worker, Python용 검증된 웹훅 핸들러.

모든 핸들러는 같은 네 가지 일을 합니다. 원본 본문을 읽고, 서명을 검증하고, 이미 처리한 전송은 무시하고, 15초 안에 2xx로 응답합니다. 페이로드와 헤더는 웹훅에 설명되어 있습니다.

Next.js 라우트 핸들러

app/api/webhooks/ikiform/route.ts

import { verifyWebhook, WebhookVerificationError } from "@ikiform/sdk/webhooks";

export async function POST(request: Request) {
  const payload = await request.text();
  try {
    const event = await verifyWebhook({
      payload,
      headers: request.headers,
      secret: process.env.IKIFORM_WEBHOOK_SECRET!,
    });
    if (event.type === "submission.created") {
      await saveLead(event.id, event.data.fields);
    }
    return new Response(null, { status: 204 });
  } catch (error) {
    if (error instanceof WebhookVerificationError) {
      return new Response(error.reason, { status: 400 });
    }
    throw error;
  }
}

request.text()로 원본 본문을 얻습니다. request.json()을 먼저 호출하지 마세요. 다시 직렬화한 JSON은 서명과 더 이상 일치하지 않습니다.

Express

server.js

import express from "express";
import { verifyWebhook, WebhookVerificationError } from "@ikiform/sdk/webhooks";

const app = express();

app.post("/webhooks/ikiform", express.text({ type: "application/json" }), async (req, res) => {
  try {
    const event = await verifyWebhook({
      payload: req.body,
      headers: req.headers,
      secret: process.env.IKIFORM_WEBHOOK_SECRET,
    });
    if (event.type === "submission.created") await saveLead(event.id, event.data.fields);
    res.status(204).end();
  } catch (error) {
    if (error instanceof WebhookVerificationError) return res.status(400).send(error.reason);
    res.status(500).end();
  }
});

app.listen(3000);

express.text({ type: "application/json" })는 본문을 Ikiform이 보낸 문자열 그대로 유지합니다. 전역 express.json()이 먼저 파싱해 버리므로, 웹훅 라우트를 그보다 먼저 마운트하거나 이 경로를 제외하세요.

Cloudflare Worker

src/index.ts

import { verifyWebhook, WebhookVerificationError } from "@ikiform/sdk/webhooks";

interface Env {
  IKIFORM_WEBHOOK_SECRET: string;
  LEADS: KVNamespace;
}

export default {
  async fetch(request: Request, env: Env, ctx: ExecutionContext) {
    if (request.method !== "POST") return new Response(null, { status: 405 });
    const payload = await request.text();
    try {
      const event = await verifyWebhook({ payload, headers: request.headers, secret: env.IKIFORM_WEBHOOK_SECRET });
      if (event.type === "submission.created") {
        ctx.waitUntil(env.LEADS.put(event.id, JSON.stringify(event.data)));
      }
      return new Response(null, { status: 204 });
    } catch (error) {
      if (error instanceof WebhookVerificationError) return new Response(error.reason, { status: 400 });
      throw error;
    }
  },
};

SDK는 Web Crypto만 사용하므로 Workers, Deno, Bun에서 폴리필 없이 실행됩니다.

Python

app.py

import base64
import hashlib
import hmac
import os
import time

from flask import Flask, abort, request

app = Flask(__name__)
secret = base64.b64decode(os.environ["IKIFORM_WEBHOOK_SECRET"].removeprefix("whsec_"))


def verify(body: bytes, headers) -> bool:
    msg_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp")
    signatures = headers.get("webhook-signature")
    if not msg_id or not timestamp or not signatures or not timestamp.isdigit():
        return False
    if abs(time.time() - int(timestamp)) > 300:
        return False
    content = f"{msg_id}.{timestamp}.".encode() + body
    expected = base64.b64encode(hmac.new(secret, content, hashlib.sha256).digest()).decode()
    for entry in signatures.split(" "):
        version, _, signature = entry.partition(",")
        if version == "v1" and hmac.compare_digest(signature, expected):
            return True
    return False


@app.post("/webhooks/ikiform")
def ikiform_webhook():
    body = request.get_data()
    if not verify(body, request.headers):
        abort(400)
    event = request.get_json()
    if event["type"] == "submission.created":
        save_lead(event["id"], event["data"]["fields"])
    return "", 204

중복 처리

서버가 처리했지만 응답이 시간 초과된 경우처럼, 전송은 두 번 이상 도착할 수 있습니다. webhook-id 헤더와 페이로드의 id는 전송 ID이며 재시도해도 바뀌지 않습니다. 고유 제약 조건과 함께 저장하고 이미 처리한 작업은 건너뛰세요.

leads.ts

async function saveLead(deliveryId: string, fields: Record<string, unknown>) {
  const inserted = await db.query(
    "insert into leads (delivery_id, fields) values ($1, $2) on conflict (delivery_id) do nothing returning id",
    [deliveryId, JSON.stringify(fields)],
  );
  if (inserted.rowCount === 0) return;
  await notifySales(fields);
}

로컬에서 테스트하기

Ikiform은 443 포트의 공개 https: URL로만 전송하므로 localhost는 작동하지 않습니다. 공개 HTTPS URL을 주는 터널로 로컬 서버를 노출하고, 그 URL을 테스트 엔드포인트로 사용하세요.
엔드포인트 페이지에서 Send test를 사용하세요. endpoint.test 이벤트를 재시도 없이 한 번 보냅니다.
실제 전송을 서명 검증기에 붙여 넣어 브라우저에서 시크릿과 헤더를 확인하세요.