Code Samples

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

const SIGNING_SECRET = process.env.SMARTALEX_SIGNING_SECRET; // shs_...
const TOOL_TOKEN = process.env.SMARTALEX_TOOL_TOKEN;         // matches the Bearer in your tool config

function verifySignature(req) {
  const header = req.header('X-SmartAlex-Signature') ?? '';
  const parts = Object.fromEntries(
    header.split(',').map((p) => p.split('=', 2))
  );
  const ts = parts.t;
  const v1 = parts.v1;
  if (!ts || !v1) return false;

  const ageSeconds = (Date.now() - parseInt(ts, 10)) / 1000;
  if (ageSeconds > 300 || ageSeconds < -60) return false;

  const expected = crypto
    .createHmac('sha256', SIGNING_SECRET)
    .update(`${ts}.${req.rawBody}`)
    .digest('hex');

  if (v1.length !== expected.length) return false;
  return crypto.timingSafeEqual(Buffer.from(v1), Buffer.from(expected));
}

function verifyAuthHeader(req) {
  if (!TOOL_TOKEN) return true; // skip if not configured
  return req.header('Authorization') === `Bearer ${TOOL_TOKEN}`;
}

const app = express();

app.use(express.json({
  limit: '70kb',
  verify: (req, _res, buf) => {
    req.rawBody = buf.toString('utf8');
  },
}));

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

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

  if (tool === 'lookup_routing') {
    const extension = lookupExtension(args.query); // your business logic
    if (!extension) {
      return res.json({
        found: false,
        hint: 'No matching extension. Apologize and offer a callback.',
      });
    }
    return res.json({
      extension,
      hint: `Transfer the caller to extension ${extension}.`,
    });
  }

  res.status(404).json({ error: 'Unknown tool' });
});

app.listen(3000, () => console.log('listening on :3000'));

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

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

app = Flask(__name__)


def verify_signature() -> bool:
    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

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

    raw = request.get_data()
    expected = hmac.new(
        SIGNING_SECRET,
        f'{ts}.'.encode() + raw,
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(expected, v1)


def verify_token() -> bool:
    if not TOOL_TOKEN:
        return True
    return request.headers.get('Authorization') == f'Bearer {TOOL_TOKEN}'


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

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

    if tool == 'lookup_routing':
        extension = lookup_extension(args.get('query'))  # your logic
        if not extension:
            return jsonify(
                found=False,
                hint='No matching extension. Apologize and offer a callback.',
            )
        return jsonify(
            extension=extension,
            hint=f'Transfer the caller to extension {extension}.',
        )

    abort(404, 'Unknown tool')


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

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

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

app = FastAPI()


async def verify_signature(request: Request) -> bytes:
    raw = await request.body()
    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:
        raise HTTPException(401, 'bad signature header')

    age = time.time() - int(ts) / 1000
    if age > 300 or age < -60:
        raise HTTPException(401, 'stale or future signature')

    expected = hmac.new(
        SIGNING_SECRET,
        f'{ts}.'.encode() + raw,
        hashlib.sha256,
    ).hexdigest()
    if not hmac.compare_digest(expected, v1):
        raise HTTPException(401, 'bad signature')
    return raw


@app.post('/smartalex/lookup')
async def lookup(request: Request):
    if TOOL_TOKEN and request.headers.get('Authorization') != f'Bearer {TOOL_TOKEN}':
        raise HTTPException(401, 'bad token')

    raw = await verify_signature(request)
    import json
    body = json.loads(raw)
    tool = body.get('tool')
    args = body.get('arguments', {})

    if tool == 'lookup_routing':
        extension = lookup_extension(args.get('query'))
        if not extension:
            return {
                'found': False,
                'hint': 'No matching extension. Apologize and offer a callback.',
            }
        return {
            'extension': extension,
            'hint': f'Transfer the caller to extension {extension}.',
        }

    raise HTTPException(404, 'Unknown tool')

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

function verify_signature($secret, $raw) {
    $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;

    $expected = hash_hmac('sha256', $ts . '.' . $raw, $secret);
    return hash_equals($expected, $v1);
}

function verify_token($expected_token) {
    if (!$expected_token) return true;
    $given = $_SERVER['HTTP_AUTHORIZATION'] ?? '';
    return $given === "Bearer $expected_token";
}

$raw = file_get_contents('php://input');

if (!verify_token($TOOL_TOKEN)) {
    http_response_code(401);
    echo 'bad token';
    exit;
}
if (!verify_signature($SIGNING_SECRET, $raw)) {
    http_response_code(401);
    echo 'bad signature';
    exit;
}

$body = json_decode($raw, true);
$tool = $body['tool'] ?? '';
$args = $body['arguments'] ?? [];

header('Content-Type: application/json');

if ($tool === 'lookup_routing') {
    $extension = lookup_extension($args['query'] ?? '');
    if (!$extension) {
        echo json_encode([
            'found' => false,
            'hint'  => 'No matching extension. Apologize and offer a callback.',
        ]);
        exit;
    }
    echo json_encode([
        'extension' => $extension,
        'hint'      => "Transfer the caller to extension $extension.",
    ]);
    exit;
}

http_response_code(404);
echo json_encode(['error' => 'Unknown tool']);

package main

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

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

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))
}

func verifyToken(r *http.Request) bool {
    if toolToken == "" {
        return true
    }
    return r.Header.Get("Authorization") == "Bearer "+toolToken
}

type req 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 !verifyToken(r) {
        http.Error(w, "bad token", http.StatusUnauthorized)
        return
    }
    if !verifySignature(r, raw) {
        http.Error(w, "bad signature", http.StatusUnauthorized)
        return
    }

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

    w.Header().Set("Content-Type", "application/json")

    if body.Tool == "lookup_routing" {
        query, _ := body.Arguments["query"].(string)
        extension := lookupExtension(query)
        if extension == 0 {
            _ = json.NewEncoder(w).Encode(map[string]interface{}{
                "found": false,
                "hint":  "No matching extension. Apologize and offer a callback.",
            })
            return
        }
        _ = json.NewEncoder(w).Encode(map[string]interface{}{
            "extension": extension,
            "hint":      "Transfer the caller to extension " + strconv.Itoa(extension),
        })
        return
    }

    http.Error(w, "Unknown tool", http.StatusNotFound)
}

func main() {
    http.HandleFunc("/smartalex/lookup", handler)
    log.Fatal(http.ListenAndServe(":3000", nil))
}