HIPAA Conversation Webhook

const crypto = require('node:crypto')
const express = require('express')

const app = express()
const SECRET = process.env.CHATBASE_HIPAA_WEBHOOK_SECRET
const MAX_AGE_SECONDS = 300

app.post(
    '/webhooks/chatbase-hipaa',
    // Raw body, NOT express.json() — the signature covers the exact bytes sent.
    express.raw({type: 'application/json'}),
    (req, res) => {
        const timestamp = req.get('X-Chatbase-Timestamp')
        const signature = req.get('X-Chatbase-Signature')

        // Reject missing headers explicitly, so they don't look like a bad signature.
        if (!timestamp || !signature) {
            return res.status(400).json({error: 'missing_headers'})
        }

        // Replay protection: the timestamp is signed, but only checking it makes
        // the protection real.
        const age = Math.abs(Date.now() / 1000 - Number(timestamp))
        if (!Number.isFinite(age) || age > MAX_AGE_SECONDS) {
            return res.status(400).json({error: 'stale_timestamp'})
        }

        // Feed the raw Buffer straight in, so the bytes are never transcoded.
        const hmac = crypto.createHmac('sha256', SECRET)
        hmac.update(`${timestamp}.`)
        hmac.update(req.body)
        const expected = `v1=${hmac.digest('hex')}`

        // timingSafeEqual throws when lengths differ, so check length first.
        const received = Buffer.from(signature, 'utf8')
        const computed = Buffer.from(expected, 'utf8')
        if (
            received.length !== computed.length ||
            !crypto.timingSafeEqual(received, computed)
        ) {
            return res.status(401).json({error: 'invalid_signature'})
        }

        // Verified — now it is safe to parse and trust the payload.
        const event = JSON.parse(req.body.toString('utf8'))

        // Verification probes carry test: true. Acknowledge without storing.
        if (event.test) {
            return res.status(200).json({ok: true})
        }

        // Acknowledge first, then process. Chatbase times out after 10 seconds.
        res.status(200).json({ok: true})
        queueForProcessing(event)
    }
)

import crypto from 'node:crypto'

const SECRET = process.env.CHATBASE_HIPAA_WEBHOOK_SECRET!
const MAX_AGE_SECONDS = 300

export async function POST(request: Request) {
    // Read the raw body as text, NOT request.json() — the signature covers the
    // exact bytes sent.
    const rawBody = await request.text()

    const timestamp = request.headers.get('x-chatbase-timestamp')
    const signature = request.headers.get('x-chatbase-signature')

    // Reject missing headers explicitly, so they don't look like a bad signature.
    if (!timestamp || !signature) {
        return Response.json({error: 'missing_headers'}, {status: 400})
    }

    // Replay protection: the timestamp is signed, but only checking it makes
    // the protection real.
    const age = Math.abs(Date.now() / 1000 - Number(timestamp))
    if (!Number.isFinite(age) || age > MAX_AGE_SECONDS) {
        return Response.json({error: 'stale_timestamp'}, {status: 400})
    }

    const expected = `v1=${crypto
        .createHmac('sha256', SECRET)
        .update(`${timestamp}.${rawBody}`)
        .digest('hex')}`

    // timingSafeEqual throws when lengths differ, so check length first.
    const received = Buffer.from(signature, 'utf8')
    const computed = Buffer.from(expected, 'utf8')
    if (
        received.length !== computed.length ||
        !crypto.timingSafeEqual(received, computed)
    ) {
        return Response.json({error: 'invalid_signature'}, {status: 401})
    }

    // Verified — now it is safe to parse and trust the payload.
    const event = JSON.parse(rawBody)

    // Verification probes carry test: true. Acknowledge without storing.
    if (event.test) {
        return Response.json({ok: true})
    }

    // Keep this fast — Chatbase times out after 10 seconds.
    await queueForProcessing(event)
    return Response.json({ok: true})
}

import hashlib
import hmac
import json
import os
import time

from flask import Flask, jsonify, request

app = Flask(__name__)
SECRET = os.environ["CHATBASE_HIPAA_WEBHOOK_SECRET"].encode()
MAX_AGE_SECONDS = 300


@app.post("/webhooks/chatbase-hipaa")
def chatbase_hipaa_webhook():
    # Raw bytes, NOT request.get_json() — the signature covers the exact bytes sent.
    raw_body = request.get_data()

    timestamp = request.headers.get("X-Chatbase-Timestamp")
    signature = request.headers.get("X-Chatbase-Signature")

    # Reject missing headers explicitly, so they don't look like a bad signature.
    if not timestamp or not signature:
        return jsonify(error="missing_headers"), 400

    # Replay protection: the timestamp is signed, but only checking it makes
    # the protection real.
    try:
        age = abs(time.time() - int(timestamp))
    except ValueError:
        return jsonify(error="stale_timestamp"), 400
    if age > MAX_AGE_SECONDS:
        return jsonify(error="stale_timestamp"), 400

    signed_input = f"{timestamp}.".encode() + raw_body
    expected = "v1=" + hmac.new(SECRET, signed_input, hashlib.sha256).hexdigest()

    # compare_digest is the constant-time comparison.
    if not hmac.compare_digest(signature, expected):
        return jsonify(error="invalid_signature"), 401

    # Verified — now it is safe to parse and trust the payload.
    event = json.loads(raw_body)

    # Verification probes carry test: true. Acknowledge without storing.
    if event.get("test"):
        return jsonify(ok=True), 200

    # Keep this fast — Chatbase times out after 10 seconds.
    queue_for_processing(event)
    return jsonify(ok=True), 200