Webhook API¶

class peyk.webhook.IPFilter(networks: Sequence[str], *, trust_forwarded_for: bool = False)[source]¶

Bases: object

Allow webhook requests only when their source IP is in configured networks.

trust_forwarded_for is deliberately opt-in. When disabled, only the transport peer address is considered, so an untrusted client cannot choose its own address through X-Forwarded-For.

allows(peer_ip: str, headers: Mapping[str, str]) → bool[source]¶: Return whether the request source belongs to an allowed network.

class peyk.webhook.WebhookOptions(allowed_updates: list[str] | None = None, drop_pending_updates: bool = False, max_connections: int | None = None)[source]¶

Bases: object

Platform-neutral webhook registration options.

allowed_updates: list[str] | None = None¶

drop_pending_updates: bool = False¶

max_connections: int | None = None¶

class peyk.webhook.WebhookProcessor(dispatcher: object, *, secret: str | None = None, max_body_size: int = MAX_DEFAULT_BODY_SIZE, handle_in_background: bool = True, endpoint_type: str | None = None)[source]¶

Bases: object

Parse and dispatch webhook requests without depending on an HTTP server.

async handle(bot: Bot[object], body: bytes, headers: Mapping[str, str]) → WebhookResult[source]¶

Verify, parse and dispatch one webhook request.

Malformed JSON and oversized bodies are rejected. Authentication failures return the same 404 used by route failures so callers cannot distinguish a secret oracle from an absent endpoint.

async shutdown() → None[source]¶: Wait for all background update tasks before server shutdown.

class peyk.webhook.WebhookRegistrar(*args, **kwargs)[source]¶

Bases: Protocol

async install(bot: Bot[object], url: str, secret: str, options: WebhookOptions) → None[source]¶

Provides the install operation for the peyk integration.

Parameters:

bot – Value used by this operation.
url – Value used by this operation.
secret – Value used by this operation.
options – Value used by this operation.

async uninstall(bot: Bot[object]) → None[source]¶

Provides the uninstall operation for the peyk integration.

Parameters:: bot – Value used by this operation.

class peyk.webhook.WebhookResult(status: int, body: bytes | None = None)[source]¶

Bases: object

HTTP result returned by a webhook processor.

body: bytes | None = None¶

status: int¶

peyk.webhook.constant_time_compare(a: str | None, b: str | None) → bool[source]¶: Compare two non-empty strings without ordinary early-exit timing.

peyk.webhook.registrar_for(bot: Bot[object]) → WebhookRegistrar[source]¶: Return the audited registrar for bot.platform.

peyk.webhook.verify_telegram_secret_header(headers: Mapping[str, str], secret_token: str) → bool[source]¶: Validate Telegram’s documented webhook secret header.