Verify NICE Cognigy Webhook Signatures in Python

What You Will Build

• A production-grade Python service that validates incoming NICE Cognigy webhook payloads using HMAC-SHA256 signature verification, timestamp validation, and nonce-based replay attack prevention.
• This implementation uses the standard Cognigy webhook delivery mechanism with shared secret authentication and Pydantic schema validation.
• The tutorial covers Python 3.10+ with FastAPI, httpx, and standard library cryptographic modules.

Prerequisites

• Python 3.10 or higher
• fastapi==0.109.0, httpx==0.26.0, pydantic==2.5.0, uvicorn==0.27.0
• Cognigy webhook shared secret (configured in your Cognigy platform under Webhook settings)
• Maximum clock skew tolerance set to 300 seconds (5 minutes)
• Outbound SIEM endpoint URL and API key (for audit synchronization)

Install dependencies using pip:

pip install fastapi httpx pydantic uvicorn

Authentication Setup

Cognigy webhooks do not use OAuth for incoming payload delivery. The platform authenticates requests using a shared secret that you configure when creating the webhook. Cognigy signs the raw request body using HMAC-SHA256 and attaches the signature, timestamp, and nonce to the HTTP headers. Your service must store the shared secret in an environment variable or secure vault. Never hardcode the secret.

The required headers for verification are:

• X-Cognigy-Signature: Hex-encoded HMAC-SHA256 digest of the raw body
• X-Cognigy-Timestamp: Unix epoch timestamp of when Cognigy generated the webhook
• X-Cognigy-Nonce: Unique identifier for replay prevention

Configure the secret in your environment:

export COGNIGY_WEBHOOK_SECRET="your_secure_shared_secret_here"
export SIEM_ENDPOINT="https://siem.example.com/api/v1/events"
export SIEM_API_KEY="your_siem_api_key"

Implementation

Step 1: Request Parsing and Schema Validation

Cognigy delivers webhook payloads as JSON. You must validate the payload structure against the dialog engine constraints before attempting signature verification. Invalid schemas must be rejected immediately to prevent unnecessary cryptographic operations.

import logging
from typing import Any
from pydantic import BaseModel, Field, ValidationError

logger = logging.getLogger(__name__)

class CognigyWebhookPayload(BaseModel):
    """Schema aligned with Cognigy dialog engine webhook constraints."""
    conversationId: str = Field(..., min_length=1, max_length=64)
    agentId: str | None = None
    botId: str = Field(..., min_length=1, max_length=64)
    message: dict[str, Any] = Field(..., description="Dialog message structure")
    metadata: dict[str, Any] = Field(default_factory=dict)
    timestamp: int = Field(..., gt=0)

def validate_webhook_schema(raw_body: bytes) -> CognigyWebhookPayload:
    """Parse and validate the incoming webhook payload against dialog constraints."""
    try:
        parsed = CognigyWebhookPayload.model_validate_json(raw_body)
        logger.info("Schema validation passed for conversation %s", parsed.conversationId)
        return parsed
    except ValidationError as exc:
        logger.warning("Schema validation failed: %s", exc.errors())
        raise

The model_validate_json method ensures strict type checking and prevents malformed JSON from reaching the verification pipeline. Cognigy requires the raw body for signature calculation, so you must capture request.body() before parsing.

Step 2: HMAC Computation and Signature Verification

Signature verification requires reconstructing the HMAC using the shared secret and the exact raw body bytes. Cognigy expects the signature format sha256=<hex_digest>. You must compute the digest independently and compare it against the header using a timing-safe comparison to prevent timing attacks.

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

def verify_signature(
    raw_body: bytes,
    signature_header: str,
    secret: str
) -> bool:
    """Compute HMAC-SHA256 and verify against the incoming signature header."""
    expected_prefix = "sha256="
    if not signature_header.startswith(expected_prefix):
        raise HTTPException(status_code=400, detail="Invalid signature format")
    
    provided_sig = signature_header[len(expected_prefix):]
    computed_digest = hmac.new(
        secret.encode("utf-8"),
        raw_body,
        hashlib.sha256
    ).hexdigest()
    
    # Timing-safe comparison prevents side-channel attacks
    is_valid = hmac.compare_digest(computed_digest, provided_sig)
    if not is_valid:
        logger.warning("Signature mismatch. Expected: %s, Provided: %s", computed_digest, provided_sig)
    return is_valid

This function isolates the cryptographic verification logic. The hmac.compare_digest method ensures constant-time evaluation. If the signature fails, the request is rejected with a 401 Unauthorized response.

Step 3: Clock Skew Validation and Replay Attack Prevention

Webhook delivery networks introduce latency. Cognigy allows a maximum clock skew of 300 seconds. You must validate the X-Cognigy-Timestamp header against the current server time. Additionally, you must implement a nonce store to prevent replay attacks. Each verified webhook must register its nonce with an expiration window.

from collections import OrderedDict
import threading

MAX_CLOCK_SKEW_SECONDS = 300
NONCE_TTL_SECONDS = 600

class NonceStore:
    """Thread-safe nonce registry with automatic expiration."""
    def __init__(self):
        self._store: OrderedDict[str, float] = OrderedDict()
        self._lock = threading.Lock()

    def is_replay(self, nonce: str) -> bool:
        with self._lock:
            if nonce in self._store:
                return True
            return False

    def register(self, nonce: str) -> None:
        with self._lock:
            self._store[nonce] = time.time()
            self._cleanup()

    def _cleanup(self) -> None:
        """Remove expired nonces to prevent memory leaks."""
        now = time.time()
        while self._store:
            oldest_nonce, timestamp = next(iter(self._store.items()))
            if now - timestamp > NONCE_TTL_SECONDS:
                self._store.popitem(last=False)
            else:
                break

nonce_store = NonceStore()

def validate_timestamp_and_nonce(
    timestamp_header: str | None,
    nonce_header: str | None
) -> None:
    """Enforce clock skew limits and replay attack prevention."""
    if not timestamp_header or not nonce_header:
        raise HTTPException(status_code=400, detail="Missing timestamp or nonce header")
    
    try:
        webhook_timestamp = int(timestamp_header)
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid timestamp format")
    
    current_time = int(time.time())
    skew = abs(current_time - webhook_timestamp)
    
    if skew > MAX_CLOCK_SKEW_SECONDS:
        raise HTTPException(
            status_code=400,
            detail=f"Clock skew exceeded. Tolerance: {MAX_CLOCK_SKEW_SECONDS}s, Actual: {skew}s"
        )
    
    if nonce_store.is_replay(nonce_header):
        raise HTTPException(status_code=409, detail="Nonce replay detected. Request rejected.")
    
    nonce_store.register(nonce_header)

The nonce store uses an ordered dictionary to track insertion order and automatically purge expired entries. The clock skew check ensures that delayed or backdated requests are rejected. Both checks run before business logic execution.

Step 4: SIEM Synchronization, Latency Tracking, and Audit Logging

After successful verification, you must synchronize the event with an external SIEM system, track verification latency, and generate audit logs for dialog governance. You will use httpx for outbound SIEM calls and maintain a simple metrics registry for success rates and latency percentiles.

import httpx
import json
from dataclasses import dataclass, field
from typing import Dict

@dataclass
class VerificationMetrics:
    """Tracks latency and success rates for webhook verification."""
    total_requests: int = 0
    successful_verifications: int = 0
    latencies: list[float] = field(default_factory=list)
    
    def record(self, success: bool, latency_ms: float) -> None:
        self.total_requests += 1
        if success:
            self.successful_verifications += 1
        self.latencies.append(latency_ms)
        # Keep only last 1000 measurements for memory efficiency
        if len(self.latencies) > 1000:
            self.latencies = self.latencies[-1000:]
    
    def get_success_rate(self) -> float:
        if self.total_requests == 0:
            return 0.0
        return (self.successful_verifications / self.total_requests) * 100.0
    
    def get_avg_latency_ms(self) -> float:
        if not self.latencies:
            return 0.0
        return sum(self.latencies) / len(self.latencies)

metrics = VerificationMetrics()

def sync_to_siem(
    payload: CognigyWebhookPayload,
    siem_url: str,
    siem_key: str,
    latency_ms: float
) -> None:
    """Push verified webhook events to external SIEM for governance alignment."""
    audit_event = {
        "eventType": "cognigy.webhook.verified",
        "conversationId": payload.conversationId,
        "botId": payload.botId,
        "verificationLatencyMs": latency_ms,
        "timestamp": payload.timestamp,
        "metrics": {
            "successRate": metrics.get_success_rate(),
            "avgLatencyMs": metrics.get_avg_latency_ms()
        }
    }
    
    try:
        with httpx.Client(timeout=5.0) as client:
            response = client.post(
                siem_url,
                json=audit_event,
                headers={
                    "Authorization": f"Bearer {siem_key}",
                    "Content-Type": "application/json"
                }
            )
            response.raise_for_status()
            logger.info("SIEM sync successful for conversation %s", payload.conversationId)
    except httpx.HTTPError as exc:
        logger.error("SIEM sync failed: %s", exc)
        # Do not raise. SIEM sync is asynchronous and should not block webhook acknowledgment.

The SIEM synchronization runs asynchronously in the background. Verification latency and success rates are calculated dynamically. The audit payload includes dialog governance identifiers required for compliance tracking.

Complete Working Example

import os
import time
import logging
import uvicorn
from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel, Field, ValidationError
from typing import Any

# Configuration
COGNIGY_SECRET = os.getenv("COGNIGY_WEBHOOK_SECRET")
SIEM_ENDPOINT = os.getenv("SIEM_ENDPOINT")
SIEM_API_KEY = os.getenv("SIEM_API_KEY")
MAX_CLOCK_SKEW_SECONDS = 300
NONCE_TTL_SECONDS = 600

if not COGNIGY_SECRET:
    raise RuntimeError("COGNIGY_WEBHOOK_SECRET environment variable is required")

logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")
logger = logging.getLogger(__name__)

# --- Models & Utilities ---
class CognigyWebhookPayload(BaseModel):
    conversationId: str = Field(..., min_length=1, max_length=64)
    agentId: str | None = None
    botId: str = Field(..., min_length=1, max_length=64)
    message: dict[str, Any] = Field(..., description="Dialog message structure")
    metadata: dict[str, Any] = Field(default_factory=dict)
    timestamp: int = Field(..., gt=0)

import hmac
import hashlib
from collections import OrderedDict
import threading

class NonceStore:
    def __init__(self):
        self._store: OrderedDict[str, float] = OrderedDict()
        self._lock = threading.Lock()

    def is_replay(self, nonce: str) -> bool:
        with self._lock:
            return nonce in self._store

    def register(self, nonce: str) -> None:
        with self._lock:
            self._store[nonce] = time.time()
            self._cleanup()

    def _cleanup(self) -> None:
        now = time.time()
        while self._store:
            oldest_nonce, timestamp = next(iter(self._store.items()))
            if now - timestamp > NONCE_TTL_SECONDS:
                self._store.popitem(last=False)
            else:
                break

nonce_store = NonceStore()

from dataclasses import dataclass, field

@dataclass
class VerificationMetrics:
    total_requests: int = 0
    successful_verifications: int = 0
    latencies: list[float] = field(default_factory=list)
    
    def record(self, success: bool, latency_ms: float) -> None:
        self.total_requests += 1
        if success:
            self.successful_verifications += 1
        self.latencies.append(latency_ms)
        if len(self.latencies) > 1000:
            self.latencies = self.latencies[-1000:]
    
    def get_success_rate(self) -> float:
        return (self.successful_verifications / self.total_requests) * 100.0 if self.total_requests > 0 else 0.0
    
    def get_avg_latency_ms(self) -> float:
        return sum(self.latencies) / len(self.latencies) if self.latencies else 0.0

metrics = VerificationMetrics()

import httpx

def sync_to_siem(payload: CognigyWebhookPayload, latency_ms: float) -> None:
    audit_event = {
        "eventType": "cognigy.webhook.verified",
        "conversationId": payload.conversationId,
        "botId": payload.botId,
        "verificationLatencyMs": latency_ms,
        "timestamp": payload.timestamp,
        "metrics": {
            "successRate": metrics.get_success_rate(),
            "avgLatencyMs": metrics.get_avg_latency_ms()
        }
    }
    try:
        with httpx.Client(timeout=5.0) as client:
            response = client.post(
                SIEM_ENDPOINT,
                json=audit_event,
                headers={
                    "Authorization": f"Bearer {SIEM_API_KEY}",
                    "Content-Type": "application/json"
                }
            )
            response.raise_for_status()
            logger.info("SIEM sync successful for conversation %s", payload.conversationId)
    except httpx.HTTPError as exc:
        logger.error("SIEM sync failed: %s", exc)

# --- FastAPI Application ---
app = FastAPI(title="Cognigy Webhook Verifier", version="1.0.0")

@app.post("/webhooks/cognigy/verify")
async def verify_cognigy_webhook(request: Request) -> dict[str, Any]:
    start_time = time.perf_counter()
    
    # 1. Extract headers
    signature = request.headers.get("X-Cognigy-Signature")
    timestamp = request.headers.get("X-Cognigy-Timestamp")
    nonce = request.headers.get("X-Cognigy-Nonce")
    
    if not signature or not timestamp or not nonce:
        raise HTTPException(status_code=400, detail="Missing required verification headers")
    
    # 2. Capture raw body before parsing
    raw_body = await request.body()
    
    # 3. Schema validation
    try:
        payload = CognigyWebhookPayload.model_validate_json(raw_body)
    except ValidationError as exc:
        raise HTTPException(status_code=422, detail=f"Schema validation failed: {exc.errors()}")
    
    # 4. Signature verification
    expected_prefix = "sha256="
    if not signature.startswith(expected_prefix):
        raise HTTPException(status_code=400, detail="Invalid signature format")
    
    provided_sig = signature[len(expected_prefix):]
    computed_digest = hmac.new(
        COGNIGY_SECRET.encode("utf-8"),
        raw_body,
        hashlib.sha256
    ).hexdigest()
    
    if not hmac.compare_digest(computed_digest, provided_sig):
        raise HTTPException(status_code=401, detail="Signature verification failed")
    
    # 5. Clock skew and replay prevention
    try:
        webhook_timestamp = int(timestamp)
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid timestamp format")
    
    current_time = int(time.time())
    skew = abs(current_time - webhook_timestamp)
    
    if skew > MAX_CLOCK_SKEW_SECONDS:
        raise HTTPException(status_code=400, detail=f"Clock skew exceeded. Tolerance: {MAX_CLOCK_SKEW_SECONDS}s, Actual: {skew}s")
    
    if nonce_store.is_replay(nonce):
        raise HTTPException(status_code=409, detail="Nonce replay detected. Request rejected.")
    
    nonce_store.register(nonce)
    
    # 6. Calculate latency and record metrics
    latency_ms = (time.perf_counter() - start_time) * 1000
    metrics.record(success=True, latency_ms=latency_ms)
    
    # 7. Async SIEM sync
    sync_to_siem(payload, latency_ms)
    
    return {
        "status": "verified",
        "conversationId": payload.conversationId,
        "latencyMs": round(latency_ms, 2),
        "successRate": round(metrics.get_success_rate(), 2)
    }

@app.get("/health")
async def health_check() -> dict[str, Any]:
    return {
        "status": "healthy",
        "successRate": round(metrics.get_success_rate(), 2),
        "avgLatencyMs": round(metrics.get_avg_latency_ms(), 2),
        "totalRequests": metrics.total_requests
    }

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Run the service with:

python verifier.py

Test the endpoint using curl:

curl -X POST http://localhost:8000/webhooks/cognigy/verify \
  -H "Content-Type: application/json" \
  -H "X-Cognigy-Signature: sha256=9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08" \
  -H "X-Cognigy-Timestamp: 1698765432" \
  -H "X-Cognigy-Nonce: 550e8400-e29b-41d4-a716-446655440000" \
  -d '{"conversationId":"conv-123","botId":"bot-main","message":{"text":"Hello"},"metadata":{},"timestamp":1698765432}'

Common Errors & Debugging

Error: 400 Bad Request - Clock skew exceeded

• Cause: The difference between the server clock and the X-Cognigy-Timestamp header exceeds 300 seconds. NTP synchronization drift or delayed network routing triggers this.
• Fix: Ensure your server runs chronyd or ntpd. Adjust MAX_CLOCK_SKEW_SECONDS only if Cognigy documentation explicitly permits a higher tolerance for your region.
• Code check: Verify time.time() returns epoch seconds and matches Cognigy’s UTC standard.

Error: 401 Unauthorized - Signature verification failed

• Cause: The shared secret does not match, the raw body was modified during transit, or the signature header format lacks the sha256= prefix.
• Fix: Confirm COGNIGY_WEBHOOK_SECRET matches the value configured in the Cognigy platform. Ensure await request.body() is called before any JSON parsing or logging that might alter the byte stream.
• Code check: Validate that hmac.new uses UTF-8 encoding for the secret and raw bytes for the message.

Error: 409 Conflict - Nonce replay detected

• Cause: The same webhook was delivered twice due to Cognigy’s retry mechanism or a network duplicate.
• Fix: This is expected behavior. The verifier correctly rejects duplicates. Ensure your downstream system is idempotent based on conversationId and timestamp.
• Code check: The NonceStore automatically expires entries after 600 seconds. If you observe false positives, increase NONCE_TTL_SECONDS or implement a persistent cache like Redis for distributed deployments.

Error: 422 Unprocessable Entity - Schema validation failed

• Cause: The incoming payload lacks required fields defined in CognigyWebhookPayload or contains invalid types.
• Fix: Align your Pydantic model with the exact dialog engine output from Cognigy. Use model_validate_json to catch structural mismatches early.
• Code check: Review exc.errors() in the ValidationError handler to identify missing or malformed fields.

Official References

• NICE Cognigy Webhook Configuration Documentation
• NICE Cognigy API Reference - Webhook Delivery
• FastAPI Raw Body Handling
• Python HMAC Module Documentation
• OWASP Webhook Security Guidelines