Webhooks

{
  "id":         "evt_a1b2c3d4",
  "type":       "email.received",
  "created_at": "2025-01-15T14:22:00Z",
  "data": {
    // varies by event type — see examples below
  }
}

{
  "id":         "evt_a1b2c3d4",
  "type":       "email.received",
  "created_at": "2025-01-15T14:22:00Z",
  "data": {
    "message_id":    "msg_abc123",
    "mailbox_id":    "mbx_xyz789",
    "mailbox_email": "support@yourdomain.com",
    "from":          "customer@example.com",
    "to":            ["support@yourdomain.com"],
    "subject":       "Help with my order",
    "snippet":       "Hi, I placed an order yesterday and...",
    "has_attachments": false,
    "received_at":   "2025-01-15T14:21:59Z"
  }
}

{
  "id":         "evt_b9c8d7e6",
  "type":       "email.bounced",
  "created_at": "2025-01-15T15:00:10Z",
  "data": {
    "message_id":    "msg_sent001",
    "recipient":     "nobody@invalid-domain.tld",
    "bounce_type":   "hard",
    "bounce_code":   "550",
    "bounce_message": "5.1.1 The email account that you tried to reach does not exist."
  }
}

const crypto = require('crypto');

function verifyWebhookSignature(req, webhookSecret) {
  const signature = req.headers['x-mailpipe-signature'];
  if (!signature) return false;

  // Compute expected signature from raw body
  const expected = crypto
    .createHmac('sha256', webhookSecret)
    .update(req.rawBody) // use the raw unparsed body buffer
    .digest('hex');

  // Use timingSafeEqual to prevent timing attacks
  const sigBuffer  = Buffer.from(signature, 'hex');
  const expBuffer  = Buffer.from(expected, 'hex');

  return (
    sigBuffer.length === expBuffer.length &&
    crypto.timingSafeEqual(sigBuffer, expBuffer)
  );
}

// Express example
app.post('/webhooks/mailpipe', (req, res) => {
  if (!verifyWebhookSignature(req, process.env.MAILPIPE_WEBHOOK_SECRET)) {
    return res.status(401).json({ error: 'Invalid signature' });
  }
  const event = req.body;
  console.log('Received event:', event.type);
  res.status(200).json({ received: true });
});

import hmac
import hashlib

def verify_webhook_signature(body: bytes, signature: str, secret: str) -> bool:
    expected = hmac.new(
        secret.encode('utf-8'),
        body,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

# Flask example
from flask import Flask, request, abort

app = Flask(__name__)

@app.route('/webhooks/mailpipe', methods=['POST'])
def webhook():
    signature = request.headers.get('X-Mailpipe-Signature', '')
    if not verify_webhook_signature(request.data, signature, WEBHOOK_SECRET):
        abort(401)
    event = request.json
    print('Received event:', event['type'])
    return {'received': True}, 200