Hmac validation bank webhooks

Adyen/util.py

@@ -7,7 +7,7 @@
 import copy
 
 
-
+# generates HMAC signature for the NotificationRequest object
 def generate_notification_sig(dict_object, hmac_key):
 
     if not isinstance(dict_object, dict):
@@ -36,7 +36,30 @@ def generate_notification_sig(dict_object, hmac_key):
     return base64.b64encode(hm.digest())
 
 
+# generates HMAC signature for the payload (bytes)
+def generate_payload_sig(payload, hmac_key):
+
+    if not isinstance(payload, bytes):
+        raise ValueError("Must Provide payload as bytes")
+
+    hmac_key = binascii.a2b_hex(hmac_key)
+
+    hm = hmac.new(hmac_key, payload, hashlib.sha256)
+    return base64.b64encode(hm.digest())
+
+
 def is_valid_hmac_notification(dict_object, hmac_key):
+    """
+    validates the HMAC signature of the NotificationRequestItem object. Use for webhooks that provide the
+    hmacSignature as part of the payload `AdditionalData` (i.e. Payments)
+    Args:
+        dict_object: object with a list of notificationItems
+        hmac_key: HMAC key to generate the signature
+
+    Returns:
+        boolean: true when HMAC signature is valid
+    """
+
     dict_object = copy.deepcopy(dict_object) 
 
     if 'notificationItems' in dict_object:
@@ -53,5 +76,24 @@ def is_valid_hmac_notification(dict_object, hmac_key):
             return hmac.compare_digest(merchant_sign_str, expected_sign)
 
 
+def is_valid_hmac_payload(hmac_signature, hmac_key, payload):
+    """
+    validates the HMAC signature of a payload against an expected signature. Use for webhooks that provide the
+    hmacSignature in the HTTP header (i.e. Banking, Management API)
+    Args:
+        hmac_signature: HMAC signature to validate
+        hmac_key: HMAC key to generate the signature
+        payload: webhook payload
+
+    Returns:
+        boolean: true when HMAC signature is valid
+    """
+
+    merchant_sign = generate_payload_sig(payload, hmac_key)
+    merchant_sign_str = merchant_sign.decode("utf-8")
+
+    return hmac.compare_digest(merchant_sign_str, hmac_signature)
+
+
 def get_query(query_parameters):
     return '?' + '&'.join(["{}={}".format(k, v) for k, v in query_parameters.items()])

test/UtilTest.py

@@ -4,6 +4,7 @@
 from Adyen import settings
 from Adyen.util import (
     generate_notification_sig,
+    is_valid_hmac_payload,
     is_valid_hmac_notification,
     get_query
 )
@@ -132,3 +133,53 @@ def test_is_valid_hmac_notification_removes_additional_data(self):
             ]}
         is_valid_hmac_notification(notification, "11aa")
         self.assertIsNotNone(notification['notificationItems'][0]['NotificationRequestItem']['additionalData'])
+
+    def test_is_valid_hmac_payload(self):
+
+        payload = '''
+        {
+            "type": "merchant.created",
+            "environment": "test",
+            "createdAt": "01-01-2024",
+            "data": {
+                "capabilities": {
+                    "sendToTransferInstrument": {
+                        "requested": true,
+                        "requestedLevel": "notApplicable"
+                    }
+                },
+                "companyId": "YOUR_COMPANY_ID",
+                "merchantId": "YOUR_MERCHANT_ACCOUNT",
+                "status": "PreActive"
+            }
+        }
+        '''
+        hmac_key = "44782DEF547AAA06C910C43932B1EB0C71FC68D9D0C057550C48EC2ACF6BA056"
+        expected_hmac = "fX74xUdztFmaXAn3IusMFFUBUSkLmDQUK0tm8xL6ZTU="
+
+        self.assertTrue(is_valid_hmac_payload(expected_hmac, hmac_key, payload.encode("utf-8")))
+
+    def test_is_invalid_hmac_payload(self):
+        payload = '''
+        {
+            "type": "merchant.created",
+            "environment": "test",
+            "createdAt": "01-01-2024",
+            "data": {
+                "capabilities": {
+                    "sendToTransferInstrument": {
+                        "requested": true,
+                        "requestedLevel": "notApplicable"
+                    }
+                },
+                "companyId": "YOUR_COMPANY_ID",
+                "merchantId": "YOUR_MERCHANT_ACCOUNT",
+                "status": "PreActive"
+            }
+        }
+        '''
+
+        hmac_key = "44782DEF547AAA06C910C43932B1EB0C71FC68D9D0C057550C48EC2ACF6BA056"
+        expected_hmac = "MismatchingHmacKey="
+
+        self.assertFalse(is_valid_hmac_payload(expected_hmac, hmac_key, payload.encode("utf-8")))