From 4b925bf2979346079eeaf74ea58cc4961d128914 Mon Sep 17 00:00:00 2001 From: kelson Date: Wed, 17 Aug 2016 16:08:05 -0400 Subject: [PATCH] added PKCS #7 v1.5 padding updated secure_dumps to use SHA256 fixed secure_dumps using truncated encryption_key hash fixed pad ignoring blocksize argument --- gluon/tests/test_utils.py | 37 ++++++++++++++++++++-- gluon/utils.py | 66 ++++++++++++++++++++++++++++++++++++--- 2 files changed, 95 insertions(+), 8 deletions(-) diff --git a/gluon/tests/test_utils.py b/gluon/tests/test_utils.py index 21db1d88..fde8d54f 100644 --- a/gluon/tests/test_utils.py +++ b/gluon/tests/test_utils.py @@ -10,7 +10,9 @@ fix_sys_path(__file__) from gluon.utils import md5_hash, compare, is_valid_ip_address, web2py_uuid +import pickle import hashlib +import gluon.utils from hashlib import md5, sha1, sha224, sha256, sha384, sha512 from gluon.utils import simple_hash, get_digest, secure_dumps, secure_loads, basestring @@ -85,7 +87,30 @@ class TestUtils(unittest.TestCase): # TODO: def test_get_callable_argspec(self): - # TODO: def test_pad(self): + def test_pad(self): + test_cases = [ + (16, b'mydata'), # verify data padding and unpad identity + (32, b'mydata '), # verify space is not stripped + (8, b'mydata\x01'), # verify "padding" bytes are ignored + (4, b'mydata'), # verify multiblock behavior + (2, b''), # verify empty string behavior + ] + for (testlen,teststr) in test_cases: + padded = gluon.utils.pad(teststr,testlen) + unpadded = gluon.utils.unpad(padded,testlen) + self.assertTrue(len(padded) > len(teststr)) + self.assertTrue(len(padded)%testlen == 0) + self.assertEqual(teststr, unpadded) + + testobj = {'a': 1, 'b': 2} + pickled = pickle.dumps(testobj) + padded = gluon.utils.pad(pickled) + unpadded = gluon.utils.unpad(padded) + unpickled = pickle.loads(unpadded) + self.assertEqual(pickled, unpadded) + self.assertEqual(testobj, unpickled) + self.assertTrue(len(padded) > len(pickled)) + self.assertTrue(len(padded)%32==0) def test_secure_dumps_and_loads(self): """ Tests secure_dumps and secure_loads""" @@ -95,7 +120,13 @@ class TestUtils(unittest.TestCase): original = secure_loads(secured, testkey) self.assertEqual(testobj, original) self.assertTrue(isinstance(secured, bytes)) - self.assertTrue(b':' in secured) + self.assertTrue(secured.count(b':') == 2) + + secured_deprecated = gluon.utils.secure_dumps_deprecated(testobj, testkey) + original_deprecated = secure_loads(secured_deprecated, testkey) + self.assertEqual(testobj, original_deprecated) + self.assertTrue(isinstance(secured_deprecated, bytes)) + self.assertTrue(secured_deprecated.count(b':') == 1) large_testobj = [x for x in range(1000)] secured_comp = secure_dumps(large_testobj, testkey, compression_level=9) @@ -115,7 +146,7 @@ class TestUtils(unittest.TestCase): self.assertEqual(wrong2, None) wrong3 = secure_loads(secured, 'wrongkey', 'wronghash') self.assertEqual(wrong3, None) - wrong4 = secure_loads('abc', 'a', 'b') + wrong4 = secure_loads(b'abc', 'a', 'b') self.assertEqual(wrong4, None) # TODO: def test_initialize_urandom(self): diff --git a/gluon/utils.py b/gluon/utils.py index 8589cd2f..a9e9f842 100644 --- a/gluon/utils.py +++ b/gluon/utils.py @@ -157,25 +157,81 @@ def get_callable_argspec(fn): return inspect.getargspec(inspectable) -def pad(s, n=32, padchar=b' '): - return s + (32 - len(s) % 32) * padchar +def pad(s, n=32): + # PKCS7v1.5 https://www.ietf.org/rfc/rfc2315.txt + padlen = n - len(s) % n + return s + bytes(bytearray(padlen * [padlen])) + + +def unpad(s, n=32): + padlen = s[-1] + if isinstance(padlen,str): + padlen = ord(padlen) # python2 + if (padlen < 1) | (padlen > n): # avoid short-circuit + # return garbage to minimize side channels + return bytes(bytearray(len(s)*[0])) + return s[:-padlen] def secure_dumps(data, encryption_key, hash_key=None, compression_level=None): + dump = pickle.dumps(data, pickle.HIGHEST_PROTOCOL) + if compression_level: + dump = zlib.compress(dump, compression_level) + encryption_key = to_bytes(encryption_key) + if not hash_key: + hash_key = hashlib.sha256(encryption_key).digest() + cipher, IV = AES_new(pad(encryption_key)[:32]) + encrypted_data = base64.urlsafe_b64encode(IV + cipher.encrypt(pad(dump))) + signature = to_bytes(hmac.new(to_bytes(hash_key), encrypted_data, hashlib.sha256).hexdigest()) + return b'hmac256:' + signature + b':' + encrypted_data + + +def secure_loads(data, encryption_key, hash_key=None, compression_level=None): + components = data.count(b':') + if components == 1: + return secure_loads_deprecated(data, encryption_key, hash_key, compression_level) + if components != 2: + return None + version,signature,encrypted_data = data.split(b':', 2) + if version != b'hmac256': + return None + encryption_key = to_bytes(encryption_key) + if not hash_key: + hash_key = hashlib.sha256(encryption_key).digest() + actual_signature = hmac.new(to_bytes(hash_key), encrypted_data, hashlib.sha256).hexdigest() + if not compare(to_native(signature), actual_signature): + return None + encrypted_data = base64.urlsafe_b64decode(encrypted_data) + IV, encrypted_data = encrypted_data[:16], encrypted_data[16:] + cipher, _ = AES_new(pad(encryption_key)[:32], IV=IV) + try: + data = unpad(cipher.decrypt(encrypted_data)) + if compression_level: + data = zlib.decompress(data) + return pickle.loads(data) + except Exception as e: + return None + + +def __pad_deprecated(s, n=32, padchar=b' '): + return s + (32 - len(s) % 32) * padchar + + +def secure_dumps_deprecated(data, encryption_key, hash_key=None, compression_level=None): encryption_key = to_bytes(encryption_key) if not hash_key: hash_key = sha1(encryption_key).hexdigest() dump = pickle.dumps(data, pickle.HIGHEST_PROTOCOL) if compression_level: dump = zlib.compress(dump, compression_level) - key = pad(encryption_key)[:32] + key = __pad_deprecated(encryption_key)[:32] cipher, IV = AES_new(key) encrypted_data = base64.urlsafe_b64encode(IV + cipher.encrypt(pad(dump))) signature = to_bytes(hmac.new(to_bytes(hash_key), encrypted_data, hashlib.md5).hexdigest()) return signature + b':' + encrypted_data -def secure_loads(data, encryption_key, hash_key=None, compression_level=None): +def secure_loads_deprecated(data, encryption_key, hash_key=None, compression_level=None): encryption_key = to_bytes(encryption_key) data = to_native(data) if ':' not in data: @@ -187,7 +243,7 @@ def secure_loads(data, encryption_key, hash_key=None, compression_level=None): actual_signature = hmac.new(to_bytes(hash_key), encrypted_data, hashlib.md5).hexdigest() if not compare(signature, actual_signature): return None - key = pad(encryption_key)[:32] + key = __pad_deprecated(encryption_key)[:32] encrypted_data = base64.urlsafe_b64decode(encrypted_data) IV, encrypted_data = encrypted_data[:16], encrypted_data[16:] cipher, _ = AES_new(key, IV=IV)