From 70933f9ed0315c1a499d1a531e0e347a1f17ab45 Mon Sep 17 00:00:00 2001 From: isra el Date: Thu, 5 Nov 2026 15:48:35 +0300 Subject: [PATCH 1/2] feat: verify the timestamped webhook signature Add verify_timestamped_webhook_signature for the X-Textbee-Signature header (t=,v1=) and verify_webhook_request, which reads it from the request headers and rejects timestamps more than 300 seconds from now. X-Signature is accepted only with allow_legacy=True. verify_webhook_signature is unchanged and marked deprecated. Release 0.3.0. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_017B6pTVUGhFgCC3D8B6wr9P --- README.md | 35 +++++-- src/textbee/__init__.py | 8 +- src/textbee/_version.py | 2 +- src/textbee/_webhooks.py | 141 ++++++++++++++++++++++++++- tests/test_webhooks.py | 206 ++++++++++++++++++++++++++++++++++++++- 5 files changed, 379 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 106ed2d..831dcc1 100644 --- a/README.md +++ b/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 `.`. `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 diff --git a/src/textbee/__init__.py b/src/textbee/__init__.py index f6d5d65..da8a535 100644 --- a/src/textbee/__init__.py +++ b/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", ] diff --git a/src/textbee/_version.py b/src/textbee/_version.py index d3ec452..493f741 100644 --- a/src/textbee/_version.py +++ b/src/textbee/_version.py @@ -1 +1 @@ -__version__ = "0.2.0" +__version__ = "0.3.0" diff --git a/src/textbee/_webhooks.py b/src/textbee/_webhooks.py index ed53c4d..54c58a1 100644 --- a/src/textbee/_webhooks.py +++ b/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,127 @@ 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=,v1=``, where the hex value is + HMAC-SHA256 over ``.``. 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: + 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: + 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 _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) diff --git a/tests/test_webhooks.py b/tests/test_webhooks.py index dc6daba..a8cbde5 100644 --- a/tests/test_webhooks.py +++ b/tests/test_webhooks.py @@ -5,7 +5,13 @@ import json from typing import Any -from textbee import verify_webhook_signature +import pytest + +from textbee import ( + verify_timestamped_webhook_signature, + verify_webhook_request, + verify_webhook_signature, +) SIGNING_SECRET = "whsec_test_secret" @@ -140,3 +146,201 @@ def test_accepts_a_bytearray_body() -> None: signature=NODE_SIGNATURE, signing_secret=SIGNING_SECRET, ) + + +NOW = 1_762_300_000 +RAW_BODY = to_js_json({**payload, "idempotencyKey": "evt_1"}) + + +def sign_timestamped(body: str, t: int | str = NOW, secret: str = SIGNING_SECRET) -> str: + """Exactly how the api builds X-Textbee-Signature.""" + v1 = hmac.new(secret.encode(), f"{t}.{body}".encode(), hashlib.sha256).hexdigest() + return f"t={t},v1={v1}" + + +def valid_v1() -> str: + return sign_timestamped(RAW_BODY).split("v1=")[1] + + +def verify(header: str | None, *, body: str | bytes = RAW_BODY, now: float = NOW) -> bool: + return verify_timestamped_webhook_signature( + payload=body, signature_header=header, signing_secret=SIGNING_SECRET, now=now + ) + + +def test_timestamped_accepts_a_header_produced_the_way_the_api_produces_it() -> None: + assert verify(sign_timestamped(RAW_BODY)) + + +def test_timestamped_accepts_raw_bytes() -> None: + assert verify(sign_timestamped(RAW_BODY), body=RAW_BODY.encode()) + + +def test_timestamped_accepts_a_non_ascii_body() -> None: + body = to_js_json({**payload, "message": "café 👍"}) + assert verify(sign_timestamped(body), body=body) + assert verify(sign_timestamped(body), body=body.encode("utf-8")) + + +def test_timestamped_rejects_the_wrong_signing_secret() -> None: + assert not verify(sign_timestamped(RAW_BODY, secret="whsec_other_secret")) + + +def test_timestamped_rejects_a_tampered_body() -> None: + assert not verify(sign_timestamped(RAW_BODY), body=RAW_BODY.replace("hello", "tampered")) + + +def test_timestamped_rejects_a_stale_timestamp() -> None: + header = sign_timestamped(RAW_BODY) + assert verify(header, now=NOW + 300) + assert not verify(header, now=NOW + 301) + + +def test_timestamped_rejects_a_future_timestamp() -> None: + header = sign_timestamped(RAW_BODY) + assert verify(header, now=NOW - 300) + assert not verify(header, now=NOW - 301) + + +def test_timestamped_honors_a_custom_tolerance() -> None: + assert not verify_timestamped_webhook_signature( + payload=RAW_BODY, + signature_header=sign_timestamped(RAW_BODY), + signing_secret=SIGNING_SECRET, + tolerance_seconds=10, + now=NOW + 11, + ) + + +def test_timestamped_rejects_a_nan_clock() -> None: + assert not verify(sign_timestamped(RAW_BODY), now=float("nan")) + + +def test_timestamped_rejects_a_signature_moved_to_another_t() -> None: + assert not verify(f"t={NOW + 1},v1={valid_v1()}") + + +@pytest.mark.parametrize( + "header", + [ + None, + "", + "garbage", + "v1=" + "a" * 64, + f"t={NOW}", + f"t={NOW},v1=", + f"t=-{NOW},v1=" + "a" * 64, + f"t={NOW}.5,v1=" + "a" * 64, + "t=abc,v1=" + "a" * 64, + "t=,v1=" + "a" * 64, + "t=\u0661\u0662,v1=" + "a" * 64, + "t=" + "9" * 5000 + ",v1=" + "a" * 64, + ], +) +def test_timestamped_rejects_malformed_headers(header: str | None) -> None: + assert not verify(header) + + +def test_timestamped_rejects_a_valid_v1_with_a_malformed_t() -> None: + assert not verify(f"t={NOW}x,v1={valid_v1()}") + + +def test_timestamped_accepts_any_matching_v1() -> None: + other = "a" * 64 + assert verify(f"t={NOW},v1={other},v1={valid_v1()}") + assert verify(f"t={NOW}, v1={valid_v1()}, v1={other}") + assert verify(f"t={NOW},v0={other},v1={valid_v1()}") + assert not verify(f"t={NOW},v1={other},v1={other}") + + +def test_timestamped_rejects_a_duplicate_t() -> None: + assert not verify(f"t={NOW},t={NOW},v1={valid_v1()}") + assert not verify(f"t={NOW},v1={valid_v1()},t={NOW - 1000}") + + +def test_timestamped_rejects_a_non_ascii_v1_without_raising() -> None: + assert not verify(f"t={NOW},v1=" + "é" * 64) + + +def test_timestamped_accepts_an_uppercase_hex_v1() -> None: + assert verify(f"t={NOW},v1={valid_v1().upper()}") + + +LEGACY_SIGNATURE = hmac.new(SIGNING_SECRET.encode(), RAW_BODY.encode(), hashlib.sha256).hexdigest() + + +def test_request_verifies_the_signature_header() -> None: + assert verify_webhook_request( + payload=RAW_BODY.encode(), + headers={"X-Textbee-Signature": sign_timestamped(RAW_BODY)}, + signing_secret=SIGNING_SECRET, + now=NOW, + ) + + +def test_request_matches_header_names_in_any_case() -> None: + for name in ("x-textbee-signature", "X-TEXTBEE-SIGNATURE", "X-Textbee-Signature"): + assert verify_webhook_request( + payload=RAW_BODY, + headers={name: sign_timestamped(RAW_BODY)}, + signing_secret=SIGNING_SECRET, + now=NOW, + ) + + +def test_request_rejects_a_stale_signature_header() -> None: + assert not verify_webhook_request( + payload=RAW_BODY, + headers={"X-Textbee-Signature": sign_timestamped(RAW_BODY)}, + signing_secret=SIGNING_SECRET, + now=NOW + 301, + ) + + +def test_request_ignores_x_signature_by_default() -> None: + assert not verify_webhook_request( + payload=RAW_BODY, + headers={"X-Signature": LEGACY_SIGNATURE}, + signing_secret=SIGNING_SECRET, + now=NOW, + ) + + +def test_request_falls_back_to_x_signature_only_with_allow_legacy() -> None: + assert verify_webhook_request( + payload=RAW_BODY.encode(), + headers={"x-signature": LEGACY_SIGNATURE}, + signing_secret=SIGNING_SECRET, + allow_legacy=True, + ) + assert not verify_webhook_request( + payload=RAW_BODY, + headers={"x-signature": sign(payload, "whsec_other_secret")}, + signing_secret=SIGNING_SECRET, + allow_legacy=True, + ) + assert not verify_webhook_request( + payload=RAW_BODY, + headers={"x-signature": "é" * 64}, + signing_secret=SIGNING_SECRET, + allow_legacy=True, + ) + + +def test_request_does_not_fall_back_when_the_signature_header_is_present() -> None: + assert not verify_webhook_request( + payload=RAW_BODY, + headers={ + "X-Textbee-Signature": sign_timestamped(RAW_BODY), + "X-Signature": LEGACY_SIGNATURE, + }, + signing_secret=SIGNING_SECRET, + now=NOW + 301, + allow_legacy=True, + ) + + +def test_request_rejects_a_request_with_no_signature_headers() -> None: + assert not verify_webhook_request( + payload=RAW_BODY, headers={}, signing_secret=SIGNING_SECRET, allow_legacy=True + ) From 6cdd76f5ab280185fb3d31919176dcc0cfa15b19 Mon Sep 17 00:00:00 2001 From: isra el Date: Thu, 5 Nov 2026 16:10:42 +0300 Subject: [PATCH 2/2] fix: return False for an unread body and tighten the legacy secret test A framework that did not read the body can pass None or a dict, which now returns False instead of raising. The wrong-secret X-Signature test signs the same body, so only the secret differs. Adds tests for a tolerance above 300 seconds, an empty signature header with allow_legacy on, and two case variants of the signature header. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_017B6pTVUGhFgCC3D8B6wr9P --- src/textbee/_webhooks.py | 9 +++++-- tests/test_webhooks.py | 54 +++++++++++++++++++++++++++++++++++++++- 2 files changed, 60 insertions(+), 3 deletions(-) diff --git a/src/textbee/_webhooks.py b/src/textbee/_webhooks.py index 54c58a1..e06bd69 100644 --- a/src/textbee/_webhooks.py +++ b/src/textbee/_webhooks.py @@ -67,7 +67,7 @@ def verify_timestamped_webhook_signature( 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: + if not signature_header or not signing_secret or not _is_raw_body(payload): return False parsed = _parse_signature_header(signature_header) @@ -119,7 +119,7 @@ def verify_webhook_request( return False legacy_signature = _get_header(headers, "x-signature") - if not legacy_signature or not signing_secret: + if not legacy_signature or not signing_secret or not _is_raw_body(payload): return False expected = _hmac_hex(signing_secret, _raw_bytes(payload)) @@ -165,6 +165,11 @@ 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") diff --git a/tests/test_webhooks.py b/tests/test_webhooks.py index a8cbde5..212055f 100644 --- a/tests/test_webhooks.py +++ b/tests/test_webhooks.py @@ -315,7 +315,11 @@ def test_request_falls_back_to_x_signature_only_with_allow_legacy() -> None: ) assert not verify_webhook_request( payload=RAW_BODY, - headers={"x-signature": sign(payload, "whsec_other_secret")}, + headers={ + "x-signature": hmac.new( + b"whsec_other_secret", RAW_BODY.encode(), hashlib.sha256 + ).hexdigest() + }, signing_secret=SIGNING_SECRET, allow_legacy=True, ) @@ -340,6 +344,54 @@ def test_request_does_not_fall_back_when_the_signature_header_is_present() -> No ) +def test_request_does_not_fall_back_when_the_signature_header_is_empty() -> None: + assert not verify_webhook_request( + payload=RAW_BODY, + headers={"X-Textbee-Signature": "", "X-Signature": LEGACY_SIGNATURE}, + signing_secret=SIGNING_SECRET, + now=NOW, + allow_legacy=True, + ) + + +def test_request_reads_every_case_variant_of_the_signature_header() -> None: + assert not verify_webhook_request( + payload=RAW_BODY, + headers={ + "X-Textbee-Signature": sign_timestamped(RAW_BODY), + "x-textbee-signature": f"t={NOW - 1000},v1={valid_v1()}", + }, + signing_secret=SIGNING_SECRET, + now=NOW, + ) + + +def test_timestamped_accepts_a_custom_tolerance_above_300_seconds() -> None: + assert verify_timestamped_webhook_signature( + payload=RAW_BODY, + signature_header=sign_timestamped(RAW_BODY), + signing_secret=SIGNING_SECRET, + tolerance_seconds=600, + now=NOW + 301, + ) + + +@pytest.mark.parametrize("body", [None, {}]) +def test_request_returns_false_when_the_body_was_not_read(body: Any) -> None: + assert not verify_webhook_request( + payload=body, + headers={"X-Textbee-Signature": sign_timestamped(RAW_BODY)}, + signing_secret=SIGNING_SECRET, + now=NOW, + ) + assert not verify_webhook_request( + payload=body, + headers={"X-Signature": LEGACY_SIGNATURE}, + signing_secret=SIGNING_SECRET, + allow_legacy=True, + ) + + def test_request_rejects_a_request_with_no_signature_headers() -> None: assert not verify_webhook_request( payload=RAW_BODY, headers={}, signing_secret=SIGNING_SECRET, allow_legacy=True