diff --git a/tests/unit/test_request_validator.py b/tests/unit/test_request_validator.py index 2159d4c91..6da183d62 100644 --- a/tests/unit/test_request_validator.py +++ b/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" diff --git a/twilio/request_validator.py b/twilio/request_validator.py index f7a0db361..682dbf414 100644 --- a/twilio/request_validator.py +++ b/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=,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)