Webhooks API | fal.ai Reference

pip install pynacl requests

import base64
import hashlib
import time
from typing import Optional
import requests
from nacl.signing import VerifyKey
from nacl.exceptions import BadSignatureError
from nacl.encoding import HexEncoder

JWKS_URL = "https://rest.alpha.fal.ai/.well-known/jwks.json"
JWKS_CACHE_DURATION = 24 * 60 * 60  # 24 hours in seconds
_jwks_cache = None
_jwks_cache_time = 0

def fetch_jwks() -> list:
    """Fetch and cache JWKS, refreshing after 24 hours."""
    global _jwks_cache, _jwks_cache_time
    current_time = time.time()
    if _jwks_cache is None or (current_time - _jwks_cache_time) > JWKS_CACHE_DURATION:
        response = requests.get(JWKS_URL, timeout=10)
        response.raise_for_status()
        _jwks_cache = response.json().get("keys", [])
        _jwks_cache_time = current_time
    return _jwks_cache

def verify_webhook_signature(
    request_id: str,
    user_id: str,
    timestamp: str,
    signature_hex: str,
    body: bytes
) -> bool:
    """
    Verify a webhook signature using provided headers and body.

    Args:
        request_id: Value of X-Fal-Webhook-Request-Id header.
        user_id: Value of X-Fal-Webhook-User-Id header.
        timestamp: Value of X-Fal-Webhook-Timestamp header.
        signature_hex: Value of X-Fal-Webhook-Signature header (hex-encoded).
        body: Raw request body as bytes.

    Returns:
        bool: True if the signature is valid, False otherwise.
    """
    # Validate timestamp (within ±5 minutes)
    try:
        timestamp_int = int(timestamp)
        current_time = int(time.time())
        if abs(current_time - timestamp_int) > 300:
            print("Timestamp is too old or in the future.")
            return False
    except ValueError:
        print("Invalid timestamp format.")
        return False

    # Construct the message to verify
    try:
        message_parts = [
            request_id,
            user_id,
            timestamp,
            hashlib.sha256(body).hexdigest()
        ]
        if any(part is None for part in message_parts):
            print("Missing required header value.")
            return False
        message_to_verify = "\n".join(message_parts).encode("utf-8")
    except Exception as e:
        print(f"Error constructing message: {e}")
        return False

    # Decode signature
    try:
        signature_bytes = bytes.fromhex(signature_hex)
    except ValueError:
        print("Invalid signature format (not hexadecimal).")
        return False

    # Fetch public keys
    try:
        public_keys_info = fetch_jwks()
        if not public_keys_info:
            print("No public keys found in JWKS.")
            return False
    except Exception as e:
        print(f"Error fetching JWKS: {e}")
        return False

    # Verify signature with each public key
    for key_info in public_keys_info:
        try:
            public_key_b64url = key_info.get("x")
            if not isinstance(public_key_b64url, str):
                continue
            public_key_bytes = base64.urlsafe_b64decode(public_key_b64url)
            verify_key = VerifyKey(public_key_bytes.hex(), encoder=HexEncoder)
            verify_key.verify(message_to_verify, signature_bytes)
            return True
        except (BadSignatureError, Exception) as e:
            print(f"Verification failed with a key: {e}")
            continue

    print("Signature verification failed with all keys.")
    return False