Patikrinkite webhook parašus

Kiekviena ThunderPhone siunčiama webhook ir įrankio užklausa pasirašoma su laiko žyma. Patikrinkite ją naudodami čia pateiktą receptą arba SDK pagalbinę priemonę ir atmeskite pakartotines užklausas, senesnes nei penkios minutės.

Kiekvienoje užklausoje, kurią siunčiame į jūsų serverį (webhook pristatymuose, senajame vieno URL webhook ir įrankių galinių taškų iškvietimuose), yra dvi parašo antraštės. Abi naudoja HMAC-SHA256 su ta pačia paslaptimi.

Naujose integracijose tikrinkite X-ThunderPhone-Signature-V2 ir perkelkite esamas integracijas į ją. X-ThunderPhone-Signature vis dar siunčiama su kiekviena užklausa nepakitusi, todėl pagal ją sukurti tikrintuvai ir toliau veiks. Prieš nustodami ją siųsti, paskelbsime datą.

Signature V2#

X-ThunderPhone-Signature-V2: t=1767225600,v1=03a89bec521f78d55727237ca4713bb0b8dcdf2e09e450865b5b5637b80c1c4e

• t yra Unix laikas sekundėmis, kai išsiuntėme užklausą. Pakartotinai pristatant užklausa pasirašoma iš naujo, naudojant jai skirtą t.
• v1 yra toliau pateiktos pasirašomos eilutės šešioliktainis HMAC-SHA256, sugeneruotas naudojant jūsų slaptąjį raktą.

Reikšmė yra kableliais atskirtas key=value porų sąrašas. Nepaisykite neatpažįstamų raktų ir priimkite užklausą, jei sutampa bet kuris v1: per būsimą pakeitimą galime pridėti antrą parašą.

Pasirašoma eilutė#

Keturi laukai, po kiekvieno jų yra viena naujos eilutės žyma (\n, baitas 0x0A), išskyrus paskutinį:

<t>\n<METHOD>\n<path and query>\n<body>

Norėdami patikrinti:

1. Prieš bet kokį analizavimą nuskaitykite antraštę, metodą, neapdorotą kelią ir užklausą bei neapdorotą turinį.
2. Iš antraštės paimkite t ir kiekvieną v1.
3. Atmeskite užklausą, jei t nuo jūsų laikrodžio skiriasi daugiau nei 300 sekundžių į praeitį arba į ateitį.
4. Naudodami savo slaptąjį raktą apskaičiuokite pasirašomos eilutės šešioliktainį HMAC-SHA256.
5. Pastoviu laiku palyginkite jį su v1. Jei niekas nesutampa, atsakykite 401.

Išsamus pavyzdys#

Slaptasis raktas whsec_example_4c1f2a9b. Šią užklausą siunčiame laiku t = 1767225600 (2026-01-01 00:00:00 UTC):

POST /tools/lookup?region=us HTTP/1.1
Host: api.example.com
Content-Type: application/json
X-ThunderPhone-Signature-V2: t=1767225600,v1=03a89bec521f78d55727237ca4713bb0b8dcdf2e09e450865b5b5637b80c1c4e
 
{"customer_id":"cust_123"}

Pasirašoma eilutė yra šios keturios eilutės, be naujos eilutės žymos po turinio:

1767225600
POST
/tools/lookup?region=us
{"customer_id":"cust_123"}

Jos HMAC-SHA256 yra 03a89bec521f78d55727237ca4713bb0b8dcdf2e09e450865b5b5637b80c1c4e. Palyginkite su juo savo įgyvendinimą arba atkurkite jį naudodami openssl:

printf '1767225600\nPOST\n/tools/lookup?region=us\n{"customer_id":"cust_123"}' \
  | openssl dgst -sha256 -hmac 'whsec_example_4c1f2a9b'

Tą pačią sekundę atliekamas GET įrankio iškvietimas į /tools/lookup?customer_id=cust_123 turi tuščią turinį. Jo pasirašoma eilutė yra 1767225600\nGET\n/tools/lookup?customer_id=cust_123\n, o jo parašas yra 3ce214cb0b8f2bf39b1d19d4e7265df9dbaff5a5f838d3cc6c753a8f7a4ad280.

Su SDK#

Python ir TypeScript SDK turi pagalbinę funkciją (thunderphone 0.2.0 arba naujesnė versija, @thunderphone/sdk 0.2.0 arba naujesnė versija). Ji tikrina 5 minučių laikotarpį, palygina pastoviu laiku ir priima slaptųjų raktų sąrašą, kad galėtumėte juos keisti be prastovų.

from thunderphone import verify_signature
 
ok = verify_signature(
    header,            # the X-ThunderPhone-Signature-V2 value
    secret=SECRET,     # or [NEW_SECRET, OLD_SECRET] while rotating
    method=method,     # "POST"
    path=path,         # "/tools/lookup?region=us", exactly as received
    body=raw_body,     # bytes, before any JSON parsing
    tolerance=300,     # optional; seconds
)

import { verifySignature } from "@thunderphone/sdk";
 
const ok = verifySignature({
  header,               // the X-ThunderPhone-Signature-V2 value
  secret: SECRET,       // or [NEW_SECRET, OLD_SECRET] while rotating
  method,               // "POST"
  path,                 // "/tools/lookup?region=us", exactly as received
  body: rawBody,        // Buffer or string, before any JSON parsing
  toleranceSeconds: 300, // optional
});

Be SDK#

import hashlib
import hmac
import time
 
TOLERANCE_SECONDS = 300
 
 
def verify_v2(header: str, secret: str, method: str, path: str, body: bytes) -> bool:
    """path is the path plus ?query exactly as received; body is the raw body."""
    timestamp, signatures = None, []
    for item in (header or "").split(","):
        key, _, value = item.strip().partition("=")
        if key == "t" and value.isdigit() and len(value) <= 15:
            timestamp = int(value)
        elif key == "v1":
            signatures.append(value)
    if timestamp is None or abs(time.time() - timestamp) > TOLERANCE_SECONDS:
        return False
    signed = f"{timestamp}\n{method.upper()}\n{path}\n".encode() + body
    expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected.encode(), s.encode()) for s in signatures)

import crypto from "node:crypto";
 
const TOLERANCE_SECONDS = 300;
 
// path is the path plus ?query exactly as received; body is the raw Buffer.
export function verifyV2(header, secret, method, path, body) {
  let timestamp;
  const signatures = [];
  for (const item of (header || "").split(",")) {
    const [key, value = ""] = item.trim().split("=", 2);
    if (key === "t" && /^\d+$/.test(value)) timestamp = Number(value);
    else if (key === "v1") signatures.push(value);
  }
  if (timestamp === undefined) return false;
  if (Math.abs(Date.now() / 1000 - timestamp) > TOLERANCE_SECONDS) return false;
  const expected = Buffer.from(
    crypto
      .createHmac("sha256", secret)
      .update(`${timestamp}\n${method.toUpperCase()}\n${path}\n`)
      .update(body)
      .digest("hex"),
  );
  return signatures.some((sig) => {
    const candidate = Buffer.from(sig);
    return (
      candidate.length === expected.length &&
      crypto.timingSafeEqual(candidate, expected)
    );
  });
}

Sisteminės sąsajos prijungimas#

Kiekviename pavyzdyje nuskaitomas neapdorotas kelias ir užklausa bei neapdorotas turinys, tada iškviečiama SDK pagalbinė funkcija.

import json
 
from fastapi import FastAPI, HTTPException, Request
from thunderphone import verify_signature
 
app = FastAPI()
 
 
def raw_path_and_query(request: Request) -> str:
    path = (request.scope.get("raw_path") or request.url.path.encode()).decode("latin-1")
    query = request.scope.get("query_string", b"").decode("latin-1")
    return f"{path}?{query}" if query else path
 
 
@app.post("/thunderphone-webhook")
async def hook(request: Request):
    body = await request.body()  # raw bytes, NOT request.json()
    if not verify_signature(
        request.headers.get("X-ThunderPhone-Signature-V2"),
        secret=SECRET,
        method=request.method,
        path=raw_path_and_query(request),
        body=body,
    ):
        raise HTTPException(status_code=401)
    event = json.loads(body)
    # … dispatch on event["type"] …
    return {"ok": True}

import express from "express";
import { verifySignature } from "@thunderphone/sdk";
 
const app = express();
 
app.post(
  "/thunderphone-webhook",
  // IMPORTANT: parse as raw; do NOT use express.json() here.
  express.raw({ type: "application/json" }),
  (req, res) => {
    const ok = verifySignature({
      header: req.header("X-ThunderPhone-Signature-V2"),
      secret: process.env.WEBHOOK_SECRET,
      method: req.method,
      path: req.originalUrl, // path and query as received
      body: req.body,
    });
    if (!ok) return res.sendStatus(401);
    const event = JSON.parse(req.body.toString("utf8"));
    // … dispatch on event.type …
    res.sendStatus(204);
  },
);

import json
 
from django.http import HttpResponseForbidden, JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from thunderphone import verify_signature
 
 
@csrf_exempt
@require_POST
def hook(request):
    if not verify_signature(
        request.headers.get("X-ThunderPhone-Signature-V2"),
        secret=SECRET,
        method=request.method,
        path=request.get_full_path(),
        body=request.body,  # raw bytes
    ):
        return HttpResponseForbidden("invalid signature")
    event = json.loads(request.body)
    # … dispatch on event["type"] …
    return JsonResponse({"ok": True})

req.originalUrl („Express“) ir ASGI raw_path bei query_string („FastAPI“) yra tikslinis adresas tiksliai toks, koks buvo gautas. „Django“ get_full_path() ir „Flask“ request.path pirmiausia dekoduoja kelią, todėl įprastiems keliams gaunama ta pati eilutė. Jei jūsų galinio taško kelyje yra procentais užkoduotų arba rezervuotų simbolių, pvz., %2F, ; ar =, vietoje to nuskaitykite neapdorotą tikslinį adresą, kurį pateikia jūsų serveris (pavyzdžiui, request.META["RAW_URI"] sistemoje „Gunicorn“).

Kurį slaptąjį raktą naudoti?#

Ta pati paslaptis pasirašo abi antraštes.

Saugokite slaptąjį raktą slaptųjų raktų tvarkytuvėje arba aplinkos kintamajame — niekada jo neįtraukite į versijų valdymą.

Įrankių iškvietimų tikrinimas#

Kai agentas tiesiogiai iškviečia vieną iš jūsų funkcijų įrankių (įrankis turi endpoint), užklausoje kartu su sukonfigūruotomis endpoint.headers perduodamos šios ThunderPhone antraštės:

• X-ThunderPhone-Call-ID — vykstančio skambučio skaitinis ID.
• X-ThunderPhone-Signature-V2 — pirmiau nurodytas laiko žyma pažymėtas parašas, susietas su jūsų organizacijos lygio žiniatinklio kabliuko slaptuoju raktu.
• X-ThunderPhone-Signature — senasis tik užklausos turinį apimantis parašas, naudojantis tą patį raktą.

Nuo žiniatinklio kabliukų skiriasi du dalykai:

1. GET / DELETE įrankiai neturi užklausos turinio. Argumentai perduodami kaip užklausos parametrai. V2 parašas juos apima per kelią ir užklausą, o užklausos turinio laukas yra tuščias. Tikrinkite naudodami neapdorotą užklausos eilutę: išanalizavus parametrus ir sukūrus naują užklausos eilutę gali pasikeisti jų tvarka arba kodavimas. Senosios antraštės parašas apima tuščią baitų eilutę.
2. Organizacijos, kuriose nesukonfigūruotas senasis žiniatinklio kabliukas, neturi organizacijos slaptojo rakto. Tokiu atveju įrankių iškvietimuose perduodama tik X-ThunderPhone-Call-ID ir nėra parašo antraščių. Sukonfigūruokite senąjį žiniatinklio kabliuką (PUT /v1/webhook), kad gautumėte pasirašymo slaptąjį raktą, arba autentifikuokite įrankių iškvietimus naudodami savo antraštę per endpoint.headers.

@app.api_route("/tools/search-appointments", methods=["GET", "POST"])
async def tool(request: Request):
    body = await request.body()  # b"" for GET/DELETE tools
    if not verify_signature(
        request.headers.get("X-ThunderPhone-Signature-V2"),
        secret=ORG_WEBHOOK_SECRET,
        method=request.method,
        path=raw_path_and_query(request),
        body=body,
    ):
        raise HTTPException(status_code=401)
    call_id = request.headers.get("X-ThunderPhone-Call-ID", "")
    args = json.loads(body) if body else dict(request.query_params)
    ...

Express sistemoje užklausos turinio analizatorius GET užklausose palieka req.body kaip tuščią objektą, todėl perduokite Buffer.isBuffer(req.body) ? req.body : Buffer.alloc(0).

Įrankių iškvietimų nukreipimas žiniatinklio kabliuko režimu (įrankiai be endpoint, pateikiami į jūsų organizacijos žiniatinklio kabliuką kaip telephony.tool / web.tool) yra įprastas pasirašytas žiniatinklio kabliukas — taikomas pirmiau pateiktas standartinis būdas. Abiejų užklausų formų aprašą rasite Funkcijų įrankiuose.

Senasis parašas: X-ThunderPhone-Signature#

Pradinė antraštė yra neapdoroto turinio hex HMAC-SHA256, naudojant tą patį slaptą raktą. Joje nėra laiko žymos, todėl bet kas, perėmęs vieną pasirašytą užklausą, gali ją pakartoti. Ji vis dar siunčiama su kiekviena užklausa, kol integracijos pereina prie V2.

1. Perskaitykite neapdorotą užklausos turinį — tikslius baitus, kuriuos jums išsiuntėme.
2. Apskaičiuokite hmac_sha256(secret, body).hexdigest().
3. Palyginkite pastoviu laiku su X-ThunderPhone-Signature.

Pasirašome tiksliai tuos baitus, kuriuos perduodame. Šie baitai taip pat yra naudingojo krūvio kanoninis JSON serializavimas — raktai surikiuoti abėcėlės tvarka, glausti skirtukai (, ir : be tarpų), UTF-8. Jei jūsų sistema pateikia tik išanalizuotą JSON, iš naujo jį kanoniškai serializuokite ir apskaičiuokite jo HMAC:

# Equivalent to hashing the raw body:
import json
canonical = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode("utf-8")

Pirmenybę teikite neapdorotam turiniui — tai vienu žingsniu mažiau ir jis apsaugotas nuo JSON skaičių pakartotinio konvertavimo ypatumų kai kuriose kalbose. Tas pats taikoma V2 turinio laukui.

import hashlib
import hmac
 
 
def verify(body: bytes, signature: str, secret: str) -> bool:
    """Constant-time HMAC-SHA256 verification."""
    expected = hmac.new(
        secret.encode("utf-8"),
        body,
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(expected, signature or "")

import crypto from "node:crypto";
 
export function verify(body, signature, secret) {
  const expected = crypto
    .createHmac("sha256", secret)
    .update(body)
    .digest("hex");
  if (!signature || expected.length !== signature.length) return false;
  return crypto.timingSafeEqual(
    Buffer.from(expected),
    Buffer.from(signature),
  );
}

package webhook
 
import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
)
 
func Verify(body []byte, signature, secret string) bool {
    mac := hmac.New(sha256.New, []byte(secret))
    mac.Write(body)
    expected := hex.EncodeToString(mac.Sum(nil))
    return hmac.Equal([]byte(expected), []byte(signature))
}

require "openssl"
 
def verify(body, signature, secret)
  expected = OpenSSL::HMAC.hexdigest("SHA256", secret, body)
  Rack::Utils.secure_compare(expected, signature.to_s)
end

Dažnos klaidos#

Tolesni veiksmai#