Merge pull request #69 from nqkdev/master

Add support for new webhook signature method

Author: marcel corso gonzalez

examples/signed_request.py

@@ -0,0 +1,18 @@
+#!/usr/bin/env python
+import argparse
+import messagebird
+from messagebird.error import ValidationError
+
+parser = argparse.ArgumentParser()
+parser.add_argument('--signingKey', help='access key for MessageBird API', type=str, required=True)
+parser.add_argument('--signature', help='the signature', type=str, required=True)
+parser.add_argument('--requestURL', help='the full request url', type=str, required=True)
+parser.add_argument('--requestBody', help='the request body', type=str, required=False)
+args = vars(parser.parse_args())
+
+request_validator = messagebird.RequestValidator(args['signingKey'])
+
+try:
+    request_validator.validate_signature(args['signature'], args['requestURL'], bytearray(args['requestBody'].encode()))
+except ValidationError as err:
+    print("The signed request cannot be verified: ", str(err))

examples/signed_request_validation.py

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

messagebird/__init__.py

@@ -0,0 +1,86 @@
+import hashlib
+import hmac
+import jwt
+
+from typing import Union, Dict
+from messagebird.error import ValidationError
+
+
+class RequestValidator:
+    """
+    RequestValidator validates request signature signed by MessageBird services.
+
+    See https://developers.messagebird.com/docs/verify-http-requests
+    """
+
+    ALLOWED_ALGOS = ["HS256", "HS384", "HS512"]
+
+    def __init__(self, signature_key: str, skip_url_validation: bool = False):
+        """
+        :param signature_key: customer signature key. Can be retrieved through
+        <a href="https://dashboard.messagebird.com/developers/settings">Developer Settings</a>. This is NOT your API key.
+        :param skip_url_validation: whether url_hash claim validation should be skipped.
+        Note that when true, no query parameters should be trusted.
+        """
+        super().__init__()
+        self._signature_key = signature_key
+        self._skip_url_validation = skip_url_validation
+
+    def __str__(self) -> str:
+        return super().__str__()
+
+    def validate_signature(self, signature: str, url: str, request_body: Union[bytes, bytearray]) -> Dict[str, str]:
+        """
+        This method validates provided request signature, which is a JWT token.
+        This JWT is signed with a MessageBird account unique secret key, ensuring the request is from MessageBird and
+        a specific account.
+
+        The JWT contains the following claims:
+
+        *   "url_hash" - the raw URL hashed with SHA256 ensuring the URL wasn't altered.
+        *   "payload_hash" - the raw payload hashed with SHA256 ensuring the payload wasn't altered.
+        *    "jti" - a unique token ID to implement an optional non-replay check (NOT validated by default).
+        *    "nbf" - the not before timestamp.
+        *    "exp" - the expiration timestamp is ensuring that a request isn't captured and used at a later time.
+        *    "iss" - the issuer name, always MessageBird.
+
+        :param signature: the actual signature taken from request header "MessageBird-Signature-JWT".
+        :param url: the raw url including the protocol, hostname and query string, e.g. "https://example.com/?example=42".
+        :param request_body: the raw request body.
+        :returns: raw signature payload.
+        :raises: ValidationError if signature is invalid.
+        """
+        if not signature:
+            raise ValidationError("Signature is empty")
+        if not self._skip_url_validation and not url:
+            raise ValidationError("URL is empty")
+
+        try:
+            claims = jwt.decode(
+                jwt=signature,
+                key=self._signature_key,
+                algorithms=RequestValidator.ALLOWED_ALGOS,
+                options={
+                    "require": ["iss", "nbf", "exp"],
+                    "verify_iat": False,
+                },
+                issuer="MessageBird",
+                leeway=1
+            )
+        except jwt.InvalidTokenError as err:
+            raise ValidationError(str(err)) from err
+
+        if not self._skip_url_validation:
+            expected_url_hash = hashlib.sha256(url.encode("utf-8")).hexdigest()
+            if not hmac.compare_digest(expected_url_hash, claims["url_hash"]):
+                raise ValidationError("invalid jwt: claim url_hash is invalid")
+
+        payload_hash = claims.get("payload_hash")
+        if not request_body and payload_hash:
+            raise ValidationError("invalid jwt: claim payload_hash is set but actual payload is missing")
+        if request_body and not payload_hash:
+            raise ValidationError("invalid jwt: claim payload_hash is not set but payload is present")
+        if request_body and not hmac.compare_digest(hashlib.sha256(request_body).hexdigest(), payload_hash):
+            raise ValidationError("invalid jwt: claim payload_hash is invalid")
+
+        return claims

messagebird/request_validator.py

@@ -2,27 +2,48 @@
 import hmac
 import base64
 import time
+from warnings import warn
 from collections import OrderedDict
-
 from urllib.parse import urlencode
 
 
 class SignedRequest:
 
     def __init__(self, requestSignature, requestTimestamp, requestBody, requestParameters):
+        """
+        DEPRECATED
+        """
+        warn("signed_request module is deprecated, "
+             "use request_validator module instead",
+             DeprecationWarning, stacklevel=2)
+
         self._requestSignature = requestSignature
         self._requestTimestamp = str(requestTimestamp)
         self._requestBody = requestBody
         self._requestParameters = requestParameters
 
     def verify(self, signing_key):
+        """
+        DEPRECATED
+        """
+        warn("signed_request.verify is deprecated, "
+             "use request_validator.validate_signature instead",
+             DeprecationWarning, stacklevel=2)
+
         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):
+        """
+        DEPRECATED
+        """
+        warn("signed_request.is_recent is deprecated, "
+             "use request_validator package instead",
+             DeprecationWarning, stacklevel=2)
+
         return int(time.time()) - int(self._requestTimestamp) < offset
 
     def _build_payload(self):

messagebird/signed_request.py

@@ -17,7 +17,7 @@
     url                           = 'https://github.com/messagebird/python-rest-api',
     download_url                  = 'https://github.com/messagebird/python-rest-api/tarball/2.0.0',
     keywords                      = ['messagebird', 'sms'],
-    install_requires              = ['requests>=2.4.1', 'python-dateutil>=2.6.0'],
+    install_requires              = ['requests>=2.4.1', 'python-dateutil>=2.6.0', 'pyjwt>=2.1.0'],
     extras_require               = {
         'dev': [
             'pytest',

setup.py

@@ -0,0 +1,49 @@
+import json
+import re
+from unittest import mock
+from pathlib import Path
+
+import pytest as pytest
+
+from messagebird.base import Base
+from messagebird.error import ValidationError
+from messagebird.request_validator import RequestValidator
+
+ERROR_MAP = {
+    "invalid jwt: claim nbf is in the future": "The token is not yet valid (nbf)",
+    "invalid jwt: claim exp is in the past": "Signature has expired",
+    "invalid jwt: signature is invalid": "Signature verification failed",
+    "invalid jwt: signing method none is invalid": "The specified alg value is not allowed"
+}
+
+
+def load_test_cases():
+    test_data_file = Path(__file__).parent / 'test_request_validator/webhook_signature_test_data.json'
+    with open(test_data_file) as f:
+        tc_data = json.loads(f.read())
+
+    return tc_data
+
+
+@mock.patch('jwt.api_jwt.datetime')
+@pytest.mark.parametrize('test_case', load_test_cases(), ids=lambda args: args['name'])
+def test_validate_signature(mock_dt, test_case):
+    mock_dt.utcnow = mock.Mock(return_value=Base.value_to_time(test_case['timestamp']))
+
+    validator = RequestValidator(test_case.get('secret'))
+
+    payload = test_case.get('payload')
+    if payload:
+        payload = bytes(payload.encode())
+
+    def run_decode():
+        return validator.validate_signature(test_case['token'], test_case['url'], payload)
+
+    if not test_case['valid']:
+        err = ERROR_MAP.get(test_case["reason"]) or test_case["reason"]
+        pytest.raises(ValidationError, run_decode).match(re.escape(err))
+        return
+
+    decoded = run_decode()
+
+    assert decoded is not None