Signature Verification

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

const SIGNING_SECRET = process.env.SMARTALEX_SIGNING_SECRET; // shs_...

function verifySignature(req) {
  const header = req.header('X-SmartAlex-Signature');
  if (!header) return false;

  const [tPart, v1Part] = header.split(',');
  const ts = tPart?.split('=')[1];
  const v1 = v1Part?.split('=')[1];
  if (!ts || !v1) return false;

  // Replay defense: reject signatures older than 5 minutes or
  // more than 1 minute in the future (clock skew tolerance).
  const ageSeconds = (Date.now() - parseInt(ts, 10)) / 1000;
  if (ageSeconds > 300 || ageSeconds < -60) return false;

  // Use raw request body bytes, not the JSON-stringified parsed object.
  const expected = crypto
    .createHmac('sha256', SIGNING_SECRET)
    .update(`${ts}.${req.rawBody}`)
    .digest('hex');

  // Constant-time compare. Lengths must match or timingSafeEqual throws.
  if (v1.length !== expected.length) return false;
  return crypto.timingSafeEqual(Buffer.from(v1), Buffer.from(expected));
}

const app = express();

// Capture raw body so we can HMAC the exact bytes that arrived.
app.use(express.json({
  verify: (req, _res, buf) => {
    req.rawBody = buf.toString('utf8');
  },
}));

app.post('/smartalex/lookup', (req, res) => {
  if (!verifySignature(req)) {
    return res.status(401).send('Invalid signature');
  }

  const { tool, arguments: args, call_id } = req.body;

  if (tool === 'lookup_routing') {
    const extension = lookupRouting(args.query);
    return res.json({
      extension,
      hint: `Transfer to extension ${extension}.`,
    });
  }

  res.status(404).send('Unknown tool');
});

app.listen(3000);

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

SIGNING_SECRET = os.environ['SMARTALEX_SIGNING_SECRET'].encode()

def verify_signature():
    header = request.headers.get('X-SmartAlex-Signature', '')
    parts = dict(p.split('=', 1) for p in header.split(',') if '=' in p)
    ts = parts.get('t')
    v1 = parts.get('v1')
    if not ts or not v1:
        return False

    # Replay defense
    age = time.time() - int(ts) / 1000
    if age > 300 or age < -60:
        return False

    # Hash the raw bytes, not the parsed body
    raw = request.get_data()  # returns bytes
    expected = hmac.new(
        SIGNING_SECRET,
        f'{ts}.'.encode() + raw,
        hashlib.sha256,
    ).hexdigest()

    return hmac.compare_digest(expected, v1)

app = Flask(__name__)

@app.post('/smartalex/lookup')
def lookup():
    if not verify_signature():
        abort(401, 'Invalid signature')

    body = request.get_json(silent=True) or {}
    tool = body.get('tool')

    if tool == 'lookup_routing':
        ext = lookup_routing(body['arguments']['query'])
        return jsonify(extension=ext, hint=f'Transfer to extension {ext}.')

    abort(404, 'Unknown tool')

<?php
$SIGNING_SECRET = getenv('SMARTALEX_SIGNING_SECRET');

function verify_signature($secret) {
    $header = $_SERVER['HTTP_X_SMARTALEX_SIGNATURE'] ?? '';
    $parts = [];
    foreach (explode(',', $header) as $p) {
        if (str_contains($p, '=')) {
            [$k, $v] = explode('=', $p, 2);
            $parts[$k] = $v;
        }
    }
    $ts = $parts['t'] ?? null;
    $v1 = $parts['v1'] ?? null;
    if (!$ts || !$v1) return false;

    $age = time() - intval($ts) / 1000;
    if ($age > 300 || $age < -60) return false;

    $raw = file_get_contents('php://input');
    $expected = hash_hmac('sha256', $ts . '.' . $raw, $secret);

    return hash_equals($expected, $v1);
}

if (!verify_signature($SIGNING_SECRET)) {
    http_response_code(401);
    exit('Invalid signature');
}

$body = json_decode(file_get_contents('php://input'), true);
$tool = $body['tool'] ?? '';

if ($tool === 'lookup_routing') {
    $ext = lookup_routing($body['arguments']['query']);
    header('Content-Type: application/json');
    echo json_encode([
        'extension' => $ext,
        'hint' => "Transfer to extension {$ext}.",
    ]);
    exit;
}

http_response_code(404);
echo 'Unknown tool';

package main

import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
    "encoding/json"
    "io"
    "net/http"
    "os"
    "strconv"
    "strings"
    "time"
)

var signingSecret = []byte(os.Getenv("SMARTALEX_SIGNING_SECRET"))

func verifySignature(r *http.Request, raw []byte) bool {
    header := r.Header.Get("X-SmartAlex-Signature")
    parts := map[string]string{}
    for _, p := range strings.Split(header, ",") {
        if kv := strings.SplitN(p, "=", 2); len(kv) == 2 {
            parts[kv[0]] = kv[1]
        }
    }
    tsStr, v1 := parts["t"], parts["v1"]
    if tsStr == "" || v1 == "" {
        return false
    }

    tsMs, err := strconv.ParseInt(tsStr, 10, 64)
    if err != nil {
        return false
    }
    age := time.Since(time.UnixMilli(tsMs)).Seconds()
    if age > 300 || age < -60 {
        return false
    }

    mac := hmac.New(sha256.New, signingSecret)
    mac.Write([]byte(tsStr + "."))
    mac.Write(raw)
    expected := hex.EncodeToString(mac.Sum(nil))

    return hmac.Equal([]byte(expected), []byte(v1))
}

type body struct {
    Tool      string                 `json:"tool"`
    Arguments map[string]interface{} `json:"arguments"`
}

func handler(w http.ResponseWriter, r *http.Request) {
    raw, _ := io.ReadAll(r.Body)
    if !verifySignature(r, raw) {
        http.Error(w, "Invalid signature", http.StatusUnauthorized)
        return
    }

    var b body
    if err := json.Unmarshal(raw, &b); err != nil {
        http.Error(w, "bad body", http.StatusBadRequest)
        return
    }

    if b.Tool == "lookup_routing" {
        query := b.Arguments["query"].(string)
        ext := lookupRouting(query)
        w.Header().Set("Content-Type", "application/json")
        json.NewEncoder(w).Encode(map[string]interface{}{
            "extension": ext,
            "hint":      "Transfer to extension " + strconv.Itoa(ext),
        })
        return
    }

    http.NotFound(w, r)
}