Webhooks

curl -X POST https://api.audital.ai/v1/webhooks \
  -H "Authorization: Bearer ak_live_xxxxxxxxxxxxxxxxxxxx" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://your-app.example.com/webhooks/audital",
    "events": [
      "audit.created",
      "chain.integrity_failure",
      "shadow_ai.detected",
      "evidence.ready",
      "alert.triggered"
    ],
    "description": "Production compliance webhook",
    "active": true
  }'

{
  "id": "wh_01HZABCDEF1234567890ABCD",
  "url": "https://your-app.example.com/webhooks/audital",
  "events": ["audit.created", "chain.integrity_failure", "shadow_ai.detected", "evidence.ready", "alert.triggered"],
  "signingSecret": "whsec_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
  "active": true,
  "createdAt": "2026-03-02T15:00:00.000Z"
}

{
  "id": "we_01HZABCDEF1234567890ABCD",
  "type": "audit.created",
  "apiVersion": "2024-04-10",
  "created": 1741000000,
  "livemode": true,
  "organisationId": "org_xyz789",
  "data": {
    "id": "evt_01HZABCDEF1234567890ABCD",
    "chainPosition": 4822,
    "blockHash": "sha256:9f86d081884c7d659a2feaa0c55ad015a3bf4f1b",
    "timestamp": "2026-03-02T14:22:11.003Z",
    "eventType": "INFERENCE",
    "severity": "INFO",
    "modelId": "mdl_abc123",
    "verified": true
  }
}

const crypto = require('crypto');

function verifyWebhookSignature(payload, signature, secret) {
  // The X-Audital-Signature header format: t=TIMESTAMP,v1=HMAC_HEX
  const [tPart, v1Part] = signature.split(',');
  const timestamp = tPart.split('=')[1];
  const providedHmac = v1Part.split('=')[1];

  // Prevent replay attacks: reject if timestamp is >5 minutes old
  const age = Math.abs(Date.now() / 1000 - parseInt(timestamp));
  if (age > 300) {
    throw new Error('Webhook timestamp too old — possible replay attack');
  }

  // Signed payload: "TIMESTAMP.RAW_BODY"
  const signedPayload = `${timestamp}.${payload}`;
  const expectedHmac = crypto
    .createHmac('sha256', secret)
    .update(signedPayload)
    .digest('hex');

  // Constant-time comparison
  if (!crypto.timingSafeEqual(
    Buffer.from(expectedHmac, 'hex'),
    Buffer.from(providedHmac, 'hex')
  )) {
    throw new Error('Invalid webhook signature');
  }

  return JSON.parse(payload);
}

// Express.js example
app.post('/webhooks/audital', express.raw({ type: 'application/json' }), (req, res) => {
  const sig = req.headers['x-audital-signature'];
  const secret = process.env.AUDITAL_WEBHOOK_SECRET;

  let event;
  try {
    event = verifyWebhookSignature(req.body, sig, secret);
  } catch (err) {
    console.error('Webhook verification failed:', err.message);
    return res.status(400).send(`Webhook Error: ${err.message}`);
  }

  // Handle the event
  switch (event.type) {
    case 'audit.created':
      console.log('New audit event:', event.data.id);
      break;
    case 'chain.integrity_failure':
      console.error('CRITICAL: Chain integrity breach!', event.data);
      break;
    case 'shadow_ai.detected':
      console.warn('Shadow AI detected:', event.data.details.aiProvider);
      break;
    case 'evidence.ready':
      console.log('Evidence package ready:', event.data.id);
      break;
    default:
      console.log('Unhandled event type:', event.type);
  }

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

import hmac
import hashlib
import time
import json
from flask import Flask, request, jsonify, abort

app = Flask(__name__)
WEBHOOK_SECRET = "whsec_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

def verify_signature(payload: bytes, signature: str, secret: str) -> dict:
    parts = {p.split('=')[0]: p.split('=')[1] for p in signature.split(',')}
    timestamp = parts.get('t', '')
    provided_hmac = parts.get('v1', '')

    # Reject stale payloads
    if abs(time.time() - int(timestamp)) > 300:
        raise ValueError("Webhook timestamp too old")

    signed_payload = f"{timestamp}.{payload.decode('utf-8')}"
    expected_hmac = hmac.new(
        secret.encode(),
        signed_payload.encode(),
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(expected_hmac, provided_hmac):
        raise ValueError("Invalid HMAC signature")

    return json.loads(payload)


@app.route('/webhooks/audital', methods=['POST'])
def handle_webhook():
    sig = request.headers.get('X-Audital-Signature', '')
    try:
        event = verify_signature(request.data, sig, WEBHOOK_SECRET)
    except ValueError as e:
        abort(400, str(e))

    if event['type'] == 'chain.integrity_failure':
        # Page on-call immediately
        trigger_pagerduty(event['data'])

    return jsonify(received=True)

# Audital retry schedule for failed webhook deliveries
# An endpoint is considered failed if it returns:
#   - Any non-2xx HTTP status
#   - No response within 30 seconds

# Retry attempts:
# Attempt 1:  Immediately (t+0s)
# Attempt 2:  After 60s   (t+1min)
# Attempt 3:  After 300s  (t+5min)
# Attempt 4:  After 1800s (t+30min)
# Attempt 5:  After 10800s (t+3h)
#
# After 5 failed attempts, the webhook delivery is marked FAILED
# and a notification is sent to the account owner's email.
#
# Webhook endpoints that fail >50% of deliveries over 7 days
# are automatically disabled to prevent resource exhaustion.
# You will receive an email notification before this happens.