Added Wycheproof tests for AES-SIV

Changelog.rst

@@ -13,6 +13,8 @@ Resolved issues
 ---------------
 
 * In certain circumstances (counter wrapping) AES GCM produced wrong ciphertexts.
+* Method ``encrypt()`` of AES SIV cipher could be called, whereas only ``encrypt_and_digest()``
+  should have been allowed.
 
 3.6.0 (8 April 2018)
 ++++++++++++++++++++

lib/Crypto/Cipher/_mode_siv.py

@@ -124,14 +124,15 @@ class SivMode(object):
         self._next = [self.update, self.encrypt, self.decrypt,
                       self.digest, self.verify]
 
-    def _create_ctr_cipher(self, mac_tag):
+    def _create_ctr_cipher(self, v):
-        """Create a new CTR cipher from the MAC in SIV mode"""
+        """Create a new CTR cipher from V in SIV mode"""
 
-        tag_int = bytes_to_long(mac_tag)
+        v_int = bytes_to_long(v)
+        q = v_int & 0xFFFFFFFFFFFFFFFF7FFFFFFF7FFFFFFF
         return self._factory.new(
                     self._subkey_cipher,
                     self._factory.MODE_CTR,
-                    initial_value=tag_int ^ (tag_int & 0x8000000080000000L),
+                    initial_value=q,
                     nonce=b"",
                     **self._cipher_params)
 
@@ -158,7 +159,7 @@ class SivMode(object):
 
         :Parameters:
           component : bytes/bytearray/memoryview
-            The next associated data component. It must not be empty.
+            The next associated data component.
         """
 
         if self.update not in self._next:
@@ -171,46 +172,18 @@ class SivMode(object):
         return self._kdf.update(component)
 
     def encrypt(self, plaintext):
-        """Encrypt data with the key and the parameters set at initialization.
+        """
+        For SIV, encryption and MAC authentication must take place at the same
+        point. This method shall not be used.
 
-        A cipher object is stateful: once you have encrypted a message
+        Use `encrypt_and_digest` instead.
-        you cannot encrypt (or decrypt) another message using the same
-        object.
-
-        This method can be called only **once**.
-
-        You cannot reuse an object for encrypting
-        or decrypting other data with the same key.
-
-        This function does not add any padding to the plaintext.
-
-        :Parameters:
-          plaintext : bytes/bytearray/memoryview
-            The piece of data to encrypt.
-            It can be of any length, but it cannot be empty.
-        :Return:
-            the encrypted data, as a byte string.
-            It is as long as *plaintext*.
         """
 
-        if self.encrypt not in self._next:
+        raise TypeError("encrypt() not allowed for SIV mode."
-            raise TypeError("encrypt() can only be called after"
+                        " Use encrypt_and_digest() instead.")
-                            " initialization or an update()")
-
-        self._next = [self.digest]
-
-        if hasattr(self, 'nonce'):
-            self._kdf.update(self.nonce)
-        self._kdf.update(plaintext)
-
-        self._mac_tag = self._kdf.derive()
-        cipher = self._create_ctr_cipher(self._mac_tag)
-
-        return cipher.encrypt(plaintext)
 
     def decrypt(self, ciphertext):
-        """Decrypt data with the key and the parameters set at initialization.
+        """
-
         For SIV, decryption and verification must take place at the same
         point. This method shall not be used.
 
@@ -308,8 +281,22 @@ class SivMode(object):
             - the encrypted data
             - the MAC
         """
+        
+        if self.encrypt not in self._next:
+            raise TypeError("encrypt() can only be called after"
+                            " initialization or an update()")
 
-        return self.encrypt(plaintext), self.digest()
+        self._next = [ self.digest ]
+
+        # Compute V (MAC)
+        if hasattr(self, 'nonce'):
+            self._kdf.update(self.nonce)
+        self._kdf.update(plaintext)
+        self._mac_tag = self._kdf.derive()
+        
+        cipher = self._create_ctr_cipher(self._mac_tag)
+
+        return cipher.encrypt(plaintext), self._mac_tag
 
     def decrypt_and_verify(self, ciphertext, mac_tag):
         """Perform decryption and verification in one step.
@@ -339,7 +326,7 @@ class SivMode(object):
         if self.decrypt not in self._next:
             raise TypeError("decrypt() can only be called"
                             " after initialization or an update()")
-        self._next = [self.verify]
+        self._next = [ self.verify ]
 
         # Take the MAC and start the cipher for decryption
         self._cipher = self._create_ctr_cipher(mac_tag)
@@ -348,10 +335,9 @@ class SivMode(object):
 
         if hasattr(self, 'nonce'):
             self._kdf.update(self.nonce)
-        if plaintext:
+        self._kdf.update(plaintext)
-            self._kdf.update(plaintext)
-
         self.verify(mac_tag)
+        
         return plaintext
 
 

lib/Crypto/Protocol/KDF.py

@@ -189,7 +189,10 @@ class _S2V(object):
         self._key = _copy_bytes(None, None, key)
         self._ciphermod = ciphermod
         self._last_string = self._cache = b'\x00' * ciphermod.block_size
+        
+        # Max number of update() call we can process
         self._n_updates = ciphermod.block_size * 8 - 1
+        
         if cipher_params is None:
             self._cipher_params = {}
         else:
@@ -224,12 +227,8 @@ class _S2V(object):
           item : byte string
             The next component of the vector.
         :Raise TypeError: when the limit on the number of components has been reached.
-        :Raise ValueError: when the component is empty
         """
 
-        if not item:
-            raise ValueError("A component cannot be empty")
-
         if self._n_updates == 0:
             raise TypeError("Too many components passed to S2V")
         self._n_updates -= 1
@@ -248,8 +247,10 @@ class _S2V(object):
         """
 
         if len(self._last_string) >= 16:
+            # xorend
             final = self._last_string[:-16] + strxor(self._last_string[-16:], self._cache)
         else:
+            # zero-pad & xor
             padded = (self._last_string + b'\x80' + b'\x00' * 15)[:16]
             final = strxor(padded, self._double(self._cache))
         mac = CMAC.new(self._key,

lib/Crypto/SelfTest/Cipher/test_GCM.py

@@ -36,7 +36,6 @@ from Crypto.Util.py3compat import unhexlify, tobytes, bchr, b, _memoryview
 from Crypto.Cipher import AES
 from Crypto.Hash import SHAKE128
 
-
 from Crypto.Util._file_system import pycryptodome_filename
 from Crypto.Util.strxor import strxor
 

lib/Crypto/SelfTest/Cipher/test_SIV.py

@@ -28,6 +28,7 @@
 # POSSIBILITY OF SUCH DAMAGE.
 # ===================================================================
 
+import json
 import unittest
 
 from Crypto.SelfTest.st_common import list_test_cases
@@ -35,6 +36,9 @@ from Crypto.Util.py3compat import unhexlify, tobytes, bchr, b, _memoryview
 from Crypto.Cipher import AES
 from Crypto.Hash import SHAKE128
 
+from Crypto.Util._file_system import pycryptodome_filename
+from Crypto.Util.strxor import strxor
+
 
 def get_tag_random(tag, length):
     return SHAKE128.new(data=tobytes(tag)).read(length)
@@ -63,10 +67,11 @@ class SivTests(unittest.TestCase):
         AES.new(self.key_256, AES.MODE_SIV)
 
         cipher = AES.new(self.key_256, AES.MODE_SIV, self.nonce_96)
-        ct = cipher.encrypt(self.data_128)
+        ct1, tag1 = cipher.encrypt_and_digest(self.data_128)
 
         cipher = AES.new(self.key_256, AES.MODE_SIV, nonce=self.nonce_96)
-        self.assertEquals(ct, cipher.encrypt(self.data_128))
+        ct2, tag2 = cipher.encrypt_and_digest(self.data_128)
+        self.assertEquals(ct1 + tag1, ct2 + tag2)
 
     def test_nonce_must_be_bytes(self):
         self.assertRaises(TypeError, AES.new, self.key_256, AES.MODE_SIV,
@@ -79,7 +84,7 @@ class SivTests(unittest.TestCase):
 
         for x in range(1, 128):
             cipher = AES.new(self.key_256, AES.MODE_SIV, nonce=bchr(1) * x)
-            cipher.encrypt(bchr(1))
+            cipher.encrypt_and_digest(b'\x01')
 
     def test_block_size_128(self):
         cipher = AES.new(self.key_256, AES.MODE_SIV, nonce=self.nonce_96)
@@ -103,21 +108,13 @@ class SivTests(unittest.TestCase):
         AES.new(self.key_256, AES.MODE_SIV, nonce=self.nonce_96,
                 use_aesni=False)
 
-    def test_invalid_null_encryption(self):
-        cipher = AES.new(self.key_256, AES.MODE_SIV, nonce=self.nonce_96)
-        self.assertRaises(ValueError, cipher.encrypt, b(""))
-
-    def test_invalid_null_component(self):
-        cipher = AES.new(self.key_256, AES.MODE_SIV, nonce=self.nonce_96)
-        self.assertRaises(ValueError, cipher.update, b(""))
-
     def test_encrypt_excludes_decrypt(self):
         cipher = AES.new(self.key_256, AES.MODE_SIV, nonce=self.nonce_96)
-        cipher.encrypt(self.data_128)
+        cipher.encrypt_and_digest(self.data_128)
         self.assertRaises(TypeError, cipher.decrypt, self.data_128)
 
         cipher = AES.new(self.key_256, AES.MODE_SIV, nonce=self.nonce_96)
-        cipher.encrypt(self.data_128)
+        cipher.encrypt_and_digest(self.data_128)
         self.assertRaises(TypeError, cipher.decrypt_and_verify,
                           self.data_128, self.data_128)
 
@@ -174,9 +171,7 @@ class SivTests(unittest.TestCase):
         nonce[:3] = b'\xFF\xFF\xFF'
         cipher2.update(header)
         header[:3] = b'\xFF\xFF\xFF'
-        ct_test = cipher2.encrypt(data)
+        ct_test, tag_test = cipher2.encrypt_and_digest(data)
-        data[:3] = b'\xFF\xFF\xFF'
-        tag_test = cipher2.digest()
 
         self.assertEqual(ct, ct_test)
         self.assertEqual(tag, tag_test)
@@ -221,9 +216,7 @@ class SivTests(unittest.TestCase):
         nonce[:3] = b'\xFF\xFF\xFF'
         cipher2.update(header)
         header[:3] = b'\xFF\xFF\xFF'
-        ct_test = cipher2.encrypt(data)
+        ct_test, tag_test= cipher2.encrypt_and_digest(data)
-        data[:3] = b'\xFF\xFF\xFF'
-        tag_test = cipher2.digest()
 
         self.assertEqual(ct, ct_test)
         self.assertEqual(tag, tag_test)
@@ -257,19 +250,12 @@ class SivFSMTests(unittest.TestCase):
     key_256 = get_tag_random("key_256", 32)
     nonce_96 = get_tag_random("nonce_96", 12)
     data_128 = get_tag_random("data_128", 16)
-
+    
-    def test_valid_init_encrypt_decrypt_verify(self):
+    def test_invalid_init_encrypt(self):
-        # No authenticated data, fixed plaintext
+        # Path INIT->ENCRYPT fails
-        # Verify path INIT->ENCRYPT->DIGEST
         cipher = AES.new(self.key_256, AES.MODE_SIV,
                          nonce=self.nonce_96)
-        ct = cipher.encrypt(self.data_128)
+        self.assertRaises(TypeError, cipher.encrypt, b("xxx"))
-        mac = cipher.digest()
-
-        # Verify path INIT->DECRYPT_AND_VERIFY
-        cipher = AES.new(self.key_256, AES.MODE_SIV,
-                         nonce=self.nonce_96)
-        cipher.decrypt_and_verify(ct, mac)
 
     def test_invalid_init_decrypt(self):
         # Path INIT->DECRYPT fails
@@ -291,21 +277,6 @@ class SivFSMTests(unittest.TestCase):
         cipher.update(self.data_128)
         cipher.verify(mac)
 
-    def test_valid_full_path(self):
-        # Fixed authenticated data, fixed plaintext
-        # Verify path INIT->UPDATE->ENCRYPT->DIGEST
-        cipher = AES.new(self.key_256, AES.MODE_SIV,
-                         nonce=self.nonce_96)
-        cipher.update(self.data_128)
-        ct = cipher.encrypt(self.data_128)
-        mac = cipher.digest()
-
-        # Verify path INIT->UPDATE->DECRYPT_AND_VERIFY
-        cipher = AES.new(self.key_256, AES.MODE_SIV,
-                         nonce=self.nonce_96)
-        cipher.update(self.data_128)
-        cipher.decrypt_and_verify(ct, mac)
-
     def test_valid_init_digest(self):
         # Verify path INIT->DIGEST
         cipher = AES.new(self.key_256, AES.MODE_SIV, nonce=self.nonce_96)
@@ -319,18 +290,6 @@ class SivFSMTests(unittest.TestCase):
         cipher = AES.new(self.key_256, AES.MODE_SIV, nonce=self.nonce_96)
         cipher.verify(mac)
 
-    def test_invalid_multiple_encrypt(self):
-        # Without AAD
-        cipher = AES.new(self.key_256, AES.MODE_SIV, nonce=self.nonce_96)
-        cipher.encrypt(b("xxx"))
-        self.assertRaises(TypeError, cipher.encrypt, b("xxx"))
-
-        # With AAD
-        cipher = AES.new(self.key_256, AES.MODE_SIV, nonce=self.nonce_96)
-        cipher.update(b("yyy"))
-        cipher.encrypt(b("xxx"))
-        self.assertRaises(TypeError, cipher.encrypt, b("xxx"))
-
     def test_valid_multiple_digest_or_verify(self):
         # Multiple calls to digest
         cipher = AES.new(self.key_256, AES.MODE_SIV, nonce=self.nonce_96)
@@ -357,27 +316,18 @@ class SivFSMTests(unittest.TestCase):
         pt = cipher.decrypt_and_verify(ct, mac)
         self.assertEqual(self.data_128, pt)
 
-    def test_invalid_encrypt_or_update_after_digest(self):
+    def test_invalid_multiple_encrypt_and_digest(self):
-        for method_name in "encrypt", "update":
-            cipher = AES.new(self.key_256, AES.MODE_SIV, nonce=self.nonce_96)
-            cipher.encrypt(self.data_128)
-            cipher.digest()
-            self.assertRaises(TypeError, getattr(cipher, method_name),
-                              self.data_128)
-
-            cipher = AES.new(self.key_256, AES.MODE_SIV, nonce=self.nonce_96)
-            cipher.encrypt_and_digest(self.data_128)
-
-    def test_invalid_decrypt_or_update_after_verify(self):
         cipher = AES.new(self.key_256, AES.MODE_SIV, nonce=self.nonce_96)
-        ct = cipher.encrypt(self.data_128)
+        ct, tag = cipher.encrypt_and_digest(self.data_128)
-        mac = cipher.digest()
+        self.assertRaises(TypeError, cipher.encrypt_and_digest, b'')
 
-        for method_name in "decrypt", "update":
+    def test_invalid_multiple_decrypt_and_verify(self):
-            cipher = AES.new(self.key_256, AES.MODE_SIV, nonce=self.nonce_96)
+        cipher = AES.new(self.key_256, AES.MODE_SIV, nonce=self.nonce_96)
-            cipher.decrypt_and_verify(ct, mac)
+        ct, tag = cipher.encrypt_and_digest(self.data_128)
-            self.assertRaises(TypeError, getattr(cipher, method_name),
+
-                              self.data_128)
+        cipher = AES.new(self.key_256, AES.MODE_SIV, nonce=self.nonce_96)
+        cipher.decrypt_and_verify(ct, tag)
+        self.assertRaises(TypeError, cipher.decrypt_and_verify, ct, tag)
 
 
 class TestVectors(unittest.TestCase):
@@ -444,11 +394,71 @@ class TestVectors(unittest.TestCase):
             self.assertEqual(pt, pt2)
 
 
+class TestVectorsWycheproof(unittest.TestCase):
+
+    def __init__(self):
+        unittest.TestCase.__init__(self)
+
+    def setUp(self):
+        file_in = open(pycryptodome_filename(
+                        "Crypto.SelfTest.Cipher.test_vectors.wycheproof".split("."),
+                        "aes_siv_cmac_test.json"), "rt")
+        tv_tree = json.load(file_in)
+
+        class TestVector(object):
+            pass
+        self.tv = []
+
+        for group in tv_tree['testGroups']:
+            for test in group['tests']:
+                tv = TestVector()
+
+                tv.id = test['tcId']
+                for attr in 'key', 'aad', 'msg', 'ct':
+                    setattr(tv, attr, unhexlify(test[attr]))
+                tv.valid = test['result'] != "invalid"
+                self.tv.append(tv)
+
+    def shortDescription(self):
+        return self._id
+
+    def test_encrypt(self, tv):
+        self._id = "Wycheproof Encrypt AES-SIV Test #" + str(tv.id)
+
+        cipher = AES.new(tv.key, AES.MODE_SIV)
+        cipher.update(tv.aad)
+        ct, tag = cipher.encrypt_and_digest(tv.msg)
+        if tv.valid:
+            self.assertEqual(tag + ct, tv.ct)
+
+    def test_decrypt(self, tv):
+        self._id = "Wycheproof Decrypt AES_SIV Test #" + str(tv.id)
+
+        cipher = AES.new(tv.key, AES.MODE_SIV)
+        cipher.update(tv.aad)
+        try:
+            pt = cipher.decrypt_and_verify(tv.ct[16:], tv.ct[:16])
+        except ValueError:
+            assert not tv.valid
+        else:
+            assert tv.valid
+            self.assertEqual(pt, tv.msg)
+
+    def runTest(self):
+
+        for tv in self.tv:
+            self.test_encrypt(tv)
+            self.test_decrypt(tv)
+
+
 def get_tests(config={}):
+    wycheproof_warnings = config.get('wycheproof_warnings')
+
     tests = []
     tests += list_test_cases(SivTests)
     tests += list_test_cases(SivFSMTests)
-    tests += [TestVectors()]
+    tests += [ TestVectors() ]
+    tests += [ TestVectorsWycheproof() ]
     return tests