Improve webhook request validation and test coverage

Author: alexjweil

lightspark/__tests__/test_webhooks.py

@@ -0,0 +1,51 @@
+from datetime import datetime
+
+import pytest
+
+import lightspark
+
+
+class TestWebhooks:
+    def test_valid_data_parses(self):
+        data = b'{"event_type": "NODE_STATUS", "event_id": "1615c8be5aa44e429eba700db2ed8ca5", "timestamp": "2023-05-17T23:56:47.874449+00:00", "entity_id": "lightning_node:01882c25-157a-f96b-0000-362d42b64397"}'
+        secret = "3gZ5oQQUASYmqQNuEk0KambNMVkOADDItIJjzUlAWjX"
+        hex_digest = "62a8829aeb48b4142533520b1f7f86cdb1ee7d718bf3ea15bc1c662d4c453b74"
+
+        webhook = lightspark.WebhookEvent.verify_and_parse(data, hex_digest, secret)
+
+        assert (
+            webhook.entity_id == "lightning_node:01882c25-157a-f96b-0000-362d42b64397"
+        )
+        assert webhook.event_id == "1615c8be5aa44e429eba700db2ed8ca5"
+        assert webhook.event_type == lightspark.WebhookEventType.NODE_STATUS
+        assert webhook.timestamp == datetime.fromisoformat(
+            "2023-05-17T23:56:47.874449+00:00"
+        )
+
+    @pytest.mark.parametrize(
+        "hex_digest",
+        [
+            pytest.param("deadbeef", id="wrong length"),
+            pytest.param("a" * 64, id="incorrect"),
+            pytest.param("NotAHexString", id="not hex"),
+            pytest.param(
+                "62a8829aeb48b4142533520b1f7f86cdb1ee7d718bf3ea15bc1c662d4c453b74"
+                + "qq",
+                id="extra bytes",
+            ),
+        ],
+    )
+    def test_invalid_data_raises_value_error(self, hex_digest: str):
+        data = b'{"event_type": "NODE_STATUS", "event_id": "1615c8be5aa44e429eba700db2ed8ca5", "timestamp": "2023-05-17T23:56:47.874449+00:00", "entity_id": "lightning_node:01882c25-157a-f96b-0000-362d42b64397"}'
+        secret = "3gZ5oQQUASYmqQNuEk0KambNMVkOADDItIJjzUlAWjX"
+
+        with pytest.raises(ValueError):
+            lightspark.WebhookEvent.verify_and_parse(data, hex_digest, secret)
+
+    def test_invalid_data_type_raises_type_error(self):
+        data = 1
+        hex_digest = "deadbeef"
+        secret = "3gZ5oQQUASYmqQNuEk0KambNMVkOADDItIJjzUlAWjX"
+
+        with pytest.raises(TypeError):
+            lightspark.WebhookEvent.verify_and_parse(data, hex_digest, secret)  # type: ignore

lightspark/webhooks.py

@@ -22,17 +22,14 @@ class WebhookEvent:
 
     @classmethod
     def verify_and_parse(
-        cls, data: bytes, hexdigest: str, webhook_secret: str
+        cls, data: bytes, hex_digest: str, webhook_secret: str
     ) -> "WebhookEvent":
-        """Verifies the signature and parses the message into a
-        WebhookEvent object.
+        """Verifies the signature and parses the message into a WebhookEvent object.
 
         Args:
           data: the POST message body received by the webhook.
-          hexdigest: the message signature sent in the
-            `lightspark-signature` header.
-          webhook_secret: the webhook secret configured at the
-            Lightspark API configuration.
+          hex_digest: the message signature sent in the `lightspark-signature` header.
+          webhook_secret: the webhook secret configured in the Lightspark API configuration.
 
         Returns:
           A parsed WebhookEvent object.
@@ -45,8 +42,9 @@ def verify_and_parse(
 
         sig = hmac.new(
             webhook_secret.encode("ascii"), msg=data, digestmod=hashlib.sha256
-        )
-        if sig.hexdigest().lower() != hexdigest.lower():
+        ) 
+
+        if not hmac.compare_digest(sig.hexdigest(), hex_digest):
             raise ValueError("Webhook message hash does not match signature")
 
         return cls.parse(data)
@@ -73,5 +71,5 @@ def parse(cls, data: bytes) -> "WebhookEvent":
             event_id=event["event_id"],
             timestamp=datetime.fromisoformat(event["timestamp"]),
             entity_id=event["entity_id"],
-            wallet_id=event["wallet_id"] if "wallet_id" in event else None,
+            wallet_id=event.get("wallet_id", None),
         )