Webhooks

import hashlib
import hmac
import json
import os
import time

from flask import Flask, request

app = Flask(__name__)
SECRET = os.environ['ATYS_WEBHOOK_SECRET'].encode()


@app.post('/atys/webhook')
def atys_webhook():
    assinatura = request.headers.get('X-Atys-Signature', '')
    timestamp = request.headers.get('X-Atys-Timestamp', '')
    corpo_cru = request.get_data()          # bytes, sem reprocessar

    if not timestamp or abs(time.time() - int(timestamp)) > 300:
        return 'timestamp fora da janela', 401

    esperada = 'sha256=' + hmac.new(
        SECRET,
        f'{timestamp}.'.encode() + corpo_cru,
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(assinatura, esperada):
        return 'assinatura inválida', 401

    evento = json.loads(corpo_cru)
    print(evento['event'], evento['delivery_id'])

    return '', 200

<?php

$secret = getenv('ATYS_WEBHOOK_SECRET');

$assinatura = $_SERVER['HTTP_X_ATYS_SIGNATURE'] ?? '';
$timestamp  = $_SERVER['HTTP_X_ATYS_TIMESTAMP'] ?? '';
$corpoCru   = file_get_contents('php://input');

if (! $timestamp || abs(time() - (int) $timestamp) > 300) {
    http_response_code(401);
    exit('timestamp fora da janela');
}

$esperada = 'sha256=' . hash_hmac('sha256', $timestamp . '.' . $corpoCru, $secret);

if (! hash_equals($esperada, $assinatura)) {
    http_response_code(401);
    exit('assinatura inválida');
}

$evento = json_decode($corpoCru, true);
error_log($evento['event'] . ' ' . $evento['delivery_id']);

http_response_code(200);

<?php

// routes/web.php  — lembre-se de isentar a rota do CSRF
Route::post('/atys/webhook', function (\Illuminate\Http\Request $request) {
    $secret = config('services.atys.webhook_secret');

    $assinatura = $request->header('X-Atys-Signature', '');
    $timestamp  = $request->header('X-Atys-Timestamp', '');
    $corpoCru   = $request->getContent();

    if (! $timestamp || abs(time() - (int) $timestamp) > 300) {
        abort(401, 'timestamp fora da janela');
    }

    $esperada = 'sha256=' . hash_hmac('sha256', $timestamp . '.' . $corpoCru, $secret);

    if (! hash_equals($esperada, $assinatura)) {
        abort(401, 'assinatura inválida');
    }

    $evento = $request->json()->all();

    // Idempotência: ignore delivery_id já processado.
    \App\Jobs\ProcessarEventoAtys::dispatch($evento);

    return response()->noContent();
});

package main

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

var secret = []byte(os.Getenv("ATYS_WEBHOOK_SECRET"))

func handler(w http.ResponseWriter, r *http.Request) {
	corpoCru, err := io.ReadAll(r.Body)
	if err != nil {
		http.Error(w, "corpo ilegível", http.StatusBadRequest)
		return
	}

	timestamp := r.Header.Get("X-Atys-Timestamp")
	ts, err := strconv.ParseInt(timestamp, 10, 64)
	if err != nil || math.Abs(float64(time.Now().Unix()-ts)) > 300 {
		http.Error(w, "timestamp fora da janela", http.StatusUnauthorized)
		return
	}

	mac := hmac.New(sha256.New, secret)
	mac.Write([]byte(timestamp + "."))
	mac.Write(corpoCru)
	esperada := "sha256=" + hex.EncodeToString(mac.Sum(nil))

	if !hmac.Equal([]byte(esperada), []byte(r.Header.Get("X-Atys-Signature"))) {
		http.Error(w, "assinatura inválida", http.StatusUnauthorized)
		return
	}

	var evento struct {
		Event      string          `json:"event"`
		DeliveryID string          `json:"delivery_id"`
		Data       json.RawMessage `json:"data"`
	}
	if err := json.Unmarshal(corpoCru, &evento); err != nil {
		http.Error(w, "json inválido", http.StatusBadRequest)
		return
	}

	w.WriteHeader(http.StatusOK)
}

func main() {
	http.HandleFunc("/atys/webhook", handler)
	http.ListenAndServe(":3000", nil)
}

# Útil para conferir sua implementação: gere a assinatura do mesmo material
# e compare com o header que o Atys enviou.

SECRET='whsec_seu_secret_aqui'
TIMESTAMP='1755436800'
BODY='{"event":"ping","delivery_id":"abc","company_id":42,"data":{}}'

printf '%s.%s' "$TIMESTAMP" "$BODY" \
  | openssl dgst -sha256 -hmac "$SECRET" -hex \
  | sed 's/^.*= /sha256=/'

import requests

response = requests.post(
    'https://api.atys.pro/api/channel-webhooks',
    headers={
        'Authorization': 'Bearer YOUR_ACCESS_TOKEN',
        'Content-Type': 'application/json',
        'Accept': 'application/json',
    },
    json={
        'name': 'Integração ERP',
        'url': 'https://meuservidor.com/atys/webhook',
        'events': ['omnichat.message.received', 'crm.business.won'],
    },
)

data = response.json()

# Guarde o secret agora: ele não é exibido de novo.
print(data['secret'])

<?php

$ch = curl_init('https://api.atys.pro/api/channel-webhooks');

curl_setopt_array($ch, [
    CURLOPT_RETURNTRANSFER => true,
    CURLOPT_POST           => true,
    CURLOPT_HTTPHEADER     => [
        'Authorization: Bearer YOUR_ACCESS_TOKEN',
        'Content-Type: application/json',
        'Accept: application/json',
    ],
    CURLOPT_POSTFIELDS => json_encode([
        'name'   => 'Integração ERP',
        'url'    => 'https://meuservidor.com/atys/webhook',
        'events' => ['omnichat.message.received', 'crm.business.won'],
    ]),
]);

$response = curl_exec($ch);
curl_close($ch);

$data = json_decode($response, true);

// Guarde o secret agora: ele não é exibido de novo.
echo $data['secret'];

<?php

use GuzzleHttp\Client;

$client = new Client();

$response = $client->post('https://api.atys.pro/api/channel-webhooks', [
    'headers' => [
        'Authorization' => 'Bearer YOUR_ACCESS_TOKEN',
        'Accept'        => 'application/json',
    ],
    'json' => [
        'name'   => 'Integração ERP',
        'url'    => 'https://meuservidor.com/atys/webhook',
        'events' => ['omnichat.message.received', 'crm.business.won'],
    ],
]);

$data = json_decode($response->getBody()->getContents(), true);

// Guarde o secret agora: ele não é exibido de novo.
echo $data['secret'];

package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"net/http"
)

func main() {
	corpo, _ := json.Marshal(map[string]any{
		"name":   "Integração ERP",
		"url":    "https://meuservidor.com/atys/webhook",
		"events": []string{"omnichat.message.received", "crm.business.won"},
	})

	req, _ := http.NewRequest(
		"POST",
		"https://api.atys.pro/api/channel-webhooks",
		bytes.NewBuffer(corpo),
	)
	req.Header.Set("Authorization", "Bearer YOUR_ACCESS_TOKEN")
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Accept", "application/json")

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		panic(err)
	}
	defer resp.Body.Close()

	var data struct {
		Secret string `json:"secret"`
	}
	json.NewDecoder(resp.Body).Decode(&data)

	// Guarde o secret agora: ele não é exibido de novo.
	fmt.Println(data.Secret)
}

curl -X POST https://api.atys.pro/api/channel-webhooks \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json" \
  -d '{
    "name": "Integração ERP",
    "url": "https://meuservidor.com/atys/webhook",
    "events": ["omnichat.message.received", "crm.business.won"]
  }'