Merge pull request #1428 from ShySec/pad_update
secure_dumps/secure_loads/pad update
This commit is contained in:
@@ -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):
|
||||
|
||||
+61
-5
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user