Signature Verification

All ATHENA webhooks are signed using HMAC-SHA256. Always verify signatures to prevent spoofed events.

Signature Header

Every webhook includes a signature header:

X-Athena-Signature: t=1703505600,v1=5257a869e7ecebeda32affa62cdca3fa51cad7e77a0e56ff536d0ce8e108d8bd

Components:

• t — Unix timestamp when signature was generated

• v1 — HMAC-SHA256 signature

Verification Steps

1. Extract Components

const header = req.headers['x-athena-signature'];
const [timestampPart, signaturePart] = header.split(',');
const timestamp = parseInt(timestampPart.split('=')[1]);
const signature = signaturePart.split('=')[1];

2. Check Timestamp

Reject events older than 5 minutes (prevents replay attacks):

const now = Math.floor(Date.now() / 1000);
const tolerance = 300; // 5 minutes

if (now - timestamp > tolerance) {
  throw new Error('Webhook timestamp too old');
}

3. Compute Expected Signature

const crypto = require('crypto');

const payload = `${timestamp}.${req.body}`;
const expected = crypto
  .createHmac('sha256', WEBHOOK_SECRET)
  .update(payload)
  .digest('hex');

4. Compare Signatures

Use constant-time comparison to prevent timing attacks:

const crypto = require('crypto');

if (!crypto.timingSafeEqual(
  Buffer.from(signature),
  Buffer.from(expected)
)) {
  throw new Error('Invalid webhook signature');
}

Complete Examples

JavaScript (Express)

const crypto = require('crypto');
const express = require('express');
const app = express();

// Use raw body for signature verification
app.post('/webhook', express.text({ type: '*/*' }), (req, res) => {
  try {
    verifySignature(req.body, req.headers['x-athena-signature']);
    
    const event = JSON.parse(req.body);
    console.log('Event:', event.event);
    
    res.sendStatus(200);
  } catch (error) {
    console.error('Verification failed:', error.message);
    res.sendStatus(401);
  }
});

function verifySignature(payload, header) {
  const [timestampPart, signaturePart] = header.split(',');
  const timestamp = parseInt(timestampPart.split('=')[1]);
  const signature = signaturePart.split('=')[1];
  
  // Check timestamp (5 min tolerance)
  const now = Math.floor(Date.now() / 1000);
  if (now - timestamp > 300) {
    throw new Error('Timestamp too old');
  }
  
  // Compute expected signature
  const expected = crypto
    .createHmac('sha256', process.env.WEBHOOK_SECRET)
    .update(`${timestamp}.${payload}`)
    .digest('hex');
  
  // Constant-time comparison
  if (!crypto.timingSafeEqual(
    Buffer.from(signature),
    Buffer.from(expected)
  )) {
    throw new Error('Invalid signature');
  }
}

Python (Flask)

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

app = Flask(__name__)
WEBHOOK_SECRET = os.environ['WEBHOOK_SECRET']

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    try:
        verify_signature(
            request.data.decode(),
            request.headers['X-Athena-Signature']
        )
        
        event = request.get_json()
        print(f"Event: {event['event']}")
        
        return '', 200
    except Exception as e:
        print(f"Verification failed: {e}")
        abort(401)

def verify_signature(payload: str, header: str):
    parts = dict(p.split('=') for p in header.split(','))
    timestamp = int(parts['t'])
    signature = parts['v1']
    
    # Check timestamp (5 min tolerance)
    if time.time() - timestamp > 300:
        raise ValueError('Timestamp too old')
    
    # Compute expected signature
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        f"{timestamp}.{payload}".encode(),
        hashlib.sha256
    ).hexdigest()
    
    # Constant-time comparison
    if not hmac.compare_digest(signature, expected):
        raise ValueError('Invalid signature')

Python (FastAPI)

from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib
import time
import os

app = FastAPI()
WEBHOOK_SECRET = os.environ['WEBHOOK_SECRET']

@app.post('/webhook')
async def handle_webhook(request: Request):
    body = await request.body()
    header = request.headers.get('X-Athena-Signature')
    
    try:
        verify_signature(body.decode(), header)
    except ValueError as e:
        raise HTTPException(status_code=401, detail=str(e))
    
    event = await request.json()
    print(f"Event: {event['event']}")
    
    return {'status': 'ok'}

def verify_signature(payload: str, header: str):
    parts = dict(p.split('=') for p in header.split(','))
    timestamp = int(parts['t'])
    signature = parts['v1']
    
    if time.time() - timestamp > 300:
        raise ValueError('Timestamp too old')
    
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        f"{timestamp}.{payload}".encode(),
        hashlib.sha256
    ).hexdigest()
    
    if not hmac.compare_digest(signature, expected):
        raise ValueError('Invalid signature')

Using the SDK

Both SDKs include built-in verification:

JavaScript

import { verifyWebhookSignature } from '@athena-ai/sdk';

app.post('/webhook', express.text({ type: '*/*' }), async (req, res) => {
  try {
    await verifyWebhookSignature(
      req.body,
      req.headers['x-athena-signature'],
      process.env.WEBHOOK_SECRET
    );
    res.sendStatus(200);
  } catch {
    res.sendStatus(401);
  }
});

Python

from athena.utils import verify_webhook_signature

@app.route('/webhook', methods=['POST'])
@athena_webhook(secret=os.environ['WEBHOOK_SECRET'])
def handle_webhook():
    # Signature already verified by decorator
    return {'status': 'ok'}

Troubleshooting

Error	Cause	Solution
Timestamp too old	Clock skew or replay attack	Sync server clock, check NTP
Invalid signature	Wrong secret or tampered payload	Verify secret matches webhook
Missing header	Spoofed request	Reject requests without signature

Secret Rotation

When rotating secrets, both old and new secrets are valid for 24 hours:

curl -X POST https://api.athenatrust.ai/v1/webhooks/:id/rotate-secret \
  -H "Authorization: Bearer YOUR_API_KEY"

During the grace period, try verifying with both secrets:

function verifySignature(payload, header) {
  try {
    verify(payload, header, NEW_SECRET);
  } catch {
    verify(payload, header, OLD_SECRET);
  }
}

---

Next: Compliance Guides