mirror of
https://github.com/Legrandin/pycryptodome.git
synced 2025-12-08 05:19:46 +00:00
Support for memoryview for SIV
This commit is contained in:
Helder Eijs
parent
53be0708d7
commit
6fd65d491f
3 changed files with 104 additions and 41 deletions
|
|
@ -36,7 +36,7 @@ __all__ = ['SivMode']
|
|||
|
||||
from binascii import hexlify
|
||||
|
||||
from Crypto.Util.py3compat import byte_string, bord, unhexlify, b, bstr
|
||||
from Crypto.Util.py3compat import byte_string, bord, unhexlify, _copy_bytes
|
||||
|
||||
from Crypto.Util.number import long_to_bytes, bytes_to_long
|
||||
from Crypto.Protocol.KDF import _S2V
|
||||
|
|
@ -93,7 +93,6 @@ class SivMode(object):
|
|||
|
||||
self._factory = factory
|
||||
|
||||
self._nonce = nonce
|
||||
self._cipher_params = kwargs
|
||||
|
||||
if len(key) not in (32, 48, 64):
|
||||
|
|
@ -106,7 +105,7 @@ class SivMode(object):
|
|||
if len(nonce) == 0:
|
||||
raise ValueError("When provided, the nonce must be non-empty")
|
||||
|
||||
self.nonce = bstr(nonce)
|
||||
self.nonce = _copy_bytes(None, None, nonce)
|
||||
"""Public attribute is only available in case of non-deterministic
|
||||
encryption."""
|
||||
|
||||
|
|
@ -128,12 +127,12 @@ class SivMode(object):
|
|||
def _create_ctr_cipher(self, mac_tag):
|
||||
"""Create a new CTR cipher from the MAC in SIV mode"""
|
||||
|
||||
tag_int = bytes_to_long(bstr(mac_tag))
|
||||
tag_int = bytes_to_long(mac_tag)
|
||||
return self._factory.new(
|
||||
self._subkey_cipher,
|
||||
self._factory.MODE_CTR,
|
||||
initial_value=tag_int ^ (tag_int & 0x8000000080000000L),
|
||||
nonce=b(""),
|
||||
nonce=b"",
|
||||
**self._cipher_params)
|
||||
|
||||
def update(self, component):
|
||||
|
|
@ -200,7 +199,7 @@ class SivMode(object):
|
|||
|
||||
self._next = [self.digest]
|
||||
|
||||
if self._nonce:
|
||||
if hasattr(self, 'nonce'):
|
||||
self._kdf.update(self.nonce)
|
||||
self._kdf.update(plaintext)
|
||||
|
||||
|
|
@ -347,7 +346,7 @@ class SivMode(object):
|
|||
|
||||
plaintext = self._cipher.decrypt(ciphertext)
|
||||
|
||||
if self._nonce:
|
||||
if hasattr(self, 'nonce'):
|
||||
self._kdf.update(self.nonce)
|
||||
if plaintext:
|
||||
self._kdf.update(plaintext)
|
||||
|
|
|
|||
|
|
@ -23,7 +23,7 @@
|
|||
|
||||
import struct
|
||||
|
||||
from Crypto.Util.py3compat import *
|
||||
from Crypto.Util.py3compat import tobytes, bord, _copy_bytes
|
||||
|
||||
from Crypto.Hash import SHA1, SHA256, HMAC, CMAC
|
||||
from Crypto.Util.strxor import strxor
|
||||
|
|
@ -144,7 +144,7 @@ def PBKDF2(password, salt, dkLen=16, count=1000, prf=None, hmac_hash_module=None
|
|||
s[0], s[1] = s[1], prf(password, s[1])
|
||||
return s[0]
|
||||
|
||||
key = b('')
|
||||
key = b''
|
||||
i = 1
|
||||
while len(key)<dkLen:
|
||||
s = [ prf(password, salt + struct.pack(">I", i)) ] * 2
|
||||
|
|
@ -153,10 +153,10 @@ def PBKDF2(password, salt, dkLen=16, count=1000, prf=None, hmac_hash_module=None
|
|||
|
||||
else:
|
||||
# Optimized implementation
|
||||
key = b('')
|
||||
key = b''
|
||||
i = 1
|
||||
while len(key)<dkLen:
|
||||
base = HMAC.new(password, b(""), hmac_hash_module)
|
||||
base = HMAC.new(password, b"", hmac_hash_module)
|
||||
first_digest = base.copy().update(salt + struct.pack(">I", i)).digest()
|
||||
key += base._pbkdf2_hmac_assist(first_digest, count)
|
||||
i += 1
|
||||
|
|
@ -186,10 +186,10 @@ class _S2V(object):
|
|||
A set of extra parameters to use to create a cipher instance.
|
||||
"""
|
||||
|
||||
self._key = key
|
||||
self._key = _copy_bytes(None, None, key)
|
||||
self._ciphermod = ciphermod
|
||||
self._last_string = self._cache = bchr(0)*ciphermod.block_size
|
||||
self._n_updates = ciphermod.block_size*8-1
|
||||
self._last_string = self._cache = b'\x00' * ciphermod.block_size
|
||||
self._n_updates = ciphermod.block_size * 8 - 1
|
||||
if cipher_params is None:
|
||||
self._cipher_params = {}
|
||||
else:
|
||||
|
|
@ -230,7 +230,7 @@ class _S2V(object):
|
|||
if not item:
|
||||
raise ValueError("A component cannot be empty")
|
||||
|
||||
if self._n_updates==0:
|
||||
if self._n_updates == 0:
|
||||
raise TypeError("Too many components passed to S2V")
|
||||
self._n_updates -= 1
|
||||
|
||||
|
|
@ -239,7 +239,7 @@ class _S2V(object):
|
|||
ciphermod=self._ciphermod,
|
||||
cipher_params=self._cipher_params)
|
||||
self._cache = strxor(self._double(self._cache), mac.digest())
|
||||
self._last_string = item
|
||||
self._last_string = _copy_bytes(None, None, item)
|
||||
|
||||
def derive(self):
|
||||
""""Derive a secret from the vector of components.
|
||||
|
|
@ -247,10 +247,10 @@ class _S2V(object):
|
|||
:Return: a byte string, as long as the block length of the cipher.
|
||||
"""
|
||||
|
||||
if len(self._last_string)>=16:
|
||||
if len(self._last_string) >= 16:
|
||||
final = self._last_string[:-16] + strxor(self._last_string[-16:], self._cache)
|
||||
else:
|
||||
padded = (self._last_string + bchr(0x80)+ bchr(0)*15)[:16]
|
||||
padded = (self._last_string + b'\x80' + b'\x00' * 15)[:16]
|
||||
final = strxor(padded, self._double(self._cache))
|
||||
mac = CMAC.new(self._key,
|
||||
msg=final,
|
||||
|
|
@ -301,24 +301,24 @@ def HKDF(master, key_len, salt, hashmod, num_keys=1, context=None):
|
|||
if output_len > (255 * hashmod.digest_size):
|
||||
raise ValueError("Too much secret data to derive")
|
||||
if not salt:
|
||||
salt = bchr(0) * hashmod.digest_size
|
||||
salt = b'\x00' * hashmod.digest_size
|
||||
if context is None:
|
||||
context = b("")
|
||||
context = b""
|
||||
|
||||
# Step 1: extract
|
||||
hmac = HMAC.new(salt, master, digestmod=hashmod)
|
||||
prk = hmac.digest()
|
||||
|
||||
# Step 2: expand
|
||||
t = [b("")]
|
||||
t = [ b"" ]
|
||||
n = 1
|
||||
tlen = 0
|
||||
while tlen < output_len:
|
||||
hmac = HMAC.new(prk, t[-1] + context + bchr(n), digestmod=hashmod)
|
||||
hmac = HMAC.new(prk, t[-1] + context + struct.pack('B', n), digestmod=hashmod)
|
||||
t.append(hmac.digest())
|
||||
tlen += hashmod.digest_size
|
||||
n += 1
|
||||
derived_output = b("").join(t)
|
||||
derived_output = b"".join(t)
|
||||
if num_keys == 1:
|
||||
return derived_output[:key_len]
|
||||
kol = [derived_output[idx:idx + key_len]
|
||||
|
|
@ -403,7 +403,7 @@ def scrypt(password, salt, key_len, N, r, p, num_keys=1):
|
|||
data_out += [ get_raw_buffer(buffer_out) ]
|
||||
|
||||
dk = PBKDF2(password,
|
||||
b("").join(data_out),
|
||||
b"".join(data_out),
|
||||
key_len * num_keys, 1,
|
||||
prf=prf_hmac_sha256)
|
||||
|
||||
|
|
|
|||
|
|
@ -31,7 +31,7 @@
|
|||
import unittest
|
||||
|
||||
from Crypto.SelfTest.st_common import list_test_cases
|
||||
from Crypto.Util.py3compat import unhexlify, tobytes, bchr, b
|
||||
from Crypto.Util.py3compat import unhexlify, tobytes, bchr, b, _memoryview
|
||||
from Crypto.Cipher import AES
|
||||
from Crypto.Hash import SHAKE128
|
||||
|
||||
|
|
@ -154,38 +154,102 @@ class SivTests(unittest.TestCase):
|
|||
cipher.hexverify(mac_hex)
|
||||
|
||||
def test_bytearray(self):
|
||||
|
||||
# Encrypt
|
||||
key = bytearray(self.key_256)
|
||||
nonce = bytearray(self.nonce_96)
|
||||
data = bytearray(self.data_128)
|
||||
header = bytearray(self.data_128)
|
||||
|
||||
cipher1 = AES.new(self.key_256,
|
||||
AES.MODE_SIV,
|
||||
nonce=self.nonce_96)
|
||||
cipher1.update(self.data_128)
|
||||
ref1 = cipher1.encrypt(self.data_128)
|
||||
ct, tag = cipher1.encrypt_and_digest(self.data_128)
|
||||
|
||||
cipher2 = AES.new(bytearray(self.key_256),
|
||||
cipher2 = AES.new(key,
|
||||
AES.MODE_SIV,
|
||||
nonce=bytearray(self.nonce_96))
|
||||
cipher2.update(bytearray(self.data_128))
|
||||
ref2 = cipher2.encrypt(bytearray(self.data_128))
|
||||
nonce=nonce)
|
||||
key[:3] = b'\xFF\xFF\xFF'
|
||||
nonce[:3] = b'\xFF\xFF\xFF'
|
||||
cipher2.update(header)
|
||||
header[:3] = b'\xFF\xFF\xFF'
|
||||
ct_test = cipher2.encrypt(data)
|
||||
data[:3] = b'\xFF\xFF\xFF'
|
||||
tag_test = cipher2.digest()
|
||||
|
||||
self.assertEqual(ref1, ref2)
|
||||
self.assertEqual(ct, ct_test)
|
||||
self.assertEqual(tag, tag_test)
|
||||
self.assertEqual(cipher1.nonce, cipher2.nonce)
|
||||
|
||||
tag = cipher1.digest()
|
||||
|
||||
# Decrypt
|
||||
cipher3 = AES.new(self.key_256,
|
||||
key = bytearray(self.key_256)
|
||||
nonce = bytearray(self.nonce_96)
|
||||
header = bytearray(self.data_128)
|
||||
ct_ba = bytearray(ct)
|
||||
tag_ba = bytearray(tag)
|
||||
|
||||
cipher3 = AES.new(key,
|
||||
AES.MODE_SIV,
|
||||
nonce=nonce)
|
||||
key[:3] = b'\xFF\xFF\xFF'
|
||||
nonce[:3] = b'\xFF\xFF\xFF'
|
||||
cipher3.update(header)
|
||||
header[:3] = b'\xFF\xFF\xFF'
|
||||
pt_test = cipher3.decrypt_and_verify(ct_ba, tag_ba)
|
||||
|
||||
self.assertEqual(self.data_128, pt_test)
|
||||
|
||||
def test_memoryview(self):
|
||||
|
||||
# Encrypt
|
||||
key = memoryview(bytearray(self.key_256))
|
||||
nonce = memoryview(bytearray(self.nonce_96))
|
||||
data = memoryview(bytearray(self.data_128))
|
||||
header = memoryview(bytearray(self.data_128))
|
||||
|
||||
cipher1 = AES.new(self.key_256,
|
||||
AES.MODE_SIV,
|
||||
nonce=self.nonce_96)
|
||||
cipher3.update(self.data_128)
|
||||
ref3 = cipher3.decrypt_and_verify(ref1, tag)
|
||||
cipher1.update(self.data_128)
|
||||
ct, tag = cipher1.encrypt_and_digest(self.data_128)
|
||||
|
||||
cipher4 = AES.new(bytearray(self.key_256),
|
||||
cipher2 = AES.new(key,
|
||||
AES.MODE_SIV,
|
||||
nonce=bytearray(self.nonce_96))
|
||||
cipher4.update(bytearray(self.data_128))
|
||||
ref4 = cipher4.decrypt_and_verify(bytearray(ref1), bytearray(tag))
|
||||
nonce=nonce)
|
||||
key[:3] = b'\xFF\xFF\xFF'
|
||||
nonce[:3] = b'\xFF\xFF\xFF'
|
||||
cipher2.update(header)
|
||||
header[:3] = b'\xFF\xFF\xFF'
|
||||
ct_test = cipher2.encrypt(data)
|
||||
data[:3] = b'\xFF\xFF\xFF'
|
||||
tag_test = cipher2.digest()
|
||||
|
||||
self.assertEqual(ref3, ref4)
|
||||
self.assertEqual(ct, ct_test)
|
||||
self.assertEqual(tag, tag_test)
|
||||
self.assertEqual(cipher1.nonce, cipher2.nonce)
|
||||
|
||||
# Decrypt
|
||||
key = memoryview(bytearray(self.key_256))
|
||||
nonce = memoryview(bytearray(self.nonce_96))
|
||||
header = memoryview(bytearray(self.data_128))
|
||||
ct_ba = memoryview(bytearray(ct))
|
||||
tag_ba = memoryview(bytearray(tag))
|
||||
|
||||
cipher3 = AES.new(key,
|
||||
AES.MODE_SIV,
|
||||
nonce=nonce)
|
||||
key[:3] = b'\xFF\xFF\xFF'
|
||||
nonce[:3] = b'\xFF\xFF\xFF'
|
||||
cipher3.update(header)
|
||||
header[:3] = b'\xFF\xFF\xFF'
|
||||
pt_test = cipher3.decrypt_and_verify(ct_ba, tag_ba)
|
||||
|
||||
self.assertEqual(self.data_128, pt_test)
|
||||
|
||||
import types
|
||||
if _memoryview is types.NoneType:
|
||||
del test_memoryview
|
||||
|
||||
|
||||
class SivFSMTests(unittest.TestCase):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue