fix: bind webhook HMAC signature to timestamp to prevent replay

Signed-off-by: Lang-Akshay <akshay.shinde26@ibm.com>

Author: Lang-Akshay

plugins/webhook_notification/webhook_notification.py

@@ -18,6 +18,7 @@
 import hashlib
 import hmac
 import logging
+import time
 from typing import Any, Dict, List, Optional
 
 # Third-Party
@@ -82,6 +83,7 @@ class AuthenticationConfig(BaseModel):
     hmac_secret: Optional[str] = None
     hmac_algorithm: str = "sha256"
     hmac_header: str = "X-Signature"
+    hmac_timestamp_header: str = "X-Timestamp"
 
 
 class WebhookConfig(BaseModel):
@@ -165,19 +167,24 @@ async def _render_template(self, template: str, context: Dict[str, Any]) -> str:
                 result = result.replace(placeholder, str(value))
         return result
 
-    def _create_hmac_signature(self, payload: str, secret: str, algorithm: str) -> str:
-        """Create HMAC signature for the payload.
+    def _create_hmac_signature(self, payload: str, secret: str, algorithm: str, timestamp: str) -> str:
+        """Create HMAC signature for the payload, bound to a timestamp.
+
+        The signed material is ``{timestamp}.{payload}`` so a captured
+        signature cannot be replayed at a different time.
 
         Args:
             payload: The payload to sign.
             secret: The secret key for HMAC.
             algorithm: The hash algorithm to use.
+            timestamp: Unix epoch seconds as a string, included in the signed material.
 
         Returns:
-            str: The HMAC signature string.
+            str: The HMAC signature string in the form ``{algorithm}={hex}``.
         """
         hash_func = getattr(hashlib, algorithm, hashlib.sha256)
-        signature = hmac.new(secret.encode("utf-8"), payload.encode("utf-8"), hash_func).hexdigest()
+        signed_material = f"{timestamp}.{payload}".encode("utf-8")
+        signature = hmac.new(secret.encode("utf-8"), signed_material, hash_func).hexdigest()
         return f"{algorithm}={signature}"
 
     async def _send_webhook(
@@ -241,8 +248,10 @@ async def _send_webhook(
         elif auth_config.type == AuthenticationType.API_KEY and auth_config.api_key:
             headers[auth_config.api_key_header] = auth_config.api_key
         elif auth_config.type == AuthenticationType.HMAC and auth_config.hmac_secret:
-            signature = self._create_hmac_signature(payload_json, auth_config.hmac_secret, auth_config.hmac_algorithm)
+            timestamp = str(int(time.time()))
+            signature = self._create_hmac_signature(payload_json, auth_config.hmac_secret, auth_config.hmac_algorithm, timestamp)
             headers[auth_config.hmac_header] = signature
+            headers[auth_config.hmac_timestamp_header] = timestamp
 
         # Attempt delivery with retry logic
         for attempt in range(webhook.retry_attempts + 1):

tests/unit/mcpgateway/plugins/plugins/webhook_notification/test_webhook_notification.py

@@ -7,6 +7,7 @@
 """
 
 import json
+import time
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -92,18 +93,72 @@ async def test_template_rendering(self):
 
     @pytest.mark.asyncio
     async def test_hmac_signature_creation(self):
-        """Test HMAC signature creation."""
+        """Test HMAC signature creation includes timestamp in signed material."""
         plugin = _create_plugin()
 
         payload = '{"test": "data"}'
         secret = "test-secret"  # pragma: allowlist secret
         algorithm = "sha256"
+        timestamp = "1700000000"
 
-        signature = plugin._create_hmac_signature(payload, secret, algorithm)
+        signature = plugin._create_hmac_signature(payload, secret, algorithm, timestamp)
 
         assert signature.startswith("sha256=")
         assert len(signature) > 10  # Basic length check
 
+        # A different timestamp must produce a different signature (timestamp is in the signed material).
+        signature_other_ts = plugin._create_hmac_signature(payload, secret, algorithm, "1700000001")
+        assert signature != signature_other_ts
+
+    @pytest.mark.asyncio
+    @patch("plugins.webhook_notification.webhook_notification.httpx.AsyncClient")
+    async def test_webhook_delivery_with_hmac_auth_sends_timestamp_header(self, mock_client_class):
+        """Test HMAC auth sends both signature and timestamp headers."""
+        mock_client = AsyncMock()
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_client.post.return_value = mock_response
+        mock_client_class.return_value = mock_client
+
+        config = {
+            "webhooks": [
+                {
+                    "url": "https://hooks.example.com/webhook",
+                    "events": ["tool_success"],
+                    "authentication": {
+                        "type": "hmac",
+                        "hmac_secret": "test-secret",  # pragma: allowlist secret
+                        "hmac_algorithm": "sha256",
+                        "hmac_header": "X-Signature",
+                        "hmac_timestamp_header": "X-Timestamp",
+                    },
+                    "retry_attempts": 0,
+                    "enabled": True,
+                }
+            ]
+        }
+
+        plugin = _create_plugin(config)
+        plugin._client = mock_client
+
+        before = int(time.time())
+        context = _create_context()
+        webhook = plugin._cfg.webhooks[0]
+
+        await plugin._send_webhook(webhook=webhook, event=EventType.TOOL_SUCCESS, context=context)
+
+        after = int(time.time())
+
+        call_args = mock_client.post.call_args
+        headers = call_args[1]["headers"]
+
+        assert "X-Signature" in headers
+        assert headers["X-Signature"].startswith("sha256=")
+
+        assert "X-Timestamp" in headers
+        ts = int(headers["X-Timestamp"])
+        assert before <= ts <= after
+
     @pytest.mark.asyncio
     @patch("plugins.webhook_notification.webhook_notification.httpx.AsyncClient")
     async def test_webhook_delivery_success(self, mock_client_class):