add basic twilio signature checking and tests

tests/gateway/test_sms.py

@@ -1,11 +1,14 @@
 """Tests for SMS (Twilio) platform integration.
 
 Covers config loading, format/truncate, echo prevention,
-requirements check, and toolset verification.
+requirements check, toolset verification, and Twilio signature validation.
 """
 
+import base64
+import hashlib
+import hmac
 import os
-from unittest.mock import patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -213,3 +216,202 @@ class TestSmsToolset:
         from tools.cronjob_tools import CRONJOB_SCHEMA
         deliver_desc = CRONJOB_SCHEMA["parameters"]["properties"]["deliver"]["description"]
         assert "sms" in deliver_desc.lower()
+
+
+# ── Webhook host configuration ─────────────────────────────────────
+
+class TestWebhookHostConfig:
+    """Verify SMS_WEBHOOK_HOST env var and default."""
+
+    def test_default_host_is_all_interfaces(self):
+        from gateway.platforms.sms import DEFAULT_WEBHOOK_HOST
+        assert DEFAULT_WEBHOOK_HOST == "0.0.0.0"
+
+    def test_host_from_env(self):
+        from gateway.platforms.sms import SmsAdapter
+
+        env = {
+            "TWILIO_ACCOUNT_SID": "ACtest",
+            "TWILIO_AUTH_TOKEN": "tok",
+            "TWILIO_PHONE_NUMBER": "+15550001111",
+            "SMS_WEBHOOK_HOST": "127.0.0.1",
+        }
+        with patch.dict(os.environ, env):
+            pc = PlatformConfig(enabled=True, api_key="tok")
+            adapter = SmsAdapter(pc)
+            assert adapter._webhook_host == "127.0.0.1"
+
+    def test_webhook_url_from_env(self):
+        from gateway.platforms.sms import SmsAdapter
+
+        env = {
+            "TWILIO_ACCOUNT_SID": "ACtest",
+            "TWILIO_AUTH_TOKEN": "tok",
+            "TWILIO_PHONE_NUMBER": "+15550001111",
+            "SMS_WEBHOOK_URL": "https://example.com/webhooks/twilio",
+        }
+        with patch.dict(os.environ, env):
+            pc = PlatformConfig(enabled=True, api_key="tok")
+            adapter = SmsAdapter(pc)
+            assert adapter._webhook_url == "https://example.com/webhooks/twilio"
+
+    def test_webhook_url_stripped(self):
+        from gateway.platforms.sms import SmsAdapter
+
+        env = {
+            "TWILIO_ACCOUNT_SID": "ACtest",
+            "TWILIO_AUTH_TOKEN": "tok",
+            "TWILIO_PHONE_NUMBER": "+15550001111",
+            "SMS_WEBHOOK_URL": "  https://example.com/webhooks/twilio  ",
+        }
+        with patch.dict(os.environ, env):
+            pc = PlatformConfig(enabled=True, api_key="tok")
+            adapter = SmsAdapter(pc)
+            assert adapter._webhook_url == "https://example.com/webhooks/twilio"
+
+
+# ── Twilio signature validation ────────────────────────────────────
+
+def _compute_twilio_signature(auth_token, url, params):
+    """Reference implementation of Twilio's signature algorithm."""
+    data_to_sign = url
+    for key in sorted(params.keys()):
+        data_to_sign += key + params[key]
+    mac = hmac.new(
+        auth_token.encode("utf-8"),
+        data_to_sign.encode("utf-8"),
+        hashlib.sha1,
+    )
+    return base64.b64encode(mac.digest()).decode("utf-8")
+
+
+class TestTwilioSignatureValidation:
+    """Unit tests for SmsAdapter._validate_twilio_signature."""
+
+    def _make_adapter(self, auth_token="test_token_secret"):
+        from gateway.platforms.sms import SmsAdapter
+
+        env = {
+            "TWILIO_ACCOUNT_SID": "ACtest",
+            "TWILIO_AUTH_TOKEN": auth_token,
+            "TWILIO_PHONE_NUMBER": "+15550001111",
+        }
+        with patch.dict(os.environ, env):
+            pc = PlatformConfig(enabled=True, api_key=auth_token)
+            adapter = SmsAdapter(pc)
+        return adapter
+
+    def test_valid_signature_accepted(self):
+        adapter = self._make_adapter()
+        url = "https://example.com/webhooks/twilio"
+        params = {"From": "+15551234567", "Body": "hello", "To": "+15550001111"}
+        sig = _compute_twilio_signature("test_token_secret", url, params)
+        assert adapter._validate_twilio_signature(url, params, sig) is True
+
+    def test_invalid_signature_rejected(self):
+        adapter = self._make_adapter()
+        url = "https://example.com/webhooks/twilio"
+        params = {"From": "+15551234567", "Body": "hello"}
+        assert adapter._validate_twilio_signature(url, params, "badsig") is False
+
+    def test_wrong_token_rejected(self):
+        adapter = self._make_adapter(auth_token="correct_token")
+        url = "https://example.com/webhooks/twilio"
+        params = {"From": "+15551234567", "Body": "hello"}
+        sig = _compute_twilio_signature("wrong_token", url, params)
+        assert adapter._validate_twilio_signature(url, params, sig) is False
+
+    def test_params_sorted_by_key(self):
+        """Signature must be computed with params sorted alphabetically."""
+        adapter = self._make_adapter()
+        url = "https://example.com/webhooks/twilio"
+        params = {"Zebra": "last", "Alpha": "first", "Middle": "mid"}
+        sig = _compute_twilio_signature("test_token_secret", url, params)
+        assert adapter._validate_twilio_signature(url, params, sig) is True
+
+    def test_empty_param_values_included(self):
+        """Blank values must be included in signature computation."""
+        adapter = self._make_adapter()
+        url = "https://example.com/webhooks/twilio"
+        params = {"From": "+15551234567", "Body": "", "SmsStatus": "received"}
+        sig = _compute_twilio_signature("test_token_secret", url, params)
+        assert adapter._validate_twilio_signature(url, params, sig) is True
+
+    def test_url_matters(self):
+        """Different URLs produce different signatures."""
+        adapter = self._make_adapter()
+        params = {"Body": "hello"}
+        sig = _compute_twilio_signature(
+            "test_token_secret", "https://a.com/webhooks/twilio", params
+        )
+        assert adapter._validate_twilio_signature(
+            "https://b.com/webhooks/twilio", params, sig
+        ) is False
+
+
+# ── Webhook signature enforcement (handler-level) ──────────────────
+
+class TestWebhookSignatureEnforcement:
+    """Integration tests for signature validation in _handle_webhook."""
+
+    def _make_adapter(self, webhook_url=""):
+        from gateway.platforms.sms import SmsAdapter
+
+        env = {
+            "TWILIO_ACCOUNT_SID": "ACtest",
+            "TWILIO_AUTH_TOKEN": "test_token_secret",
+            "TWILIO_PHONE_NUMBER": "+15550001111",
+            "SMS_WEBHOOK_URL": webhook_url,
+        }
+        with patch.dict(os.environ, env):
+            pc = PlatformConfig(enabled=True, api_key="test_token_secret")
+            adapter = SmsAdapter(pc)
+        adapter._message_handler = AsyncMock()
+        return adapter
+
+    def _mock_request(self, body, headers=None):
+        request = MagicMock()
+        request.read = AsyncMock(return_value=body)
+        request.headers = headers or {}
+        return request
+
+    @pytest.mark.asyncio
+    async def test_no_webhook_url_skips_validation(self):
+        """Without SMS_WEBHOOK_URL, all requests are accepted."""
+        adapter = self._make_adapter(webhook_url="")
+        body = b"From=%2B15551234567&To=%2B15550001111&Body=hello&MessageSid=SM123"
+        request = self._mock_request(body)
+        resp = await adapter._handle_webhook(request)
+        assert resp.status == 200
+
+    @pytest.mark.asyncio
+    async def test_missing_signature_returns_403(self):
+        adapter = self._make_adapter(webhook_url="https://example.com/webhooks/twilio")
+        body = b"From=%2B15551234567&To=%2B15550001111&Body=hello&MessageSid=SM123"
+        request = self._mock_request(body, headers={})
+        resp = await adapter._handle_webhook(request)
+        assert resp.status == 403
+
+    @pytest.mark.asyncio
+    async def test_invalid_signature_returns_403(self):
+        adapter = self._make_adapter(webhook_url="https://example.com/webhooks/twilio")
+        body = b"From=%2B15551234567&To=%2B15550001111&Body=hello&MessageSid=SM123"
+        request = self._mock_request(body, headers={"X-Twilio-Signature": "invalid"})
+        resp = await adapter._handle_webhook(request)
+        assert resp.status == 403
+
+    @pytest.mark.asyncio
+    async def test_valid_signature_returns_200(self):
+        webhook_url = "https://example.com/webhooks/twilio"
+        adapter = self._make_adapter(webhook_url=webhook_url)
+        params = {
+            "From": "+15551234567",
+            "To": "+15550001111",
+            "Body": "hello",
+            "MessageSid": "SM123",
+        }
+        sig = _compute_twilio_signature("test_token_secret", webhook_url, params)
+        body = b"From=%2B15551234567&To=%2B15550001111&Body=hello&MessageSid=SM123"
+        request = self._mock_request(body, headers={"X-Twilio-Signature": sig})
+        resp = await adapter._handle_webhook(request)
+        assert resp.status == 200