From 8ec668b3cc8b9289d63f02fa6abca3d67e993eb3 Mon Sep 17 00:00:00 2001 From: Alex Honecker Date: Sun, 4 Oct 2026 11:35:03 +0300 Subject: [PATCH] fix(security): treat whitespace-only HMAC secrets as unset A blank or whitespace-only secret now fails closed at webhook startup, on the request path, and when a dynamic route is admitted. (cherry picked from commit 05e918d52c41e905d88e6c0a0f55e7612f4d899d) --- gateway/platforms/webhook.py | 7 ++++--- gateway/relay/auth.py | 10 ++++++++++ tests/gateway/relay/test_auth.py | 14 +++++++++++++- tests/gateway/test_webhook_adapter.py | 12 ++++++++++++ 4 files changed, 39 insertions(+), 4 deletions(-) diff --git a/gateway/platforms/webhook.py b/gateway/platforms/webhook.py index a3c72d3c03..aaa6465fbd 100644 --- a/gateway/platforms/webhook.py +++ b/gateway/platforms/webhook.py @@ -35,6 +35,7 @@ except ImportError: web = None # type: ignore[assignment] from gateway.config import Platform, PlatformConfig +from gateway.relay.auth import is_usable_secret from gateway.platforms.base import BasePlatformAdapter, SendResult from gateway.platforms.event import MessageEvent, MessageType from gateway.platforms.tcp_site import start_tcp_site @@ -225,7 +226,7 @@ class WebhookAdapter(BasePlatformAdapter): def _validate_route(self, name: str, route: dict) -> None: """Startup validation: secret required; INSECURE_NO_AUTH only on loopback (crash early on a public footgun).""" secret = route.get("secret", self._global_secret) - if not secret: + if not is_usable_secret(secret): raise ValueError(f"[webhook] Route '{name}' has no HMAC secret. Set 'secret' on the route or globally. " f"For testing without auth, set secret to '{_INSECURE_NO_AUTH}'.") if secret == _INSECURE_NO_AUTH and not _is_loopback_host(self._host): @@ -378,7 +379,7 @@ class WebhookAdapter(BasePlatformAdapter): """An empty effective secret would make _handle_webhook skip HMAC validation → reject such dynamic routes; INSECURE_NO_AUTH is loopback-only.""" effective_secret = route.get("secret", self._global_secret) - if not effective_secret: + if not is_usable_secret(effective_secret): logger.warning("[webhook] Dynamic route '%s' skipped: 'secret' is missing or empty. Set a valid HMAC " "secret, or use '%s' to explicitly disable auth (testing only).", name, _INSECURE_NO_AUTH) return False @@ -503,7 +504,7 @@ class WebhookAdapter(BasePlatformAdapter): # Missing/empty secrets fail closed here too (not only in connect()), so direct handler reuse # cannot become an unauthenticated dispatch surface. secret = route_config.get("secret", self._global_secret) - if not secret: + if not is_usable_secret(secret): logger.error("[webhook] Route %s has no HMAC secret; refusing request", route_name) return None, _json_error("Webhook route is missing an HMAC secret", 403) if secret != _INSECURE_NO_AUTH and not self._validate_signature(request, raw_body, secret): diff --git a/gateway/relay/auth.py b/gateway/relay/auth.py index ca23ab1c5d..cfa6d2ccea 100644 --- a/gateway/relay/auth.py +++ b/gateway/relay/auth.py @@ -22,6 +22,16 @@ def sign(payload: str, secret: str) -> str: return hmac.new(secret.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256).hexdigest() +def is_usable_secret(secret: object) -> bool: + """True when ``secret`` is a string with at least one non-space character. + + A whitespace-only value is what an unset key looks like in config. It is + falsy to a person and truthy to ``if not secret``, so a bare falsy check + lets it through. Non-strings are rejected for the same reason. + """ + return isinstance(secret, str) and bool(secret.strip()) + + def make_token(payload: str, secret: str, ttl_seconds: int = 0) -> str: """``base64url(f"{payload}:{exp}:{sig}")``; ``exp`` unix seconds (0 = never). diff --git a/tests/gateway/relay/test_auth.py b/tests/gateway/relay/test_auth.py index 04ee7f1386..5ecd9c2a71 100644 --- a/tests/gateway/relay/test_auth.py +++ b/tests/gateway/relay/test_auth.py @@ -13,7 +13,9 @@ drift). Regenerate with: from __future__ import annotations -from gateway.relay.auth import make_token, make_upgrade_token +import pytest + +from gateway.relay.auth import is_usable_secret, make_token, make_upgrade_token # A fixed 256-bit hex secret used for the frozen connector vector below. _SECRET = "00112233445566778899aabbccddeeff00112233445566778899aabbccddeeff" @@ -28,3 +30,13 @@ def test_upgrade_token_is_make_token_of_gateway_id(): def test_python_make_token_matches_connector_byte_for_byte(): assert make_token("gw-instance-1", _SECRET, 0) == _CONN_TOKEN + + +@pytest.mark.parametrize( + "value,usable", + [(_SECRET, True), ("x", True), (" x ", True), + ("", False), (" ", False), ("\t\n", False), + (None, False), (b"bytes", False), (123, False)], +) +def test_is_usable_secret_classifies_configured_secrets(value, usable): + assert is_usable_secret(value) is usable diff --git a/tests/gateway/test_webhook_adapter.py b/tests/gateway/test_webhook_adapter.py index 443f2d0cd8..ea087dd780 100644 --- a/tests/gateway/test_webhook_adapter.py +++ b/tests/gateway/test_webhook_adapter.py @@ -1243,3 +1243,15 @@ def test_route_profile_validation_fails_closed(): assert WebhookAdapter._route_allows_profile( {"profile": malformed}, "worker" ) is False + + +class TestBlankRouteSecretFailsClosed: + @pytest.mark.parametrize("blank", [" ", "\t", "\n"]) + def test_connect_rejects_a_whitespace_only_route_secret(self, blank): + adapter = _make_adapter(routes={"hook": {"secret": blank}}) + with pytest.raises(ValueError, match="HMAC secret"): + asyncio.run(adapter.connect()) + + def test_connect_still_accepts_a_real_secret(self): + adapter = _make_adapter(routes={"hook": {"secret": "s3cr3t"}}) + assert asyncio.run(adapter.connect()) is True