Add request body validation (#436)

Author: cjcodes

tests/unit/test_request_validator.py

@@ -10,58 +10,57 @@
 class ValidationTest(unittest.TestCase):
 
     def setUp(self):
-        token = "1c892n40nd03kdnc0112slzkl3091j20"
+        token = "12345"
         self.validator = RequestValidator(token)
 
-        self.uri = "http://www.postbin.org/1ed898x"
+        self.uri = "https://mycompany.com/myapp.php?foo=1&bar=2"
         self.params = {
-            "AccountSid": "AC9a9f9392lad99kla0sklakjs90j092j3",
-            "ApiVersion": "2010-04-01",
-            "CallSid": "CAd800bb12c0426a7ea4230e492fef2a4f",
-            "CallStatus": "ringing",
-            "Called": "+15306384866",
-            "CalledCity": "OAKLAND",
-            "CalledCountry": "US",
-            "CalledState": "CA",
-            "CalledZip": "94612",
-            "Caller": "+15306666666",
-            "CallerCity": "SOUTH LAKE TAHOE",
-            "CallerCountry": "US",
-            "CallerName": "CA Wireless Call",
-            "CallerState": "CA",
-            "CallerZip": "89449",
-            "Direction": "inbound",
-            "From": "+15306666666",
-            "FromCity": "SOUTH LAKE TAHOE",
-            "FromCountry": "US",
-            "FromState": "CA",
-            "FromZip": "89449",
-            "To": "+15306384866",
-            "ToCity": "OAKLAND",
-            "ToCountry": "US",
-            "ToState": "CA",
-            "ToZip": "94612",
+            "CallSid": "CA1234567890ABCDE",
+            "Digits": "1234",
+            "From": "+14158675309",
+            "To": "+18005551212",
+            "Caller": "+14158675309",
         }
+        self.expected = "RSOYDt4T1cUTdK1PDd93/VVr8B8="
+        self.body = "{\"property\": \"value\", \"boolean\": true}"
+        self.bodyHash = "Ch/3Y02as7ldtcmi3+lBbkFQKyg6gMfPGWMmMvluZiA="
+        self.encodedBodyHash = self.bodyHash.replace("+", "%2B").replace("=", "%3D")
+        self.uriWithBody = self.uri + "&bodySHA256=" + self.encodedBodyHash
 
     def test_compute_signature_bytecode(self):
-        expected = b("fF+xx6dTinOaCdZ0aIeNkHr/ZAA=")
+        expected = b(self.expected)
         signature = self.validator.compute_signature(self.uri,
                                                      self.params,
                                                      utf=False)
         assert_equal(signature, expected)
 
     def test_compute_signature_unicode(self):
-        expected = u("fF+xx6dTinOaCdZ0aIeNkHr/ZAA=")
+        expected = u(self.expected)
         signature = self.validator.compute_signature(self.uri,
                                                      self.params,
                                                      utf=True)
         assert_equal(signature, expected)
 
+    def test_compute_hash_bytecode(self):
+        expected = b(self.bodyHash)
+        body_hash = self.validator.compute_hash(self.body, utf=False)
+
+        assert_equal(expected, body_hash)
+
+    def test_compute_hash_unicode(self):
+        expected = u(self.bodyHash)
+        body_hash = self.validator.compute_hash(self.body, utf=True)
+
+        assert_equal(expected, body_hash)
+
     def test_validation(self):
-        expected = "fF+xx6dTinOaCdZ0aIeNkHr/ZAA="
-        assert_true(self.validator.validate(self.uri, self.params, expected))
+        assert_true(self.validator.validate(self.uri, self.params, self.expected))
 
     def test_validation_removes_port_on_https(self):
-        self.uri = "https://www.postbin.org:1234/1ed898x"
-        expected = "Y7MeICc5ECftd1G11Fc8qoxAn0A="
-        assert_true(self.validator.validate(self.uri, self.params, expected))
+        uri = self.uri.replace(".com", ".com:1234")
+        assert_true(self.validator.validate(uri, self.params, self.expected))
+
+    def test_validation_of_body_succeeds(self):
+        uri = self.uriWithBody
+        is_valid = self.validator.validate(uri, self.body, "afcFvPLPYT8mg/JyIVkdnqQKa2s=")
+        assert_true(is_valid)

twilio/.request_validator.py.swp

@@ -1,13 +1,13 @@
 # Those are not supported by the six library and needs to be done manually
 try:
     # python 3
-    from urllib.parse import urlencode, urlparse, urljoin, urlunparse
+    from urllib.parse import urlencode, urlparse, urljoin, urlunparse, parse_qs
 except ImportError:
     # python 2 backward compatibility
     # noinspection PyUnresolvedReferences
     from urllib import urlencode
     # noinspection PyUnresolvedReferences
-    from urlparse import urlparse, urljoin, urlunparse
+    from urlparse import urlparse, urljoin, urlunparse, parse_qs
 
 try:
     # python 2

twilio/compat.py

@@ -1,10 +1,10 @@
 import base64
 import hmac
-from hashlib import sha1
+from hashlib import sha1, sha256
 
-from six import PY3
+from six import PY3, string_types
 
-from twilio.compat import izip, urlparse
+from twilio.compat import izip, urlparse, parse_qs
 
 
 def compare(string1, string2):
@@ -64,16 +64,38 @@ def compute_signature(self, uri, params, utf=PY3):
 
         return computed.strip()
 
+    def compute_hash(self, body, utf=PY3):
+        computed = base64.b64encode(sha256(body.encode("utf-8")).digest())
+
+        if utf:
+            computed = computed.decode('utf-8')
+
+        return computed.strip()
+
     def validate(self, uri, params, signature):
         """Validate a request from Twilio
 
         :param uri: full URI that Twilio requested on your server
-        :param params: post vars that Twilio sent with the request
+        :param params: dictionary of POST variables or string of POST body for JSON requests
         :param signature: expected signature in HTTP X-Twilio-Signature header
 
         :returns: True if the request passes validation, False if not
         """
+        if params is None:
+            params = {}
+
         parsed_uri = urlparse(uri)
         if parsed_uri.scheme == "https" and parsed_uri.port:
             uri = remove_port(parsed_uri)
-        return compare(self.compute_signature(uri, params), signature)
+
+        valid_signature = False  # Default fail
+        valid_body_hash = True  # May not receive body hash, so default succeed
+
+        query = parse_qs(parsed_uri.query)
+        if "bodySHA256" in query and isinstance(params, string_types):
+            valid_body_hash = compare(self.compute_hash(params), query["bodySHA256"][0])
+            valid_signature = compare(self.compute_signature(uri, {}), signature)
+        else:
+            valid_signature = compare(self.compute_signature(uri, params), signature)
+
+        return valid_signature and valid_body_hash