Webhooks

import { createHmac, timingSafeEqual } from 'node:crypto';

/** Pass the original HTTP body, before JSON parsing, and the literal secret. */
export function verifyMelsoWebhook(
  raw: Buffer,
  signature: string,
  secret: string,
  now = Math.floor(Date.now() / 1000),
  tolerance = 300,
): boolean {
  if (!secret) return false;
  const timestamps: string[] = [];
  const signatures: string[] = [];
  for (const part of signature.split(',')) {
    const match = /^([^=]+)=(.*)$/.exec(part.trim());
    if (!match) return false;
    if (match[1] === 't') timestamps.push(match[2]);
    if (match[1] === 'v1') signatures.push(match[2]);
  }
  if (timestamps.length !== 1 || !/^\d+$/.test(timestamps[0])) return false;
  const timestamp = Number(timestamps[0]);
  if (!Number.isSafeInteger(timestamp) || Math.abs(now - timestamp) > tolerance) return false;
  const expected = createHmac('sha256', secret).update(timestamps[0] + '.').update(raw).digest();
  return signatures.some(value => /^[a-f0-9]{64}$/i.test(value) && timingSafeEqual(expected, Buffer.from(value, 'hex')));
}

import hashlib, hmac, re, time

def verify(raw:bytes,signature:str,secret:str,now:int|None=None,tolerance:int=300)->bool:
    if not secret:return False
    parts=signature.split(',');timestamps=[];digests=[]
    for part in parts:
        key,sep,value=part.strip().partition('=')
        if not sep:return False
        if key=='t':timestamps.append(value)
        elif key=='v1':digests.append(value)
    if len(timestamps)!=1 or not re.fullmatch(r'[0-9]{1,20}',timestamps[0]):return False
    stamp=timestamps[0];now=int(time.time()) if now is None else now
    if abs(now-int(stamp))>tolerance:return False
    expected=hmac.new(secret.encode(),stamp.encode()+b'.'+raw,hashlib.sha256).hexdigest()
    return any(re.fullmatch(r'[a-fA-F0-9]{64}',d) and hmac.compare_digest(expected,d.lower()) for d in digests)

const raw = Buffer.from("{\"id\":\"11111111-1111-4111-8111-111111111111\",\"type\":\"task.delivered\"}", 'utf8');
const secret = "whsec_documentation_test_only";
const signature = "t=1800000000,v1=d7dffdbfbd0b3a34eeed08ae82ad32776a90ed6587d92e421637c1186822f534";
if (!verifyMelsoWebhook(raw, signature, secret, 1800000000)) throw new Error('Valid vector rejected');
if (verifyMelsoWebhook(Buffer.concat([raw, Buffer.from('\n')]), signature, secret, 1800000000)) throw new Error('Changed body accepted');
if (verifyMelsoWebhook(raw, signature, secret, 1800000301)) throw new Error('Expired vector accepted');
console.log('Signature vector passed');

raw = b'{"id":"11111111-1111-4111-8111-111111111111","type":"task.delivered"}'
secret = 'whsec_documentation_test_only'
signature = 't=1800000000,v1=d7dffdbfbd0b3a34eeed08ae82ad32776a90ed6587d92e421637c1186822f534'
assert verify(raw, signature, secret, now=1800000000)
assert not verify(raw + b'\n', signature, secret, now=1800000000)
assert not verify(raw, signature, secret, now=1800000301)
print('Signature vector passed')