Function validate_webhook_headers

evi/evi-python-webhooks/utils.py:53–98

Validates the HMAC signature and timestamp of an incoming webhook request. Ensures the request was sent by Hume and has not been tampered with or replayed. Args: payload: The raw request payload as a string. headers: The headers from the incoming request. Raises:

(payload: str, headers: Headers)

def validate_webhook_headers(payload: str, headers: Headers) -> None:
    """
    Validates the HMAC signature and timestamp of an incoming webhook request.
    Ensures the request was sent by Hume and has not been tampered with or replayed.

    Args:
        payload: The raw request payload as a string.
        headers: The headers from the incoming request.

    Raises:
        ValueError: If headers are missing, the signature is invalid, or the timestamp is stale.
    """
    timestamp = headers.get("X-Hume-AI-Webhook-Timestamp")
    signature = headers.get("X-Hume-AI-Webhook-Signature")

    if not signature:
        raise ValueError("Missing HMAC signature")

    if not timestamp:
        raise ValueError("Missing timestamp")

    # Validate HMAC signature
    signing_key = os.environ.get("HUME_WEBHOOK_SIGNING_KEY")
    if not signing_key:
        raise ValueError("HUME_WEBHOOK_SIGNING_KEY is not set in environment variables")

    message = (payload + "." + timestamp).encode("utf-8")
    expected_sig = hmac.new(
        key=signing_key.encode("utf-8"),
        msg=message,
        digestmod=hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(signature, expected_sig):
        raise ValueError("Invalid HMAC signature")

    # Validate timestamp to prevent replay attacks
    try:
        timestamp_int = int(timestamp)
    except ValueError:
        raise ValueError("Invalid timestamp format")

    current_time = int(time.time())
    TIMESTAMP_VALIDATION_WINDOW = 180
    if current_time - timestamp_int > TIMESTAMP_VALIDATION_WINDOW:
        raise ValueError("The timestamp on the request is too old")


async def fetch_weather(parameters: str) -> str: