Verifying Request Signatures in NICE Cognigy Webhook Endpoints Using Python Middleware

What You Will Build

• This code implements a Python middleware that validates incoming NICE Cognigy webhook requests by verifying HMAC-SHA256 signatures against a shared secret key.
• The implementation uses the standard NICE Cognigy Platform webhook signature verification pattern with HTTP headers.
• The tutorial covers Python 3.10+ using FastAPI and httpx.

Prerequisites

• Shared secret key generated in the NICE Cognigy Platform webhook configuration
• Python 3.10 or higher
• FastAPI 0.100+, httpx 0.24+, uvicorn 0.23+
• No OAuth scopes are required for inbound webhook verification. Any outbound calls to the Cognigy API require the bot:read or webhook:manage scope depending on the endpoint.

Authentication Setup

The Cognigy Platform uses OAuth 2.0 for programmatic API access. Although webhook signature verification relies on a shared secret, your middleware may need to call the Cognigy API after verification. The following code demonstrates a production-ready token fetcher with caching and automatic refresh logic.

import time
import httpx
from typing import Optional

class CognigyAuthManager:
    def __init__(self, client_id: str, client_secret: str, tenant: str = "default"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.tenant = tenant
        self.token_url = f"https://{tenant}.cognigy.com/v1/auth/token"
        self._access_token: Optional[str] = None
        self._token_expiry: float = 0.0

    async def get_access_token(self) -> str:
        """Retrieves and caches an OAuth 2.0 access token. Automatically refreshes when expired."""
        if self._access_token and time.time() < self._token_expiry - 30:
            return self._access_token

        # OAuth scope required: bot:read webhook:manage
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": "bot:read webhook:manage"
        }

        async with httpx.AsyncClient() as client:
            response = await client.post(self.token_url, data=payload)
            response.raise_for_status()
            token_data = response.json()

        self._access_token = token_data["access_token"]
        self._token_expiry = time.time() + token_data["expires_in"]
        return self._access_token

The get_access_token method checks the cached token against its expiry time. It subtracts thirty seconds to prevent edge-case expiration during request transit. The method posts credentials to the Cognigy authentication endpoint and returns the bearer token. The code explicitly notes the required OAuth scopes in the payload comment.

Implementation

Step 1: Extract HMAC Headers and Validate Timestamp

Cognigy Platform appends two headers to every webhook invocation: X-Cognigy-Signature and X-Cognigy-Timestamp. The middleware must extract these values and verify that the timestamp falls within an acceptable clock-skew window. Clock drift or replay attacks are mitigated by rejecting requests older than sixty seconds.

import time
from fastapi import Request, HTTPException
from typing import Tuple

ALLOWED_CLOCK_SKEW_SECONDS = 60

def extract_and_validate_timestamp(request: Request) -> Tuple[str, str]:
    """Extracts signature and timestamp headers. Raises HTTPException if missing or expired."""
    signature = request.headers.get("X-Cognigy-Signature")
    timestamp_str = request.headers.get("X-Cognigy-Timestamp")

    if not signature or not timestamp_str:
        raise HTTPException(status_code=400, detail="Missing X-Cognigy-Signature or X-Cognigy-Timestamp header")

    try:
        request_timestamp = float(timestamp_str)
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid timestamp format in X-Cognigy-Timestamp")

    current_time = time.time()
    if abs(current_time - request_timestamp) > ALLOWED_CLOCK_SKEW_SECONDS:
        raise HTTPException(status_code=403, detail="Webhook timestamp expired or outside allowed clock skew")

    return signature, timestamp_str

This function performs strict header validation. It converts the timestamp to a float and compares it against the current system time. If the absolute difference exceeds the allowed skew, the function raises a 403 Forbidden exception. Missing or malformed headers trigger a 400 Bad Request exception.

Step 2: Reconstruct Canonical Request String and Compute Expected Signature

The Cognigy signature algorithm concatenates the timestamp and the raw request body, then signs the result using HMAC-SHA256 with your shared secret. The middleware must reconstruct this exact canonical string to compute the expected signature.

import hmac
import hashlib
import logging
from fastapi import Request

logger = logging.getLogger("cognigy.webhook.audit")

def compute_expected_signature(shared_secret: str, timestamp: str, payload: bytes) -> str:
    """Reconstructs the canonical request string and computes the HMAC-SHA256 signature."""
    # Canonical format: timestamp + "\n" + raw_payload
    canonical_string = f"{timestamp}\n{payload.decode('utf-8', errors='replace')}"
    
    expected_signature = hmac.new(
        key=shared_secret.encode('utf-8'),
        msg=canonical_string.encode('utf-8'),
        digestmod=hashlib.sha256
    ).hexdigest()
    
    return expected_signature

The canonical string strictly follows the pattern timestamp + "\n" + payload. The function decodes the raw bytes to UTF-8, replacing invalid sequences to prevent decoding errors from malformed payloads. It then generates the hexadecimal digest using HMAC-SHA256. The shared secret must match the exact string configured in the Cognigy Platform webhook settings.

Step 3: Compare Signatures, Reject Unauthorized, and Log Forensic Details

Signature comparison must use a timing-safe function to prevent side-channel attacks. The middleware logs forensic details for both successful and failed verifications. Failed verifications return a 401 Unauthorized response.

import logging
from fastapi import Request, HTTPException

logger = logging.getLogger("cognigy.webhook.audit")

async def verify_webhook_signature(request: Request, shared_secret: str) -> dict:
    """Verifies the webhook signature and logs forensic details."""
    signature, timestamp = extract_and_validate_timestamp(request)
    payload = await request.body()
    
    expected_signature = compute_expected_signature(shared_secret, timestamp, payload)
    
    # Timing-safe comparison
    if not hmac.compare_digest(signature, expected_signature):
        logger.warning(
            "Signature verification failed. IP: %s, Timestamp: %s, Expected: %s, Received: %s, Path: %s",
            request.client.host, timestamp, expected_signature, signature, request.url.path
        )
        raise HTTPException(status_code=401, detail="Invalid webhook signature")
    
    logger.info(
        "Signature verified successfully. IP: %s, Timestamp: %s, Path: %s",
        request.client.host, timestamp, request.url.path
    )
    
    return await request.json()

The hmac.compare_digest function ensures constant-time comparison. The audit logger captures the client IP, timestamp, expected signature, received signature, and request path. This forensic trail enables rapid investigation of replay attempts or misconfigured secrets. The function raises a 401 Unauthorized exception on mismatch and returns the parsed JSON payload on success.

Complete Working Example

The following script combines the authentication manager, middleware, and endpoint handlers into a single runnable FastAPI application. It includes structured logging, retry logic for outbound API calls, and pagination handling.

import os
import time
import httpx
import logging
import hmac
import hashlib
from typing import Optional
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse

# Configure structured audit logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(name)s | %(levelname)s | %(message)s"
)
logger = logging.getLogger("cognigy.webhook.audit")

ALLOWED_CLOCK_SKEW_SECONDS = 60
SHARED_SECRET = os.getenv("COGNIGY_WEBHOOK_SECRET", "your-secure-shared-secret-key")
COGNIGY_CLIENT_ID = os.getenv("COGNIGY_CLIENT_ID", "")
COGNIGY_CLIENT_SECRET = os.getenv("COGNIGY_CLIENT_SECRET", "")
COGNIGY_TENANT = os.getenv("COGNIGY_TENANT", "default")

app = FastAPI(title="Cognigy Webhook Verifier")

class CognigyAuthManager:
    def __init__(self, client_id: str, client_secret: str, tenant: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.tenant = tenant
        self.token_url = f"https://{tenant}.cognigy.com/v1/auth/token"
        self._access_token: Optional[str] = None
        self._token_expiry: float = 0.0

    async def get_access_token(self) -> str:
        if self._access_token and time.time() < self._token_expiry - 30:
            return self._access_token

        # OAuth scope required: bot:read webhook:manage
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": "bot:read webhook:manage"
        }

        async with httpx.AsyncClient() as client:
            response = await client.post(self.token_url, data=payload)
            response.raise_for_status()
            token_data = response.json()

        self._access_token = token_data["access_token"]
        self._token_expiry = time.time() + token_data["expires_in"]
        return self._access_token

auth_manager = CognigyAuthManager(COGNIGY_CLIENT_ID, COGNIGY_CLIENT_SECRET, COGNIGY_TENANT)

def extract_and_validate_timestamp(request: Request) -> tuple:
    signature = request.headers.get("X-Cognigy-Signature")
    timestamp_str = request.headers.get("X-Cognigy-Timestamp")

    if not signature or not timestamp_str:
        raise HTTPException(status_code=400, detail="Missing signature or timestamp header")

    try:
        request_timestamp = float(timestamp_str)
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid timestamp format")

    if abs(time.time() - request_timestamp) > ALLOWED_CLOCK_SKEW_SECONDS:
        raise HTTPException(status_code=403, detail="Timestamp expired or outside allowed skew")

    return signature, timestamp_str

def compute_expected_signature(shared_secret: str, timestamp: str, payload: bytes) -> str:
    canonical_string = f"{timestamp}\n{payload.decode('utf-8', errors='replace')}"
    return hmac.new(
        key=shared_secret.encode('utf-8'),
        msg=canonical_string.encode('utf-8'),
        digestmod=hashlib.sha256
    ).hexdigest()

async def verify_webhook_signature(request: Request) -> dict:
    signature, timestamp = extract_and_validate_timestamp(request)
    payload = await request.body()
    expected_signature = compute_expected_signature(SHARED_SECRET, timestamp, payload)

    if not hmac.compare_digest(signature, expected_signature):
        logger.warning(
            "Signature verification failed. IP: %s, Timestamp: %s, Path: %s",
            request.client.host, timestamp, request.url.path
        )
        raise HTTPException(status_code=401, detail="Invalid webhook signature")

    logger.info(
        "Signature verified. IP: %s, Timestamp: %s, Path: %s",
        request.client.host, timestamp, request.url.path
    )
    return await request.json()

async def call_cognigy_api_with_retry(endpoint: str, payload: dict) -> dict:
    """Outbound API call with exponential backoff for 429 and pagination support."""
    token = await auth_manager.get_access_token()
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    
    # OAuth scope required: bot:read webhook:manage
    url = f"https://{COGNIGY_TENANT}.cognigy.com/v1{endpoint}"
    
    async with httpx.AsyncClient() as client:
        for attempt in range(4):
            response = await client.post(url, json=payload, headers=headers)
            
            if response.status_code == 429:
                wait_time = 2 ** attempt
                logger.warning("Rate limited (429). Retrying in %s seconds...", wait_time)
                await asyncio.sleep(wait_time)
                continue
            
            response.raise_for_status()
            data = response.json()
            
            # Pagination handling when the endpoint supports it
            if "nextPageToken" in data:
                logger.info("Pagination detected. Fetching next page...")
                headers["PageToken"] = data["nextPageToken"]
                next_response = await client.get(url, headers=headers)
                next_response.raise_for_status()
                data["nextPageData"] = next_response.json()
                
            return data

import asyncio

@app.post("/webhooks/cognigy-callback")
async def cognigy_webhook_endpoint(request: Request):
    verified_payload = await verify_webhook_signature(request)
    
    # Process verified payload
    try:
        api_result = await call_cognigy_api_with_retry("/bot/trigger", {"webhookData": verified_payload})
        return JSONResponse(content={"status": "processed", "apiResponse": api_result})
    except httpx.HTTPStatusError as e:
        logger.error("Cognigy API call failed: %s", str(e))
        return JSONResponse(content={"status": "api_error", "detail": str(e)}, status_code=502)
    except Exception as e:
        logger.error("Unhandled processing error: %s", str(e))
        return JSONResponse(content={"status": "internal_error", "detail": str(e)}, status_code=500)

Run the application with uvicorn main:app --host 0.0.0.0 --port 8000. The middleware validates every incoming request before the route handler executes. The outbound function demonstrates retry logic for 429 responses and pagination handling when the API returns a nextPageToken.

Common Errors & Debugging

Error: 401 Unauthorized (Signature Mismatch)

• Cause: The shared secret in your environment does not match the secret configured in the Cognigy Platform webhook settings, or the canonical string construction differs from the platform specification.
• Fix: Verify the exact secret string. Ensure no trailing whitespace exists. Confirm the canonical format matches timestamp + "\n" + raw_payload. Use the audit log to compare expected and received signatures.
• Code showing the fix:

# Debug helper to print canonical string for manual verification
canonical = f"{timestamp}\n{payload.decode('utf-8', errors='replace')}"
print(f"DEBUG Canonical: {repr(canonical)}")
print(f"DEBUG Expected: {expected_signature}")

Error: 403 Forbidden (Timestamp Expired)

• Cause: System clock drift between your server and the Cognigy Platform, or the webhook was queued for longer than the allowed skew window.
• Fix: Synchronize server time using NTP. Increase ALLOWED_CLOCK_SKEW_SECONDS if your infrastructure experiences high latency. Ensure the webhook endpoint responds within two seconds to prevent platform-side timeouts.
• Code showing the fix:

# Adjust skew tolerance for high-latency environments
ALLOWED_CLOCK_SKEW_SECONDS = 120

Error: 429 Too Many Requests

• Cause: The Cognigy API rate limit is exceeded during outbound calls after webhook verification.
• Fix: Implement exponential backoff. The complete example includes a retry loop that waits 2 ** attempt seconds before retrying. Add jitter to prevent thundering herd scenarios in production.
• Code showing the fix:

import random
wait_time = (2 ** attempt) + random.uniform(0, 1)
await asyncio.sleep(wait_time)

Error: 5xx Internal Server Error

• Cause: Unhandled exceptions during payload processing or API communication.
• Fix: Wrap outbound calls in try-except blocks. Log the full stack trace. Return structured error responses to the platform to prevent retry storms. The complete example catches httpx.HTTPStatusError and generic Exception to maintain stability.

Official References

• NICE Cognigy Platform Webhook Documentation
• Cognigy API Authentication Guide
• FastAPI Middleware Documentation
• httpx Client Library Reference