Verify Webhook Signatures with HMAC

import hashlib
import hmac
import time
from dataclasses import dataclass


class SignatureError(ValueError):
    pass


@dataclass(frozen=True)
class SignedRequest:
    timestamp: int
    signature: str


def parse_header(header: str) -> SignedRequest:
    fields: dict[str, str] = {}
    for part in header.split(","):
        key, separator, value = part.strip().partition("=")
        if not separator:
            raise SignatureError(f"malformed header segment: {part!r}")
        fields[key] = value

    try:
        timestamp = int(fields["t"])
        signature = fields["v1"]
    except (KeyError, ValueError) as error:
        raise SignatureError("header must contain integer t and v1") from error

    if len(signature) != 64:
        raise SignatureError("v1 must be a 64-character hex digest")
    try:
        bytes.fromhex(signature)
    except ValueError as error:
        raise SignatureError("v1 must be hexadecimal") from error
    return SignedRequest(timestamp=timestamp, signature=signature)


def sign(payload: bytes, secret: bytes, timestamp: int) -> str:
    message = f"{timestamp}.".encode("utf-8") + payload
    digest = hmac.new(secret, message, hashlib.sha256).hexdigest()
    return f"t={timestamp},v1={digest}"


def verify(
    payload: bytes,
    header: str,
    secret: bytes,
    *,
    tolerance_seconds: int = 300,
    now: int | None = None,
) -> None:
    signed = parse_header(header)
    current = int(time.time()) if now is None else now
    if abs(current - signed.timestamp) > tolerance_seconds:
        raise SignatureError("timestamp is outside the tolerance window")

    message = f"{signed.timestamp}.".encode("utf-8") + payload
    expected = hmac.new(secret, message, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, signed.signature):
        raise SignatureError("signature mismatch")


def main() -> None:
    secret = b"whsec_demo_key"
    payload = b'{"event":"payment.succeeded","id":"evt_1"}'
    timestamp = 1_800_000_000
    header = sign(payload, secret, timestamp)
    print(header)

    verify(payload, header, secret, now=timestamp + 10)
    print("valid signature accepted")

    try:
        verify(payload + b" ", header, secret, now=timestamp + 10)
    except SignatureError as error:
        print("tampered payload:", error)

    try:
        verify(payload, header, secret, now=timestamp + 600)
    except SignatureError as error:
        print("expired timestamp:", error)


if __name__ == "__main__":
    main()