Use hmac.compare_digest to compare signature

hmac.compare_digest must be used to compare sample
signature to avoid timing-attack.

We also provide a best-effort version of hmac.compare_digest
for python version that doesn't have the C version.

Closes bug #1332390

Change-Id: Ia313c98b7cccf0e8aed4fb933292bd7e6141b801

ceilometer/publisher/utils.py

@@ -60,15 +60,52 @@ def compute_signature(message, secret):
     return digest_maker.hexdigest()
 
 
+def besteffort_compare_digest(first, second):
+    """Returns True if both string inputs are equal, otherwise False.
+
+    This function should take a constant amount of time regardless of
+    how many characters in the strings match.
+
+    """
+    # NOTE(sileht): compare_digest method protected for timing-attacks
+    # exists since python >= 2.7.7 and python >= 3.3
+    # this a bit less-secure python fallback version
+    # taken from https://github.com/openstack/python-keystoneclient/blob/
+    # master/keystoneclient/middleware/memcache_crypt.py#L88
+    if len(first) != len(second):
+        return False
+    result = 0
+    if six.PY3 and isinstance(first, bytes) and isinstance(second, bytes):
+        for x, y in zip(first, second):
+            result |= x ^ y
+    else:
+        for x, y in zip(first, second):
+            result |= ord(x) ^ ord(y)
+    return result == 0
+
+
+if hasattr(hmac, 'compare_digest'):
+    compare_digest = hmac.compare_digest
+else:
+    compare_digest = besteffort_compare_digest
+
+
 def verify_signature(message, secret):
     """Check the signature in the message.
 
     Message is verified against the value computed from the rest of the
     contents.
     """
-    old_sig = message.get('message_signature')
+    old_sig = message.get('message_signature', '')
     new_sig = compute_signature(message, secret)
-    return new_sig == old_sig
+
+    if isinstance(old_sig, six.text_type):
+        try:
+            old_sig = old_sig.encode('ascii')
+        except UnicodeDecodeError:
+            return False
+
+    return compare_digest(new_sig, old_sig)
 
 
 def meter_message_from_counter(sample, secret):

ceilometer/tests/publisher/test_utils.py

@@ -74,6 +74,16 @@ class TestSignature(test.BaseTestCase):
                 'message_signature': 'Not the same'}
         self.assertFalse(utils.verify_signature(data, 'not-so-secret'))
 
+    def test_verify_signature_invalid_encoding(self):
+        data = {'a': 'A', 'b': 'B',
+                'message_signature': ''}
+        self.assertFalse(utils.verify_signature(data, 'not-so-secret'))
+
+    def test_verify_signature_unicode(self):
+        data = {'a': 'A', 'b': 'B',
+                'message_signature': u''}
+        self.assertFalse(utils.verify_signature(data, 'not-so-secret'))
+
     def test_verify_signature_nested(self):
         data = {'a': 'A',
                 'b': 'B',
@@ -100,3 +110,16 @@ class TestSignature(test.BaseTestCase):
             'not-so-secret')
         jsondata = jsonutils.loads(jsonutils.dumps(data))
         self.assertTrue(utils.verify_signature(jsondata, 'not-so-secret'))
+
+    def test_besteffort_compare_digest(self):
+        hash1 = "f5ac3fe42b80b80f979825d177191bc5"
+        hash2 = "f5ac3fe42b80b80f979825d177191bc5"
+        hash3 = "1dece7821bf3fd70fe1309eaa37d52a2"
+        hash4 = b"f5ac3fe42b80b80f979825d177191bc5"
+        hash5 = b"f5ac3fe42b80b80f979825d177191bc5"
+        hash6 = b"1dece7821bf3fd70fe1309eaa37d52a2"
+
+        self.assertTrue(utils.besteffort_compare_digest(hash1, hash2))
+        self.assertFalse(utils.besteffort_compare_digest(hash1, hash3))
+        self.assertTrue(utils.besteffort_compare_digest(hash4, hash5))
+        self.assertFalse(utils.besteffort_compare_digest(hash4, hash6))