Tra 63 optimus python package for webhook

CHANGELOG.md

@@ -4,6 +4,9 @@ Changelog
 [Unreleased]
 ------------
 
+[v0.2.0] - 2022-12-04
+------------------
+
 [v0.1.3] - 2022-09-17
 ------------------
 - add README.md

README.md

@@ -32,3 +32,36 @@ response = client.Payment.request_payment({
 
 print(response.text)
 ```
+
+## Verify webhook
+
+Before you respond to a Webhook request, you need first to verify that it is coming from Transfa.
+
+```python
+from rest_framework.decorators import api_view
+from rest_framework import status
+from rest_framework.response import Response
+
+from transfa.webhook import Webhook
+
+# Do not save the secret key in plain text in your code, set it instead as an environment variable.
+secret_key = 'ps_test:...'
+
+@api_view(["POST"])
+def webhook_endpoint(request):
+    body = request.data
+
+    # Will return True or False and the body of the request that has been slightly modified
+    # You should use that data when processing the webhook instead of the previous one
+    webhook = Webhook(webhook_token=secret_key, body=body, headers=request.headers)
+    verified = webhook.verify()
+
+    if verified is None:
+        return Response({"detail": "unauthorized"}, status=status.HTTP_401_UNAUTHORIZED)
+
+    # Process Webhook payload
+    # ...
+    # ...
+
+    return Response({"detail": True}, status=status.HTTP_200_OK)
+```

pyproject.toml

@@ -6,4 +6,4 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "transfa"
-version = "0.1.3"
+version = "0.2.0"

setup.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.1.3
+current_version = 0.2.0
 commit = True
 tag = True
 

tests/client.py

@@ -1,3 +1,3 @@
 """
 
-"""
+"""

tests/conftest.py

@@ -17,3 +17,23 @@ def payment(payments):
     :return: a payment object
     """
     return payments[0]
+
+
+@pytest.fixture()
+def valid_webhook_payload():
+    """
+    :return: a valid webhook payload
+    """
+
+    with open("tests/valid_webhook_payload.json") as f:
+        return json.load(f)
+
+
+@pytest.fixture()
+def invalid_webhook_payload():
+    """
+    :return: an invalid webhook payload
+    """
+
+    with open("tests/invalid_webhook_payload.json") as f:
+        return json.load(f)

tests/invalid_webhook_payload.json

@@ -0,0 +1,7 @@
+{
+        "id": "2588efb42dc",
+        "event": "payment:success",
+        "payment": "0QL4KX",
+        "created": "2021-12-02T10:54:08.511427+00:00",
+        "updated": "2021-12-02T10:54:08.511435+00:00"
+}

tests/test_payment.py

@@ -14,10 +14,7 @@
 
 
 def paginator_response(data):
-    return {
-        "count": len(data),
-        "results": data
-    }
+    return {"count": len(data), "results": data}
 
 
 @responses.activate
@@ -32,30 +29,29 @@ def test_request_payment(payment):
         API_URL,
         json=payment,
         status=HTTPStatus.CREATED,
-        match=[matchers.header_matcher({
-            "Authorization": f"{default_auth_header_bearer} {client.api_key}",
-            'accept': 'application/json',
-            'content-type': 'application/json;charset=utf-8',
-            'Idempotency-Key': idempotency_key,
-            "user-agent": "Transfa API SDK-Python/%s" % VERSION,
-        }
-        )]
+        match=[
+            matchers.header_matcher(
+                {
+                    "Authorization": f"{default_auth_header_bearer} {client.api_key}",
+                    "accept": "application/json",
+                    "content-type": "application/json;charset=utf-8",
+                    "Idempotency-Key": idempotency_key,
+                    "user-agent": "Transfa API SDK-Python/%s" % VERSION,
+                }
+            )
+        ],
     )
-    data = {
-        "account_alias": "60201010",
-        "amount": 5000,
-        "mode": "mtn-benin"
-    }
+    data = {"account_alias": "60201010", "amount": 5000, "mode": "mtn-benin"}
 
     response = client.Payment.request_payment(data, idempotency_key=idempotency_key)
 
     assert response.status_code == HTTPStatus.CREATED
     response_data = response.json()
 
-    assert response_data['account_alias'] == data['account_alias']
-    assert response_data['amount'] == data['amount']
-    assert response_data['mode'] == data['mode']
-    assert response_data['type'] == "request-payment"
+    assert response_data["account_alias"] == data["account_alias"]
+    assert response_data["amount"] == data["amount"]
+    assert response_data["mode"] == data["mode"]
+    assert response_data["type"] == "request-payment"
 
 
 @responses.activate
@@ -69,19 +65,22 @@ def test_list_payments(payments):
         API_URL,
         json=paginator_response(payments),
         status=HTTPStatus.OK,
-        match=[matchers.header_matcher({
-            "Authorization": f"{default_auth_header_bearer} {client.api_key}",
-            "user-agent": "Transfa API SDK-Python/%s" % VERSION,
-        }
-        )]
+        match=[
+            matchers.header_matcher(
+                {
+                    "Authorization": f"{default_auth_header_bearer} {client.api_key}",
+                    "user-agent": "Transfa API SDK-Python/%s" % VERSION,
+                }
+            )
+        ],
     )
 
     response = client.Payment.list()
 
     assert response.status_code == HTTPStatus.OK
     response_data = response.json()
 
-    assert response_data['count'] == 1
+    assert response_data["count"] == 1
 
 
 @responses.activate
@@ -95,19 +94,23 @@ def test_retrieve_payment(payment):
         f"{API_URL}{payment['id']}/",
         json=payment,
         status=HTTPStatus.OK,
-        match=[matchers.header_matcher({
-            "Authorization": f"{default_auth_header_bearer} {client.api_key}",
-            "user-agent": "Transfa API SDK-Python/%s" % VERSION,
-        }
-        ), matchers.query_param_matcher({})]
+        match=[
+            matchers.header_matcher(
+                {
+                    "Authorization": f"{default_auth_header_bearer} {client.api_key}",
+                    "user-agent": "Transfa API SDK-Python/%s" % VERSION,
+                }
+            ),
+            matchers.query_param_matcher({}),
+        ],
     )
 
-    response = client.Payment.retrieve(payment['id'])
+    response = client.Payment.retrieve(payment["id"])
 
     assert response.status_code == HTTPStatus.OK
     response_data = response.json()
 
-    assert response_data['id'] == payment['id']
+    assert response_data["id"] == payment["id"]
 
 
 @responses.activate
@@ -121,16 +124,19 @@ def test_refund_payment(payment):
         f"{API_URL}{payment['id']}/refund/",
         json=payment,
         status=HTTPStatus.OK,
-        match=[matchers.header_matcher({
-            "Authorization": f"{default_auth_header_bearer} {client.api_key}",
-            "user-agent": "Transfa API SDK-Python/%s" % VERSION,
-        }
-        )]
+        match=[
+            matchers.header_matcher(
+                {
+                    "Authorization": f"{default_auth_header_bearer} {client.api_key}",
+                    "user-agent": "Transfa API SDK-Python/%s" % VERSION,
+                }
+            )
+        ],
     )
 
-    response = client.Payment.refund(payment['id'])
+    response = client.Payment.refund(payment["id"])
 
     assert response.status_code == HTTPStatus.OK
     response_data = response.json()
 
-    assert response_data['id'] == payment['id']
+    assert response_data["id"] == payment["id"]

tests/test_webhook.py

@@ -0,0 +1,76 @@
+import hmac
+import hashlib
+import json
+
+from transfa.webhook import Webhook
+from transfa.utils import get_random_string
+
+
+class WebhookData:
+
+    def __init__(self):
+        self.valid_webhook_token = get_random_string(64)
+        self.invalid_webhook_token = get_random_string(64)
+
+    def generate_signature(self, body, webhook_token, algorithm=hashlib.sha512):
+        body = json.dumps(body)
+        secret = webhook_token.encode("utf-8")
+
+        if not isinstance(body, bytes):
+            body = body.encode("utf-8")
+
+        signature = hmac.new(secret, body, digestmod=algorithm)
+
+        return signature.hexdigest()
+
+
+def test_valid_webhook_signature(valid_webhook_payload):
+    webhook_data = WebhookData()
+    signature = webhook_data.generate_signature(valid_webhook_payload, webhook_data.valid_webhook_token)
+
+    webhook = Webhook(webhook_token=webhook_data.valid_webhook_token, body=valid_webhook_payload, headers={
+        "X-Webhook-Transfa-Signature": signature
+    })
+
+    verified = webhook.verify()
+
+    assert verified == valid_webhook_payload
+
+
+def test_invalid_webhook_signature(valid_webhook_payload, invalid_webhook_payload):
+    webhook_data = WebhookData()
+    invalid_signature = webhook_data.generate_signature(invalid_webhook_payload, webhook_data.valid_webhook_token)
+
+    webhook = Webhook(webhook_token=webhook_data.valid_webhook_token, body=valid_webhook_payload, headers={
+        "X-Webhook-Transfa-Signature": invalid_signature
+    })
+
+    verified = webhook.verify()
+
+    assert not verified
+
+
+def test_valid_webhook_token(valid_webhook_payload):
+    webhook_data = WebhookData()
+    valid_signature = webhook_data.generate_signature(valid_webhook_payload, webhook_data.valid_webhook_token)
+
+    webhook = Webhook(webhook_token=webhook_data.valid_webhook_token, body=valid_webhook_payload, headers={
+        "X-Webhook-Transfa-Signature": valid_signature
+    })
+
+    verified = webhook.verify()
+
+    assert verified == valid_webhook_payload
+
+
+def test_invalid_webhook_token(valid_webhook_payload):
+    webhook_data = WebhookData()
+    valid_signature = webhook_data.generate_signature(valid_webhook_payload, webhook_data.valid_webhook_token)
+
+    webhook = Webhook(webhook_token=webhook_data.invalid_webhook_token, body=valid_webhook_payload, headers={
+        "X-Webhook-Transfa-Signature": valid_signature
+    })
+
+    verified = webhook.verify()
+
+    assert not verified

tests/valid_webhook_payload.json

@@ -0,0 +1,7 @@
+{
+        "id": "2588efb42dc7436f890c1b8a0aabd72f",
+        "event": "payment:processing",
+        "payment": "GYKU0QL4KX",
+        "created": "2022-12-02T10:54:08.511427+00:00",
+        "updated": "2022-12-02T10:54:08.511435+00:00"
+}

transfa/__init__.py

@@ -10,7 +10,7 @@
 private_secret = None
 api_base = "https://api.transfapp.com"
 verify_ssl = True
-default_auth_header_bearer = "Api-Optimus-Key"
+default_auth_header_bearer = "Api-Transfa-Key"
 
 
 # API Resources

transfa/api_resources/__init__.py

@@ -1 +1 @@
-from transfa.api_resources.payments import Payment
+from transfa.api_resources.payments import Payment

transfa/api_resources/payments.py

@@ -7,7 +7,7 @@
 
 class PaymentResource:
     """
-        This class contains methods and utils used for interacting with the payment API.
+    This class contains methods and utils used for interacting with the payment API.
     """
 
     def __init__(self, api):
@@ -32,13 +32,9 @@ def request_payment(self, data, idempotency_key=None, **kwargs):
 
         # Idempotency key setup in headers
         if idempotency_key:
-            kwargs['headers'] = {
-                "Idempotency-Key": idempotency_key
-            }
+            kwargs["headers"] = {"Idempotency-Key": idempotency_key}
         else:
-            kwargs['headers'] = {
-                "Idempotency-Key": uuid.uuid4().hex
-            }
+            kwargs["headers"] = {"Idempotency-Key": uuid.uuid4().hex}
 
         data["type"] = PaymentTypeEnum.request_payment.value
 
@@ -80,7 +76,7 @@ def status(self, payment_id):
 
         response_data = response.json()
 
-        return response_data.get('status'), response_data.get('financial_status')
+        return response_data.get("status"), response_data.get("financial_status")
 
 
 Payment = PaymentResource

transfa/enums.py

@@ -30,3 +30,7 @@ class APIKeyPrefix(Enum):
 class PrivateSecretPrefix(Enum):
     test = "ps_test"
     live = "ps_live"
+
+
+class TransfaHeadersIdentifiers(Enum):
+    webhook_signature = "X-Webhook-Transfa-Signature"