Verify the timestamped webhook signature and release 0.3.0

README.md

@@ -142,33 +142,54 @@ Responses are plain dicts typed as `TypedDict`s, so keys keep the API's names (`
 
 ## Verifying webhooks
 
-textbee signs each webhook delivery with HMAC-SHA256 and sends the hex digest in the `X-Signature` header. Pass the raw request body, not a re-serialized dict, whenever your framework gives you access to it.
+textbee signs each webhook delivery with your signing secret and sends a timestamped signature in the `X-Textbee-Signature` header:
+
+```
+X-Textbee-Signature: t=1762300000,v1=5f2b...
+```
+
+`v1` is the hex HMAC-SHA256 of `<t>.<raw body>`. `verify_webhook_request` reads the signature header, checks `v1` against the raw body in constant time, and rejects the request when `t` is more than 300 seconds from now. The signed timestamp makes a captured delivery useless after that window.
+
+Verify the raw body exactly as received. A parsed and re-serialized body can differ by one byte and fail.
 
 ```python
 import json
 import os
 
 from flask import Flask, request
-from textbee import verify_webhook_signature
+from textbee import verify_webhook_request
 
 app = Flask(__name__)
 
 
 @app.post("/webhooks/textbee")
 def textbee_webhook():
-    valid = verify_webhook_signature(
-        payload=request.get_data(),
-        signature=request.headers.get("X-Signature"),
+    raw_body = request.get_data()
+    valid = verify_webhook_request(
+        payload=raw_body,
+        headers=request.headers,
         signing_secret=os.environ["TEXTBEE_WEBHOOK_SECRET"],
     )
     if not valid:
         return "", 401
 
-    event = json.loads(request.get_data())
+    event = json.loads(raw_body)
+    # Retries carry the same idempotencyKey. Skip keys you have already handled.
     return "", 200
 ```
 
-`payload` takes `bytes`, `str`, or a dict. A dict is re-serialized the way the API serializes it, which only matches while key order and number formatting survive the round trip, so prefer the raw body.
+`payload` takes `bytes` or `str`. `headers` takes any mapping or headers object with `.items()`, such as Flask's `request.headers`, Django's `request.headers` or a plain dict, and header names match in any case.
+
+Options:
+
+- `tolerance_seconds`: the largest allowed distance between `t` and now, in either direction. Defaults to `300`. Keep your server clock in sync (NTP).
+- `allow_legacy`: an explicit opt-in for old receivers. Off by default. When it is `True` and the signature header is absent, the request is checked against `X-Signature` instead, which has no replay protection.
+
+To check the signature header value yourself, call `verify_timestamped_webhook_signature(payload=..., signature_header=..., signing_secret=...)`. It takes the same `tolerance_seconds` option.
+
+Retries and duplicates: a failed delivery is retried with a fresh `t`, so each attempt passes the freshness check. Every payload carries an `idempotencyKey`. Store the keys you have processed and skip repeats.
+
+**Deprecated:** `verify_webhook_signature` checks only `X-Signature`. It still works because textbee still sends that header, but it has no replay protection. See [textbee.dev/docs/webhooks](https://textbee.dev/docs/webhooks) for the upgrade steps.
 
 ## SMS utilities
 

src/textbee/__init__.py

@@ -34,7 +34,11 @@
     WebhookEvent,
 )
 from ._version import __version__
-from ._webhooks import verify_webhook_signature
+from ._webhooks import (
+    verify_timestamped_webhook_signature,
+    verify_webhook_request,
+    verify_webhook_signature,
+)
 
 __all__ = [
     "Device",
@@ -67,5 +71,7 @@
     "is_valid_e164",
     "normalize_phone_number",
     "sanitize_for_gsm7",
+    "verify_timestamped_webhook_signature",
+    "verify_webhook_request",
     "verify_webhook_signature",
 ]

src/textbee/_version.py

@@ -1 +1 @@
-__version__ = "0.2.0"
+__version__ = "0.3.0"

src/textbee/_webhooks.py

@@ -4,8 +4,16 @@
 import hmac
 import json
 import re
-from collections.abc import Mapping
-from typing import Any
+import time
+from collections.abc import Iterable, Mapping
+from typing import Any, Protocol
+
+DEFAULT_TOLERANCE_SECONDS = 300
+"""Default maximum distance, in seconds, between a signed timestamp and now."""
+
+
+class _Headers(Protocol):
+    def items(self) -> Iterable[tuple[str, str]]: ...
 
 
 def verify_webhook_signature(
@@ -14,11 +22,17 @@ def verify_webhook_signature(
     signature: str | None,
     signing_secret: str | None,
 ) -> bool:
-    """Check that a webhook really came from textbee.
+    """Check that a webhook really came from textbee, using ``X-Signature``.
 
     textbee signs deliveries with HMAC-SHA256 over the request body and sends
     the hex digest in the ``X-Signature`` header. Comparison is constant time.
 
+    .. deprecated:: 0.3.0
+        ``X-Signature`` has no timestamp, so a captured delivery can be
+        replayed. textbee still sends it, so this keeps working. Use
+        :func:`verify_webhook_request` or
+        :func:`verify_timestamped_webhook_signature`.
+
     ``payload`` is the request body. Prefer the raw body (``bytes`` or ``str``).
     A parsed dict works too, re-serialized the way the API serializes it, but
     only while key order and number formatting survive the round trip.
@@ -36,6 +50,132 @@ def verify_webhook_signature(
     return hmac.compare_digest(expected.encode("ascii"), signature.lower().encode("utf-8"))
 
 
+def verify_timestamped_webhook_signature(
+    *,
+    payload: str | bytes | bytearray,
+    signature_header: str | None,
+    signing_secret: str | None,
+    tolerance_seconds: float = DEFAULT_TOLERANCE_SECONDS,
+    now: float | None = None,
+) -> bool:
+    """Check the timestamped signature in the ``X-Textbee-Signature`` header.
+
+    The header is ``t=<unix seconds>,v1=<hex>``, where the hex value is
+    HMAC-SHA256 over ``<t>.<raw body>``. Any matching ``v1`` entry passes.
+    Returns False when the header is malformed, no ``v1`` entry matches, or
+    ``t`` is more than ``tolerance_seconds`` away from ``now`` in either
+    direction. ``payload`` is the raw request body, exactly as received.
+    ``now`` defaults to the system clock, in unix seconds.
+    """
+    if not signature_header or not signing_secret or not _is_raw_body(payload):
+        return False
+
+    parsed = _parse_signature_header(signature_header)
+    if parsed is None:
+        return False
+    timestamp, candidates = parsed
+
+    current = time.time() if now is None else now
+    # Written as a negated <= so a NaN tolerance or clock rejects instead of passing.
+    if not abs(current - int(timestamp)) <= tolerance_seconds:
+        return False
+
+    expected = _hmac_hex(signing_secret, timestamp.encode("ascii") + b"." + _raw_bytes(payload))
+
+    matched = False
+    for candidate in candidates:
+        if hmac.compare_digest(expected, candidate.lower().encode("utf-8")):
+            matched = True
+    return matched
+
+
+def verify_webhook_request(
+    *,
+    payload: str | bytes | bytearray,
+    headers: Mapping[str, str] | _Headers,
+    signing_secret: str | None,
+    tolerance_seconds: float = DEFAULT_TOLERANCE_SECONDS,
+    now: float | None = None,
+    allow_legacy: bool = False,
+) -> bool:
+    """Check that a webhook request came from textbee, from its headers.
+
+    Verifies the timestamped signature in ``X-Textbee-Signature``. Header
+    names match in any case. When that header is absent and ``allow_legacy``
+    is True, falls back to ``X-Signature``, which has no replay protection.
+    ``payload`` is the raw request body, exactly as received.
+    """
+    signature_header = _get_header(headers, "x-textbee-signature")
+    if signature_header is not None:
+        return verify_timestamped_webhook_signature(
+            payload=payload,
+            signature_header=signature_header,
+            signing_secret=signing_secret,
+            tolerance_seconds=tolerance_seconds,
+            now=now,
+        )
+
+    if not allow_legacy:
+        return False
+
+    legacy_signature = _get_header(headers, "x-signature")
+    if not legacy_signature or not signing_secret or not _is_raw_body(payload):
+        return False
+
+    expected = _hmac_hex(signing_secret, _raw_bytes(payload))
+    return hmac.compare_digest(expected, legacy_signature.lower().encode("utf-8"))
+
+
+# 1 to 15 digits: ample for unix seconds, and short enough that int() never
+# hits Python's limit on converting long digit strings.
+_TIMESTAMP = re.compile("[0-9]{1,15}")
+
+
+def _parse_signature_header(header: str) -> tuple[str, list[str]] | None:
+    timestamp: str | None = None
+    candidates: list[str] = []
+
+    for part in header.split(","):
+        key, separator, value = part.partition("=")
+        if not separator:
+            continue
+        key = key.strip()
+        value = value.strip()
+
+        if key == "t":
+            if timestamp is not None or not _TIMESTAMP.fullmatch(value):
+                return None
+            timestamp = value
+        elif key == "v1" and value:
+            candidates.append(value)
+
+    if timestamp is None or not candidates:
+        return None
+    return timestamp, candidates
+
+
+def _get_header(headers: Mapping[str, str] | _Headers, name: str) -> str | None:
+    values = [value for key, value in headers.items() if key.lower() == name]
+    if not values:
+        return None
+    return ",".join(values)
+
+
+def _hmac_hex(secret: str, message: bytes) -> bytes:
+    return hmac.new(secret.encode("utf-8"), message, hashlib.sha256).hexdigest().encode("ascii")
+
+
+def _is_raw_body(payload: object) -> bool:
+    # A framework that did not read the body can hand over None or a dict.
+    return isinstance(payload, (str, bytes, bytearray))
+
+
+def _raw_bytes(payload: str | bytes | bytearray) -> bytes:
+    if isinstance(payload, str):
+        return payload.encode("utf-8")
+    return bytes(payload)
+
+
 def _body_bytes(payload: str | bytes | bytearray | Mapping[str, Any]) -> bytes:
     if isinstance(payload, (bytes, bytearray)):
         return bytes(payload)