Add support for SIV (Synthetic IV) mode

This patch add supports for SIV, an AEAD block cipher
mode defined in RFC5297. SIV is only valid for AES.

The PRF of SIV (S2V) is factored out in the Protocol.KDF module.

See the following example to get a feeling of the API (slightly
different than other AEAD mode, during decryption).

Encryption (Python 2):

	>>> from Crypto.Cipher import AES
	>>> key = b'0'*32
	>>> siv = AES.new(key, AES.MODE_SIV)
	>>> ct  = siv.encrypt(b'Message')
	>>> mac = siv.digest()

Decryption (Python 2):

	>>> from Crypto.Cipher import AES, MacMismatchError
	>>> key = b'0'*32
	>>> siv = AES.new(key, AES.MODE_SIV)
	>>> pt  = siv.decrypt(ct + mac)
	>>> try:
	>>>	siv.verify(mac)
	>>>	print "Plaintext", pt
	>>> except MacMismatchError:
	>>>     print "Error"

This change also fixes the description/design of AEAD API.

With SIV (RFC5297), decryption can only start when the MAC is known.
The original AEAD API did not support that.

For SIV the MAC is now exceptionally passed together with the ciphertext
to the decrypt() method.

[dlitz@dlitz.net: Included changes from the following commits from the author's pull request:]
- [9c13f9c] Rename 'IV' parameter to 'nonce' for AEAD modes.
- [d7727fb] Fix description/design of AEAD API.
- [fb62fae] ApiUsageError becomes TypeError [whitespace]
- [4ec64d8] Removed last references to ApiUsageError [whitespace]
- [ee46922] Removed most 'import *' statements
- [ca460a7] Made blockalgo.py more PEP-8 compliant;
            The second parameter of the _GHASH constructor
            is now the length of the block (block_size)
            and not the full module.
[dlitz@dlitz.net: A conflict that was not resolved in the previous
                  commit was originally resolved here.  Moved the
                  resolution to the previous commit.]
[dlitz@dlitz.net: Replaced MacMismatchError with ValueError]
[dlitz@dlitz.net: Replaced ApiUsageError with TypeError]
[dlitz@dlitz.net: Whitespace fixed with "git rebase --whitespace=fix"]

3 files changed, 147 insertions, 23 deletions

diff --git a/lib/Crypto/SelfTest/Cipher/common.py b/lib/Crypto/SelfTest/Cipher/common.py
index 603ab54..ed612b4 100644
--- a/lib/Crypto/SelfTest/Cipher/common.py
+++ b/lib/Crypto/SelfTest/Cipher/common.py
@@ -29,8 +29,12 @@ from __future__ import nested_scopes
 __revision__ = "$Id$"
 
 import sys
+if sys.version_info[0] == 2 and sys.version_info[1] == 1:
+    from Crypto.Util.py21compat import *
+
 import unittest
 from binascii import a2b_hex, b2a_hex, hexlify
+
 from Crypto.Util.py3compat import *
 from Crypto.Util.strxor import strxor_c
 
@@ -70,8 +74,6 @@ class CipherSelfTest(unittest.TestCase):
         self.ciphertext = b(_extract(params, 'ciphertext'))
         self.module_name = _extract(params, 'module_name', None)
         self.assoc_data = _extract(params, 'assoc_data', None)
-        if self.assoc_data:
-            self.assoc_data = b(self.assoc_data)
         self.mac = _extract(params, 'mac', None)
         if self.assoc_data:
             self.mac = b(self.mac)
@@ -130,12 +132,17 @@ class CipherSelfTest(unittest.TestCase):
             else:
                 return self.module.new(a2b_hex(self.key), self.mode, a2b_hex(self.iv), **params)
 
+    def isMode(self, name):
+        if not hasattr(self.module, "MODE_"+name):
+            return False
+        return self.mode == getattr(self.module, "MODE_"+name)
+
     def runTest(self):
         plaintext = a2b_hex(self.plaintext)
         ciphertext = a2b_hex(self.ciphertext)
-        assoc_data = None
+        assoc_data = []
         if self.assoc_data:
-            assoc_data = a2b_hex(self.assoc_data)
+            assoc_data = [ a2b_hex(b(x)) for x in self.assoc_data]
 
         ct = None
         pt = None
@@ -149,19 +156,22 @@ class CipherSelfTest(unittest.TestCase):
             decipher = self._new(1)
 
             # Only AEAD modes
-            if self.assoc_data:
-                cipher.update(assoc_data)
-                decipher.update(assoc_data)
+            for comp in assoc_data:
+                cipher.update(comp)
+                decipher.update(comp)
 
             ctX = b2a_hex(cipher.encrypt(plaintext))
-            ptX = b2a_hex(decipher.decrypt(ciphertext))
+            if self.isMode("SIV"):
+                ptX = b2a_hex(decipher.decrypt(ciphertext+a2b_hex(self.mac)))
+            else:
+                ptX = b2a_hex(decipher.decrypt(ciphertext))
 
             if ct:
                 self.assertEqual(ct, ctX)
                 self.assertEqual(pt, ptX)
             ct, pt = ctX, ptX
 
-        if hasattr(self.module, "MODE_OPENPGP") and self.mode == self.module.MODE_OPENPGP:
+        if self.isMode("OPENPGP"):
             # In PGP mode, data returned by the first encrypt()
             # is prefixed with the encrypted IV.
             # Here we check it and then remove it from the ciphertexts.
@@ -387,10 +397,18 @@ class AEADTests(unittest.TestCase):
         self.module = module
         self.mode_name = mode_name
         self.mode = getattr(module, mode_name)
-        self.key = b('\xFF')*key_size
+        if not self.isMode("SIV"):
+            self.key = b('\xFF')*key_size
+        else:
+            self.key = b('\xFF')*key_size*2
         self.iv = b('\x00')*10
         self.description = "AEAD Test"
 
+    def isMode(self, name):
+        if not hasattr(self.module, "MODE_"+name):
+            return False
+        return self.mode == getattr(self.module, "MODE_"+name)
+
     def right_mac_test(self):
         """Positive tests for MAC"""
 
@@ -409,7 +427,10 @@ class AEADTests(unittest.TestCase):
         # Decrypt and verify that MAC is accepted
         decipher = self.module.new(self.key, self.mode, self.iv)
         decipher.update(ad_ref)
-        pt = decipher.decrypt(ct_ref)
+        if not self.isMode("SIV"):
+            pt = decipher.decrypt(ct_ref)
+        else:
+            pt = decipher.decrypt(ct_ref+mac_ref)
         decipher.verify(mac_ref)
         self.assertEqual(pt, pt_ref)
 
@@ -452,6 +473,15 @@ class AEADTests(unittest.TestCase):
         self.description = "Test for multiple updates in %s of %s" % \
             (self.mode_name, self.module.__name__)
 
+        # In all modes other than SIV, the associated data is a single
+        # component that can be arbitrarilly split and submitted to update().
+        #
+        # In SIV, associated data is instead organized in a vector or multiple
+        # components. Each component is passed to update() as a whole.
+        # This test is therefore not meaningful to SIV.
+        if self.isMode("SIV"):
+            return
+
         ad = b("").join([bchr(x) for x in xrange(0,128)])
 
         mac1, mac2, mac3 = (None,)*3
@@ -489,23 +519,23 @@ class AEADTests(unittest.TestCase):
 
         # Calling decrypt after encrypt raises an exception
         cipher = self.module.new(self.key, self.mode, self.iv)
-        cipher.encrypt(b("PT"))
-        self.assertRaises(TypeError, cipher.decrypt, b("XYZ"))
+        cipher.encrypt(b("PT")*40)
+        self.assertRaises(TypeError, cipher.decrypt, b("XYZ")*40)
 
         # Calling encrypt after decrypt raises an exception
         cipher = self.module.new(self.key, self.mode, self.iv)
-        cipher.decrypt(b("CT"))
-        self.assertRaises(TypeError, cipher.encrypt, b("XYZ"))
+        cipher.decrypt(b("CT")*40)
+        self.assertRaises(TypeError, cipher.encrypt, b("XYZ")*40)
 
         # Calling verify after encrypt raises an exception
         cipher = self.module.new(self.key, self.mode, self.iv)
-        cipher.encrypt(b("PT"))
+        cipher.encrypt(b("PT")*40)
         self.assertRaises(TypeError, cipher.verify, b("XYZ"))
         self.assertRaises(TypeError, cipher.hexverify, "12")
 
         # Calling digest after decrypt raises an exception
         cipher = self.module.new(self.key, self.mode, self.iv)
-        cipher.decrypt(b("CT"))
+        cipher.decrypt(b("CT")*40)
         self.assertRaises(TypeError, cipher.digest)
         self.assertRaises(TypeError, cipher.hexdigest)
 
@@ -518,13 +548,13 @@ class AEADTests(unittest.TestCase):
         # Calling update after encrypt raises an exception
         cipher = self.module.new(self.key, self.mode, self.iv)
         cipher.update(b("XX"))
-        cipher.encrypt(b("PT"))
+        cipher.encrypt(b("PT")*40)
         self.assertRaises(TypeError, cipher.update, b("XYZ"))
 
         # Calling update after decrypt raises an exception
         cipher = self.module.new(self.key, self.mode, self.iv)
         cipher.update(b("XX"))
-        cipher.decrypt(b("CT"))
+        cipher.decrypt(b("CT")*40)
         self.assertRaises(TypeError, cipher.update, b("XYZ"))
 
     def runTest(self):
@@ -658,10 +688,10 @@ def make_block_tests(module, module_name, test_data, additional_params=dict()):
             extra_tests_added = 1
 
         # Extract associated data and MAC for AEAD modes
-        if p_mode in ('CCM', 'EAX'):
+        if p_mode in ('CCM', 'EAX', 'SIV'):
             assoc_data, params['plaintext'] = params['plaintext'].split('|')
             assoc_data2, params['ciphertext'], params['mac'] = params['ciphertext'].split('|')
-            params['assoc_data'] = assoc_data
+            params['assoc_data'] = assoc_data.split("-")
             params['mac_len'] = len(params['mac'])>>1
 
         # Add the current test to the test suite
@@ -687,7 +717,7 @@ def make_block_tests(module, module_name, test_data, additional_params=dict()):
             CCMMACLengthTest(module),
             CCMSplitEncryptionTest(module),
         ]
-    for aead_mode in ("MODE_CCM","MODE_EAX"):
+    for aead_mode in ("MODE_CCM","MODE_EAX", "MODE_SIV"):
         if hasattr(module, aead_mode):
             key_sizes = []
             try:
diff --git a/lib/Crypto/SelfTest/Cipher/test_AES.py b/lib/Crypto/SelfTest/Cipher/test_AES.py
index 53d60c0..89243c0 100644
--- a/lib/Crypto/SelfTest/Cipher/test_AES.py
+++ b/lib/Crypto/SelfTest/Cipher/test_AES.py
@@ -1759,6 +1759,42 @@ test_data = [
       'EAX spec Appendix G',
       dict(mode='EAX', nonce='22E7ADD93CFC6393C57EC0B3C17D6B44')
     ),
+
+    # Test vectors for SIV taken from RFC5297
+    # This is a list of tuples with 5 items:
+    #
+    #  1. Header + '|' + plaintext
+    #  2. Header + '|' + ciphertext + '|' + MAC
+    #  3. AES-128 key
+    #  4. Description
+    #  5. Dictionary of parameters to be passed to AES.new().
+    #     It must include the nonce.
+    #
+    #  A "Header" is a dash ('-') separated sequece of components.
+    #
+    ( '101112131415161718191a1b1c1d1e1f2021222324252627|112233445566778899aabbccddee',
+      '101112131415161718191a1b1c1d1e1f2021222324252627|40c02b9690c4dc04daef7f6afe5c|' +
+      '85632d07c6e8f37f950acd320a2ecc93',
+      'fffefdfcfbfaf9f8f7f6f5f4f3f2f1f0f0f1f2f3f4f5f6f7f8f9fafbfcfdfeff',
+      'RFC5297 A.1',
+      dict(mode='SIV', nonce=None)
+    ),
+
+    ( '00112233445566778899aabbccddeeffdeaddadadeaddadaffeeddccbbaa9988' +
+      '7766554433221100-102030405060708090a0|' +
+      '7468697320697320736f6d6520706c61696e7465787420746f20656e63727970' +
+      '74207573696e67205349562d414553',
+
+      '00112233445566778899aabbccddeeffdeaddadadeaddadaffeeddccbbaa9988' +
+      '7766554433221100-102030405060708090a0|' +
+      'cb900f2fddbe404326601965c889bf17dba77ceb094fa663b7a3f748ba8af829' +
+      'ea64ad544a272e9c485b62a3fd5c0d|' +
+      '7bdb6e3b432667eb06f4d14bff2fbd0f',
+
+      '7f7e7d7c7b7a79787776757473727170404142434445464748494a4b4c4d4e4f',
+      'RFC5297 A.2',
+      dict(mode='SIV', nonce='09f911029d74e35bd84156c5635688c0')
+    ),
 ]
 
 def get_tests(config={}):
diff --git a/lib/Crypto/SelfTest/Protocol/test_KDF.py b/lib/Crypto/SelfTest/Protocol/test_KDF.py
index 75bb8f2..37cfb55 100644
--- a/lib/Crypto/SelfTest/Protocol/test_KDF.py
+++ b/lib/Crypto/SelfTest/Protocol/test_KDF.py
@@ -29,8 +29,9 @@ from Crypto.Util.py3compat import *
 
 from Crypto.SelfTest.st_common import list_test_cases
 from Crypto.Hash import SHA1, HMAC
+from Crypto.Cipher import AES, DES3
 
-from Crypto.Protocol.KDF import PBKDF1, PBKDF2
+from Crypto.Protocol.KDF import PBKDF1, PBKDF2, S2V
 
 def t2b(t): return unhexlify(b(t))
 
@@ -87,10 +88,67 @@ class PBKDF2_Tests(unittest.TestCase):
             self.assertEqual(res, t2b(v[4]))
             self.assertEqual(res, res2)
 
+class S2V_Tests(unittest.TestCase):
+
+    # Sequence of test vectors.
+    # Each test vector is made up by:
+    #   Item #0: a tuple of strings
+    #   Item #1: an AES key
+    #   Item #2: the result
+    #   Item #3: the cipher module S2V is based on
+    # Everything is hex encoded
+    _testData = [
+
+            # RFC5297, A.1
+            (
+             (  '101112131415161718191a1b1c1d1e1f2021222324252627',
+                '112233445566778899aabbccddee' ),
+            'fffefdfcfbfaf9f8f7f6f5f4f3f2f1f0',
+            '85632d07c6e8f37f950acd320a2ecc93',
+            AES
+            ),
+
+            # RFC5297, A.2
+            (
+             (  '00112233445566778899aabbccddeeffdeaddadadeaddadaffeeddcc'+
+                'bbaa99887766554433221100',
+                '102030405060708090a0',
+                '09f911029d74e35bd84156c5635688c0',
+                '7468697320697320736f6d6520706c61'+
+                '696e7465787420746f20656e63727970'+
+                '74207573696e67205349562d414553'),
+            '7f7e7d7c7b7a79787776757473727170',
+            '7bdb6e3b432667eb06f4d14bff2fbd0f',
+            AES
+            ),
+
+        ]
+
+    def test1(self):
+        """Verify correctness of test vector"""
+        for tv in self._testData:
+            s2v = S2V.new(t2b(tv[1]), tv[3])
+            for s in tv[0]:
+                s2v.update(t2b(s))
+            result = s2v.derive()
+            self.assertEqual(result, t2b(tv[2]))
+
+    def test2(self):
+        """Verify that no more than 127(AES) and 63(TDES)
+        components are accepted."""
+        key = bchr(0)*16
+        for module in (AES, DES3):
+            s2v = S2V.new(key, module)
+            max_comps = module.block_size*8-1
+            for i in xrange(max_comps):
+                s2v.update(b("XX"))
+            self.assertRaises(TypeError, s2v.update, b("YY"))
+
 def get_tests(config={}):
     tests = []
     tests += list_test_cases(PBKDF1_Tests)
     tests += list_test_cases(PBKDF2_Tests)
+    tests += list_test_cases(S2V_Tests)
     return tests
 
 if __name__ == '__main__':