Webhooks

import time
import hashlib
import hmac
import json

class VerificationError(Exception):
"""Custom exception to handle verification errors."""
def **init**(self, message="Verification failed", status_code=400):
super().**init**(message)
self.status_code = status_code

def verify_hmac(received_data, received_signature, secret_key): # Ensure the received data is in bytes, necessary for HMAC
if isinstance(received_data, str):
received_data = received_data.encode()

    # Generate the HMAC new object with the secret key and SHA-256
    calculated_signature = hmac.new(secret_key.encode(), received_data, hashlib.sha256).hexdigest()

    # Safely compare the computed HMAC with the received HMAC
    return hmac.compare_digest(calculated_signature, received_signature)

def verify_message(received_data, received_signature, secret_key):
if not received_data:
raise VerificationError("Received request without body", 400)
try:
data_json = json.loads(received_data.decode('utf-8'))
except json.JSONDecodeError as e:
raise VerificationError("Failed to decode json: " + str(e), 400)

    # Check if the expiration is sent
    expires_at = data_json.get("expires_at")
    if not expires_at:
        raise VerificationError("No expiration date sent", 400)

    # Verify HMAC first
    if not verify_hmac(received_data, received_signature, secret_key):
        raise VerificationError("Invalid HMAC signature.", 401)

    # Check if the message is not expired
    if time.time() > expires_at:
        raise VerificationError("Webhook message expired.", 400)

    return data_json

const crypto = require("crypto");

class VerificationError extends Error {
  constructor(message = "Verification failed", statusCode = 400) {
    super(message);
    this.statusCode = statusCode;
    this.name = "VerificationError";
  }
}

function verifyHmac(receivedData, receivedSignature, secretKey) {
  const hmac = crypto.createHmac("sha256", secretKey);
  hmac.update(receivedData);
  const calculatedSignature = hmac.digest("hex");

  // Remove the 'Sha256=' prefix from the receivedSignature before comparing
  const formattedReceivedSignature = receivedSignature.split("=")[1];

  // Ensure both signatures are of equal length and hex encoded
  if (formattedReceivedSignature.length !== calculatedSignature.length) {
    return false;
  }

  return crypto.timingSafeEqual(
    Buffer.from(calculatedSignature, "hex"),
    Buffer.from(formattedReceivedSignature, "hex")
  );
}

function verifyMessage(receivedData, receivedSignature, secretKey) {
  if (!receivedData) {
    throw new VerificationError("Received request without body", 400);
  }

  let dataJson;
  try {
    dataJson = JSON.parse(receivedData.toString("utf-8"));
  } catch (error) {
    throw new VerificationError("Failed to decode json: " + error.message, 400);
  }

  const expiresAt = dataJson.expires_at;
  if (!expiresAt) {
    throw new VerificationError("No expiration date sent", 400);
  }

  if (!verifyHmac(receivedData, receivedSignature, secretKey)) {
    console.log(
      "Invalid HMAC signature.",
      receivedData,
      receivedSignature,
      secretKey
    );
    throw new VerificationError("Invalid HMAC signature.", 401);
  }

  if (Date.now() / 1000 > expiresAt) {
    throw new VerificationError("Webhook message expired.", 400);
  }

  return dataJson;
}