diff --git a/CHANGELOG.md b/CHANGELOG.md index 8fbc51b..37fabf2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- `tango.webhooks` subpackage with HMAC-SHA256 signing helpers (`verify_signature`, `generate_signature`, `parse_signature_header`) that mirror the canonical Tango server scheme byte-for-byte. Importable from a default `pip install tango-python` (pure stdlib). +- `WebhookReceiver`: a stdlib-based local HTTP listener for development and integration tests. Verifies signatures, optionally forwards each delivery to a downstream URL, and records deliveries in memory for inspection. Usable as a context manager (`with WebhookReceiver(secret=...).run() as rx: ...`). +- `tango.webhooks.simulate.deliver(...)`: locally sign and POST a payload to any URL — no Tango involvement. Useful for offline iteration on receiver code. +- New `tango[webhooks]` extra (adds `click`) ships a `tango` console script covering the full webhook lifecycle for developer integrations: + - `listen` — local receiver + - `simulate` — sign a payload locally; with `--to`, also POST it + - `trigger` — ask Tango to send a real test delivery + - `fetch-sample` — print the canonical payload Tango emits for an event type + - `list-event-types` — discover what's subscribable + - `endpoints list|get|create|delete` — manage delivery endpoints + - `subscriptions list|get|create|delete` — manage what events you receive + Together these let a developer go from zero to receiving real Tango webhooks without leaving the shell or dropping into Python. + +### Notes +- Console script name `tango` may be revisited before the next release if it conflicts with sibling tooling (`tango-scripts` reuses the bare name). + +### Documentation +- New `docs/WEBHOOKS.md` — comprehensive guide covering install, concepts, a zero-to-receiving quickstart, full CLI reference, and programmatic patterns for `WebhookReceiver` / `simulate.sign` / `simulate.deliver` in pytest fixtures. +- `docs/API_REFERENCE.md`: filled in `get_webhook_subscription`, replaced the hand-rolled signature-verification snippet with a pointer to `tango.webhooks.verify_signature`, and added a new "Webhook tooling (`tango.webhooks`)" section that documents every importable from the new subpackage. +- `README.md`: new "Webhook Tooling" section under Advanced Features, plus the new guide is linked from the Documentation index. + ## [0.5.0] - 2026-04-08 ### Added diff --git a/README.md b/README.md index d154abf..3cc0d5b 100644 --- a/README.md +++ b/README.md @@ -327,6 +327,46 @@ contracts = client.list_contracts( # Returns: {"key": "...", "transactions.0.action_date": "...", "transactions.0.obligated": "..."} ``` +### Webhook Tooling + +The SDK ships first-class tooling for **building and testing webhook integrations against the Tango API** — including signing helpers, a local receiver, and a command-line tool covering the full lifecycle: + +```bash +pip install 'tango-python[webhooks]' +``` + +This adds a `tango` console script with subcommands for the full webhook lifecycle: + +```bash +# Discover what's available +tango webhooks list-event-types +tango webhooks fetch-sample --event-type entities.updated + +# Local development +tango webhooks listen --port 8011 --secret $SECRET # receiver +tango webhooks simulate --secret $SECRET --event-type entities.updated # sign + print +tango webhooks simulate --secret $SECRET --event-type entities.updated \ + --to http://127.0.0.1:8011/tango/webhooks # also POST + +# Manage real subscriptions and endpoints +tango webhooks endpoints create|list|get|delete +tango webhooks subscriptions create|list|get|delete + +# Force a real test delivery from Tango +tango webhooks trigger +``` + +The signing helpers (`verify_signature`, `generate_signature`) are pure stdlib and importable from the default install — your receiver code doesn't need the extra: + +```python +from tango.webhooks import verify_signature + +if not verify_signature(raw_body, secret, request.headers.get("X-Tango-Signature")): + return 401, "invalid signature" +``` + +For the full guide — workflow, CLI reference, and programmatic patterns for pytest fixtures — see [`docs/WEBHOOKS.md`](docs/WEBHOOKS.md). + ### Type Hints with IDE Support Import TypedDict types for IDE autocomplete: @@ -476,6 +516,7 @@ tango-python/ - [Shape System Guide](docs/SHAPES.md) - Comprehensive guide to response shaping - [API Reference](docs/API_REFERENCE.md) - Detailed API documentation - [Developer Guide](docs/DEVELOPERS.md) - Technical documentation for developers +- [Webhooks Guide](docs/WEBHOOKS.md) - Workflow, CLI reference, and programmatic patterns for webhook integrations - [Quick Start Notebook](docs/quick_start.ipynb) - Interactive Jupyter notebook with examples ## Requirements diff --git a/docs/API_REFERENCE.md b/docs/API_REFERENCE.md index e1d58ee..e41e039 100644 --- a/docs/API_REFERENCE.md +++ b/docs/API_REFERENCE.md @@ -1227,6 +1227,8 @@ for code in naics.results: Webhook APIs let **Large / Enterprise** users manage subscription filters for outbound Tango webhooks. +> **For testing, signing, and a CLI tool**, see [`docs/WEBHOOKS.md`](WEBHOOKS.md). This section covers SDK method signatures only. + ### list_webhook_event_types() Discover supported `event_type` values and subject types. @@ -1246,6 +1248,12 @@ Notes: - This endpoint uses `page` + `page_size` (tier-capped) rather than `limit`. +### get_webhook_subscription() + +```python +sub = client.get_webhook_subscription("SUBSCRIPTION_UUID") +``` + ### create_webhook_subscription() ```python @@ -1335,21 +1343,89 @@ Every delivery includes an HMAC signature header: Compute the digest over the **raw request body bytes** using your shared secret. +The SDK ships a stdlib-only verifier that mirrors the Tango server's signing scheme byte-for-byte. Use it instead of hand-rolling — it's importable from a default install (no extras needed): + ```python -import hashlib -import hmac +from tango.webhooks import verify_signature + +if not verify_signature(raw_body, secret, request.headers.get("X-Tango-Signature")): + return 401 +``` +`verify_signature` returns `False` for missing/empty/malformed headers — it never raises. Comparison is constant-time. -def verify_tango_webhook_signature(secret: str, raw_body: bytes, signature_header: str | None) -> bool: - if not signature_header: - return False - sig = signature_header.strip() - if sig.startswith("sha256="): - sig = sig[len("sha256=") :] - expected = hmac.new(secret.encode(), raw_body, hashlib.sha256).hexdigest() - return hmac.compare_digest(expected, sig) +--- + +## Webhook tooling (`tango.webhooks`) + +The `tango.webhooks` subpackage adds testing and developer-tooling primitives on top of the API methods above. Signing helpers ship with the default install; the receiver and CLI ship with `pip install 'tango-python[webhooks]'`. See [`docs/WEBHOOKS.md`](WEBHOOKS.md) for usage guides; this section is the import-level reference. + +### Signing (default install) + +```python +from tango.webhooks import ( + verify_signature, # (body: bytes, secret: str, header: str | None) -> bool + generate_signature, # (body: bytes, secret: str) -> str (lowercase hex) + parse_signature_header, # (header: str | None) -> str | None (strips "sha256=") + SIGNATURE_HEADER, # "X-Tango-Signature" + SIGNATURE_PREFIX, # "sha256=" +) ``` +### `WebhookReceiver` (with `[webhooks]` extra) + +A stdlib-based local HTTP receiver, useful in tests and during local development. + +```python +from tango.webhooks import WebhookReceiver, Delivery + +with WebhookReceiver(secret="dev").run() as rx: + # ... cause something to POST to rx.url ... + deliveries: list[Delivery] = rx.deliveries +``` + +Constructor (all keyword arguments): + +| Arg | Default | Meaning | +|---|---|---| +| `secret` | `""` | Shared secret. Empty means signatures are not verified. | +| `path` | `/tango/webhooks` | URL path to accept POSTs on. | +| `host` | `127.0.0.1` | Bind address. | +| `port` | `0` | TCP port. `0` = OS picks a free port. | +| `forward_to` | `None` | Optional URL to mirror each delivery to. | +| `max_history` | `256` | Cap on the in-memory `deliveries` deque. | +| `on_delivery` | `None` | Callback fired for every delivery (verified or not). | +| `require_signature` | `None` | Override default (require iff `secret` is set). | + +Each `Delivery` is a dataclass: `received_at`, `path`, `signature_header`, `body_bytes`, `body_json`, `verified`, `remote_addr`, `forward_status`, `forward_error`. + +### `simulate.sign` and `simulate.deliver` + +```python +from tango.webhooks import sign, SignedRequest +from tango.webhooks import simulate + +# Offline — produce the signed wire form without POSTing: +signed: SignedRequest = sign({"events": [{"event_type": "..."}]}, secret="s") +signed.body # bytes you would put on the wire +signed.signature # bare lowercase hex +signed.headers # {"Content-Type": ..., "X-Tango-Signature": "sha256=..."} + +# With delivery — sign and POST to a target URL: +result = simulate.deliver(target_url="http://localhost:8011/tango/webhooks", + payload={...}, secret="s") +result.status_code # status from the receiver +result.signature # bare hex +result.sent_bytes # exact bytes that were POSTed +result.response_body # body the receiver returned +``` + +`simulate.deliver` and `simulate.sign` accept payloads as `dict`, `list`, `str`, or raw `bytes`. Dicts/lists are serialized via `json.dumps(..., sort_keys=True, separators=(",", ":"))` so signatures are reproducible across runs. + +### CLI entry point + +The `tango[webhooks]` extra also installs a `tango` console script. See [`docs/WEBHOOKS.md` § CLI reference](WEBHOOKS.md#cli-reference) for the full command list. + --- ## Response Objects diff --git a/docs/WEBHOOKS.md b/docs/WEBHOOKS.md new file mode 100644 index 0000000..817b22c --- /dev/null +++ b/docs/WEBHOOKS.md @@ -0,0 +1,430 @@ +# Webhooks Guide + +This guide covers everything `tango-python` provides for **building, testing, and operating webhook integrations against the Tango API**: signing helpers, a local receiver, a command-line tool, and management commands for the underlying endpoints and subscriptions. + +If you only need the SDK method signatures, see [`API_REFERENCE.md` § Webhooks](API_REFERENCE.md#webhooks). For the API-level contract (signing scheme, event taxonomy, retry behavior), see the [Tango Webhooks Partner Guide](https://docs.makegov.com/webhooks-user-guide/). + +--- + +## Contents + +- [Install](#install) +- [Concepts in 60 seconds](#concepts-in-60-seconds) +- [Quickstart: zero to receiving](#quickstart-zero-to-receiving) +- [CLI reference](#cli-reference) + - [`tango webhooks listen`](#tango-webhooks-listen) + - [`tango webhooks simulate`](#tango-webhooks-simulate) + - [`tango webhooks trigger`](#tango-webhooks-trigger) + - [`tango webhooks fetch-sample`](#tango-webhooks-fetch-sample) + - [`tango webhooks list-event-types`](#tango-webhooks-list-event-types) + - [`tango webhooks endpoints`](#tango-webhooks-endpoints) + - [`tango webhooks subscriptions`](#tango-webhooks-subscriptions) +- [Programmatic use](#programmatic-use) + - [Signature verification in your handler](#signature-verification-in-your-handler) + - [`WebhookReceiver` in pytest fixtures](#webhookreceiver-in-pytest-fixtures) + - [`simulate.sign` and `simulate.deliver`](#simulatesign-and-simulatedeliver) +- [Common workflows](#common-workflows) +- [Troubleshooting](#troubleshooting) + +--- + +## Install + +The signing helpers ship with the default install: + +```bash +pip install tango-python +``` + +The CLI (`tango webhooks ...`) and the local receiver class are gated behind an optional extra: + +```bash +pip install 'tango-python[webhooks]' +``` + +This adds [`click`](https://palletsprojects.com/projects/click) as a runtime dependency. The base SDK install stays unchanged. + +After installing the extra, the `tango` console script is on your `PATH`: + +```bash +tango webhooks --help +``` + +--- + +## Concepts in 60 seconds + +Tango webhooks have three pieces of state: + +| Concept | What it is | Tango term | +|---|---|---| +| **Endpoint** | The URL Tango POSTs to, plus a generated signing secret | `WebhookEndpoint` | +| **Subscription** | A filter saying *which events* you want delivered to that endpoint | `WebhookSubscription` | +| **Delivery** | A single signed POST Tango makes when a matching event fires | (the request itself) | + +A typical setup: + +1. **Create an endpoint** (`POST /api/webhooks/endpoints/`) with the public URL of your handler. Tango returns a `secret` — save it; it's used to sign every delivery. +2. **Create one or more subscriptions** (`POST /api/webhooks/subscriptions/`) describing the events your handler cares about (e.g. `entities.updated` for specific UEIs). +3. **Tango POSTs** to your endpoint when matching events fire. The body is JSON; the header `X-Tango-Signature: sha256=` is the HMAC-SHA256 of the raw body bytes keyed by your endpoint's secret. +4. **Your handler verifies the signature**, parses the body, and acts on it. + +--- + +## Quickstart: zero to receiving + +Assumes you have a `TANGO_API_KEY` and want to receive entity-update webhooks for a specific UEI. + +### 1. See what you can subscribe to + +```bash +export TANGO_API_KEY=... +tango webhooks list-event-types +# entities.updated An entity record was updated +# awards.created A new award was published +# ... +``` + +### 2. See what a payload looks like + +```bash +tango webhooks fetch-sample --event-type entities.updated +``` + +Prints the canonical JSON shape Tango will deliver. No POST, no signature — just the body. + +### 3. Run a local receiver + +In one shell, start a listener with a chosen secret: + +```bash +export TANGO_WEBHOOK_SECRET=dev_secret +tango webhooks listen --port 8011 +``` + +In another shell, drive it with the canonical sample, signed locally: + +```bash +tango webhooks simulate \ + --secret $TANGO_WEBHOOK_SECRET \ + --event-type entities.updated \ + --to http://127.0.0.1:8011/tango/webhooks +``` + +The listener should print a `verified` delivery with the entities-updated body. You now have a feedback loop: edit your handler, re-run `simulate`, see the result. + +### 4. Wire up the real Tango → your handler path + +When you're ready for end-to-end testing against Tango itself, expose your local listener via a tunnel (`ngrok http 8011`, `cloudflared tunnel`, etc.) and register that public URL with Tango: + +```bash +# Use the public URL the tunnel gave you. +tango webhooks endpoints create --url https://.ngrok.io/tango/webhooks +# Save the `secret` from the response — that's what your handler uses to verify. + +tango webhooks subscriptions create \ + --name "watch UEI ABC123" \ + --event-type entities.updated \ + --subject-type entity \ + --subject-id ABC123 +``` + +To force a real test delivery from Tango (without waiting for an actual event): + +```bash +tango webhooks trigger +``` + +You should see a `verified` delivery in your local listener with the signature value generated by Tango — not by `simulate`. + +--- + +## CLI reference + +All commands live under `tango webhooks`. Options that talk to Tango's API (`--api-key`, `--base-url`) read `TANGO_API_KEY` and `TANGO_BASE_URL` if not passed explicitly. + +### `tango webhooks listen` + +Run a local HTTP receiver. Verifies signatures, optionally forwards each delivery downstream, prints a one-line summary plus the JSON body for each delivery. + +```bash +tango webhooks listen \ + --port 8011 \ + --host 127.0.0.1 \ + --path /tango/webhooks \ + --secret $TANGO_WEBHOOK_SECRET \ + --forward-to http://127.0.0.1:4242/wh +``` + +Options: + +- `--port` (default `8011`) +- `--host` (default `127.0.0.1` — loopback only, by design) +- `--path` (default `/tango/webhooks`) +- `--secret` / `TANGO_WEBHOOK_SECRET` — if empty, signatures are not verified (the listener accepts everything; useful for inspecting payloads when you don't have the right secret yet) +- `--forward-to URL` — mirror each delivery to a downstream URL, preserving body bytes and the `X-Tango-Signature` header +- `--require-signature / --allow-unsigned` — override the default policy (default: require when `--secret` is set) + +Press Ctrl+C to stop. Rejected (signature-mismatch) deliveries are still printed with the label `UNVERIFIED` so you can debug what arrived. + +### `tango webhooks simulate` + +Sign a payload locally with the same scheme Tango uses, then either print the signed request or POST it to a receiver. + +**Without `--to`** — just print the headers + body a real Tango delivery would have: + +```bash +tango webhooks simulate --secret dev_secret --event-type entities.updated +``` + +Output includes `delivered: false`, the headers (`Content-Type`, `X-Tango-Signature`), and the JSON payload. + +**With `--to`** — also POST the signed body to a receiver: + +```bash +tango webhooks simulate \ + --secret dev_secret \ + --event-type entities.updated \ + --to http://127.0.0.1:8011/tango/webhooks +``` + +Output includes `delivered: true`, the receiver's status code, and the receiver's response body. + +Three sources for the payload (mutually exclusive): + +| Flag | Source | When to use | +|---|---|---| +| `--event-type X` | Fetches the canonical sample for `X` from Tango | You want a realistic body without setting up a subscription | +| `--payload-file PATH` | Reads a JSON file | You're testing a specific shape (regression, edge case) | +| *(neither)* | A built-in placeholder envelope | Smoke-testing the wiring | + +### `tango webhooks trigger` + +Ask Tango to send a real test delivery to your configured endpoint. Wraps `POST /api/webhooks/endpoints/test-delivery/`. Requires `--api-key`. + +```bash +tango webhooks trigger +tango webhooks trigger --endpoint-id +``` + +Output is JSON: `success`, `status_code` (the HTTP code Tango got from your endpoint), `response_time_ms`, `endpoint_url`, `message`, `error`. Exit code is non-zero if delivery failed. + +### `tango webhooks fetch-sample` + +Print the canonical sample payload for one event type, or the full mapping if `--event-type` is omitted. Wraps `GET /api/webhooks/endpoints/sample-payload/`. Read-only. + +```bash +tango webhooks fetch-sample --event-type entities.updated +tango webhooks fetch-sample # all event types +``` + +### `tango webhooks list-event-types` + +List every event type Tango supports with a one-line description. + +```bash +tango webhooks list-event-types +``` + +### `tango webhooks endpoints` + +Manage **where Tango delivers**. + +```bash +tango webhooks endpoints list [--page N] [--limit N] +tango webhooks endpoints get ENDPOINT_ID +tango webhooks endpoints create --url URL [--inactive] +tango webhooks endpoints delete ENDPOINT_ID [--yes] +``` + +`create` returns the generated `secret` once — save it. `delete` prompts for confirmation; `--yes` skips. `--inactive` registers the endpoint disabled (no deliveries until you re-enable it). + +### `tango webhooks subscriptions` + +Manage **what Tango delivers**. + +```bash +tango webhooks subscriptions list [--page N] [--page-size N] +tango webhooks subscriptions get SUBSCRIPTION_ID +tango webhooks subscriptions create \ + --name "watch UEI ABC123" \ + --event-type entities.updated \ + --subject-type entity \ + --subject-id ABC123 +tango webhooks subscriptions delete SUBSCRIPTION_ID [--yes] +``` + +`create` builds a single-record subscription (one event type, one subject type, one or more subject IDs). For multi-record subscriptions, call `client.create_webhook_subscription(...)` directly with a hand-crafted `payload` dict. + +--- + +## Programmatic use + +The CLI is built on top of small importable pieces. You can use them directly in your own code — most usefully, in tests. + +### Signature verification in your handler + +`verify_signature` is pure stdlib (no SDK dependencies, no `click`). Call it on the raw request body, not on a re-serialized parsed body — the HMAC is computed over exact bytes. + +```python +from tango.webhooks import verify_signature + +# In your Flask / FastAPI / Django / Starlette / whatever handler: +def handle_webhook(request): + body = request.body # raw bytes + signature = request.headers.get("X-Tango-Signature") + if not verify_signature(body, secret=ENDPOINT_SECRET, signature_header=signature): + return 401, {"error": "invalid_signature"} + payload = json.loads(body) + # ... act on the events ... + return 200, {"ok": True} +``` + +`verify_signature` returns `False` for missing/empty/malformed headers — it never raises. Comparison is constant-time (`hmac.compare_digest`). + +### `WebhookReceiver` in pytest fixtures + +The CLI's `listen` command is a thin wrapper around `tango.webhooks.WebhookReceiver`, which is a context-manager-friendly local HTTP server. Use it directly in tests to verify your code emits webhook calls correctly, or to drive your handler with realistic deliveries. + +```python +from tango.webhooks import WebhookReceiver, verify_signature +import httpx + +def test_my_handler_processes_entity_update(): + with WebhookReceiver(secret="test_secret").run() as rx: + # Trigger whatever in your code-under-test should send a webhook + # (e.g. a publisher, or in this case a manual POST). + body = b'{"events":[{"event_type":"entities.updated","uei":"ABC"}]}' + from tango.webhooks import generate_signature + sig = generate_signature(body, "test_secret") + httpx.post(rx.url, content=body, headers={"X-Tango-Signature": f"sha256={sig}"}) + + assert len(rx.deliveries) == 1 + assert rx.deliveries[0].verified + assert rx.deliveries[0].body_json["events"][0]["uei"] == "ABC" +``` + +`WebhookReceiver` options: + +- `secret: str = ""` — shared secret. Empty means "don't verify." +- `path: str = "/tango/webhooks"` — URL path to accept. +- `host: str = "127.0.0.1"` / `port: int = 0` — bind address. `0` lets the OS pick a free port. +- `forward_to: str | None = None` — mirror each delivery to a downstream URL. +- `max_history: int = 256` — cap on the in-memory `deliveries` deque. +- `on_delivery: Callable[[Delivery], None] | None = None` — fires for every recorded delivery, including signature-failed ones. +- `require_signature: bool | None = None` — override default (require iff `secret` is set). + +Each `Delivery` has: `received_at`, `path`, `signature_header`, `body_bytes`, `body_json`, `verified`, `remote_addr`, `forward_status`, `forward_error`. + +### `simulate.sign` and `simulate.deliver` + +`simulate.sign` is the offline counterpart — it produces the exact wire form a Tango delivery would have, so you can drive your handler from a unit test: + +```python +from tango.webhooks import sign + +signed = sign({"events": [{"event_type": "entities.updated"}]}, secret="s") +assert signed.headers["X-Tango-Signature"].startswith("sha256=") + +# Use `signed.body` as the raw bytes and `signed.headers` directly: +response = my_app.test_client().post( + "/webhooks", data=signed.body, headers=signed.headers +) +``` + +`simulate.deliver` does the same but POSTs the result to a URL — `WebhookReceiver` works as a target: + +```python +from tango.webhooks import simulate, WebhookReceiver + +with WebhookReceiver(secret="s").run() as rx: + result = simulate.deliver(target_url=rx.url, payload={...}, secret="s") + assert result.status_code == 200 +``` + +--- + +## Common workflows + +### "I'm starting fresh — set me up to receive entity updates" + +```bash +export TANGO_API_KEY=... +# 1. Confirm event types +tango webhooks list-event-types +# 2. Stand up a tunnel so Tango can reach you +ngrok http 8011 & +# 3. Register your endpoint and subscription +tango webhooks endpoints create --url https://.ngrok.io/tango/webhooks +# (save the `secret` from the response into TANGO_WEBHOOK_SECRET) +tango webhooks subscriptions create \ + --name "entities" --event-type entities.updated \ + --subject-type entity --subject-id +# 4. Run the listener pointed at your downstream handler +tango webhooks listen --port 8011 --secret $TANGO_WEBHOOK_SECRET \ + --forward-to http://localhost:4242/wh +# 5. Force a test delivery +tango webhooks trigger +``` + +### "I want to develop my handler offline" + +You don't need a Tango account or any tunnel: + +```bash +# Run the handler however you normally would on, e.g., :4242 +tango webhooks listen --port 8011 --secret dev --forward-to http://127.0.0.1:4242/wh + +# In another shell, drive it. Use Tango-shaped bodies if you have an API key: +tango webhooks simulate --secret dev --event-type entities.updated \ + --to http://127.0.0.1:8011/tango/webhooks + +# Or use a custom shape from a file (no API key required): +tango webhooks simulate --secret dev --payload-file ./fixtures/edge.json \ + --to http://127.0.0.1:8011/tango/webhooks +``` + +### "I want to test my handler in CI, no network" + +In pytest, use `WebhookReceiver` and `simulate.deliver` together — both are pure-Python and don't talk to Tango: + +```python +from tango.webhooks import simulate, WebhookReceiver + +def test_handler_round_trip(): + with WebhookReceiver(secret="s").run() as rx: + result = simulate.deliver( + target_url=rx.url, + payload={"events": [{"event_type": "entities.updated", "uei": "X"}]}, + secret="s", + ) + assert result.status_code == 200 + assert rx.deliveries[0].verified +``` + +### "I need to inspect what bytes Tango actually sends" + +```bash +tango webhooks simulate --secret $TANGO_WEBHOOK_SECRET --event-type entities.updated +# Prints { "delivered": false, "headers": {...}, "sent_payload": {...} } +``` + +This is the shape your handler will receive — including the exact `X-Tango-Signature` value it should verify. + +--- + +## Troubleshooting + +**Signature always fails.** Verify on raw bytes, not on a re-serialized parsed body. The HMAC is over exact bytes; reformatting whitespace or reordering keys breaks it. Most web frameworks expose the raw body separately from a parsed JSON shortcut — use the raw one. + +**`tango: command not found`.** Install the extra: `pip install 'tango-python[webhooks]'`. The console script is registered only when `click` is available. + +**Listener prints `WARNING: no --secret provided`.** You started `listen` without `--secret` and without `TANGO_WEBHOOK_SECRET` set. Every delivery will be accepted with `verified=False`. Useful for inspecting payloads when you don't have the secret yet, but unsafe in any shared environment. + +**`fetch-sample` returns 401.** Set `TANGO_API_KEY` (or pass `--api-key`). `fetch-sample` reads from Tango's API. + +**`endpoints create` returns 403 or "endpoint already exists".** Tango limits one endpoint per user. Use `endpoints list` to find the existing one, then either reuse it or delete it first. + +**`simulate --event-type X` fails with HTTP 4xx.** Tango doesn't recognize the event type. Run `list-event-types` to see the current list. + +**`trigger` returns `success: false`.** Tango reached your endpoint but got a non-2xx response. Check `endpoint_url` and `response_body` in the output, then look at your handler's logs. diff --git a/pyproject.toml b/pyproject.toml index 640b5c2..e99dd68 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -41,6 +41,12 @@ notebooks = [ "jupyter>=1.0.0", "ipykernel>=6.25.0", ] +webhooks = [ + "click>=8.1", +] + +[project.scripts] +tango = "tango.webhooks.cli:main" [project.urls] Homepage = "https://github.com/makegov/tango-python" @@ -116,6 +122,10 @@ exclude_lines = [ [tool.hatch.build.targets.wheel] packages = ["tango"] +[[tool.mypy.overrides]] +module = "tango.webhooks.cli" +disallow_untyped_decorators = false + [dependency-groups] dev = [ "python-dotenv>=1.2.1", diff --git a/tango/__init__.py b/tango/__init__.py index ff0d6f5..60ecd49 100644 --- a/tango/__init__.py +++ b/tango/__init__.py @@ -28,6 +28,12 @@ ShapeParser, TypeGenerator, ) +from .webhooks import ( + generate_signature, + parse_signature_header, + verify_signature, +) +from .webhooks.receiver import Delivery, WebhookReceiver __version__ = "0.5.0" __all__ = [ @@ -53,4 +59,9 @@ "ModelFactory", "TypeGenerator", "SchemaRegistry", + "Delivery", + "WebhookReceiver", + "generate_signature", + "parse_signature_header", + "verify_signature", ] diff --git a/tango/webhooks/__init__.py b/tango/webhooks/__init__.py new file mode 100644 index 0000000..e0b8d18 --- /dev/null +++ b/tango/webhooks/__init__.py @@ -0,0 +1,27 @@ +"""Tango webhooks: signature helpers and developer tooling. + +The signing helpers (:func:`verify_signature`, :func:`generate_signature`, +:func:`parse_signature_header`) are pure stdlib and importable from a default +``pip install tango``. The CLI (``tango webhooks ...``) and the in-process +:class:`~tango.webhooks.receiver.WebhookReceiver` ship with the +``tango[webhooks]`` extra. +""" + +from tango.webhooks.signing import ( + SIGNATURE_HEADER, + SIGNATURE_PREFIX, + generate_signature, + parse_signature_header, + verify_signature, +) +from tango.webhooks.simulate import SignedRequest, sign + +__all__ = [ + "SIGNATURE_HEADER", + "SIGNATURE_PREFIX", + "SignedRequest", + "generate_signature", + "parse_signature_header", + "sign", + "verify_signature", +] diff --git a/tango/webhooks/cli.py b/tango/webhooks/cli.py new file mode 100644 index 0000000..6c20d1c --- /dev/null +++ b/tango/webhooks/cli.py @@ -0,0 +1,519 @@ +"""Command-line interface for Tango webhook tooling. + +This module is the entry point for the ``tango`` console script. Click is an +optional dependency installed via the ``tango[webhooks]`` extra; importing +this module without it raises a friendly error that points users at the +right install command. +""" + +from __future__ import annotations + +import json +import sys +import threading +from pathlib import Path +from typing import Any + +try: + import click +except ImportError as _import_error: # pragma: no cover - tested via subprocess + sys.stderr.write( + "tango CLI requires the 'webhooks' extra. Install it with:\n" + " pip install 'tango-python[webhooks]'\n" + ) + raise SystemExit(1) from _import_error + +from tango.webhooks import simulate +from tango.webhooks.receiver import Delivery, WebhookReceiver +from tango.webhooks.signing import SIGNATURE_HEADER + + +@click.group() +@click.version_option(package_name="tango-python", prog_name="tango") +def main() -> None: + """Tango developer tooling.""" + + +@main.group() +def webhooks() -> None: + """Receive, trigger, and simulate Tango webhook deliveries.""" + + +@webhooks.command("listen") +@click.option("--port", type=int, default=8011, show_default=True, help="TCP port to bind.") +@click.option("--host", default="127.0.0.1", show_default=True, help="Bind address.") +@click.option( + "--path", + default="/tango/webhooks", + show_default=True, + help="URL path to accept deliveries on.", +) +@click.option( + "--secret", + envvar="TANGO_WEBHOOK_SECRET", + default="", + help="Shared secret. Reads TANGO_WEBHOOK_SECRET if unset. " + "If empty, deliveries are accepted without signature verification.", +) +@click.option( + "--forward-to", + default=None, + help="Optional URL to mirror each delivery to (preserves body and signature).", +) +@click.option( + "--require-signature/--allow-unsigned", + default=None, + help="Override default policy. Default: require when --secret is set.", +) +def listen_cmd( + port: int, + host: str, + path: str, + secret: str, + forward_to: str | None, + require_signature: bool | None, +) -> None: + """Run a local receiver and stream deliveries to stdout.""" + receiver = WebhookReceiver( + secret=secret, + path=path, + host=host, + port=port, + forward_to=forward_to, + require_signature=require_signature, + on_delivery=_print_delivery, + ) + receiver.start() + try: + click.echo(f"Listening on {receiver.url}") + if not secret: + click.echo( + " WARNING: no --secret provided; signatures will not be verified.", + err=True, + ) + if forward_to: + click.echo(f" Forwarding to {forward_to}") + click.echo(" Press Ctrl+C to stop.") + threading.Event().wait() # block until interrupted + except KeyboardInterrupt: + click.echo("\nStopping...") + finally: + receiver.stop() + + +@webhooks.command("trigger") +@click.option( + "--endpoint-id", + default=None, + help="Endpoint UUID. If omitted, the server's default endpoint is used.", +) +@click.option( + "--api-key", + envvar="TANGO_API_KEY", + help="Tango API key (or set TANGO_API_KEY).", +) +@click.option( + "--base-url", + envvar="TANGO_BASE_URL", + default="https://tango.makegov.com", + show_default=True, + help="Tango base URL (or set TANGO_BASE_URL).", +) +def trigger_cmd(endpoint_id: str | None, api_key: str | None, base_url: str) -> None: + """Ask Tango to send a real test delivery to your configured endpoint.""" + from tango import TangoClient + + client = TangoClient(api_key=api_key, base_url=base_url) + result = client.test_webhook_delivery(endpoint_id=endpoint_id) + click.echo( + json.dumps( + { + "success": result.success, + "status_code": result.status_code, + "response_time_ms": result.response_time_ms, + "endpoint_url": result.endpoint_url, + "message": result.message, + "error": result.error, + }, + indent=2, + ) + ) + if not result.success: + raise SystemExit(1) + + +@webhooks.command("simulate") +@click.option( + "--to", + "target_url", + default=None, + help="Receiver URL to POST to. If omitted, the signed request is printed but not sent.", +) +@click.option( + "--secret", + envvar="TANGO_WEBHOOK_SECRET", + required=True, + help="Shared secret used to sign the payload (or TANGO_WEBHOOK_SECRET).", +) +@click.option( + "--payload-file", + type=click.Path(exists=True, dir_okay=False, path_type=Path), + default=None, + help="Path to a JSON file with the body to send. Mutually exclusive with --event-type.", +) +@click.option( + "--event-type", + default=None, + help="Fetch a canonical sample for this event type from Tango and send that.", +) +@click.option( + "--api-key", + envvar="TANGO_API_KEY", + help="Tango API key, only needed with --event-type (or TANGO_API_KEY).", +) +@click.option( + "--base-url", + envvar="TANGO_BASE_URL", + default="https://tango.makegov.com", + show_default=True, + help="Tango base URL, only needed with --event-type.", +) +def simulate_cmd( + target_url: str | None, + secret: str, + payload_file: Path | None, + event_type: str | None, + api_key: str | None, + base_url: str, +) -> None: + """Sign a payload like Tango would. With --to, also POST it to a receiver.""" + if payload_file and event_type: + raise click.UsageError("Use either --payload-file or --event-type, not both.") + + payload: dict[str, Any] | list[Any] + if payload_file: + payload = json.loads(payload_file.read_text(encoding="utf-8")) + elif event_type: + from tango import TangoClient + + client = TangoClient(api_key=api_key, base_url=base_url) + payload = client.get_webhook_sample_payload(event_type=event_type) + else: + payload = {"events": [{"event_type": "tango.cli.simulated", "subject_ids": []}]} + + if target_url is None: + signed = simulate.sign(payload, secret) + click.echo( + json.dumps( + { + "delivered": False, + "headers": signed.headers, + "sent_payload": payload, + }, + indent=2, + sort_keys=True, + ) + ) + return + + result = simulate.deliver(target_url=target_url, payload=payload, secret=secret) + click.echo( + json.dumps( + { + "delivered": True, + "target_url": target_url, + "status_code": result.status_code, + "signature": f"sha256={result.signature}", + "sent_payload": payload, + "receiver_response": result.response_body[:500], + }, + indent=2, + sort_keys=True, + ) + ) + if result.status_code >= 400: + raise SystemExit(1) + + +@webhooks.command("fetch-sample") +@click.option( + "--event-type", + default=None, + help="If set, fetch the canonical sample for that event type. " + "Otherwise return the full samples mapping for every event type.", +) +@click.option( + "--api-key", + envvar="TANGO_API_KEY", + help="Tango API key (or TANGO_API_KEY).", +) +@click.option( + "--base-url", + envvar="TANGO_BASE_URL", + default="https://tango.makegov.com", + show_default=True, + help="Tango base URL (or TANGO_BASE_URL).", +) +def fetch_sample_cmd(event_type: str | None, api_key: str | None, base_url: str) -> None: + """Print the canonical sample payload Tango emits for a given event type.""" + from tango import TangoClient + + client = TangoClient(api_key=api_key, base_url=base_url) + payload = client.get_webhook_sample_payload(event_type=event_type) + click.echo(json.dumps(payload, indent=2, sort_keys=True)) + + +@webhooks.command("list-event-types") +@click.option( + "--api-key", + envvar="TANGO_API_KEY", + help="Tango API key (or TANGO_API_KEY).", +) +@click.option( + "--base-url", + envvar="TANGO_BASE_URL", + default="https://tango.makegov.com", + show_default=True, + help="Tango base URL (or TANGO_BASE_URL).", +) +def list_event_types_cmd(api_key: str | None, base_url: str) -> None: + """List webhook event types Tango supports, with descriptions.""" + from tango import TangoClient + + client = TangoClient(api_key=api_key, base_url=base_url) + response = client.list_webhook_event_types() + width = max((len(et.event_type) for et in response.event_types), default=0) + for et in response.event_types: + click.echo(f"{et.event_type:<{width}} {et.description}") + + +# --------------------------------------------------------------------------- +# Endpoint and subscription management +# --------------------------------------------------------------------------- +# +# These commands wrap the SDK's CRUD methods. Common --api-key / --base-url +# options are repeated per command (rather than factored into a parent group) +# so envvars resolve correctly per click's normal precedence and `--help` +# output stays self-contained. + + +def _tango_client(api_key: str | None, base_url: str) -> Any: + from tango import TangoClient + + return TangoClient(api_key=api_key, base_url=base_url) + + +def _common_api_options(fn: Any) -> Any: + """Stack the --api-key / --base-url options used by every API command.""" + fn = click.option( + "--base-url", + envvar="TANGO_BASE_URL", + default="https://tango.makegov.com", + show_default=True, + help="Tango base URL (or TANGO_BASE_URL).", + )(fn) + fn = click.option( + "--api-key", + envvar="TANGO_API_KEY", + help="Tango API key (or TANGO_API_KEY).", + )(fn) + return fn + + +@webhooks.group("endpoints") +def endpoints_group() -> None: + """Manage webhook endpoints (where Tango delivers).""" + + +@endpoints_group.command("list") +@click.option("--page", type=int, default=1, show_default=True) +@click.option("--limit", type=int, default=25, show_default=True, help="Max per page (cap 100).") +@_common_api_options +def endpoints_list_cmd(page: int, limit: int, api_key: str | None, base_url: str) -> None: + """List webhook endpoints configured for your account.""" + from dataclasses import asdict + + client = _tango_client(api_key, base_url) + resp = client.list_webhook_endpoints(page=page, limit=limit) + click.echo( + json.dumps( + { + "count": resp.count, + "results": [asdict(e) for e in resp.results], + }, + indent=2, + sort_keys=True, + ) + ) + + +@endpoints_group.command("get") +@click.argument("endpoint_id") +@_common_api_options +def endpoints_get_cmd(endpoint_id: str, api_key: str | None, base_url: str) -> None: + """Show one endpoint by id.""" + from dataclasses import asdict + + client = _tango_client(api_key, base_url) + endpoint = client.get_webhook_endpoint(endpoint_id) + click.echo(json.dumps(asdict(endpoint), indent=2, sort_keys=True)) + + +@endpoints_group.command("create") +@click.option("--url", "callback_url", required=True, help="Receiver URL Tango will POST to.") +@click.option("--inactive", is_flag=True, default=False, help="Create the endpoint disabled.") +@_common_api_options +def endpoints_create_cmd( + callback_url: str, inactive: bool, api_key: str | None, base_url: str +) -> None: + """Create a webhook endpoint. Output includes the generated secret — save it.""" + from dataclasses import asdict + + client = _tango_client(api_key, base_url) + endpoint = client.create_webhook_endpoint(callback_url=callback_url, is_active=not inactive) + click.echo(json.dumps(asdict(endpoint), indent=2, sort_keys=True)) + + +@endpoints_group.command("delete") +@click.argument("endpoint_id") +@click.option("--yes", is_flag=True, help="Skip the confirmation prompt.") +@_common_api_options +def endpoints_delete_cmd(endpoint_id: str, yes: bool, api_key: str | None, base_url: str) -> None: + """Delete a webhook endpoint.""" + if not yes: + click.confirm(f"Delete endpoint {endpoint_id}?", abort=True) + client = _tango_client(api_key, base_url) + client.delete_webhook_endpoint(endpoint_id) + click.echo(json.dumps({"deleted": endpoint_id})) + + +@webhooks.group("subscriptions") +def subscriptions_group() -> None: + """Manage webhook subscriptions (what Tango delivers).""" + + +@subscriptions_group.command("list") +@click.option("--page", type=int, default=1, show_default=True) +@click.option("--page-size", type=int, default=None) +@_common_api_options +def subscriptions_list_cmd( + page: int, page_size: int | None, api_key: str | None, base_url: str +) -> None: + """List webhook subscriptions configured for your account.""" + from dataclasses import asdict + + client = _tango_client(api_key, base_url) + resp = client.list_webhook_subscriptions(page=page, page_size=page_size) + click.echo( + json.dumps( + { + "count": resp.count, + "results": [asdict(s) for s in resp.results], + }, + indent=2, + sort_keys=True, + ) + ) + + +@subscriptions_group.command("get") +@click.argument("subscription_id") +@_common_api_options +def subscriptions_get_cmd(subscription_id: str, api_key: str | None, base_url: str) -> None: + """Show one subscription by id.""" + from dataclasses import asdict + + client = _tango_client(api_key, base_url) + sub = client.get_webhook_subscription(subscription_id) + click.echo(json.dumps(asdict(sub), indent=2, sort_keys=True)) + + +@subscriptions_group.command("create") +@click.option("--name", "subscription_name", required=True, help="Human-readable name.") +@click.option("--event-type", required=True, help="Event type to subscribe to.") +@click.option( + "--subject-type", + required=True, + help="Subject type (e.g. 'entity', 'opportunity'). See `list-event-types`.", +) +@click.option( + "--subject-id", + "subject_ids", + multiple=True, + required=True, + help="One or more subject ids. Repeat the flag for multiple.", +) +@_common_api_options +def subscriptions_create_cmd( + subscription_name: str, + event_type: str, + subject_type: str, + subject_ids: tuple[str, ...], + api_key: str | None, + base_url: str, +) -> None: + """Create a webhook subscription with a single records[] entry. + + For multi-record subscriptions, use the SDK's + `create_webhook_subscription` directly with a custom payload. + """ + from dataclasses import asdict + + client = _tango_client(api_key, base_url) + sub = client.create_webhook_subscription( + subscription_name=subscription_name, + payload={ + "records": [ + { + "event_type": event_type, + "subject_type": subject_type, + "subject_ids": list(subject_ids), + } + ] + }, + ) + click.echo(json.dumps(asdict(sub), indent=2, sort_keys=True)) + + +@subscriptions_group.command("delete") +@click.argument("subscription_id") +@click.option("--yes", is_flag=True, help="Skip the confirmation prompt.") +@_common_api_options +def subscriptions_delete_cmd( + subscription_id: str, yes: bool, api_key: str | None, base_url: str +) -> None: + """Delete a webhook subscription.""" + if not yes: + click.confirm(f"Delete subscription {subscription_id}?", abort=True) + client = _tango_client(api_key, base_url) + client.delete_webhook_subscription(subscription_id) + click.echo(json.dumps({"deleted": subscription_id})) + + +def _print_delivery(delivery: Delivery) -> None: + """Default ``listen`` callback: write a one-line summary plus body.""" + summary = _summarize(delivery.body_json) + status = "verified" if delivery.verified else "UNVERIFIED" + parts = [delivery.received_at, status, summary] + if delivery.forward_status is not None: + parts.append(f"forwarded={delivery.forward_status}") + if delivery.forward_error: + parts.append(f"forward_error={delivery.forward_error}") + click.echo(" | ".join(parts)) + if delivery.body_json is not None: + click.echo(json.dumps(delivery.body_json, indent=2, sort_keys=True)) + click.echo("") + + +def _summarize(body: Any) -> str: + if isinstance(body, dict): + events = body.get("events") + if isinstance(events, list) and events and isinstance(events[0], dict): + event_type = events[0].get("event_type", "?") + count = len(events) + return f"{event_type} (n={count})" + return "(no events)" + + +# Make ``X-Tango-Signature`` accessible from this module for IDE completion. +__all__ = ["main", "SIGNATURE_HEADER"] diff --git a/tango/webhooks/receiver.py b/tango/webhooks/receiver.py new file mode 100644 index 0000000..8679252 --- /dev/null +++ b/tango/webhooks/receiver.py @@ -0,0 +1,232 @@ +"""Local webhook receiver for development and integration testing. + +A small stdlib-based HTTP server that accepts Tango-style POSTs, verifies +the ``X-Tango-Signature`` header against a shared secret, optionally +forwards the request to a downstream URL (e.g. your real handler running on +another port), and records each delivery in memory for later inspection. + +Typical use from the CLI:: + + tango webhooks listen --port 8011 --secret $TANGO_WEBHOOK_SECRET \\ + --forward-to http://localhost:4242/webhooks + +Or programmatically inside an integration test:: + + from tango.webhooks import WebhookReceiver + + with WebhookReceiver(secret="dev_secret").run() as rx: + # ... cause a webhook to fire at rx.url ... + deliveries = rx.deliveries +""" + +from __future__ import annotations + +import contextlib +import json +import threading +from collections import deque +from collections.abc import Callable, Iterator +from dataclasses import dataclass, field +from datetime import UTC, datetime +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any + +from tango.webhooks.signing import SIGNATURE_HEADER, verify_signature + +DEFAULT_PATH = "/tango/webhooks" +DEFAULT_MAX_HISTORY = 256 + + +@dataclass +class Delivery: + """A recorded webhook delivery.""" + + received_at: str + path: str + signature_header: str | None + body_bytes: bytes + body_json: Any + verified: bool + remote_addr: str | None = None + forward_status: int | None = None + forward_error: str | None = None + + +@dataclass +class WebhookReceiver: + """A configurable local receiver for Tango webhook deliveries. + + Args: + secret: Shared secret. If empty, signatures are not verified and + every delivery is recorded with ``verified=False`` — useful for + inspecting payloads without a configured endpoint. + path: URL path to accept deliveries on. Defaults to ``/tango/webhooks``. + host: Bind address. Defaults to ``127.0.0.1`` (loopback only). + port: TCP port. ``0`` lets the OS choose a free port. + forward_to: Optional URL to mirror each delivery to, preserving body + bytes and the signature header. + max_history: Cap on the in-memory ``deliveries`` deque. + on_delivery: Optional callback invoked for each recorded delivery. + require_signature: If True (the default when a secret is set), + unsigned or invalid deliveries get a 401 response. + """ + + secret: str = "" + path: str = DEFAULT_PATH + host: str = "127.0.0.1" + port: int = 0 + forward_to: str | None = None + max_history: int = DEFAULT_MAX_HISTORY + on_delivery: Callable[[Delivery], None] | None = None + require_signature: bool | None = None + + _server: ThreadingHTTPServer | None = field(default=None, init=False, repr=False) + _thread: threading.Thread | None = field(default=None, init=False, repr=False) + _deliveries: deque[Delivery] = field(default_factory=deque, init=False, repr=False) + + @property + def deliveries(self) -> list[Delivery]: + """Snapshot of recorded deliveries, oldest first.""" + return list(self._deliveries) + + @property + def url(self) -> str: + """Full URL the receiver is bound to (only valid while running).""" + if self._server is None: + raise RuntimeError("Receiver is not running") + host_addr, port = self._server.server_address[:2] + host = host_addr.decode() if isinstance(host_addr, bytes) else str(host_addr) + return f"http://{host}:{port}{self.path}" + + def start(self) -> None: + """Bind the socket and start serving in a background thread.""" + if self._server is not None: + raise RuntimeError("Receiver already started") + receiver = self + deliveries = self._deliveries + max_history = self.max_history + + class Handler(BaseHTTPRequestHandler): + def log_message(self, format: str, *args: Any) -> None: # noqa: A002 + # Suppress stderr access logging; users see deliveries through + # the on_delivery callback or the deliveries list instead. + return + + def do_POST(self) -> None: # noqa: N802 (stdlib API) + if self.path != receiver.path: + self._write_json(404, {"ok": False, "error": "not_found"}) + return + length = int(self.headers.get("Content-Length", "0") or 0) + body = self.rfile.read(length) if length > 0 else b"" + signature = self.headers.get(SIGNATURE_HEADER) + verified = bool(receiver.secret) and verify_signature( + body, receiver.secret, signature + ) + + require = ( + receiver.require_signature + if receiver.require_signature is not None + else bool(receiver.secret) + ) + if require and not verified: + self._record(body, signature, verified=False) + self._write_json(401, {"ok": False, "error": "invalid_signature"}) + return + + forward_status: int | None = None + forward_error: str | None = None + if receiver.forward_to: + forward_status, forward_error = _forward(receiver.forward_to, body, signature) + + self._record( + body, + signature, + verified=verified, + forward_status=forward_status, + forward_error=forward_error, + ) + self._write_json(200, {"ok": True}) + + def _write_json(self, status: int, body: dict[str, Any]) -> None: + payload = json.dumps(body).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + def _record( + self, + body: bytes, + signature: str | None, + *, + verified: bool, + forward_status: int | None = None, + forward_error: str | None = None, + ) -> None: + try: + parsed: Any = json.loads(body.decode("utf-8")) if body else None + except (UnicodeDecodeError, json.JSONDecodeError): + parsed = None + delivery = Delivery( + received_at=datetime.now(UTC).isoformat().replace("+00:00", "Z"), + path=self.path, + signature_header=signature, + body_bytes=body, + body_json=parsed, + verified=verified, + remote_addr=self.client_address[0] if self.client_address else None, + forward_status=forward_status, + forward_error=forward_error, + ) + while len(deliveries) >= max_history: + deliveries.popleft() + deliveries.append(delivery) + if receiver.on_delivery is not None: + receiver.on_delivery(delivery) + + self._server = ThreadingHTTPServer((self.host, self.port), Handler) + self._thread = threading.Thread( + target=self._server.serve_forever, + name="tango-webhook-receiver", + daemon=True, + ) + self._thread.start() + + def stop(self) -> None: + """Stop the server and join the background thread.""" + if self._server is None: + return + self._server.shutdown() + self._server.server_close() + if self._thread is not None: + self._thread.join(timeout=5) + self._server = None + self._thread = None + + @contextlib.contextmanager + def run(self) -> Iterator[WebhookReceiver]: + """Context manager that starts the receiver and stops it on exit.""" + self.start() + try: + yield self + finally: + self.stop() + + +def _forward(url: str, body: bytes, signature: str | None) -> tuple[int | None, str | None]: + """POST ``body`` to ``url`` preserving the signature header. + + Returns ``(status, error_message)``. httpx is imported lazily so unit + tests that don't exercise forwarding don't pay the import cost. + """ + import httpx + + headers = {"Content-Type": "application/json"} + if signature: + headers[SIGNATURE_HEADER] = signature + try: + resp = httpx.post(url, content=body, headers=headers, timeout=10.0) + except httpx.HTTPError as exc: + return None, str(exc) + return resp.status_code, None diff --git a/tango/webhooks/signing.py b/tango/webhooks/signing.py new file mode 100644 index 0000000..37f9338 --- /dev/null +++ b/tango/webhooks/signing.py @@ -0,0 +1,50 @@ +"""HMAC-SHA256 signing for Tango webhook deliveries. + +Tango signs each delivery with:: + + X-Tango-Signature: sha256= + +These helpers mirror the canonical implementation in the tango server +(``webhooks/utils.py``). Verifiers must operate on the **raw request body +bytes** — re-serializing parsed JSON will produce a different signature. +""" + +from __future__ import annotations + +import hashlib +import hmac + +SIGNATURE_HEADER = "X-Tango-Signature" +SIGNATURE_PREFIX = "sha256=" + + +def generate_signature(body: bytes, secret: str) -> str: + """Return the lowercase hex HMAC-SHA256 of ``body`` keyed by ``secret``.""" + return hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest() + + +def parse_signature_header(value: str | None) -> str | None: + """Strip the ``sha256=`` prefix from a header value; return ``None`` if empty. + + Accepts the bare hex form too, for forward compatibility. + """ + if not value: + return None + stripped = value.strip() + if stripped.startswith(SIGNATURE_PREFIX): + return stripped[len(SIGNATURE_PREFIX) :] + return stripped + + +def verify_signature(body: bytes, secret: str, signature_header: str | None) -> bool: + """Return True if ``signature_header`` matches the HMAC of ``body``. + + Uses :func:`hmac.compare_digest` for constant-time comparison. + Returns False for an absent or malformed header rather than raising — let + callers decide how to respond (typically a 401 / 403). + """ + received = parse_signature_header(signature_header) + if not received: + return False + expected = generate_signature(body, secret) + return hmac.compare_digest(expected, received) diff --git a/tango/webhooks/simulate.py b/tango/webhooks/simulate.py new file mode 100644 index 0000000..064a78e --- /dev/null +++ b/tango/webhooks/simulate.py @@ -0,0 +1,102 @@ +"""Locally sign and POST a webhook payload to a URL. + +This module is the offline counterpart to ``test_webhook_delivery``: it +never talks to the Tango API. Use it when you want to drive a downstream +receiver without provisioning a real subscription, or when you want to +fuzz event shapes that Tango wouldn't naturally emit. + +Example:: + + from tango.webhooks import simulate + + result = simulate.deliver( + target_url="http://localhost:4242/webhooks", + payload={"events": [{"event_type": "entities.updated", "uei": "ABC123"}]}, + secret="dev_secret", + ) + assert result.status_code == 200 +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import Any + +from tango.webhooks.signing import SIGNATURE_HEADER, SIGNATURE_PREFIX, generate_signature + + +@dataclass(frozen=True) +class SignedRequest: + """A Tango-shaped signed request, ready to be POSTed.""" + + body: bytes + signature: str # bare lowercase hex + headers: dict[str, str] # includes Content-Type and X-Tango-Signature + + +@dataclass(frozen=True) +class SimulationResult: + """Outcome of a simulated delivery.""" + + status_code: int + response_body: str + signature: str + sent_bytes: bytes + + +def sign(payload: dict[str, Any] | list[Any] | bytes | str, secret: str) -> SignedRequest: + """Serialize and sign ``payload`` without sending it. + + Useful for showing devs the exact wire form their handler would + receive, or for hand-rolling deliveries with a custom HTTP client. + """ + body = _to_bytes(payload) + signature_hex = generate_signature(body, secret) + return SignedRequest( + body=body, + signature=signature_hex, + headers={ + "Content-Type": "application/json", + SIGNATURE_HEADER: f"{SIGNATURE_PREFIX}{signature_hex}", + }, + ) + + +def deliver( + *, + target_url: str, + payload: dict[str, Any] | list[Any] | bytes | str, + secret: str, + extra_headers: dict[str, str] | None = None, + timeout: float = 10.0, +) -> SimulationResult: + """Sign ``payload`` with ``secret`` and POST it to ``target_url``. + + ``payload`` may be a ``dict``/``list`` (serialized via :func:`json.dumps` + with ``sort_keys=True`` to keep signatures reproducible across runs), + a pre-serialized ``str``, or raw ``bytes``. Signing is computed over the + exact bytes that go on the wire. + """ + import httpx + + signed = sign(payload, secret) + headers = dict(signed.headers) + if extra_headers: + headers.update(extra_headers) + + resp = httpx.post(target_url, content=signed.body, headers=headers, timeout=timeout) + return SimulationResult( + status_code=resp.status_code, + response_body=resp.text, + signature=signed.signature, + sent_bytes=signed.body, + ) + + +def _to_bytes(payload: dict[str, Any] | list[Any] | bytes | str) -> bytes: + if isinstance(payload, bytes): + return payload + if isinstance(payload, str): + return payload.encode("utf-8") + return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") diff --git a/tests/test_webhooks_cli.py b/tests/test_webhooks_cli.py new file mode 100644 index 0000000..bacd84f --- /dev/null +++ b/tests/test_webhooks_cli.py @@ -0,0 +1,348 @@ +"""Smoke tests for the `tango webhooks` CLI.""" + +from __future__ import annotations + +import json + +from click.testing import CliRunner + +from tango.webhooks.cli import main +from tango.webhooks.receiver import WebhookReceiver + + +def test_cli_help() -> None: + runner = CliRunner() + result = runner.invoke(main, ["webhooks", "--help"]) + assert result.exit_code == 0 + assert "listen" in result.output + assert "trigger" in result.output + assert "simulate" in result.output + assert "fetch-sample" in result.output + assert "list-event-types" in result.output + + +def test_cli_simulate_without_to_prints_signed_request() -> None: + """Without --to, simulate signs and prints — no POST, no listener required.""" + runner = CliRunner() + result = runner.invoke( + main, + ["webhooks", "simulate", "--secret", "dev"], + ) + assert result.exit_code == 0, result.output + body = json.loads(result.output) + assert body["delivered"] is False + assert body["headers"]["Content-Type"] == "application/json" + assert body["headers"]["X-Tango-Signature"].startswith("sha256=") + assert "events" in body["sent_payload"] + + +def test_cli_simulate_signs_and_posts(tmp_path: object) -> None: + runner = CliRunner() + secret = "cli-secret" + payload = {"events": [{"event_type": "cli.smoke"}]} + with WebhookReceiver(secret=secret).run() as rx: + result = runner.invoke( + main, + [ + "webhooks", + "simulate", + "--to", + rx.url, + "--secret", + secret, + ], + input=json.dumps(payload), # ignored by current command, harmless + ) + assert result.exit_code == 0, result.output + # The default body is the built-in placeholder envelope. + assert len(rx.deliveries) == 1 + assert rx.deliveries[0].verified is True + body = json.loads(result.output) + assert body["delivered"] is True + assert body["status_code"] == 200 + assert body["signature"].startswith("sha256=") + assert body["target_url"] == rx.url + # Output now includes the actual payload that was sent (the dev's + # main artifact of interest), not just its byte length. + assert isinstance(body["sent_payload"], dict) + assert "events" in body["sent_payload"] + assert body["receiver_response"] == '{"ok": true}' + + +def test_cli_simulate_with_payload_file(tmp_path: object) -> None: + import pathlib + + p = pathlib.Path(str(tmp_path)) / "payload.json" + payload = {"events": [{"event_type": "from.file", "subject_ids": ["S1"]}]} + p.write_text(json.dumps(payload), encoding="utf-8") + + runner = CliRunner() + secret = "file-secret" + with WebhookReceiver(secret=secret).run() as rx: + result = runner.invoke( + main, + [ + "webhooks", + "simulate", + "--to", + rx.url, + "--secret", + secret, + "--payload-file", + str(p), + ], + ) + assert result.exit_code == 0, result.output + assert rx.deliveries[0].body_json == payload + + +def test_cli_fetch_sample_prints_payload() -> None: + """fetch-sample hits the SDK's get_webhook_sample_payload and pretty-prints.""" + from unittest.mock import Mock, patch + + sample = {"events": [{"event_type": "entities.updated", "uei": "ABC"}]} + mock_response = Mock() + mock_response.status_code = 200 + mock_response.headers = {} + mock_response.json.return_value = sample + mock_response.raise_for_status = Mock() + + runner = CliRunner() + with patch("tango.client.httpx.Client.request", return_value=mock_response): + result = runner.invoke( + main, + [ + "webhooks", + "fetch-sample", + "--event-type", + "entities.updated", + "--api-key", + "k", + ], + ) + assert result.exit_code == 0, result.output + assert json.loads(result.output) == sample + + +def test_cli_list_event_types_prints_table() -> None: + from unittest.mock import Mock, patch + + api_response = { + "event_types": [ + { + "event_type": "entities.updated", + "default_subject_type": "entity", + "description": "Entity updated", + "schema_version": 1, + }, + { + "event_type": "awards.created", + "default_subject_type": "award", + "description": "New award", + "schema_version": 1, + }, + ], + "subject_types": [], + "subject_type_definitions": [], + } + mock_response = Mock() + mock_response.status_code = 200 + mock_response.headers = {} + mock_response.json.return_value = api_response + mock_response.raise_for_status = Mock() + + runner = CliRunner() + with patch("tango.client.httpx.Client.request", return_value=mock_response): + result = runner.invoke(main, ["webhooks", "list-event-types", "--api-key", "k"]) + assert result.exit_code == 0, result.output + assert "entities.updated" in result.output + assert "Entity updated" in result.output + assert "awards.created" in result.output + + +def _mock_response(api_response: dict[str, object]) -> object: + from unittest.mock import Mock + + mock_response = Mock() + mock_response.status_code = 200 + mock_response.headers = {} + mock_response.json.return_value = api_response + mock_response.raise_for_status = Mock() + return mock_response + + +def test_cli_endpoints_list() -> None: + from unittest.mock import patch + + api = { + "count": 1, + "next": None, + "previous": None, + "results": [ + { + "id": "ep-1", + "name": "default", + "callback_url": "https://example/webhooks", + "is_active": True, + "created_at": "2026-01-01T00:00:00Z", + "updated_at": "2026-01-01T00:00:00Z", + "secret": None, + } + ], + } + runner = CliRunner() + with patch("tango.client.httpx.Client.request", return_value=_mock_response(api)): + result = runner.invoke(main, ["webhooks", "endpoints", "list", "--api-key", "k"]) + assert result.exit_code == 0, result.output + body = json.loads(result.output) + assert body["count"] == 1 + assert body["results"][0]["callback_url"] == "https://example/webhooks" + + +def test_cli_endpoints_create_returns_secret() -> None: + from unittest.mock import patch + + api = { + "id": "ep-2", + "name": "default", + "callback_url": "https://example/wh", + "is_active": True, + "created_at": "2026-05-07T00:00:00Z", + "updated_at": "2026-05-07T00:00:00Z", + "secret": "whsec_redacted_in_test", + } + runner = CliRunner() + with patch("tango.client.httpx.Client.request", return_value=_mock_response(api)): + result = runner.invoke( + main, + [ + "webhooks", + "endpoints", + "create", + "--url", + "https://example/wh", + "--api-key", + "k", + ], + ) + assert result.exit_code == 0, result.output + body = json.loads(result.output) + assert body["id"] == "ep-2" + assert body["secret"] == "whsec_redacted_in_test" + + +def test_cli_endpoints_delete_requires_confirmation() -> None: + from unittest.mock import patch + + api: dict[str, object] = {} + runner = CliRunner() + with patch("tango.client.httpx.Client.request", return_value=_mock_response(api)): + # Without --yes, abort if user says n. + result = runner.invoke( + main, + ["webhooks", "endpoints", "delete", "ep-1", "--api-key", "k"], + input="n\n", + ) + assert result.exit_code != 0 # aborted + # With --yes, proceeds. + result = runner.invoke( + main, + ["webhooks", "endpoints", "delete", "ep-1", "--yes", "--api-key", "k"], + ) + assert result.exit_code == 0, result.output + assert json.loads(result.output) == {"deleted": "ep-1"} + + +def test_cli_subscriptions_create_builds_records_payload() -> None: + """Verify the `--event-type / --subject-type / --subject-id` flags get folded + into the right `payload.records[0]` shape Tango expects.""" + from unittest.mock import patch + + api = { + "id": "sub-1", + "endpoint": "ep-1", + "subscription_name": "ent-watch", + "payload": { + "records": [ + { + "event_type": "entities.updated", + "subject_type": "entity", + "subject_ids": ["UEI1", "UEI2"], + } + ] + }, + "created_at": "2026-05-07T00:00:00Z", + } + runner = CliRunner() + with patch( + "tango.client.httpx.Client.request", return_value=_mock_response(api) + ) as mock_request: + result = runner.invoke( + main, + [ + "webhooks", + "subscriptions", + "create", + "--name", + "ent-watch", + "--event-type", + "entities.updated", + "--subject-type", + "entity", + "--subject-id", + "UEI1", + "--subject-id", + "UEI2", + "--api-key", + "k", + ], + ) + assert result.exit_code == 0, result.output + # The SDK was called with the constructed payload. + sent_json = mock_request.call_args.kwargs["json"] + assert sent_json["subscription_name"] == "ent-watch" + assert sent_json["payload"]["records"][0]["subject_ids"] == ["UEI1", "UEI2"] + + +def test_cli_subscriptions_list() -> None: + from unittest.mock import patch + + api = { + "count": 0, + "next": None, + "previous": None, + "results": [], + } + runner = CliRunner() + with patch("tango.client.httpx.Client.request", return_value=_mock_response(api)): + result = runner.invoke(main, ["webhooks", "subscriptions", "list", "--api-key", "k"]) + assert result.exit_code == 0, result.output + assert json.loads(result.output) == {"count": 0, "results": []} + + +def test_cli_simulate_rejects_both_modes(tmp_path: object) -> None: + import pathlib + + # Click validates --payload-file exists before the command body runs, so + # we need a real path here. /dev/null worked on POSIX but not Windows CI. + p = pathlib.Path(str(tmp_path)) / "p.json" + p.write_text("{}", encoding="utf-8") + + runner = CliRunner() + result = runner.invoke( + main, + [ + "webhooks", + "simulate", + "--to", + "http://example.invalid/", + "--secret", + "x", + "--payload-file", + str(p), + "--event-type", + "entities.updated", + ], + ) + assert result.exit_code != 0 + assert "either --payload-file or --event-type" in result.output diff --git a/tests/test_webhooks_receiver.py b/tests/test_webhooks_receiver.py new file mode 100644 index 0000000..63c389c --- /dev/null +++ b/tests/test_webhooks_receiver.py @@ -0,0 +1,127 @@ +"""Tests for tango.webhooks.receiver.WebhookReceiver.""" + +from __future__ import annotations + +import json + +import httpx +import pytest + +from tango.webhooks import generate_signature +from tango.webhooks.receiver import WebhookReceiver + +SECRET = "test_secret" +PAYLOAD = {"events": [{"event_type": "entities.updated", "uei": "TEST123"}]} + + +def _post_signed(url: str, body: bytes, secret: str) -> httpx.Response: + sig = generate_signature(body, secret) + return httpx.post( + url, + content=body, + headers={ + "Content-Type": "application/json", + "X-Tango-Signature": f"sha256={sig}", + }, + timeout=5.0, + ) + + +def test_receiver_records_verified_delivery() -> None: + body = json.dumps(PAYLOAD).encode("utf-8") + with WebhookReceiver(secret=SECRET).run() as rx: + resp = _post_signed(rx.url, body, SECRET) + assert resp.status_code == 200 + assert resp.headers["content-type"] == "application/json" + assert resp.json() == {"ok": True} + assert rx.deliveries[0].verified is True + assert rx.deliveries[0].body_bytes == body + assert rx.deliveries[0].body_json == PAYLOAD + + +def test_receiver_rejects_bad_signature_with_401() -> None: + body = json.dumps(PAYLOAD).encode("utf-8") + with WebhookReceiver(secret=SECRET).run() as rx: + resp = httpx.post( + rx.url, + content=body, + headers={"X-Tango-Signature": "sha256=deadbeef"}, + timeout=5.0, + ) + assert resp.status_code == 401 + assert resp.headers["content-type"] == "application/json" + assert resp.json() == {"ok": False, "error": "invalid_signature"} + # The bad delivery is still recorded, marked unverified, so devs + # can debug what arrived. + assert len(rx.deliveries) == 1 + assert rx.deliveries[0].verified is False + + +def test_receiver_rejects_missing_signature_with_401() -> None: + body = json.dumps(PAYLOAD).encode("utf-8") + with WebhookReceiver(secret=SECRET).run() as rx: + resp = httpx.post(rx.url, content=body, timeout=5.0) + assert resp.status_code == 401 + + +def test_receiver_with_no_secret_accepts_unsigned() -> None: + body = json.dumps(PAYLOAD).encode("utf-8") + with WebhookReceiver(secret="").run() as rx: + resp = httpx.post(rx.url, content=body, timeout=5.0) + assert resp.status_code == 200 + assert rx.deliveries[0].verified is False + + +def test_receiver_404s_on_unknown_path() -> None: + with WebhookReceiver(secret=SECRET, path="/tango/webhooks").run() as rx: + wrong = rx.url.replace("/tango/webhooks", "/elsewhere") + resp = httpx.post(wrong, content=b"{}", timeout=5.0) + assert resp.status_code == 404 + assert resp.headers["content-type"] == "application/json" + assert resp.json() == {"ok": False, "error": "not_found"} + + +def test_receiver_invokes_on_delivery_callback() -> None: + seen: list[str] = [] + body = json.dumps(PAYLOAD).encode("utf-8") + with WebhookReceiver( + secret=SECRET, on_delivery=lambda d: seen.append(d.body_bytes.decode()) + ).run() as rx: + _post_signed(rx.url, body, SECRET) + assert seen == [body.decode()] + + +def test_receiver_max_history_caps_deliveries() -> None: + body = json.dumps(PAYLOAD).encode("utf-8") + with WebhookReceiver(secret=SECRET, max_history=3).run() as rx: + for _ in range(5): + _post_signed(rx.url, body, SECRET) + assert len(rx.deliveries) == 3 + + +def test_receiver_forwards_to_downstream() -> None: + body = json.dumps(PAYLOAD).encode("utf-8") + with WebhookReceiver(secret=SECRET, max_history=10).run() as downstream: + with WebhookReceiver(secret=SECRET, forward_to=downstream.url, port=0).run() as upstream: + resp = _post_signed(upstream.url, body, SECRET) + assert resp.status_code == 200 + # Downstream should have received the same bytes with the same signature. + assert len(downstream.deliveries) == 1 + assert downstream.deliveries[0].body_bytes == body + assert downstream.deliveries[0].verified is True + + +def test_url_property_raises_before_start() -> None: + rx = WebhookReceiver(secret=SECRET) + with pytest.raises(RuntimeError): + _ = rx.url + + +def test_double_start_raises() -> None: + rx = WebhookReceiver(secret=SECRET) + rx.start() + try: + with pytest.raises(RuntimeError): + rx.start() + finally: + rx.stop() diff --git a/tests/test_webhooks_signing.py b/tests/test_webhooks_signing.py new file mode 100644 index 0000000..a136324 --- /dev/null +++ b/tests/test_webhooks_signing.py @@ -0,0 +1,73 @@ +"""Tests for tango.webhooks.signing. + +The signing scheme has to match the tango server byte-for-byte. The +``KNOWN_VECTORS`` constants pin a few payload/secret/signature triples that +were computed independently against tango's reference implementation +(``webhooks/utils.py::generate_signature``). Drift in either direction +should fail this test. +""" + +from __future__ import annotations + +import hashlib +import hmac + +from tango.webhooks import generate_signature, parse_signature_header, verify_signature + +KNOWN_VECTORS: list[tuple[bytes, str, str]] = [ + # (body_bytes, secret, expected_lowercase_hex_hmac_sha256) + (b"", "dev_secret", hmac.new(b"dev_secret", b"", hashlib.sha256).hexdigest()), + ( + b'{"events":[{"event_type":"entities.updated","uei":"ABC123"}]}', + "shh", + hmac.new( + b"shh", + b'{"events":[{"event_type":"entities.updated","uei":"ABC123"}]}', + hashlib.sha256, + ).hexdigest(), + ), +] + + +def test_generate_signature_matches_reference_algorithm() -> None: + for body, secret, expected in KNOWN_VECTORS: + assert generate_signature(body, secret) == expected + + +def test_generate_signature_is_lowercase_hex() -> None: + sig = generate_signature(b"payload", "secret") + assert sig == sig.lower() + int(sig, 16) # must parse as hex + + +def test_verify_signature_round_trip() -> None: + body = b'{"events":[{"event_type":"awards.created"}]}' + secret = "rotating-secret" + sig = generate_signature(body, secret) + assert verify_signature(body, secret, f"sha256={sig}") is True + assert verify_signature(body, secret, sig) is True # bare hex also accepted + + +def test_verify_signature_rejects_tampered_body() -> None: + secret = "secret" + sig = generate_signature(b"original", secret) + assert verify_signature(b"tampered", secret, f"sha256={sig}") is False + + +def test_verify_signature_rejects_wrong_secret() -> None: + sig = generate_signature(b"body", "right") + assert verify_signature(b"body", "wrong", f"sha256={sig}") is False + + +def test_verify_signature_handles_missing_or_empty_header() -> None: + assert verify_signature(b"body", "secret", None) is False + assert verify_signature(b"body", "secret", "") is False + assert verify_signature(b"body", "secret", "sha256=") is False + + +def test_parse_signature_header_strips_prefix() -> None: + assert parse_signature_header("sha256=abc123") == "abc123" + assert parse_signature_header(" sha256=abc ") == "abc" + assert parse_signature_header("abc123") == "abc123" + assert parse_signature_header(None) is None + assert parse_signature_header("") is None diff --git a/tests/test_webhooks_simulate.py b/tests/test_webhooks_simulate.py new file mode 100644 index 0000000..cbbcaee --- /dev/null +++ b/tests/test_webhooks_simulate.py @@ -0,0 +1,62 @@ +"""Tests for tango.webhooks.simulate.""" + +from __future__ import annotations + +from tango.webhooks import simulate, verify_signature +from tango.webhooks.receiver import WebhookReceiver + +SECRET = "shared" + + +def test_sign_returns_ready_to_post_request() -> None: + """`sign` produces a SignedRequest that a downstream verifier accepts.""" + payload = {"events": [{"event_type": "entities.updated"}]} + signed = simulate.sign(payload, SECRET) + assert signed.headers["Content-Type"] == "application/json" + assert signed.headers["X-Tango-Signature"] == f"sha256={signed.signature}" + # Round-trip: verify the produced signature against the produced body. + assert verify_signature(signed.body, SECRET, signed.headers["X-Tango-Signature"]) + + +def test_sign_does_not_make_http_request(monkeypatch: object) -> None: + """`sign` is purely local; importing httpx isn't required.""" + # If anyone tries to call httpx.post here we'd want to know — but the + # easier signal is that sign() returns synchronously with no target_url + # parameter at all. This test just documents that contract. + signed = simulate.sign({"x": 1}, "s") + assert isinstance(signed.body, bytes) + + +def test_deliver_signs_and_posts_dict_payload() -> None: + payload = {"events": [{"event_type": "awards.created", "award_key": "X"}]} + with WebhookReceiver(secret=SECRET).run() as rx: + result = simulate.deliver(target_url=rx.url, payload=payload, secret=SECRET) + assert result.status_code == 200 + assert len(result.signature) == 64 # hex sha256 + # Receiver verified the signature, so the bytes round-tripped intact. + assert rx.deliveries[0].verified is True + assert rx.deliveries[0].body_json == payload + + +def test_deliver_accepts_raw_bytes() -> None: + raw = b'{"events":[{"event_type":"x"}]}' + with WebhookReceiver(secret=SECRET).run() as rx: + result = simulate.deliver(target_url=rx.url, payload=raw, secret=SECRET) + assert result.status_code == 200 + assert result.sent_bytes == raw + + +def test_deliver_dict_serialization_is_deterministic() -> None: + """Same dict in two calls produces the same signature (sort_keys + compact).""" + payload = {"b": 2, "a": 1} + with WebhookReceiver(secret=SECRET).run() as rx: + first = simulate.deliver(target_url=rx.url, payload=payload, secret=SECRET) + second = simulate.deliver(target_url=rx.url, payload=payload, secret=SECRET) + assert first.signature == second.signature + assert first.sent_bytes == second.sent_bytes + + +def test_deliver_wrong_secret_yields_401() -> None: + with WebhookReceiver(secret=SECRET).run() as rx: + result = simulate.deliver(target_url=rx.url, payload={"x": 1}, secret="not-the-secret") + assert result.status_code == 401 diff --git a/uv.lock b/uv.lock index 0edfd6b..218b45d 100644 --- a/uv.lock +++ b/uv.lock @@ -276,6 +276,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/0a/4c/925909008ed5a988ccbb72dcc897407e5d6d3bd72410d69e051fc0c14647/charset_normalizer-3.4.4-py3-none-any.whl", hash = "sha256:7a32c560861a02ff789ad905a2fe94e3f840803362c84fecf1851cb4cf3dc37f", size = 53402, upload-time = "2025-10-14T04:42:31.76Z" }, ] +[[package]] +name = "click" +version = "8.3.3" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/bb/63/f9e1ea081ce35720d8b92acde70daaedace594dc93b693c869e0d5910718/click-8.3.3.tar.gz", hash = "sha256:398329ad4837b2ff7cbe1dd166a4c0f8900c3ca3a218de04466f38f6497f18a2", size = 328061, upload-time = "2026-04-22T15:11:27.506Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ae/44/c1221527f6a71a01ec6fbad7fa78f1d50dfa02217385cf0fa3eec7087d59/click-8.3.3-py3-none-any.whl", hash = "sha256:a2bf429bb3033c89fa4936ffb35d5cb471e3719e1f3c8a7c3fff0b8314305613", size = 110502, upload-time = "2026-04-22T15:11:25.044Z" }, +] + [[package]] name = "colorama" version = "0.4.6" @@ -1852,6 +1864,9 @@ notebooks = [ { name = "ipykernel" }, { name = "jupyter" }, ] +webhooks = [ + { name = "click" }, +] [package.dev-dependencies] dev = [ @@ -1860,6 +1875,7 @@ dev = [ [package.metadata] requires-dist = [ + { name = "click", marker = "extra == 'webhooks'", specifier = ">=8.1" }, { name = "httpx", specifier = ">=0.27.0" }, { name = "ipykernel", marker = "extra == 'notebooks'", specifier = ">=6.25.0" }, { name = "jupyter", marker = "extra == 'notebooks'", specifier = ">=1.0.0" }, @@ -1872,7 +1888,7 @@ requires-dist = [ { name = "pyyaml", marker = "extra == 'dev'", specifier = ">=6.0" }, { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.3.0" }, ] -provides-extras = ["dev", "notebooks"] +provides-extras = ["dev", "notebooks", "webhooks"] [package.metadata.requires-dev] dev = [{ name = "python-dotenv", specifier = ">=1.2.1" }]