Validate legacy webhook signatures

const crypto = require('node:crypto')
const express = require('express')

const app = express()

// The signing key from "Reveal signing secret" in the Quo app.
const signingKey = Buffer.from(process.env.QUO_LEGACY_SIGNING_KEY ?? '', 'base64')
const toleranceMs = 5 * 60 * 1000

// Keep the raw body so the signed bytes are exactly what Quo sent.
app.post('/webhooks/quo', express.raw({ type: 'application/json' }), (req, res) => {
  const header = req.get('openphone-signature') ?? ''
  const [scheme, version, timestamp, providedDigest] = header.split(';')

  if (scheme !== 'hmac' || version !== '1' || !timestamp || !providedDigest) {
    return res.status(401).send('Invalid signature header')
  }

  if (Math.abs(Date.now() - Number(timestamp)) > toleranceMs) {
    return res.status(401).send('Stale timestamp')
  }

  const signedData = Buffer.concat([Buffer.from(`${timestamp}.`), req.body])
  const computedDigest = crypto.createHmac('sha256', signingKey).update(signedData).digest('base64')

  const provided = Buffer.from(providedDigest)
  const computed = Buffer.from(computedDigest)

  if (provided.length !== computed.length || !crypto.timingSafeEqual(provided, computed)) {
    return res.status(401).send('Invalid signature')
  }

  const event = JSON.parse(req.body.toString('utf8'))
  // Process the event here.

  res.sendStatus(200)
})

app.listen(8000)

import base64
import hashlib
import hmac
import os
import time

from flask import Flask, request

app = Flask(__name__)

# The signing key from "Reveal signing secret" in the Quo app.
SIGNING_KEY = base64.b64decode(os.environ.get('QUO_LEGACY_SIGNING_KEY', ''))
TOLERANCE_MS = 5 * 60 * 1000


@app.route('/webhooks/quo', methods=['POST'])
def handle_webhook():
    fields = request.headers.get('openphone-signature', '').split(';')
    if len(fields) != 4 or fields[0] != 'hmac' or fields[1] != '1':
        return 'Invalid signature header', 401

    timestamp, provided_digest = fields[2], fields[3]

    if not timestamp.isdigit() or abs(time.time() * 1000 - int(timestamp)) > TOLERANCE_MS:
        return 'Stale timestamp', 401

    # request.get_data() returns the raw body bytes.
    signed_data = timestamp.encode() + b'.' + request.get_data()
    digest = hmac.new(SIGNING_KEY, signed_data, hashlib.sha256).digest()
    computed_digest = base64.b64encode(digest).decode()

    if not hmac.compare_digest(provided_digest, computed_digest):
        return 'Invalid signature', 401

    event = request.get_json()
    # Process the event here.

    return '', 200


if __name__ == '__main__':
    app.run(port=8000)