Generate Novu HMAC header signature for bridge endpoint: https://docs.novu.co/framework/endpoint

novu-signature.py

import hashlib
import hmac
import json
import time
from os import getenv
from typing import Any


def generate_novu_signature(secret_key: str, payload: Any) -> str:
    """
    Generate Novu signature header.

    Args:
        secret_key: Your Novu secret key
        payload: The request payload (will be JSON stringified)

    Returns:
        Signature header in format: "t=TIMESTAMP,v1=SIGNATURE"
    """
    timestamp = str(int(time.time() * 1000))  # Unix timestamp in milliseconds

    # Create the message to sign: timestamp.payload
    message = f"{timestamp}.{json.dumps(payload, separators=(',', ':'))}"

    # Generate HMAC-SHA256 signature
    signature = hmac.new(
        secret_key.encode("utf-8"), message.encode("utf-8"), hashlib.sha256
    ).hexdigest()

    return f"t={timestamp},v1={signature}"


def verify_novu_signature(
    secret_key: str, payload: Any, signature_header: str, tolerance_ms: int = 300000
) -> bool:
    """
    Verify Novu signature header.

    Args:
        secret_key: Your Novu secret key
        payload: The request payload
        signature_header: The "t=TIMESTAMP,v1=SIGNATURE" header
        tolerance_ms: Timestamp tolerance in milliseconds (default 5 minutes)

    Returns:
        True if signature is valid, False otherwise
    """
    try:
        # Parse signature header
        parts = signature_header.split(",")
        if len(parts) != 2:
            return False

        timestamp_part = parts[0]
        signature_part = parts[1]

        if not timestamp_part.startswith("t=") or not signature_part.startswith("v1="):
            return False

        timestamp = int(timestamp_part[2:])
        provided_signature = signature_part[3:]

        # Check timestamp tolerance
        current_time = int(time.time() * 1000)
        if abs(current_time - timestamp) > tolerance_ms:
            return False

        # Generate expected signature
        message = f"{timestamp}.{json.dumps(payload, separators=(',', ':'))}"
        expected_signature = hmac.new(
            secret_key.encode("utf-8"), message.encode("utf-8"), hashlib.sha256
        ).hexdigest()

        return hmac.compare_digest(expected_signature, provided_signature)

    except (ValueError, KeyError):
        return False


# Generate signature for a request payload
secret_key = getenv("NOVU_SECRET_KEY")

if not secret_key:
    raise ValueError(
        "Did you forget to set the environment variable `NOVU_SECRET_KEY`?"
    )

payload = {
    "to": "user@example.com",
    "payload": {"name": "John"},
    "workflowId": "welcome-email",
}

signature = generate_novu_signature(secret_key, payload)

# Replace `example.com` with your bridge endpoint
print(
    f"curl -X GET 'https://example.com/api/novu?action=discover' -H 'Novu-Signature: {signature}'"
)