Webhook Events

import crypto from 'crypto';

function verifyWebhookSignature(
  payload: string,
  signature: string,
  secret: string
): boolean {
  const expectedSig = crypto
    .createHmac('sha256', secret)
    .update(payload, 'utf8')
    .digest('hex');

  const expectedHeader = `sha256=${expectedSig}`;

  // Use timing-safe comparison to prevent timing attacks
  return crypto.timingSafeEqual(
    Buffer.from(signature),
    Buffer.from(expectedHeader)
  );
}

// Express handler example
app.post('/webhooks/casehug', express.raw({ type: 'application/json' }), (req, res) => {
  const signature = req.headers['x-casehug-signature'] as string;
  const rawBody = req.body.toString('utf8');

  if (!verifyWebhookSignature(rawBody, signature, process.env.WEBHOOK_SECRET!)) {
    return res.status(401).json({ error: 'Invalid signature' });
  }

  const event = JSON.parse(rawBody);

  switch (event.type) {
    case 'matter.created':
      await handleMatterCreated(event.data);
      break;
    case 'document.uploaded':
      await handleDocumentUploaded(event.data);
      break;
    case 'intake.completed':
      await handleIntakeCompleted(event.data);
      break;
  }

  res.status(200).json({ received: true });
});

import hmac
import hashlib
import json

def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
    expected = hmac.new(
        secret.encode('utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()
    expected_header = f"sha256={expected}"
    return hmac.compare_digest(signature, expected_header)


# Flask handler example
from flask import Flask, request, abort

app = Flask(__name__)

@app.route('/webhooks/casehug', methods=['POST'])
def handle_casehug_webhook():
    signature = request.headers.get('X-CaseHug-Signature', '')
    raw_body = request.get_data()

    if not verify_webhook_signature(raw_body, signature, WEBHOOK_SECRET):
        abort(401, 'Invalid signature')

    event = json.loads(raw_body)
    event_type = event.get('type')

    if event_type == 'matter.created':
        handle_matter_created(event['data'])
    elif event_type == 'document.uploaded':
        handle_document_uploaded(event['data'])
    elif event_type == 'intake.completed':
        handle_intake_completed(event['data'])

    return {'received': True}, 200