Webhooks

import crypto from 'node:crypto';
import express from 'express';

function verifyAthosSignature(rawBody: string, header: string | undefined, secret: string): boolean {
  if (!header) return false;
  const parts = Object.fromEntries(header.split(',').map((kv) => kv.split('=')));
  const { t, v1 } = parts;
  if (!t || !v1) return false;
  if (Math.abs(Date.now() / 1000 - Number(t)) > 300) return false; // 5-min replay window
  const expected = crypto.createHmac('sha256', secret).update(`${t}.${rawBody}`).digest('hex');
  return v1.length === expected.length &&
    crypto.timingSafeEqual(Buffer.from(v1, 'hex'), Buffer.from(expected, 'hex'));
}

// Mount with a RAW body parser so the bytes are untouched.
app.post('/athos/webhook', express.raw({ type: 'application/json' }), (req, res) => {
  const raw = req.body.toString('utf8');
  if (!verifyAthosSignature(raw, req.header('X-Athos-Signature'), process.env.ATHOS_WHSEC!)) {
    return res.sendStatus(401);
  }

  res.sendStatus(200); // ACK fast — do the work afterwards

  const evt = JSON.parse(raw);
  if (!alreadyProcessed(req.header('X-Athos-Webhook-Id'))) {
    if (evt.success) void fetchAndStoreScore(evt.callId);
    else logScoringFailure(evt.callId, evt.reason, evt.reasonMessage);
  }
});

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

def verify_athos_signature(raw_body: bytes, header: str, secret: str) -> bool:
    if not header:
        return False
    parts = dict(p.split("=", 1) for p in header.split(","))
    t, v1 = parts.get("t"), parts.get("v1")
    if not t or not v1:
        return False
    if abs(time.time() - int(t)) > 300:           # 5-min replay window
        return False
    expected = hmac.new(secret.encode(), f"{t}.".encode() + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, v1)

@app.post("/athos/webhook")
def athos_webhook():
    raw = request.get_data()                       # RAW bytes — do not use request.json first
    if not verify_athos_signature(raw, request.headers.get("X-Athos-Signature", ""), os.environ["ATHOS_WHSEC"]):
        return "", 401

    evt = request.get_json()
    if not already_processed(request.headers.get("X-Athos-Webhook-Id")):
        if evt["success"]:
            enqueue_fetch_score(evt["callId"])     # do heavy work async
        else:
            log_scoring_failure(evt["callId"], evt.get("reason"), evt.get("reasonMessage"))
    return "", 200                                 # ACK fast