Webhooks

// node webhook.mjs (no dependencies)
import http from 'node:http';
import crypto from 'node:crypto';

const SECRET = process.env.GUIDINGHAND_WEBHOOK_SECRET; // whsec_…
const TOLERANCE_S = 300; // 5 minutes

function verify(rawBody, header, secret) {
  const parts = String(header ?? '').split(',').map((p) => p.split('='));
  const t = parts.find(([k]) => k === 't')?.[1];
  const sigs = parts.filter(([k]) => k === 'v1').map(([, v]) => v);
  if (!/^\d+$/.test(t ?? '') || sigs.length === 0) return false;
  if (Math.abs(Date.now() / 1000 - Number(t)) > TOLERANCE_S) return false;
  const expected = crypto.createHmac('sha256', secret).update(`${t}.`).update(rawBody).digest();
  return sigs.some((s) => {
    const got = Buffer.from(s, 'hex');
    return got.length === expected.length && crypto.timingSafeEqual(got, expected);
  });
}

http.createServer((req, res) => {
  const chunks = [];
  req.on('data', (c) => chunks.push(c));
  req.on('end', () => {
    const raw = Buffer.concat(chunks);
    if (!verify(raw, req.headers['guidinghand-signature'], SECRET)) {
      res.writeHead(400).end('bad signature');
      return;
    }
    res.writeHead(200).end('ok');

    const event = JSON.parse(raw);
    if (event.type === 'task.waiting_for_user') {
      const { task } = event.data;
      console.log(`Task ${task.task_id} asks: ${task.pending.question}`);
    }
    if (event.type === 'task.question_answered' && event.data.answer.answered_by === 'customer') {
      console.log(`The customer answered on their screen: ${event.data.answer.answer}`);
    }
    if (event.type === 'task.approval_decided' && event.data.decision.answered_by === 'customer') {
      console.log(`The customer chose to ${event.data.decision.decision} on their screen`);
    }
  });
}).listen(3000);

# pip install flask && flask --app webhook run --port 3000
import hashlib
import hmac
import os
import time

from flask import Flask, abort, request

SECRET = os.environ["GUIDINGHAND_WEBHOOK_SECRET"]  # whsec_…
TOLERANCE_S = 300  # 5 minutes


def verify(raw: bytes, header: str, secret: str) -> bool:
    parts = [p.split("=", 1) for p in (header or "").split(",") if "=" in p]
    t = next((v for k, v in parts if k == "t"), "")
    sigs = [v for k, v in parts if k == "v1"]
    if not t.isdigit() or not sigs:
        return False
    if abs(time.time() - int(t)) > TOLERANCE_S:
        return False
    expected = hmac.new(secret.encode(), t.encode() + b"." + raw, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected.encode(), s.encode()) for s in sigs)


app = Flask(__name__)


@app.post("/guidinghand")
def guidinghand():
    raw = request.get_data()
    if not verify(raw, request.headers.get("GuidingHand-Signature"), SECRET):
        abort(400)

    event = request.get_json()
    if event["type"] == "task.waiting_for_user":
        task = event["data"]["task"]
        print(f"Task {task['task_id']} asks: {task['pending']['question']}")
    if event["type"] == "task.question_answered" and event["data"]["answer"]["answered_by"] == "customer":
        print(f"The customer answered on their screen: {event['data']['answer']['answer']}")
    if event["type"] == "task.approval_decided" and event["data"]["decision"]["answered_by"] == "customer":
        print(f"The customer chose to {event['data']['decision']['decision']} on their screen")
    return "ok", 200