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)
This commit is contained in:
@@ -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):
|
||||
|
||||
@@ -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).
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user