Webhook receiver: a sample service

// A Kommio webhook receiver for Node 22, no dependencies:
//
//   KOMMIO_WEBHOOK_SECRET=<endpoint secret> PORT=4600 node --experimental-strip-types receiver.ts
//
// It verifies the Kommio-Signature header exactly as Kommio signs it,
// answers 200 at once and handles the event afterwards (Kommio waits at
// most 15 seconds and retries anything but a 2xx), and handles each
// delivery once: a retry or a replay carries the same delivery id.
import { createServer } from 'node:http';
import { createHmac, timingSafeEqual } from 'node:crypto';
import { pathToFileURL } from 'node:url';

/** The envelope every webhook delivery carries (docs: kommio.io/docs/extend/webhook-receiver/). */
export type Delivery = {
  id: number; // the delivery: the same on a retry or a replay
  event_id: number;
  type: string;
  account_id: number;
  object_type?: string;
  object_id?: number;
  occurred_at: string;
  data: Record<string, unknown>;
  visitor?: Record<string, unknown>;
};

/**
 * Kommio-Signature: t=<unix seconds>,v1=<hex HMAC-SHA256 of "<t>.<raw body>">
 * with the endpoint's secret (it starts with whsec_ and is used as it is). Reject a timestamp
 * more than five minutes from your clock and compare in constant time.
 */
export function verifySignature(secret: string, header: string | undefined, rawBody: string, nowMs = Date.now()): boolean {
  if (!secret || !header) return false;
  const parts = Object.fromEntries(header.split(',').map((kv) => kv.trim().split('=') as [string, string]));
  const t = Number(parts.t);
  if (!Number.isInteger(t) || Math.abs(nowMs / 1000 - t) > 300) return false;
  const expected = Buffer.from(createHmac('sha256', secret).update(`${t}.${rawBody}`).digest('hex'));
  const given = Buffer.from(parts.v1 ?? '');
  return given.length === expected.length && timingSafeEqual(given, expected);
}

/** What the example does with each event; replace with your own work. */
export function describe(d: Delivery): string | null {
  switch (d.type) {
    case 'conversation.created':
      return `New conversation ${d.object_id} in inbox ${d.data.inbox_id} from contact ${d.data.contact_id}`;
    case 'message.created':
      // The event carries no message text; read it with an API key:
      // GET /api/conversations/{conversation_id}/messages
      return d.data.direction === 'inbound' ? `Customer wrote in conversation ${d.data.conversation_id} (message ${d.object_id})` : null;
    case 'csat.submitted':
      return d.data.low
        ? `LOW RATING ${d.data.rating} on conversation ${d.data.conversation_id}: ${d.data.comment || '(no comment)'}`
        : `Rating ${d.data.rating} on conversation ${d.data.conversation_id}`;
    default:
      return null; // subscribed to more than this example handles
  }
}

export type Answer = { status: number; body: Record<string, unknown> };

/**
 * The receiver without the HTTP server, so it can be tested: handle() checks
 * the signature, drops a delivery it has already accepted, and queues the
 * rest for process(), which runs after the answer has gone out.
 */
export function createReceiver(opts: { secret: string; process: (d: Delivery) => void | Promise<void>; now?: () => number; rememberMs?: number }) {
  // Delivery ids already accepted. In production keep them where every
  // instance sees them (a table with a unique key, Redis SET NX) for at
  // least a day: Kommio retries for about 24 hours.
  const seen = new Map<string, number>();
  const rememberMs = opts.rememberMs ?? 48 * 3600_000;
  const now = opts.now ?? Date.now;
  return {
    seen,
    handle(headers: Record<string, string | undefined>, rawBody: string): Answer {
      if (!verifySignature(opts.secret, headers['kommio-signature'], rawBody, now())) return { status: 401, body: { error: 'bad signature' } };
      let d: Delivery;
      try { d = JSON.parse(rawBody) as Delivery; } catch { return { status: 400, body: { error: 'not JSON' } }; }
      const id = headers['kommio-delivery'] ?? String(d.id);
      for (const [k, at] of seen) if (now() - at > rememberMs) seen.delete(k);
      if (seen.has(id)) return { status: 200, body: { ok: true, duplicate: true } };
      seen.set(id, now());
      // Answer first, work after: a slow handler must not make Kommio retry.
      setImmediate(() => { void Promise.resolve(opts.process(d)).catch((e) => console.error('webhook handler failed', id, e)); });
      return { status: 200, body: { ok: true } };
    },
  };
}

export function start(port = Number(process.env.PORT || 4600), secret = process.env.KOMMIO_WEBHOOK_SECRET || '') {
  if (!secret) throw new Error('set KOMMIO_WEBHOOK_SECRET to the endpoint secret Kommio showed (it starts with whsec_)');
  const receiver = createReceiver({ secret, process: (d) => { const line = describe(d); if (line) console.log(`[${d.type}] ${line}`); } });
  return createServer((req, res) => {
    if (req.method !== 'POST' || req.url !== '/webhooks') { res.writeHead(404); res.end(); return; }
    const chunks: Buffer[] = [];
    let size = 0;
    req.on('data', (c: Buffer) => { size += c.length; if (size <= 1 << 20) chunks.push(c); });
    req.on('end', () => {
      const headers = Object.fromEntries(Object.entries(req.headers).map(([k, v]) => [k, Array.isArray(v) ? v[0] : v]));
      const a = size > 1 << 20 ? { status: 413, body: { error: 'too large' } } : receiver.handle(headers, Buffer.concat(chunks).toString('utf8'));
      res.writeHead(a.status, { 'Content-Type': 'application/json' });
      res.end(JSON.stringify(a.body));
    });
  }).listen(port, () => console.log(`webhook receiver on http://127.0.0.1:${port}/webhooks`));
}

if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) start();

"""A Kommio webhook receiver for Python 3.9+, standard library only.

    KOMMIO_WEBHOOK_SECRET=<endpoint secret> PORT=4600 python3 receiver.py

It verifies the Kommio-Signature header exactly as Kommio signs it, answers
200 at once and handles the event on a worker thread (Kommio waits at most
15 seconds and retries anything but a 2xx), and handles each delivery once:
a retry or a replay carries the same delivery id.
"""

import hashlib
import hmac
import json
import os
import queue
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer


def verify_signature(secret, header, raw_body, now=None):
    """Kommio-Signature: t=<unix seconds>,v1=<hex HMAC-SHA256 of "<t>.<raw body>">
    with the endpoint's secret (it starts with whsec_ and is used as it is). Reject a timestamp
    more than five minutes from your clock and compare in constant time."""
    if not secret or not header:
        return False
    parts = dict(kv.strip().split("=", 1) for kv in header.split(",") if "=" in kv)
    try:
        t = int(parts.get("t", ""))
    except ValueError:
        return False
    if abs((time.time() if now is None else now) - t) > 300:
        return False
    expected = hmac.new(secret.encode(), f"{t}.".encode() + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, parts.get("v1", ""))


def describe(d):
    """What the example does with each event; replace with your own work."""
    data = d.get("data") or {}
    if d["type"] == "conversation.created":
        return f"New conversation {d.get('object_id')} in inbox {data.get('inbox_id')} from contact {data.get('contact_id')}"
    if d["type"] == "message.created":
        # The event carries no message text; read it with an API key:
        # GET /api/conversations/{conversation_id}/messages
        if data.get("direction") == "inbound":
            return f"Customer wrote in conversation {data.get('conversation_id')} (message {d.get('object_id')})"
        return None
    if d["type"] == "csat.submitted":
        if data.get("low"):
            return f"LOW RATING {data.get('rating')} on conversation {data.get('conversation_id')}: {data.get('comment') or '(no comment)'}"
        return f"Rating {data.get('rating')} on conversation {data.get('conversation_id')}"
    return None  # subscribed to more than this example handles


class Receiver:
    """The receiver without the HTTP server, so it can be tested."""

    def __init__(self, secret, process, now=time.time, remember_seconds=48 * 3600):
        self.secret = secret
        self.process = process
        self.now = now
        self.remember = remember_seconds
        # Delivery ids already accepted. In production keep them where every
        # instance sees them (a table with a unique key, Redis SET NX) for at
        # least a day: Kommio retries for about 24 hours.
        self.seen = {}
        self.lock = threading.Lock()
        self.work = queue.Queue()

    def handle(self, headers, raw_body):
        """Returns (status, body). headers: a case-insensitive mapping or a dict with lower-case keys."""
        get = headers.get
        if not verify_signature(self.secret, get("kommio-signature") or get("Kommio-Signature"), raw_body, self.now()):
            return 401, {"error": "bad signature"}
        try:
            d = json.loads(raw_body)
        except ValueError:
            return 400, {"error": "not JSON"}
        delivery = get("kommio-delivery") or get("Kommio-Delivery") or str(d.get("id"))
        with self.lock:
            now = self.now()
            for k in [k for k, at in self.seen.items() if now - at > self.remember]:
                del self.seen[k]
            if delivery in self.seen:
                return 200, {"ok": True, "duplicate": True}
            self.seen[delivery] = now
        # Answer first, work after: a slow handler must not make Kommio retry.
        self.work.put(d)
        return 200, {"ok": True}

    def run_worker(self):
        while True:
            d = self.work.get()
            try:
                self.process(d)
            except Exception as e:  # noqa: BLE001 - keep the worker alive
                print("webhook handler failed", d.get("id"), e)


def serve(port, secret):
    def process(d):
        line = describe(d)
        if line:
            print(f"[{d['type']}] {line}", flush=True)

    receiver = Receiver(secret, process)
    threading.Thread(target=receiver.run_worker, daemon=True).start()

    class Handler(BaseHTTPRequestHandler):
        def do_POST(self):
            if self.path != "/webhooks":
                self.send_error(404)
                return
            length = int(self.headers.get("Content-Length") or 0)
            if length > 1 << 20:
                status, body = 413, {"error": "too large"}
            else:
                status, body = receiver.handle(self.headers, self.rfile.read(length))
            out = json.dumps(body).encode()
            self.send_response(status)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(out)))
            self.end_headers()
            self.wfile.write(out)

        def log_message(self, *args):
            pass

    print(f"webhook receiver on http://127.0.0.1:{port}/webhooks", flush=True)
    ThreadingHTTPServer(("", port), Handler).serve_forever()


if __name__ == "__main__":
    secret = os.environ.get("KOMMIO_WEBHOOK_SECRET", "")
    if not secret:
        raise SystemExit("set KOMMIO_WEBHOOK_SECRET to the endpoint secret Kommio showed (it starts with whsec_)")
    serve(int(os.environ.get("PORT", "4600")), secret)

// Command webhook-receiver receives Kommio webhooks, standard library only:
//
//	KOMMIO_WEBHOOK_SECRET=<endpoint secret> PORT=4600 go run main.go
//
// It verifies the Kommio-Signature header exactly as Kommio signs it,
// answers 200 at once and handles the event on a worker (Kommio waits at
// most 15 seconds and retries anything but a 2xx), and handles each
// delivery once: a retry or a replay carries the same delivery id.
package main

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

// Delivery is the envelope every webhook delivery carries.
type Delivery struct {
	ID         int64          `json:"id"` // the delivery: the same on a retry or a replay
	EventID    int64          `json:"event_id"`
	Type       string         `json:"type"`
	AccountID  int64          `json:"account_id"`
	ObjectType string         `json:"object_type,omitempty"`
	ObjectID   int64          `json:"object_id,omitempty"`
	OccurredAt time.Time      `json:"occurred_at"`
	Data       map[string]any `json:"data"`
	Visitor    map[string]any `json:"visitor,omitempty"`
}

// VerifySignature checks Kommio-Signature: t=<unix seconds>,v1=<hex
// HMAC-SHA256 of "<t>.<raw body>"> with the endpoint's secret (it starts with whsec_,
// used as it is), at most five minutes from now, in constant time.
func VerifySignature(secret, header string, body []byte, now time.Time) bool {
	var ts int64
	var v1 string
	for part := range strings.SplitSeq(header, ",") {
		k, v, _ := strings.Cut(strings.TrimSpace(part), "=")
		switch k {
		case "t":
			ts, _ = strconv.ParseInt(v, 10, 64)
		case "v1":
			v1 = v
		}
	}
	if secret == "" || ts == 0 || v1 == "" {
		return false
	}
	if d := now.Unix() - ts; d > 300 || d < -300 {
		return false
	}
	mac := hmac.New(sha256.New, []byte(secret))
	fmt.Fprintf(mac, "%d.", ts)
	mac.Write(body)
	return hmac.Equal([]byte(hex.EncodeToString(mac.Sum(nil))), []byte(v1))
}

// Describe is what the example does with each event; replace it with your own work.
func Describe(d Delivery) string {
	switch d.Type {
	case "conversation.created":
		return fmt.Sprintf("New conversation %d in inbox %v from contact %v", d.ObjectID, d.Data["inbox_id"], d.Data["contact_id"])
	case "message.created":
		// The event carries no message text; read it with an API key:
		// GET /api/conversations/{conversation_id}/messages
		if d.Data["direction"] == "inbound" {
			return fmt.Sprintf("Customer wrote in conversation %v (message %d)", d.Data["conversation_id"], d.ObjectID)
		}
	case "csat.submitted":
		if low, _ := d.Data["low"].(bool); low {
			return fmt.Sprintf("LOW RATING %v on conversation %v: %v", d.Data["rating"], d.Data["conversation_id"], d.Data["comment"])
		}
		return fmt.Sprintf("Rating %v on conversation %v", d.Data["rating"], d.Data["conversation_id"])
	}
	return "" // subscribed to more than this example handles
}

// Receiver verifies, deduplicates and queues deliveries.
type Receiver struct {
	Secret   string
	Now      func() time.Time
	Remember time.Duration
	Work     chan Delivery

	mu sync.Mutex
	// Delivery ids already accepted. In production keep them where every
	// instance sees them (a table with a unique key, Redis SET NX) for at
	// least a day: Kommio retries for about 24 hours.
	seen map[string]time.Time
}

func NewReceiver(secret string) *Receiver {
	return &Receiver{Secret: secret, Now: time.Now, Remember: 48 * time.Hour, Work: make(chan Delivery, 1024), seen: map[string]time.Time{}}
}

func (r *Receiver) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	body, err := io.ReadAll(io.LimitReader(req.Body, 1<<20+1))
	if err != nil || len(body) > 1<<20 {
		http.Error(w, "too large", http.StatusRequestEntityTooLarge)
		return
	}
	if !VerifySignature(r.Secret, req.Header.Get("Kommio-Signature"), body, r.Now()) {
		http.Error(w, "bad signature", http.StatusUnauthorized)
		return
	}
	var d Delivery
	if err := json.Unmarshal(body, &d); err != nil {
		http.Error(w, "not JSON", http.StatusBadRequest)
		return
	}
	id := req.Header.Get("Kommio-Delivery")
	if id == "" {
		id = strconv.FormatInt(d.ID, 10)
	}
	r.mu.Lock()
	now := r.Now()
	for k, at := range r.seen {
		if now.Sub(at) > r.Remember {
			delete(r.seen, k)
		}
	}
	_, dup := r.seen[id]
	if !dup {
		r.seen[id] = now
	}
	r.mu.Unlock()
	w.Header().Set("Content-Type", "application/json")
	if dup {
		_, _ = io.WriteString(w, `{"ok":true,"duplicate":true}`)
		return
	}
	// Answer first, work after: a slow handler must not make Kommio retry.
	r.Work <- d
	_, _ = io.WriteString(w, `{"ok":true}`)
}

func main() {
	secret := os.Getenv("KOMMIO_WEBHOOK_SECRET")
	if secret == "" {
		log.Fatal("set KOMMIO_WEBHOOK_SECRET to the endpoint secret Kommio showed (it starts with whsec_)")
	}
	port := os.Getenv("PORT")
	if port == "" {
		port = "4600"
	}
	r := NewReceiver(secret)
	go func() {
		for d := range r.Work {
			if line := Describe(d); line != "" {
				log.Printf("[%s] %s", d.Type, line)
			}
		}
	}()
	mux := http.NewServeMux()
	mux.Handle("POST /webhooks", r)
	log.Printf("webhook receiver on http://127.0.0.1:%s/webhooks", port)
	srv := &http.Server{Addr: ":" + port, Handler: mux, ReadHeaderTimeout: 10 * time.Second}
	log.Fatal(srv.ListenAndServe())
}