Metadata-Version: 2.4
Name: webhook-cannon
Version: 0.1.0
Summary: Lightweight async webhook delivery with HMAC signing, SSRF protection, and auto-disable for failing endpoints.
Project-URL: Homepage, https://github.com/tahakotil/webhook-cannon
Project-URL: Repository, https://github.com/tahakotil/webhook-cannon
Project-URL: Issues, https://github.com/tahakotil/webhook-cannon/issues
Author-email: Taha Kotil <devtahakotil@gmail.com>
License-Expression: MIT
License-File: LICENSE
Keywords: async,fastapi,hmac,security,ssrf,webhook
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Framework :: FastAPI
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Internet :: WWW/HTTP
Classifier: Topic :: Security
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: httpx>=0.25.0
Provides-Extra: dev
Requires-Dist: mypy>=1.10.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.24.0; extra == 'dev'
Requires-Dist: pytest>=8.0.0; extra == 'dev'
Requires-Dist: ruff>=0.5.0; extra == 'dev'
Description-Content-Type: text/markdown

# webhook-cannon

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

[![PyPI version](https://img.shields.io/pypi/v/webhook-cannon.svg)](https://pypi.org/project/webhook-cannon/)
[![Python 3.10+](https://img.shields.io/badge/python-3.10+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE)

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

```bash
pip install webhook-cannon
```

## Quick Start

### Sending Webhooks

```python
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)

```python
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](https://www.svix.com/) 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`

```python
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

```python
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

```python
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](LICENSE) for details.
