Webhooks¶

verify_signature ¶

verify_signature(
    payload: bytes,
    secret: str | bytes,
    signature: str,
    *,
    header_prefix: str = "sha256=",
) -> None

Verify an HMAC-SHA256 webhook signature using constant-time comparison.

Parameters:

Name	Type	Description	Default
`payload`	`bytes`	Raw request body bytes.	required
`secret`	`str \| bytes`	Webhook signing secret (str or bytes).	required
`signature`	`str`	The value from the provider's signature header (e.g. `"sha256=abc123…"`).	required
`header_prefix`	`str`	Expected prefix on the signature value (default: `"sha256="`).	`'sha256='`

Raises:

Type	Description
`WebhookVerificationError`	If the signature does not match.

Source code in merchants/webhooks.py

def verify_signature(
    payload: bytes,
    secret: str | bytes,
    signature: str,
    *,
    header_prefix: str = "sha256=",
) -> None:
    """Verify an HMAC-SHA256 webhook signature using constant-time comparison.

    Args:
        payload: Raw request body bytes.
        secret: Webhook signing secret (str or bytes).
        signature: The value from the provider's signature header
            (e.g. ``"sha256=abc123…"``).
        header_prefix: Expected prefix on the signature value
            (default: ``"sha256="``).

    Raises:
        WebhookVerificationError: If the signature does not match.
    """
    if isinstance(secret, str):
        secret = secret.encode()

    expected_hex = hmac.new(secret, payload, hashlib.sha256).hexdigest()

    # Strip optional prefix from provided signature
    provided = signature
    if provided.startswith(header_prefix):
        provided = provided[len(header_prefix) :]

    if not hmac.compare_digest(expected_hex, provided):
        raise WebhookVerificationError("Webhook signature verification failed.")

verify_khipu_signature ¶

verify_khipu_signature(
    payload: bytes, secret: str | bytes, header_value: str
) -> str

Verify a Khipu v3.0 webhook signature (x-khipu-signature header).

Khipu signs webhooks with HMAC-SHA256 using the format t=<unix_ms>,s=<base64_signature>. The signed message is "<timestamp>.<body>".

Parameters:

Name	Type	Description	Default
`payload`	`bytes`	Raw request body bytes (the JSON as sent by Khipu — must not be re-serialised or whitespace-modified).	required
`secret`	`str \| bytes`	Merchant secret (the Khipu receiver secret key).	required
`header_value`	`str`	The full `x-khipu-signature` header value, e.g. `"t=1711965600393,s=GYzp…Tdg="`.	required

Returns:

Type	Description
`str`	The extracted timestamp string (useful for replay-attack checks).

Raises:

Type	Description
`WebhookVerificationError`	If the header is malformed or the signature does not match.

Source code in merchants/webhooks.py

def verify_khipu_signature(
    payload: bytes,
    secret: str | bytes,
    header_value: str,
) -> str:
    """Verify a Khipu v3.0 webhook signature (``x-khipu-signature`` header).

    Khipu signs webhooks with HMAC-SHA256 using the format
    ``t=<unix_ms>,s=<base64_signature>``.  The signed message is
    ``"<timestamp>.<body>"``.

    Args:
        payload: Raw request body bytes (the JSON as sent by Khipu — must
            not be re-serialised or whitespace-modified).
        secret: Merchant secret (the Khipu receiver secret key).
        header_value: The full ``x-khipu-signature`` header value,
            e.g. ``"t=1711965600393,s=GYzp…Tdg="``.

    Returns:
        The extracted timestamp string (useful for replay-attack checks).

    Raises:
        WebhookVerificationError: If the header is malformed or the
            signature does not match.
    """
    if isinstance(secret, str):
        secret = secret.encode()

    # Parse "t=<timestamp>,s=<signature>"
    t_value: str | None = None
    s_value: str | None = None
    for part in header_value.split(","):
        key, _, val = part.partition("=")
        if key == "t":
            t_value = val
        elif key == "s":
            s_value = val

    if not t_value or not s_value:
        raise WebhookVerificationError(
            "Malformed x-khipu-signature header: missing t= or s= component."
        )

    # Build the signed string: "<timestamp>.<body>"
    to_hash = f"{t_value}.".encode() + payload
    expected_digest = hmac.new(secret, to_hash, hashlib.sha256).digest()
    expected_b64 = base64.b64encode(expected_digest).decode()

    if not hmac.compare_digest(expected_b64, s_value):
        raise WebhookVerificationError("Khipu webhook signature verification failed.")

    return t_value

parse_event ¶

parse_event(
    payload: bytes, *, provider: str = "unknown"
) -> WebhookEvent

Best-effort parse and normalisation of a raw webhook payload.

Tries to extract common fields (id, event_type/type, payment_id, status) regardless of provider format.

Parameters:

Name	Type	Description	Default
`payload`	`bytes`	Raw request body bytes.	required
`provider`	`str`	Provider name hint for the returned event.	`'unknown'`

Returns:

Name	Type	Description
`A`	`WebhookEvent`	class:`~merchants.models.WebhookEvent`. Fields that cannot be
	`WebhookEvent`	extracted are left as `None` / `PaymentState.UNKNOWN`.

Source code in merchants/webhooks.py

def parse_event(
    payload: bytes,
    *,
    provider: str = "unknown",
) -> WebhookEvent:
    """Best-effort parse and normalisation of a raw webhook payload.

    Tries to extract common fields (``id``, ``event_type``/``type``,
    ``payment_id``, ``status``) regardless of provider format.

    Args:
        payload: Raw request body bytes.
        provider: Provider name hint for the returned event.

    Returns:
        A :class:`~merchants.models.WebhookEvent`.  Fields that cannot be
        extracted are left as ``None`` / ``PaymentState.UNKNOWN``.
    """
    try:
        data: dict[str, Any] = json.loads(payload)
    except (ValueError, TypeError):
        data = {}

    event_id = data.get("id") or data.get("event_id")
    event_type = str(data.get("type") or data.get("event_type") or "unknown")
    payment_id = (
        data.get("payment_id")
        or data.get("resource", {}).get("id")
        or data.get("data", {}).get("object", {}).get("id")
    )

    # Try to extract a raw status from common structures
    raw_status: str = (
        data.get("status")
        or data.get("data", {}).get("object", {}).get("status")
        or data.get("resource", {}).get("status")
        or "unknown"
    )
    state: PaymentState = normalise_state(str(raw_status))

    return WebhookEvent(
        event_id=event_id,
        event_type=event_type,
        payment_id=payment_id,
        state=state,
        provider=provider,
        raw=data,
    )

WebhookVerificationError ¶

Bases: Exception

Raised when webhook HMAC signature verification fails.

Source code in merchants/webhooks.py

class WebhookVerificationError(Exception):
    """Raised when webhook HMAC signature verification fails."""