Add request signing

examples/signed_request.py

@@ -0,0 +1,22 @@
+#!/usr/bin/env python
+import argparse
+import messagebird
+
+parser = argparse.ArgumentParser()
+parser.add_argument('--signingKey', help='access key for MessageBird API', type=str, required=True)
+parser.add_argument('--requestTimestamp', help='the request timestamp', type=str, required=True)
+parser.add_argument('--requestBody', help='the request body', type=str, required=True)
+parser.add_argument('--signature', help='the signature', type=str, required=True)
+args = vars(parser.parse_args())
+
+signed_request = messagebird.SignedRequest(args['signature'], args['timestamp'], args['requestBody'], {})
+
+if signed_request.verify(args['signingKey']):
+    print("The signed request has been verified.")
+else:
+    print("The signed request cannot be verified.")
+
+if signed_request.is_recent():
+    print("The signed request is recent.")
+else:
+    print("The signed request is not recent.")

messagebird/__init__.py

@@ -1 +1,2 @@
 from messagebird.client import Client, ErrorException
+from messagebird.signed_request import SignedRequest

messagebird/signed_request.py

@@ -0,0 +1,43 @@
+import hashlib
+import hmac
+import base64
+import time
+from collections import OrderedDict
+
+try:
+    from urllib.parse import urlencode
+except ImportError:
+    from urllib import urlencode
+
+
+class SignedRequest:
+
+    def __init__(self, requestSignature, requestTimestamp, requestBody, requestParameters):
+        self._requestSignature = requestSignature
+        self._requestTimestamp = str(requestTimestamp)
+        self._requestBody = requestBody
+        self._requestParameters = requestParameters
+
+    def verify(self, signing_key):
+        payload = self._build_payload()
+        expected_signature = base64.b64decode(self._requestSignature)
+        calculated_signature = hmac.new(signing_key.encode('latin-1'), payload.encode('latin-1'),
+                                        hashlib.sha256).digest()
+        return expected_signature == calculated_signature
+
+    def is_recent(self, offset=10):
+        return int(time.time()) - int(self._requestTimestamp) < offset
+
+    def _build_payload(self):
+        checksum_body = hashlib.sha256(self._requestBody.encode('latin-1')).digest()
+        str_checksum_body = checksum_body.decode('latin-1')
+        parts = [self._requestTimestamp, urlencode(self._sort_dict(self._requestParameters), True), str_checksum_body]
+        return "\n".join(parts)
+
+    def _sort_dict(self, dict):
+        sorted_dict = OrderedDict()
+
+        for key in sorted(dict):
+            sorted_dict[key] = dict[key]
+
+        return sorted_dict

tests/test_signed_request.py

@@ -0,0 +1,87 @@
+import unittest
+import time
+from messagebird.signed_request import SignedRequest
+
+try:
+    from unittest.mock import Mock
+except ImportError:
+    # mock was added to unittest in Python 3.3, but was an external library
+    # before.
+    from mock import Mock
+
+SIGNING_KEY = 'PlLrKaqvZNRR5zAjm42ZT6q1SQxgbbGd'
+
+
+class TestMessage(unittest.TestCase):
+
+    def test_signed_request_withou_body(self):
+        query = {
+            'recipient': '31612345678',
+            'reference': 'FOO',
+            'statusDatetime': '2019-01-11T09:17:11+00:00',
+            'id': 'eef0ab57a9e049be946f3821568c2b2e',
+            'status': 'delivered',
+            'mccmnc': '20408',
+            'ported': '1',
+        }
+
+        signature = 'KVBdcVdz2lYMwcBLZCRITgxUfA/WkwSi+T3Wxl2HL6w='
+        timestamp = 1547198231
+        body = ''
+
+        signed_request = SignedRequest(signature, timestamp, body, query)
+        self.assertTrue(signed_request.verify(SIGNING_KEY))
+
+    def test_signed_request_with_body(self):
+        query = {
+            'recipient': '31612345678',
+            'reference': 'FOO',
+            'statusDatetime': '2019-01-11T09:17:11+00:00',
+            'id': 'eef0ab57a9e049be946f3821568c2b2e',
+            'status': 'delivered',
+            'mccmnc': '20408',
+            'ported': '1',
+        }
+
+        signature = '2bl+38H4oHVg03pC3bk2LvCB0IHFgfC4cL5HPQ0LdmI='
+        timestamp = 1547198231
+        body = '{"foo":"bar"}'
+
+        signed_request = SignedRequest(signature, timestamp, body, query)
+        self.assertTrue(signed_request.verify(SIGNING_KEY))
+
+    def test_incorrectly_signed_request(self):
+        query = {
+            'recipient': '31612345678',
+            'reference': 'BAR',
+            'statusDatetime': '2019-01-11T09:17:11+00:00',
+            'id': 'eef0ab57a9e049be946f3821568c2b2e',
+            'status': 'delivered',
+            'mccmnc': '20408',
+            'ported': '1',
+        }
+
+        signature = 'KVBdcVdz2lYMwcBLZCRITgxUfA/WkwSi+T3Wxl2HL6w='
+        timestamp = 1547198231
+        body = ''
+
+        signed_request = SignedRequest(signature, timestamp, body, query)
+        self.assertFalse(signed_request.verify(SIGNING_KEY))
+
+    def test_recent_signed_request(self):
+        query = {}
+        signature = 'KVBdcVdz2lYMwcBLZCRITgxUfA/WkwSi+T3Wxl2HL6w='
+        timestamp = int(time.time()) - 1
+        body = ''
+
+        signed_request = SignedRequest(signature, timestamp, body, query)
+        self.assertTrue(signed_request.is_recent())
+
+    def test_not_recent_signed_request(self):
+        query = {}
+        signature = 'KVBdcVdz2lYMwcBLZCRITgxUfA/WkwSi+T3Wxl2HL6w='
+        timestamp = int(time.time()) - 100
+        body = ''
+
+        signed_request = SignedRequest(signature, timestamp, body, query)
+        self.assertFalse(signed_request.is_recent())