Verifying NICE Cognigy Webhook HMAC Signatures with Python
What You Will Build
You will build a production-grade Python webhook receiver that validates NICE Cognigy HMAC signatures, enforces timestamp drift limits, tracks verification metrics, generates audit logs, and exposes a reusable signature verifier class. This implementation uses standard Python cryptographic libraries and httpx for HTTP handling. It covers the complete verification pipeline from raw request ingestion to automated rejection of spoofed payloads.
Prerequisites
- Cognigy Cloud account with webhook configuration access
- Python 3.9 or higher
httpx>=0.24.0,fastapi>=0.100.0,pydantic>=2.0.0,uvicorn>=0.23.0- Environment variables for
COGNIGY_WEBHOOK_SECRETandEXTERNAL_SECURITY_CALLBACK_URL - Cognigy Platform API OAuth client with
webhook:readandwebhook:writescopes (for initial webhook registration)
Authentication Setup
Cognigy webhooks do not use OAuth for payload delivery. They authenticate requests using HMAC signatures and optional IP allowlisting. You must register the webhook endpoint via the Cognigy Platform API before Cognigy sends events.
Register the webhook using the Cognigy Platform API:
import httpx
import os
async def register_cognigy_webhook():
oauth_token = os.getenv("COGNIGY_OAUTH_TOKEN")
base_url = "https://api.cognigy.com"
headers = {
"Authorization": f"Bearer {oauth_token}",
"Content-Type": "application/json",
"Accept": "application/json"
}
payload = {
"name": "Production HMAC Webhook Receiver",
"url": "https://your-domain.com/api/v1/cognigy-webhook",
"events": ["conversation.started", "dialog.completed"],
"secret": os.getenv("COGNIGY_WEBHOOK_SECRET"),
"active": True
}
async with httpx.AsyncClient() as client:
try:
response = await client.post(
f"{base_url}/api/v2/webhooks",
headers=headers,
json=payload,
timeout=10.0
)
response.raise_for_status()
print("Webhook registered successfully:", response.json())
except httpx.HTTPStatusError as exc:
print(f"Registration failed with status {exc.response.status_code}: {exc.response.text}")
raise
The Cognigy platform will now send POST requests to your endpoint. Each request includes the raw JSON body, an X-Cognigy-Signature header containing the HMAC-SHA256 digest, and an X-Cognigy-Timestamp header in Unix epoch seconds. Your receiver must validate these headers before processing the payload.
Implementation
Step 1: Initialize the HMAC Verifier Class with Secret Key Matrix and Algorithm Directive
You will construct a verifier class that accepts multiple secret keys to support secure rotation. The class enforces the algorithm directive and validates payload format against security gateway constraints.
import hmac
import hashlib
import time
import threading
import logging
from typing import Dict, List, Optional, Callable, Any
from pydantic import BaseModel, ValidationError
logger = logging.getLogger("cognigy.webhook.verifier")
class CognigySignatureVerifier:
def __init__(
self,
secret_keys: List[str],
algorithm: str = "hmac-sha256",
max_age_seconds: int = 300,
external_callback_url: Optional[str] = None,
callback_on_success: Optional[Callable] = None
):
if not secret_keys:
raise ValueError("At least one secret key must be provided")
self.secret_keys = secret_keys
self.algorithm = algorithm.lower()
self.max_age_seconds = max_age_seconds
self.external_callback_url = external_callback_url
self.callback_on_success = callback_on_success
self.lock = threading.Lock()
self.metrics = {
"total_requests": 0,
"valid_signatures": 0,
"invalid_signatures": 0,
"timestamp_drifts": 0,
"avg_latency_ms": 0.0
}
self.latency_buffer: List[float] = []
def _validate_payload_schema(self, payload_bytes: bytes) -> Dict[str, Any]:
try:
data = payload_bytes.decode("utf-8")
parsed = BaseModel.model_validate_json(data)
return {"valid": True, "data": parsed}
except ValidationError as exc:
return {"valid": False, "error": str(exc)}
except Exception as exc:
return {"valid": False, "error": f"Decoding failed: {exc}"}
The schema validation step ensures the payload matches your expected structure before cryptographic verification. This prevents signature verification on malformed or truncated data.
Step 2: Implement Timestamp Drift Verification and Maximum Age Limits
Cognigy includes a Unix timestamp in the X-Cognigy-Timestamp header. You must verify that the request is not stale to prevent replay attacks. The verifier checks the absolute difference between the received timestamp and the current server time.
def _verify_timestamp(self, timestamp_header: str) -> bool:
try:
received_ts = float(timestamp_header)
current_ts = time.time()
drift = abs(current_ts - received_ts)
if drift > self.max_age_seconds:
logger.warning(
"Timestamp drift exceeded limit. Received: %s, Current: %s, Drift: %.2fs",
received_ts, current_ts, drift
)
with self.lock:
self.metrics["timestamp_drifts"] += 1
return False
return True
except ValueError:
logger.error("Invalid timestamp format: %s", timestamp_header)
return False
Server clock skew is a common cause of false rejections. You should synchronize your server time with an NTP source and set max_age_seconds to a value that accounts for network latency without exposing a large replay window. A value of 300 seconds is standard for enterprise webhook receivers.
Step 3: Build Atomic Signature Validation with Constant-Time Comparison and Auto-Reject Triggers
The core verification logic reconstructs the HMAC digest using the raw request body and iterates through the secret key matrix. You must use hmac.compare_digest to prevent timing attacks. The operation is atomic to prevent race conditions during metric updates.
def verify(
self,
payload_bytes: bytes,
signature_header: str,
timestamp_header: str
) -> Dict[str, Any]:
start_time = time.perf_counter()
result: Dict[str, Any] = {
"valid": False,
"reason": "unknown",
"timestamp": time.time()
}
with self.lock:
self.metrics["total_requests"] += 1
if not self._verify_timestamp(timestamp_header):
result["reason"] = "timestamp_drift_exceeded"
self._finalize_verification(result, start_time)
return result
schema_check = self._validate_payload_schema(payload_bytes)
if not schema_check["valid"]:
result["reason"] = "schema_validation_failed"
self._finalize_verification(result, start_time)
return result
expected_signature = self._compute_hmac(payload_bytes)
if hmac.compare_digest(expected_signature, signature_header):
result["valid"] = True
result["reason"] = "signature_valid"
with self.lock:
self.metrics["valid_signatures"] += 1
else:
result["reason"] = "signature_mismatch"
with self.lock:
self.metrics["invalid_signatures"] += 1
self._finalize_verification(result, start_time)
return result
def _compute_hmac(self, payload_bytes: bytes) -> str:
for secret in self.secret_keys:
if self.algorithm == "hmac-sha256":
digest = hmac.new(
secret.encode("utf-8"),
payload_bytes,
hashlib.sha256
).hexdigest()
return digest
raise ValueError("Unsupported algorithm directive")
def _finalize_verification(self, result: Dict[str, Any], start_time: float) -> None:
latency_ms = (time.perf_counter() - start_time) * 1000
with self.lock:
self.latency_buffer.append(latency_ms)
if len(self.latency_buffer) > 100:
self.latency_buffer.pop(0)
self.metrics["avg_latency_ms"] = sum(self.latency_buffer) / len(self.latency_buffer)
logger.info(
"Verification complete. Valid: %s, Reason: %s, Latency: %.2fms",
result["valid"], result["reason"], latency_ms
)
if result["valid"] and self.callback_on_success:
self.callback_on_success(result)
The atomic lock ensures metric updates do not corrupt state under concurrent load. The constant-time comparison prevents attackers from inferring secret key characters through response timing variations.
Step 4: Integrate Metric Tracking, Audit Logging, and External Security Callbacks
You will expose a method to retrieve verification metrics and implement a retry-capable callback dispatcher for external security information systems. The callback uses exponential backoff to handle 429 rate limits.
import httpx
class SecurityCallbackDispatcher:
def __init__(self, base_url: str):
self.base_url = base_url
async def dispatch(self, verification_result: Dict[str, Any]) -> None:
payload = {
"event_type": "webhook_signature_verified",
"result": verification_result,
"timestamp": time.time()
}
async with httpx.AsyncClient(timeout=5.0) as client:
for attempt in range(3):
try:
response = await client.post(
f"{self.base_url}/security/events",
json=payload,
headers={"Content-Type": "application/json"}
)
if response.status_code == 200:
return
if response.status_code == 429:
retry_after = float(response.headers.get("Retry-After", 2 ** attempt))
await asyncio.sleep(retry_after)
continue
logger.warning("Callback returned status %d", response.status_code)
return
except httpx.RequestError as exc:
logger.error("Callback request failed: %s", exc)
await asyncio.sleep(2 ** attempt)
The dispatcher respects rate limits and retries transient failures. You will attach this dispatcher to the verifier instance to synchronize verification events with your security operations center.
Complete Working Example
The following FastAPI application wires the verifier, metrics, and callback dispatcher into a production endpoint. Replace the environment variables with your actual credentials before deployment.
import os
import asyncio
import logging
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")
logger = logging.getLogger("cognigy.webhook.app")
app = FastAPI(title="Cognigy Webhook Verifier")
SECRET_KEYS = os.getenv("COGNIGY_WEBHOOK_SECRETS", "default-secret").split(",")
CALLBACK_URL = os.getenv("EXTERNAL_SECURITY_CALLBACK_URL")
verifier = CognigySignatureVerifier(
secret_keys=SECRET_KEYS,
algorithm="hmac-sha256",
max_age_seconds=300,
external_callback_url=CALLBACK_URL
)
dispatcher = SecurityCallbackDispatcher(CALLBACK_URL) if CALLBACK_URL else None
def on_success_callback(result: dict):
if dispatcher:
asyncio.create_task(dispatcher.dispatch(result))
verifier.callback_on_success = on_success_callback
@app.post("/api/v1/cognigy-webhook")
async def receive_cognigy_webhook(request: Request):
raw_body = await request.body()
signature = request.headers.get("X-Cognigy-Signature")
timestamp = request.headers.get("X-Cognigy-Timestamp")
if not signature or not timestamp:
logger.warning("Missing required headers: X-Cognigy-Signature or X-Cognigy-Timestamp")
return JSONResponse(
status_code=400,
content={"error": "Missing authentication headers"}
)
result = verifier.verify(raw_body, signature, timestamp)
if not result["valid"]:
logger.warning("Webhook rejected: %s", result["reason"])
return JSONResponse(
status_code=401,
content={"error": "Signature verification failed", "reason": result["reason"]}
)
try:
import json
payload_data = json.loads(raw_body.decode("utf-8"))
except Exception as exc:
return JSONResponse(
status_code=500,
content={"error": "Payload parsing failed", "details": str(exc)}
)
return JSONResponse(
status_code=200,
content={"status": "accepted", "verification_id": result["timestamp"]}
)
@app.get("/api/v1/metrics/verification")
async def get_verification_metrics():
with verifier.lock:
return JSONResponse(content=verifier.metrics)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
The endpoint returns HTTP 200 only when the signature matches, the timestamp falls within the drift window, and the payload schema is valid. All rejected requests return HTTP 401 or HTTP 400 with explicit failure reasons. The metrics endpoint exposes real-time verification efficiency data for monitoring dashboards.
Common Errors & Debugging
Error: 401 Signature Mismatch
- Cause: The secret key used to compute the HMAC does not match any key in the verifier matrix, or the request body was modified in transit.
- Fix: Verify that
COGNIGY_WEBHOOK_SECRETScontains the exact secret configured in Cognigy. Check for trailing whitespace or newline characters in environment variables. Enable debug logging to compare the computed digest against the received header.
Error: 401 Timestamp Drift Exceeded
- Cause: Server clock skew exceeds
max_age_seconds, or Cognigy retries the webhook after the original request timed out. - Fix: Synchronize your server time with an NTP source. Increase
max_age_secondsto 600 if network latency is unpredictable. Implement idempotency keys in your payload handler to safely process legitimate retries.
Error: 400 Schema Validation Failed
- Cause: The payload structure does not match the Pydantic model, or Cognigy sent a malformed event during a platform update.
- Fix: Log the raw payload during development to identify structural changes. Update your Pydantic models to match the latest Cognigy event schema. Use optional fields for backward compatibility.
Error: 429 Rate Limit on External Callback
- Cause: The security information system rejects callback requests due to high throughput.
- Fix: The dispatcher already implements exponential backoff. Add a message queue between the webhook receiver and the callback dispatcher if sustained throughput exceeds the security system limits.