Python verifier

Save as verify-signature.py. Setup and usage are on Verifier samples.

"""Verify a Tidenda webhook delivery end-to-end.

Two signatures travel with every delivery:

  1. HMAC-SHA256 over ``f"{X-Tidenda-Timestamp}.{raw-body}"`` keyed with the
     shared webhook secret. Header: ``X-Tidenda-Signature: sha256=<hex>``.
  2. (When ``document.snapshotBytes`` is present) Ed25519 over the byte
     sequence ``b"tidenda-snapshot-v1\\0" + decoded_snapshot_bytes``.

The HMAC proves the HTTP request came from Tidenda. The Ed25519 signature
travels with the snapshot itself, so any downstream consumer can verify
the content's authenticity without trusting your hub.

Run::

    pip install cryptography
    TIDENDA_WEBHOOK_SECRET=wh_... \\
    TIDENDA_SIGNING_KEY=<base64-SPKI-DER> \\
    python verify-signature.py

Then POST something to http://localhost:8787/webhook.
"""

from __future__ import annotations

import base64
import hmac
import json
import os
import time
from hashlib import sha256
from http.server import BaseHTTPRequestHandler, HTTPServer

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

# In real deployments, both of these come from a secret manager. They are
# hardcoded placeholders here so the sample focuses on verification.
SECRET = os.environ.get("TIDENDA_WEBHOOK_SECRET", "wh_00000000_replace_me").encode("utf-8")
SIGNING_KEY_B64 = os.environ.get("TIDENDA_SIGNING_KEY", "")  # base64 SPKI DER

# The Ed25519 domain-separation prefix. Note the trailing NUL byte - the
# Ed25519 message is exactly these 20 bytes followed by the decoded
# snapshot bytes. See docs/RECEIVING_WEBHOOKS.md section 4.
SNAPSHOT_DOMAIN = b"tidenda-snapshot-v1\x00"

# Tolerance window for X-Tidenda-Timestamp drift, in seconds. The sender
# re-stamps every retry, so this only catches genuinely stale traffic.
MAX_CLOCK_SKEW_SECONDS = 300


def verify_hmac(raw_body: bytes, timestamp: str, header: str) -> bool:
    # The header is `sha256=<lowercase hex>`. Anything else is invalid.
    if not header.startswith("sha256="):
        return False
    given = header[len("sha256=") :]

    # Replay protection: reject if the dispatch timestamp is too far from
    # the local wall clock. Symmetric window tolerates skew in either direction.
    try:
        ts = int(timestamp)
    except ValueError:
        return False
    if abs(time.time() - ts) > MAX_CLOCK_SKEW_SECONDS:
        return False

    # Signed material is exactly `${timestamp}.${raw-body}`. The body
    # joins as UTF-8 text - webhook bodies are application/json and thus
    # valid UTF-8 by definition.
    signed = f"{ts}.".encode("utf-8") + raw_body
    expected = hmac.new(SECRET, signed, sha256).hexdigest()

    # compare_digest is the constant-time comparator. Don't use `==`;
    # string equality short-circuits and leaks bytes through timing.
    return hmac.compare_digest(expected, given)


def verify_snapshot(snapshot_b64: str, signature: dict) -> bool:
    if not SIGNING_KEY_B64:
        print("TIDENDA_SIGNING_KEY not set - skipping snapshot signature check")
        return True
    if signature.get("alg") != "ed25519":
        return False

    # Decode the snapshot to raw bytes. Verify against THESE bytes -
    # not against `json.dumps(json.loads(bytes))`, which would not be
    # byte-equal (JSON has no canonical form).
    snapshot_bytes = base64.b64decode(snapshot_b64)
    sig = base64.b64decode(signature["sig"])

    # The signed message is the 20-byte domain prefix followed by the
    # raw snapshot bytes. Domain separation prevents a signature minted
    # for one purpose being replayed as another.
    message = SNAPSHOT_DOMAIN + snapshot_bytes

    # Public keys are published as base64-encoded SPKI DER at
    # GET /api/keys. In production you would fetch /api/keys, cache for
    # ~1h, and look the matching key up by signature["keyId"].
    public_key = serialization.load_der_public_key(base64.b64decode(SIGNING_KEY_B64))
    if not isinstance(public_key, Ed25519PublicKey):
        return False

    try:
        public_key.verify(sig, message)
        return True
    except InvalidSignature:
        return False


class WebhookHandler(BaseHTTPRequestHandler):
    def do_POST(self) -> None:  # noqa: N802 - BaseHTTPRequestHandler API
        if self.path != "/webhook":
            self.send_response(404)
            self.end_headers()
            return

        length = int(self.headers.get("Content-Length", "0"))
        raw_body = self.rfile.read(length)
        signature_header = self.headers.get("X-Tidenda-Signature", "")
        timestamp_header = self.headers.get("X-Tidenda-Timestamp", "")

        if not verify_hmac(raw_body, timestamp_header, signature_header):
            self.send_response(401)
            self.end_headers()
            return

        # HMAC is good - safe to parse JSON now.
        try:
            payload = json.loads(raw_body.decode("utf-8"))
        except (json.JSONDecodeError, UnicodeDecodeError):
            self.send_response(400)
            self.end_headers()
            return

        # Optional: verify the snapshot signature if one is attached.
        # webhook.test / document.draft etc. have null snapshots and
        # nothing to verify here.
        #
        # On a mismatch, drop the event and alert, but still answer 2xx:
        # any other status triggers a retry, and the retry carries the
        # same snapshot bytes, which would fail the same way.
        doc = payload.get("document") or {}
        snapshot = doc.get("snapshotBytes")
        snap_sig = doc.get("snapshotSignature")
        if snapshot and snap_sig and not verify_snapshot(snapshot, snap_sig):
            print(f"ALERT: snapshot signature mismatch, event {payload.get('id')} dropped")
            self.send_response(200)
            self.end_headers()
            return

        self.send_response(200)
        self.end_headers()

    def log_message(self, format: str, *args: object) -> None:  # noqa: A002
        # Silence the default access log; replace with structured logging
        # in production. Never log the signature header or the secret.
        return


if __name__ == "__main__":
    port = int(os.environ.get("PORT", "8787"))
    print(f"Webhook receiver listening on http://localhost:{port}/webhook")
    HTTPServer(("127.0.0.1", port), WebhookHandler).serve_forever()