Signature Verification

const crypto = require('crypto');

function verifyWebhookSignature(payload, signature, secret) {
  const expectedSignature = 'sha256=' +
    crypto.createHmac('sha256', secret)
      .update(JSON.stringify(payload))
      .digest('hex');

  return crypto.timingSafeEqual(
    Buffer.from(signature),
    Buffer.from(expectedSignature)
  );
}

// Express.js webhook handler
app.post('/webhook', (req, res) => {
  const isValid = verifyWebhookSignature(
    req.body,
    req.headers['x-m2p-signature'],
    process.env.WEBHOOK_SECRET
  );

  if (!isValid) {
    return res.status(401).send('Invalid signature');
  }

  // Process the event
  handleEvent(req.body);
  res.status(200).send('OK');
});

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.security.MessageDigest;

public class WebhookVerifier {

    public static boolean verifySignature(
        String payload, String signature, String secret
    ) throws Exception {
        Mac mac = Mac.getInstance("HmacSHA256");
        mac.init(new SecretKeySpec(
            secret.getBytes(), "HmacSHA256"
        ));

        String expected = "sha256=" +
            bytesToHex(mac.doFinal(payload.getBytes()));

        return MessageDigest.isEqual(
            expected.getBytes(),
            signature.getBytes()
        );
    }

    private static String bytesToHex(byte[] bytes) {
        StringBuilder sb = new StringBuilder();
        for (byte b : bytes) {
            sb.append(String.format("%02x", b));
        }
        return sb.toString();
    }
}

import hmac
import hashlib

def verify_webhook_signature(payload: bytes,
                              signature: str,
                              secret: str) -> bool:
    expected = 'sha256=' + hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(expected, signature)

# Flask webhook handler
@app.route('/webhook', methods=['POST'])
def webhook():
    is_valid = verify_webhook_signature(
        request.data,
        request.headers.get('x-m2p-signature', ''),
        os.environ['WEBHOOK_SECRET']
    )

    if not is_valid:
        return 'Invalid signature', 401

    handle_event(request.json)
    return 'OK', 200