Signature Verification

import crypto from 'crypto'

function verifyWebhookSignature(req, secret) {
  const msgId = req.headers['svix-id']
  const timestamp = req.headers['svix-timestamp']
  const signature = req.headers['svix-signature']

  // Check timestamp to prevent replay attacks (5 min tolerance)
  const now = Math.floor(Date.now() / 1000)
  if (Math.abs(now - parseInt(timestamp)) > 300) {
    throw new Error('Timestamp too old')
  }

  // Decode the secret (remove whsec_ prefix, base64url decode)
  const secretBytes = Buffer.from(secret.replace('whsec_', ''), 'base64url')

  // Build the signed content
  const signedContent = `${msgId}.${timestamp}.${req.body}`

  // Compute expected signature
  const expected = crypto
    .createHmac('sha256', secretBytes)
    .update(signedContent)
    .digest('base64')

  // Extract actual signature (remove v1, prefix)
  const actual = signature.split(',')[1]

  // Constant-time comparison
  const expectedBytes = Buffer.from(expected)
  const actualBytes = Buffer.from(actual)
  if (expectedBytes.length !== actualBytes.length || !crypto.timingSafeEqual(expectedBytes, actualBytes)) {
    throw new Error('Invalid signature')
  }

  return JSON.parse(req.body)
}

// Express route
app.post('/webhooks/agentref', express.raw({ type: 'application/json' }), (req, res) => {
  try {
    const event = verifyWebhookSignature(req, process.env.AGENTREF_WEBHOOK_SECRET)
    console.log('Verified event:', event.type)
    res.status(200).send('OK')
  } catch (err) {
    console.error('Verification failed:', err.message)
    res.status(400).send('Invalid signature')
  }
})

import hmac
import hashlib
import base64
import json
import os
import time
from flask import Flask, request

app = Flask(__name__)

def verify_webhook_signature(payload, headers, secret):
    msg_id = headers.get("svix-id")
    timestamp = headers.get("svix-timestamp")
    signature = headers.get("svix-signature")

    # Check timestamp (5 min tolerance)
    now = int(time.time())
    if abs(now - int(timestamp)) > 300:
        raise ValueError("Timestamp too old")

    # Decode secret (remove whsec_ prefix, base64url decode)
    secret_bytes = base64.urlsafe_b64decode(secret.removeprefix("whsec_") + "==")

    # Build signed content
    signed_content = f"{msg_id}.{timestamp}.{payload}"

    # Compute expected signature
    expected = base64.b64encode(
        hmac.new(secret_bytes, signed_content.encode(), hashlib.sha256).digest()
    ).decode()

    # Extract actual (remove v1, prefix)
    actual = signature.split(",")[1]

    if not hmac.compare_digest(expected, actual):
        raise ValueError("Invalid signature")

    return json.loads(payload)

@app.route("/webhooks/agentref", methods=["POST"])
def webhook():
    try:
        event = verify_webhook_signature(
            request.get_data(as_text=True),
            request.headers,
            os.environ["AGENTREF_WEBHOOK_SECRET"],
        )
        print(f"Verified event: {event['type']}")
        return "OK", 200
    except ValueError as e:
        return str(e), 400