{
  "notification_id": "550e8400-e29b-41d4-a716-446655440000",
  "action_id": "approve",
  "response_data": null,
  "responded_at": "2025-05-25T10:35:12Z",
  "responder": {
    "id": "user_123",
    "type": "human"
  }
}

X-ATP-Signature: t=1622145123,v1=5257a869e7ecebb07c954661a7577d13c99705955e814274785a3174a535d5c7

const crypto = require('crypto');

function verifySignature(body, signature, secret) {
  // Extract timestamp and signature from header
  const [timeComponent, signatureComponent] = signature.split(',');
  const timestamp = timeComponent.split('=')[1];
  const signatureValue = signatureComponent.split('=')[1];
  
  // Check if timestamp is within acceptable range (5 minutes)
  const currentTime = Math.floor(Date.now() / 1000);
  if (currentTime - parseInt(timestamp) > 300) {
    return false; // Signature too old, possible replay attack
  }
  
  // Compute expected signature
  const expectedSignature = crypto
    .createHmac('sha256', secret)
    .update(body)
    .digest('hex');
  
  // Compare signatures using constant-time comparison
  return crypto.timingSafeEqual(
    Buffer.from(signatureValue, 'hex'),
    Buffer.from(expectedSignature, 'hex')
  );
}

import hmac
import hashlib
import time

def verify_signature(payload_body, signature_header, secret):
    # Parse the signature header
    header_parts = signature_header.split(',')
    timestamp_str = header_parts[0].split('=')[1]
    signature = header_parts[1].split('=')[1]
    
    # Check if signature is recent (within 5 minutes)
    timestamp = int(timestamp_str)
    current_time = int(time.time())
    if current_time - timestamp > 300:
        return False  # Signature too old
    
    # Compute expected signature
    expected_signature = hmac.new(
        secret.encode('utf-8'),
        payload_body.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()
    
    # Compare signatures using constant-time comparison
    return hmac.compare_digest(signature, expected_signature)

{
  "code": "RESOURCE_LOCKED",
  "message": "Cannot apply changes because resource is currently locked",
  "user_message": "The system is currently processing another change. Please try again in a few moments.",
  "retriable": true
}

const express = require('express');
const bodyParser = require('body-parser');
const crypto = require('crypto');

const app = express();
const WEBHOOK_SECRET = process.env.ATP_WEBHOOK_SECRET;

// Use raw body parser to get the raw request body for signature verification
app.use(bodyParser.json({
  verify: (req, res, buf) => {
    req.rawBody = buf;
  }
}));

app.post('/atp/webhook', async (req, res) => {
  // Verify signature
  const signature = req.headers['x-atp-signature'];
  if (!signature) {
    return res.status(401).json({ error: 'Missing signature header' });
  }
  
  if (!verifySignature(req.rawBody, signature, WEBHOOK_SECRET)) {
    return res.status(401).json({ error: 'Invalid signature' });
  }
  
  const { notification_id, action_id, response_data, responder } = req.body;
  
  try {
    // Process the response based on action type
    switch (action_id) {
      case 'approve_deployment':
        await triggerDeployment(notification_id);
        break;
      case 'reject_deployment':
        await cancelDeployment(notification_id, response_data);
        break;
      default:
        console.log(`Unknown action_id: ${action_id}`);
    }
    
    // Respond with success
    res.status(200).json({ status: 'processed' });
  } catch (error) {
    console.error('Error processing webhook:', error);
    
    // Determine if this error is retriable
    const retriable = error.code !== 'PERMANENT_FAILURE';
    
    // Respond with error
    res.status(500).json({
      code: error.code || 'INTERNAL_ERROR',
      message: error.message,
      user_message: 'The system encountered an error processing your response.',
      retriable
    });
  }
});

app.listen(3000, () => console.log('Webhook server running on port 3000'));

from flask import Flask, request, jsonify
import hmac
import hashlib
import json
import os

app = Flask(__name__)
WEBHOOK_SECRET = os.environ.get('ATP_WEBHOOK_SECRET')

def verify_signature(payload_body, signature_header):
    # Parse the signature header
    header_parts = signature_header.split(',')
    timestamp_str = header_parts[0].split('=')[1]
    signature = header_parts[1].split('=')[1]
    
    # Compute expected signature
    expected_signature = hmac.new(
        WEBHOOK_SECRET.encode('utf-8'),
        payload_body,
        hashlib.sha256
    ).hexdigest()
    
    # Compare signatures using constant-time comparison
    return hmac.compare_digest(signature, expected_signature)

@app.route('/atp/webhook', methods=['POST'])
def handle_webhook():
    # Get the raw request body for signature verification
    payload_body = request.data
    
    # Verify signature
    signature = request.headers.get('X-ATP-Signature')
    if not signature:
        return jsonify({"error": "Missing signature header"}), 401
    
    if not verify_signature(payload_body, signature):
        return jsonify({"error": "Invalid signature"}), 401
    
    # Parse the payload
    data = request.json
    notification_id = data.get('notification_id')
    action_id = data.get('action_id')
    response_data = data.get('response_data')
    responder = data.get('responder')
    
    try:
        # Process the response based on action type
        if action_id == 'approve_deployment':
            # Trigger deployment
            trigger_deployment(notification_id)
        elif action_id == 'reject_deployment':
            # Cancel deployment with reason
            cancel_deployment(notification_id, response_data)
        else:
            print(f"Unknown action_id: {action_id}")
        
        # Respond with success
        return jsonify({"status": "processed"}), 200
        
    except Exception as e:
        # Determine if this error is retriable
        retriable = getattr(e, 'retriable', True)
        
        # Respond with error
        return jsonify({
            "code": getattr(e, 'code', 'INTERNAL_ERROR'),
            "message": str(e),
            "user_message": "The system encountered an error processing your response.",
            "retriable": retriable
        }), 500

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