feat: add verify_webhook_signature

README.md

@@ -30,10 +30,13 @@ your API key.
 <br/>
 
 ## Installation
+
 [![PyPI Latest Release](https://img.shields.io/pypi/v/blockfrost-python.svg)](https://pypi.org/project/blockfrost-python/)
+
 ```console
 $ pip install blockfrost-python
 ```
+
 <br/>
 
 ## Usage
@@ -59,7 +62,7 @@ try:
     print(health)   # prints Dataframe:         is_healthy
                     #                       0         True
 
-    
+
     account_rewards = api.account_rewards(
         stake_address='stake1ux3g2c9dx2nhhehyrezyxpkstartcqmu9hk63qgfkccw5rqttygt7',
         count=20,
@@ -107,4 +110,61 @@ try:
         file.write(file_data)
 except ApiError as e:
     print(e)
-```
+```
+
+### Verifying Secure Webhook signature
+
+Webhooks enable Blockfrost to push real-time notifications to your application. In order to prevent malicious actor from pretending to be Blockfrost every webhook request is signed. The signature is included in a request's `Blockfrost-Signature` header. This allows you to verify that the events were sent by Blockfrost, not by a third party.
+To learn more about Secure Webhooks, see [Secure Webhooks Docs](https://blockfrost.dev/docs/start-building/webhooks/).
+
+You can verify the signature using `verifyWebhookSignature` function.
+
+Example:
+
+```python
+# Example of Python Flask app with /webhook endpoint
+# for processing events sent by Blockfrost Secure Webhooks
+from flask import Flask, request, json
+from blockfrost import verify_webhook_signature, SignatureVerificationError
+
+SECRET_AUTH_TOKEN = "SECRET-WEBHOOK-AUTH-TOKEN"
+
+app = Flask(__name__)
+
+@app.route('/webhook', methods=['POST'])
+def webhook():
+    if request.method == 'POST':
+        # Validate webhook signature
+        request_bytes = request.get_data()
+        try:
+            verify_webhook_signature(
+                request_bytes, request.headers['Blockfrost-Signature'], SECRET_AUTH_TOKEN)
+        except SignatureVerificationError as e:
+            # for easier debugging you can access passed header and request_body values (e.header, e.request_body)
+            print('Webhook signature is invalid.', e)
+            return 'Invalid signature', 403
+
+        # Get the payload as JSON
+        event = request.json
+
+        print('Received request id {}, webhook_id: {}'.format(
+            event['id'], event['webhook_id']))
+
+        if event['type'] == "block":
+            # process Block event
+            print('Received block hash {}'.format(event['payload']['hash']))
+        elif event['type'] == "...":
+            # truncated
+        else:
+            # Unexpected event type
+            print('Unexpected event type {}'.format(event['type']))
+
+        return 'Webhook received', 200
+    else:
+        return 'POST Method not supported', 405
+
+
+
+if __name__ == "__main__":
+    app.run(host='0.0.0.0', port=6666)
+```

blockfrost/__init__.py

@@ -2,3 +2,4 @@
 from blockfrost.ipfs import BlockFrostIPFS
 from blockfrost.config import ApiUrls
 from blockfrost.utils import ApiError, Namespace
+from blockfrost.helpers import SignatureVerificationError, verify_webhook_signature

blockfrost/helpers.py

@@ -0,0 +1,68 @@
+import hmac
+import hashlib
+import time
+
+
+class SignatureVerificationError(Exception):
+    def __init__(self, message, header, request_body):
+        self.message = message
+        self.header = header
+        self.request_body = request_body
+        super().__init__(self.message)
+
+
+def get_unix_timestamp():
+    return int(time.time())
+
+
+def verify_webhook_signature(request_body, signature_header, secret, timestamp_tolerance_seconds=600):
+    # Parse signature header
+    # Example of Blockfrost-Signature header: t=1648550558,v1=162381a59040c97d9b323cdfec02facdfce0968490ec1732f5d938334c1eed4e,v1=...)
+    tokens = signature_header.split(',')
+    timestamp = None
+    signatures = []
+    for token in tokens:
+        key, value = token.split('=')
+        if key == 't':
+            timestamp = value
+        elif key == 'v1':
+            signatures.append(value)
+        else:
+            print('Cannot parse part of the Blockfrost-Signature header, key "{}" is not supported by this version of Blockfrost SDK. Please upgrade.'.format(key))
+
+    if timestamp is None or timestamp.isnumeric() is False or len(tokens) < 2:
+        # timestamp and at least one signature must be present
+        raise SignatureVerificationError(
+            'Invalid signature header format.', signature_header, request_body)
+
+    if len(signatures) == 0:
+        # There are no signatures that this version of SDK supports
+        raise SignatureVerificationError(
+            'No signatures with supported version scheme.', signature_header, request_body)
+
+    has_valid_signature = False
+    for signature in signatures:
+        # Recreate signature by concatenating the timestamp with the payload (all in bytes),
+        # then compute HMAC using sha256 and provided secret (webhook auth token)
+        signature_payload = timestamp.encode() + b"." + request_body
+        local_signature = hmac.new(
+            secret.encode(), signature_payload, hashlib.sha256).hexdigest()
+
+        # computed signature should match at least one signature parsed from a signature header
+        if (hmac.compare_digest(signature, local_signature)):
+            has_valid_signature = True
+            break
+
+    if has_valid_signature == False:
+        raise SignatureVerificationError(
+            'No signature matches the expected signature for the payload.', signature_header, request_body)
+
+    current_timestamp = get_unix_timestamp()
+
+    if (current_timestamp - int(timestamp) > timestamp_tolerance_seconds):
+        # Event is older than timestamp_tolerance_seconds
+        raise SignatureVerificationError(
+            'Signature\'s timestamp is outside of the time tolerance.', signature_header, request_body)
+    else:
+        # Successfully validate the signature only if it is within timestamp_tolerance_seconds tolerance
+        return True

tests/test_helpers.py

@@ -0,0 +1,78 @@
+import os
+import mock
+import pytest
+from blockfrost import SignatureVerificationError, verify_webhook_signature
+
+request_body = b'{"id":"47668401-c3a4-42d4-bac1-ad46515924a3","webhook_id":"cf68eb9c-635f-415e-a5a8-6233638f28d7","created":1650013856,"type":"block","payload":{"time":1650013853,"height":7126256,"hash":"f49521b67b440e5030adf124aee8f88881b7682ba07acf06c2781405b0f806a4","slot":58447562,"epoch":332,"epoch_slot":386762,"slot_leader":"pool1njjr0zn7uvydjy8067nprgwlyxqnznp9wgllfnag24nycgkda25","size":34617,"tx_count":13,"output":"13403118309871","fees":"4986390","block_vrf":"vrf_vk197w95j9alkwt8l4g7xkccknhn4pqwx65c5saxnn5ej3cpmps72msgpw69d","previous_block":"9e3f5bfc9f0be44cf6e14db9ed5f1efb6b637baff0ea1740bb6711786c724915","next_block":null,"confirmations":0}}'
+success_fixtures_list = [
+    {
+        'description': 'valid signature',
+        'request_body': request_body,
+        'signature_header': 't=1650013856,v1=f4c3bb2a8b0c8e21fa7d5fdada2ee87c9c6f6b0b159cc22e483146917e195c3e',
+        'secret': '59a1eb46-96f4-4f0b-8a03-b4d26e70593a',
+        'current_timestamp_mock': 1650013856 + 1,
+        'result': True
+    },
+    {
+        'description': '2 signatures, one valid and one invalid',
+        'request_body': request_body,
+        'signature_header': 't=1650013856,v1=abc,v1=f4c3bb2a8b0c8e21fa7d5fdada2ee87c9c6f6b0b159cc22e483146917e195c3e',
+        'secret': '59a1eb46-96f4-4f0b-8a03-b4d26e70593a',
+        'current_timestamp_mock': 1650013856 + 1,
+        'result': True
+    }
+]
+
+error_fixtures_list = [
+    {
+        'description': 'throws due to invalid header fromat',
+        'request_body': request_body,
+        'signature_header': 'v1=f4c3bb2a8b0c8e21fa7d5fdada2ee87c9c6f6b0b159cc22e483146917e195c3e',
+        'secret': '59a1eb46-96f4-4f0b-8a03-b4d26e70593a',
+        'current_timestamp_mock': 1650013856 + 1,
+        'result_error': 'Invalid signature header format.'
+    },
+    {
+        'description': 'throws due to sig version not supported by this sdk',
+        'request_body': request_body,
+        'signature_header': 't=1650013856,v42=abc',
+        'secret': '59a1eb46-96f4-4f0b-8a03-b4d26e70593a',
+        'current_timestamp_mock': 1650013856 + 1,
+        'result_error': 'No signatures with supported version scheme.'
+    },
+    {
+        'description': 'throws due to no signature match',
+        'request_body': request_body,
+        'signature_header': 't=1650013856,v1=abc',
+        'secret': '59a1eb46-96f4-4f0b-8a03-b4d26e70593a',
+        'current_timestamp_mock': 1650013856 + 1,
+        'result_error': 'No signature matches the expected signature for the payload.'
+    },
+    {
+        'description': 'throws due to timestamp out of tolerance zone',
+        'request_body': request_body,
+        'signature_header': 't=1650013856,v1=f4c3bb2a8b0c8e21fa7d5fdada2ee87c9c6f6b0b159cc22e483146917e195c3e',
+        'secret': '59a1eb46-96f4-4f0b-8a03-b4d26e70593a',
+        'current_timestamp_mock': 1650013856 + 7200,
+        'result_error': 'Signature\'s timestamp is outside of the time tolerance.'
+    }
+]
+
+
+@pytest.mark.parametrize("fixture", success_fixtures_list)
+def test_verify_webhook_signature(fixture):
+    with mock.patch('blockfrost.helpers.get_unix_timestamp', return_value=fixture['current_timestamp_mock']):
+        res = verify_webhook_signature(
+            fixture['request_body'], fixture['signature_header'], fixture['secret'])
+        assert res == fixture['result']
+
+
+@pytest.mark.parametrize("fixture", error_fixtures_list)
+def test_verify_webhook_signature_fails(fixture):
+    with mock.patch('blockfrost.helpers.get_unix_timestamp', return_value=fixture['current_timestamp_mock']):
+        with pytest.raises(SignatureVerificationError) as e_info:
+            verify_webhook_signature(
+                fixture['request_body'], fixture['signature_header'], fixture['secret'])
+        assert str(e_info.value) == fixture['result_error']
+        assert e_info.value.header == fixture['signature_header']
+        assert e_info.value.request_body == fixture['request_body']