kindgi.webhooks

Verify the webhooks Kindgi sends (run.finished, webhook.test).

Requests are signed in the Standard Webhooks format (https://www.standardwebhooks.com), symmetric variant v1:

webhook-id:        the event id (the same on every retry)
webhook-timestamp: Unix seconds when the request was signed
webhook-signature: `v1,<base64 HMAC-SHA256>`, space-separated when
                   several secrets sign (secret rotation)

The signed content is {id}.{timestamp}.{body}, with body the raw request body exactly as sent. Secrets are whsec_ + base64 of the key bytes. This is the Python counterpart of verifyWebhook in @kindgi/crypto; any Standard Webhooks library verifies the same requests.

from kindgi import webhooks

body = await request.body()  # the raw bytes, never re-serialized JSON
try:
    delivery = webhooks.verify(secret, request.headers, body)
except webhooks.WebhookVerificationError:
    return Response(status_code=401)
event = webhooks.parse_event(body)

Delivery is at least once: deduplicate on delivery.id.

DEFAULT_TOLERANCE_SECONDS

DEFAULT_TOLERANCE_SECONDS = 300

ID_HEADER

ID_HEADER = 'webhook-id'

SECRET_MIN_BYTES

SECRET_MIN_BYTES = 24

SECRET_PREFIX

SECRET_PREFIX = 'whsec_'

SIGNATURE_HEADER

SIGNATURE_HEADER = 'webhook-signature'

TIMESTAMP_HEADER

TIMESTAMP_HEADER = 'webhook-timestamp'

RequestHeaders

class RequestHeaders(*args, **kwargs)

Request headers: a mapping (Starlette, Django, aiohttp, a dict), Werkzeug's Headers, or http.server's — anything whose items() gives name/value pairs.

RequestHeaders.items()

items() -> Iterable[tuple[str, str]]

VerifiedWebhook

class VerifiedWebhook(id: str, timestamp: int)

A request whose signature holds.

VerifyFailure

VerifyFailure(*args, **kwargs)

WebhookEvent

WebhookEvent(*args, **kwargs)

Runtime representation of an annotated type.

At its core 'Annotated[t, dec1, dec2, ...]' is an alias for the type 't' with extra annotations. The alias behaves like a normal typing alias. Instantiating is the same as instantiating the underlying type; binding it to types is also the same.

The metadata itself is stored in a 'metadata' attribute as a tuple.

WebhookVerificationError

class WebhookVerificationError(reason: VerifyFailure)

The request isn't a webhook signed with this secret; reason says which check failed.

generate_secret

generate_secret() -> str

A new random signing secret: whsec_ + base64 of 32 random bytes.

is_strong_secret

is_strong_secret(secret: str) -> bool

Whether a secret is whsec_ + base64 (or the bare base64) of at least 24 bytes.

parse_event

parse_event(body: bytes | str) -> WebhookEvent

The typed event in a verified request's body: a RunFinishedEvent or a WebhookTestEvent.

Raises pydantic.ValidationError for a body that isn't an event this version of the SDK knows.

sign

sign(secret: str | Sequence[str], *, id: str, timestamp: int, body: bytes | str) -> str

The webhook-signature header value: v1,<base64> per secret, space-separated.

Pass the new and the previous secret during a rotation: a receiver holding either one accepts the request. For tests of a receiver, or a service of your own that sends webhooks in the same format.

signed_headers

signed_headers(
    secret: str | Sequence[str],
    *,
    id: str,
    timestamp: int,
    body: bytes | str,
) -> dict[str, str]

The three signed headers for a request, ready to send.

verify

verify(
    secret: str,
    headers: RequestHeaders,
    body: bytes | str,
    *,
    tolerance_seconds: int = 300,
    now: Callable[[], float] = <built-in function time>,
) -> VerifiedWebhook

Verify a received webhook, or raise WebhookVerificationError.

The timestamp must be within tolerance_seconds of now() (Unix seconds) and at least one v1 signature in the header must match; comparison is constant-time. body is the raw request body, exactly as received (a str is taken as its UTF-8 bytes). Header names match case-insensitively.