Webhook Security

const crypto = require('crypto');
 
function verifyWebhook(req, secret, toleranceSec = 300) {
  const sig = req.headers['x-tracktile-signature'];
  const [tPart, v1Part] = sig.split(',');
  const timestamp = parseInt(tPart.replace('t=', ''), 10);
  const signature = v1Part.replace('v1=', '');
 
  // Check timestamp freshness (prevents replay attacks)
  const now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - timestamp) > toleranceSec) {
    throw new Error('Webhook timestamp too old');
  }
 
  // Verify signature
  const payload = `${timestamp}.${req.rawBody}`;
  const expected = crypto
    .createHmac('sha256', secret)
    .update(payload)
    .digest('hex');
 
  if (!crypto.timingSafeEqual(Buffer.from(signature), Buffer.from(expected))) {
    throw new Error('Invalid webhook signature');
  }
 
  return JSON.parse(req.rawBody);
}
 
// Express middleware to capture raw body
app.use('/webhook', express.json({
  verify: (req, res, buf) => {
    req.rawBody = buf.toString();
  }
}));
 
app.post('/webhook', (req, res) => {
  try {
    const event = verifyWebhook(req, process.env.WEBHOOK_SECRET);
    // Process verified event
    res.status(200).send('OK');
  } catch (err) {
    res.status(401).send('Invalid signature');
  }
});
 

 
def verify_webhook(headers, body, secret, tolerance_sec=300):
    sig_header = headers.get('X-Tracktile-Signature', '')
    parts = dict(p.split('=', 1) for p in sig_header.split(','))
 
    timestamp = int(parts['t'])
    signature = parts['v1']
 
    # Check timestamp freshness (prevents replay attacks)
    if abs(time.time() - timestamp) > tolerance_sec:
        raise ValueError('Webhook timestamp too old')
 
    # Verify signature
    payload = f"{timestamp}.{body}"
    expected = hmac.new(
        secret.encode(),
        payload.encode(),
        hashlib.sha256
    ).hexdigest()
 
    if not hmac.compare_digest(signature, expected):
        raise ValueError('Invalid webhook signature')
 
    return json.loads(body)
 
# Flask example
@app.route('/webhook', methods=['POST'])
def handle_webhook():
    try:
        body = request.get_data(as_text=True)
        event = verify_webhook(request.headers, body, WEBHOOK_SECRET)
        # Process verified event
        return 'OK', 200
    except ValueError as e:
        return str(e), 401

package main
 
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
    "fmt"
    "io"
    "math"
    "net/http"
    "strconv"
    "strings"
    "time"
)
 
func verifyWebhook(r *http.Request, secret string, toleranceSec int64) ([]byte, error) {
    sig := r.Header.Get("X-Tracktile-Signature")
    parts := strings.Split(sig, ",")
 
    var timestamp int64
    var signature string
    for _, part := range parts {
        if strings.HasPrefix(part, "t=") {
            timestamp, _ = strconv.ParseInt(strings.TrimPrefix(part, "t="), 10, 64)
        } else if strings.HasPrefix(part, "v1=") {
            signature = strings.TrimPrefix(part, "v1=")
        }
    }
 
    // Check timestamp freshness
    now := time.Now().Unix()
    if math.Abs(float64(now-timestamp)) > float64(toleranceSec) {
        return nil, fmt.Errorf("webhook timestamp too old")
    }
 
    // Read body
    body, err := io.ReadAll(r.Body)
    if err != nil {
        return nil, err
    }
 
    // Verify signature
    payload := fmt.Sprintf("%d.%s", timestamp, string(body))
    mac := hmac.New(sha256.New, []byte(secret))
    mac.Write([]byte(payload))
    expected := hex.EncodeToString(mac.Sum(nil))
 
    if !hmac.Equal([]byte(signature), []byte(expected)) {
        return nil, fmt.Errorf("invalid webhook signature")
    }
 
    return body, nil
 
}