Aller au contenu
SynapsAI Docs

Auth Webhook — Python

Ce contenu n’est pas encore disponible dans votre langue.

A complete Python Lambda template that handles all three channels (WebChat, WhatsApp, Instagram) with challenge-response support and HMAC signature verification.

Full example

Section titled “Full example”
import hashlib
import hmac
import json
import os
import re
import uuid


import bcrypt


# Replace with your actual database and cache clients
from db import find_by_email, find_by_document_id  # Your user database
from cache import cache_get, cache_set, cache_del   # Redis, DynamoDB, or similar


AUTH_SECRET = os.environ.get("AUTH_SECRET")  # Your webhook secret from SynapsAI




def handler(event, _context):
    body = json.loads(event["body"])
    credentials = body.get("credentials")
    challenge_response = body.get("challengeResponse")
    ctx = body["context"]


    # 1. Verify HMAC signature (if secret is configured)
    if AUTH_SECRET:
        signature = event["headers"].get("x-synapsai-signature", "")
        if not verify_signature(body, signature, AUTH_SECRET):
            return response(401, {"verified": False, "error": "Invalid signature"})


    # 2. Handle challenge answer
    if challenge_response:
        return handle_challenge_response(challenge_response, ctx)


    # 3. Route by channel
    channel = ctx.get("channel")
    if channel == "CHAT":
        return handle_web_chat(credentials, ctx)
    elif channel in ("WHATSAPP", "INSTAGRAM"):
        return handle_messaging(credentials, ctx)
    else:
        return response(200, {"verified": False, "error": "Unsupported channel"})




# --- WebChat: email + password login ---
def handle_web_chat(credentials, ctx):
    user = find_by_email(credentials.get("email"))


    if not user or not bcrypt.checkpw(
        credentials.get("password", "").encode(), user["password_hash"].encode()
    ):
        return response(200, {
            "verified": False,
            "error": "Invalid email or password",
            "retriesLeft": 2,
        })


    return response(200, {
        "verified": True,
        "userId": user["id"],
        "identity": {"name": user["name"], "email": user["email"]},
    })




# --- WhatsApp / Instagram: document ID + phone challenge ---
def handle_messaging(credentials, ctx):
    customer = find_by_document_id(credentials.get("documentId"))


    if not customer:
        return response(200, {
            "verified": False,
            "error": "Document not found",
            "retriesLeft": 2,
        })


    # Auto-verify if WhatsApp phone matches a registered phone
    implicit_identity = ctx.get("implicitIdentity", {})
    whatsapp_phone = implicit_identity.get("phone")
    if whatsapp_phone and normalize_phone(whatsapp_phone) in customer["phones"]:
        return response(200, {
            "verified": True,
            "userId": customer["id"],
            "identity": {"name": customer["name"], "email": customer["email"]},
        })


    # Otherwise: challenge with masked phones
    challenge_id = f"ch_{uuid.uuid4().hex[:8]}"
    masked_phones = [
        {"label": mask_phone(p), "value": p} for p in customer["phones"]
    ]


    # Store challenge with 5-minute TTL
    cache_set(challenge_id, {
        "userId": customer["id"],
        "phones": customer["phones"],
        "identity": {"name": customer["name"], "email": customer["email"]},
    }, ttl=300)


    return response(200, {
        "verified": False,
        "challenge": {
            "challengeId": challenge_id,
            "type": "SELECT_OPTION",
            "prompt": "Confirm one of your registered phones:",
            "options": masked_phones,
            "hint": "Type the full number with area code",
        },
    })




# --- Challenge response handler ---
def handle_challenge_response(challenge_response, ctx):
    challenge_id = challenge_response["challengeId"]
    answer = challenge_response["answer"]


    challenge = cache_get(challenge_id)


    # Challenge expired (TTL exceeded)
    if not challenge:
        return response(200, {
            "verified": False,
            "error": "Challenge expired. Please start verification again.",
            "retriesLeft": 0,
        })


    # Check answer
    if normalize_phone(answer) in challenge["phones"]:
        cache_del(challenge_id)  # Clean up
        return response(200, {
            "verified": True,
            "userId": challenge["userId"],
            "identity": challenge["identity"],
        })


    return response(200, {
        "verified": False,
        "error": "Incorrect phone number",
        "retriesLeft": 1,
    })




# --- Helpers ---


def verify_signature(body: dict, signature: str, secret: str) -> bool:
    if not signature:
        return False
    expected = hmac.new(
        secret.encode(), json.dumps(body, separators=(",", ":")).encode(), hashlib.sha256
    ).hexdigest()
    return signature == f"sha256={expected}"




def normalize_phone(phone: str) -> str:
    return re.sub(r"[^0-9+]", "", phone)




def mask_phone(phone: str) -> str:
    """'+5551981975196' → '555198xxxxx96'"""
    digits = re.sub(r"\D", "", phone)
    if len(digits) < 8:
        return digits
    return digits[:6] + "x" * (len(digits) - 8) + digits[-2:]




def response(status_code: int, body: dict) -> dict:
    return {
        "statusCode": status_code,
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps(body),
    }

Key points

Section titled “Key points”
  • Signature verification comes first — reject unauthorized requests before processing credentials.
  • Channel routing uses context["channel"] to apply the right verification logic.
  • Auto-verify on WhatsApp when implicitIdentity.phone matches a registered phone — no challenge needed.
  • Challenge TTL is your responsibility. Use Redis TTL or DynamoDB TTL (recommended: 5 minutes).
  • Never log credentials, document IDs, full phone numbers, or webhook response bodies.

See also

Section titled “See also”