chore: support HMAC-SHA256 webhook signatures in RequestValidator

tests/unit/test_request_validator.py

@@ -5,7 +5,18 @@
 from django.http import QueryDict
 from multidict import MultiDict
 
-from twilio.request_validator import RequestValidator
+import base64
+import hashlib
+import hmac
+
+from twilio.request_validator import (
+    HMAC_SHA1,
+    HMAC_SHA256,
+    SIGNATURE_HEADER,
+    SIGNATURE_SETTINGS_HEADER,
+    RequestValidator,
+    get_signature_algorithm,
+)
 
 
 class ValidationTest(unittest.TestCase):
@@ -93,3 +104,68 @@ def test_validation_of_body_succeeds(self):
             uri, self.body, "a9nBmqA0ju/hNViExpshrM61xv4="
         )
         assert is_valid
+
+    def _sha256_signature(self, data):
+        mac = hmac.new(b"12345", data.encode("utf-8"), hashlib.sha256)
+        return base64.b64encode(mac.digest()).decode("utf-8")
+
+    def test_compute_signature_sha256(self):
+        data = self.uri + "".join(k + v for k, v in sorted(self.params.items()))
+        signature = self.validator.compute_signature(self.uri, self.params, HMAC_SHA256)
+        assert signature == self._sha256_signature(data)
+
+    def test_validate_sha256(self):
+        signature = self.validator.compute_signature(self.uri, self.params, HMAC_SHA256)
+        assert self.validator.validate(self.uri, self.params, signature, HMAC_SHA256)
+        assert self.validator.validate(self.uri, self.params, signature, "hmac_sha256")
+
+    def test_sha256_and_sha1_are_not_interchangeable(self):
+        sha256_signature = self.validator.compute_signature(
+            self.uri, self.params, HMAC_SHA256
+        )
+        assert not self.validator.validate(self.uri, self.params, sha256_signature)
+        assert not self.validator.validate(
+            self.uri, self.params, self.expected, HMAC_SHA256
+        )
+
+    def test_validate_defaults_to_sha1(self):
+        assert self.validator.validate(self.uri, self.params, self.expected)
+        assert self.validator.validate(self.uri, self.params, self.expected, None)
+        assert self.validator.validate(self.uri, self.params, self.expected, "")
+        assert self.validator.validate(self.uri, self.params, self.expected, HMAC_SHA1)
+
+    def test_validate_unsupported_algorithm_returns_false_and_warns(self):
+        with self.assertLogs("twilio.request_validator", level="WARNING") as logs:
+            assert not self.validator.validate(
+                self.uri, self.params, self.expected, "HMAC_SHA512"
+            )
+        assert "HMAC_SHA512" in logs.output[0]
+
+    def test_compute_signature_unsupported_algorithm_raises(self):
+        with self.assertRaises(ValueError):
+            self.validator.compute_signature(self.uri, self.params, "MD5")
+
+    def test_validate_body_sha256(self):
+        uri = self.uriWithBody
+        signature = self.validator.compute_signature(uri, {}, HMAC_SHA256)
+        assert self.validator.validate(uri, self.body, signature, HMAC_SHA256)
+        assert not self.validator.validate(uri, "{}", signature, HMAC_SHA256)
+        assert not self.validator.validate(uri, self.body, signature)
+
+    def test_get_signature_algorithm(self):
+        assert (
+            get_signature_algorithm("key=webhooks_sharedkey_1,algorithm=HMAC_SHA256")
+            == HMAC_SHA256
+        )
+        assert get_signature_algorithm("algorithm=HMAC_SHA1,key=k") == HMAC_SHA1
+        assert (
+            get_signature_algorithm(" key=k , algorithm = HMAC_SHA256 ") == HMAC_SHA256
+        )
+
+    def test_get_signature_algorithm_defaults_to_sha1(self):
+        for header in (None, "", "key=k", "key=k,algorithm="):
+            assert get_signature_algorithm(header) == HMAC_SHA1
+
+    def test_header_constants(self):
+        assert SIGNATURE_HEADER == "X-Twilio-Signature"
+        assert SIGNATURE_SETTINGS_HEADER == "X-Twilio-Webhooks-Signature-Settings"

twilio/request_validator.py

@@ -1,9 +1,21 @@
 import base64
 import hmac
+import logging
 from hashlib import sha1, sha256
 
 from urllib.parse import urlparse, parse_qs
 
+logger = logging.getLogger(__name__)
+
+SIGNATURE_HEADER = "X-Twilio-Signature"
+# Header Twilio adds to webhooks signed with a SharedKey:
+# key=<sharedKeyId>,algorithm=<algorithm>
+SIGNATURE_SETTINGS_HEADER = "X-Twilio-Webhooks-Signature-Settings"
+HMAC_SHA1 = "HMAC_SHA1"
+HMAC_SHA256 = "HMAC_SHA256"
+
+_HASH_FUNCTIONS = {HMAC_SHA1: sha1, HMAC_SHA256: sha256}
+
 
 def compare(string1, string2):
     """Compare two strings while protecting against timing attacks
@@ -58,18 +70,47 @@ def add_port(uri):
     return new_uri.geturl()
 
 
+def get_signature_algorithm(signature_settings_header):
+    """Get the signing algorithm named in the X-Twilio-Webhooks-Signature-Settings header
+
+    A missing or empty header (or one without ``algorithm=``) means the legacy
+    Auth Token signature, HMAC_SHA1.
+
+    :param str signature_settings_header: header value, e.g.
+        ``key=webhooks_sharedkey_...,algorithm=HMAC_SHA256``
+
+    :returns: the algorithm name
+    :rtype: str
+    """
+    for part in (signature_settings_header or "").split(","):
+        name, separator, value = part.partition("=")
+        if separator and name.strip() == "algorithm" and value.strip():
+            return value.strip()
+    return HMAC_SHA1
+
+
 class RequestValidator(object):
     def __init__(self, token):
+        """
+        :param str token: the secret the webhook was signed with: the account Auth Token,
+            or the SharedKey secret when the request carries X-Twilio-Webhooks-Signature-Settings
+        """
         self.token = token.encode("utf-8")
 
-    def compute_signature(self, uri, params):
+    def compute_signature(self, uri, params, algorithm=None):
         """Compute the signature for a given request
 
         :param uri: full URI that Twilio requested on your server
         :param params: post vars that Twilio sent with the request
+        :param str algorithm: HMAC_SHA1 (default) or HMAC_SHA256
 
         :returns: The computed signature
+        :raises ValueError: if the algorithm is not supported
         """
+        hash_function = _HASH_FUNCTIONS.get((algorithm or HMAC_SHA1).upper())
+        if hash_function is None:
+            raise ValueError("Unsupported webhook signature algorithm: %s" % algorithm)
+
         s = uri
         if params:
             for param_name in sorted(set(params)):
@@ -79,7 +120,7 @@ def compute_signature(self, uri, params):
                     s += param_name + value
 
         # compute signature and compare signatures
-        mac = hmac.new(self.token, s.encode("utf-8"), sha1)
+        mac = hmac.new(self.token, s.encode("utf-8"), hash_function)
         computed = base64.b64encode(mac.digest())
         computed = computed.decode("utf-8")
 
@@ -102,15 +143,28 @@ def compute_hash(self, body):
 
         return computed.strip()
 
-    def validate(self, uri, params, signature):
+    def validate(self, uri, params, signature, algorithm=None):
         """Validate a request from Twilio
 
         :param uri: full URI that Twilio requested on your server
         :param params: dictionary of POST variables or string of POST body for JSON requests
         :param signature: expected signature in HTTP X-Twilio-Signature header
+        :param str algorithm: HMAC_SHA1 (default) or HMAC_SHA256, see
+            :func:`get_signature_algorithm`. An unsupported algorithm fails validation
+            and logs a warning.
 
         :returns: True if the request passes validation, False if not
         """
+        if (algorithm or HMAC_SHA1).upper() not in _HASH_FUNCTIONS:
+            logger.warning(
+                "Unsupported webhook signature algorithm '%s'. Supported: %s, %s. "
+                "Upgrade twilio if Twilio added a new one.",
+                algorithm,
+                HMAC_SHA1,
+                HMAC_SHA256,
+            )
+            return False
+
         if params is None:
             params = {}
 
@@ -128,10 +182,10 @@ def validate(self, uri, params, signature):
         #  check signature of uri with and without port,
         #  since sig generation on back end is inconsistent
         valid_signature = compare(
-            self.compute_signature(uri_without_port, params), signature
+            self.compute_signature(uri_without_port, params, algorithm), signature
         )
         valid_signature_with_port = compare(
-            self.compute_signature(uri_with_port, params), signature
+            self.compute_signature(uri_with_port, params, algorithm), signature
         )
 
         return valid_body_hash and (valid_signature or valid_signature_with_port)