webhook-cannon 0.1.0

Lightweight async webhook delivery with HMAC signing, SSRF protection, and auto-disable for failing endpoints.

webhook-cannon

Lightweight async webhook delivery with HMAC signing, SSRF protection, and auto-disable.

A drop-in Python library for sending webhooks securely. No external services, no infrastructure — just pip install and fire.

Features

• HMAC-SHA256 Signing — Every delivery is signed with a timestamp and HMAC hash. Receivers can verify authenticity and reject replay attacks.
• SSRF Protection — DNS resolution validates that target IPs are not in private/internal ranges (RFC-1918, loopback, link-local, CGNAT, IPv4-mapped IPv6).
• Auto-Disable — Endpoints that fail consecutively are automatically disabled. Reset manually or configure a cooldown for auto-recovery.
• Async-First — Built on httpx.AsyncClient for non-blocking delivery. Works with FastAPI, Starlette, Django, or any async Python app.

Installation

pip install webhook-cannon

Quick Start

Sending Webhooks

import asyncio
from webhook_cannon import WebhookCannon

cannon = WebhookCannon(
    signing_secret="whsec_your_secret_here",
    timeout=10,
    max_failures=5,
)

async def main():
    result = await cannon.fire(
        url="https://customer.com/webhooks",
        event="order.completed",
        payload={"order_id": 123, "total": 99.90},
    )
    print(f"Delivered: {result.success} ({result.status_code})")

asyncio.run(main())

Verifying Webhooks (Receiver Side)

from webhook_cannon import verify_signature, SignatureVerificationFailed

try:
    verify_signature(
        payload=request_body_bytes,
        signature=request.headers["X-Webhook-Signature"],
        secret="whsec_your_secret_here",
        tolerance=300,  # Reject signatures older than 5 minutes
    )
    print("Signature valid!")
except SignatureVerificationFailed as e:
    print(f"Invalid: {e}")

Why Not Svix?

Svix is a full webhook infrastructure service with queues, retries, a dashboard, and managed hosting. If you need all of that, use Svix.

webhook-cannon is for developers who want:

• A simple pip install with zero infrastructure
• Webhook delivery as a library call, not a service
• SSRF protection built-in (Svix doesn't handle this — it's your job)
• Full control over delivery timing and retry logic

API Reference

WebhookCannon

cannon = WebhookCannon(
    signing_secret="whsec_...",    # Required. HMAC-SHA256 signing key.
    timeout=10,                     # HTTP timeout in seconds (default: 10).
    max_failures=5,                 # Consecutive failures to auto-disable (default: 5).
    cooldown_seconds=0,             # Auto-recovery delay. 0 = manual reset only.
    user_agent="MyApp/1.0",        # User-Agent header (default: "webhook-cannon/0.1").
    blocked_ip_ranges=None,         # Extra CIDR strings to block (e.g. ["203.0.113.0/24"]).
    follow_redirects=False,         # Follow HTTP redirects (default: False for SSRF safety).
)

await cannon.fire(url, event, payload, *, custom_headers=None, signing_secret=None)

Deliver a signed webhook. Returns a DeliveryResult.

Parameter	Type	Description
url	str	Target webhook URL
event	str	Event type (e.g. "order.completed")
payload	dict	JSON-serializable event data
custom_headers	dict | None	Additional HTTP headers
signing_secret	str | None	Override signing secret for this delivery

Raises: SSRFBlocked if the URL resolves to an internal IP. EndpointDisabled if the endpoint has been auto-disabled.

cannon.get_endpoint_status(url) -> EndpointStatus

Returns EndpointStatus with failures, disabled, disabled_at, last_failure, last_success_at.

await cannon.reset_endpoint(url)

Re-enable a disabled endpoint and reset its failure counter.

await cannon.remove_endpoint(url)

Remove all tracking state for an endpoint.

DeliveryResult

Field	Type	Description
success	bool	Whether delivery succeeded (2xx response)
status_code	int	HTTP status code (0 on network error)
duration_ms	float	Round-trip time in milliseconds
attempt_number	int	Attempt number for this endpoint
url	str	Target URL
event	str	Event type
error	str | None	Error message on failure
response_body	str | None	Truncated response body (max 500 chars)

Signing & Verification

from webhook_cannon import sign, verify, SIGNATURE_HEADER

# Sign a payload (sender side).
signature = sign(payload_bytes, secret, timestamp=None)
# -> "t=1234567890,v1=abc123..."

# Verify a signature (receiver side).
verify(payload_bytes, signature_header, secret, tolerance=300)
# -> True or raises SignatureVerificationFailed

SSRF Validation

from webhook_cannon import resolve_and_validate, SSRFBlocked

try:
    safe_ip = resolve_and_validate("https://example.com/hook")
    print(f"Safe to connect: {safe_ip}")
except SSRFBlocked as e:
    print(f"Blocked: {e}")

SSRF Protection Details

Before every delivery, the target hostname is resolved via DNS and every returned IP address is checked against these blocked ranges:

Range	Type
127.0.0.0/8	Loopback
10.0.0.0/8	RFC-1918 private
172.16.0.0/12	RFC-1918 private
192.168.0.0/16	RFC-1918 private
169.254.0.0/16	Link-local
100.64.0.0/10	CGNAT (RFC 6598)
0.0.0.0/8	"This network"
198.18.0.0/15	Benchmarking
240.0.0.0/4	Reserved
::1/128	IPv6 loopback
fc00::/7	IPv6 ULA
fe80::/10	IPv6 link-local
::ffff:0:0/96	IPv4-mapped IPv6

The last one (::ffff:0:0/96) is critical — it catches IPv4-mapped IPv6 addresses like ::ffff:127.0.0.1, a common SSRF bypass vector.

DNS rebinding prevention: The resolved IP is returned for direct connection, so subsequent DNS lookups cannot return a different (internal) IP.

Redirect blocking: HTTP redirects are disabled by default to prevent redirect-based SSRF bypasses.

Configuration Options

Option	Default	Description
signing_secret	(required)	HMAC-SHA256 signing key
timeout	10	HTTP timeout in seconds
max_failures	5	Consecutive failures before auto-disable
cooldown_seconds	0	Auto-recovery delay (0 = manual only)
user_agent	"webhook-cannon/0.1"	User-Agent header value
blocked_ip_ranges	None	Extra CIDR blocks (added to defaults)
follow_redirects	False	Follow HTTP redirects

Signature Format

X-Webhook-Signature: t=1234567890,v1=5257a869e7ecebeda32affa62cdca3fa51cad7e77a0e56ff536d0ce8e108d8bd

The signed message is {timestamp}.{raw_body} — concatenating the Unix timestamp with the raw JSON body. This prevents:

• Replay attacks: Receivers reject signatures older than the tolerance window (default: 5 minutes)
• Payload tampering: Any change to the body invalidates the HMAC

Contributing

1. Fork the repository
2. Create a feature branch (git checkout -b feature/my-feature)
3. Run tests (pytest -v)
4. Run linter (ruff check src/ tests/)
5. Submit a pull request

License

MIT License. See LICENSE for details.