diff --git a/src/openjarvis/server/webhook_routes.py b/src/openjarvis/server/webhook_routes.py index 0c3db1da..ecd9983d 100644 --- a/src/openjarvis/server/webhook_routes.py +++ b/src/openjarvis/server/webhook_routes.py @@ -34,15 +34,25 @@ def _validate_twilio_signature( params: dict, signature: str, ) -> bool: - """Validate Twilio webhook signature using the SDK.""" + """Validate Twilio webhook signature using the SDK. + + Fails closed: returns ``False`` if the SDK is not installed or + if no auth_token is configured. + """ + if not auth_token: + logger.error("Twilio auth token not configured — rejecting webhook") + return False try: from twilio.request_validator import RequestValidator validator = RequestValidator(auth_token) return validator.validate(url, params, signature) except ImportError: - logger.warning("twilio SDK not installed — skipping signature validation") - return True + logger.error( + "twilio SDK not installed — rejecting webhook. " + "Install it: pip install twilio" + ) + return False def _format_for_sms(text: str) -> str: @@ -236,6 +246,11 @@ def create_webhook_router( header_secret = request.headers.get("x-sendblue-secret", "") if header_secret != sb.webhook_secret: return Response("Invalid secret", status_code=403) + elif sb: + logger.warning( + "SendBlue webhook received without secret verification. " + "Set webhook_secret for HMAC validation." + ) # Ignore outbound status callbacks if payload.get("is_outbound", False): diff --git a/tests/security/test_webhook_validation.py b/tests/security/test_webhook_validation.py new file mode 100644 index 00000000..25b35fec --- /dev/null +++ b/tests/security/test_webhook_validation.py @@ -0,0 +1,32 @@ +"""Tests for webhook fail-closed validation (Section 3).""" + +from __future__ import annotations + +from unittest.mock import patch + + +class TestTwilioValidationFailClosed: + """Twilio validation must reject when SDK is unavailable.""" + + def test_missing_sdk_returns_false(self) -> None: + from openjarvis.server.webhook_routes import _validate_twilio_signature + + with patch.dict("sys.modules", {"twilio": None, "twilio.request_validator": None}): + result = _validate_twilio_signature( + auth_token="test_token", + url="https://example.com/webhooks/twilio", + params={"Body": "hello"}, + signature="invalid", + ) + assert result is False + + def test_empty_auth_token_returns_false(self) -> None: + from openjarvis.server.webhook_routes import _validate_twilio_signature + + result = _validate_twilio_signature( + auth_token="", + url="https://example.com/webhooks/twilio", + params={}, + signature="", + ) + assert result is False