Add MODE_KW (KeyWrap) for AES

Author: Legrandin

lib/Crypto/Cipher/AES.py

@@ -41,7 +41,8 @@
 MODE_SIV = 10       #: Synthetic Initialization Vector (:ref:`siv_mode`)
 MODE_GCM = 11       #: Galois Counter Mode (:ref:`gcm_mode`)
 MODE_OCB = 12       #: Offset Code Book (:ref:`ocb_mode`)
-
+MODE_KW = 13        #: Key Wrap
+MODE_KWP = 14       #: Key Wrap with Padding
 
 _cproto = """
         int AES_start_operation(const uint8_t key[],

lib/Crypto/Cipher/__init__.py

@@ -69,15 +69,18 @@ def _create_cipher(factory, key, mode, *args, **kwargs):
         if mode == 8:
             from Crypto.Cipher._mode_ccm import _create_ccm_cipher
             res = _create_ccm_cipher(factory, **kwargs)
-        if mode == 10:
+        elif mode == 10:
             from Crypto.Cipher._mode_siv import _create_siv_cipher
             res = _create_siv_cipher(factory, **kwargs)
-        if mode == 11:
+        elif mode == 11:
             from Crypto.Cipher._mode_gcm import _create_gcm_cipher
             res = _create_gcm_cipher(factory, **kwargs)
-        if mode == 12:
+        elif mode == 12:
             from Crypto.Cipher._mode_ocb import _create_ocb_cipher
             res = _create_ocb_cipher(factory, **kwargs)
+        elif mode == 13:
+            from Crypto.Cipher._mode_kw import _create_kw_cipher
+            res = _create_kw_cipher(factory, **kwargs)
 
     if res is None:
         raise ValueError("Mode not supported")

lib/Crypto/Cipher/_mode_kw.py

@@ -0,0 +1,147 @@
+import struct
+from collections import deque
+
+from types import ModuleType
+from typing import Union
+
+from Crypto.Util.strxor import strxor
+
+
+def W(cipher: ModuleType,
+      plaintext: Union[bytes, bytearray]) -> bytes:
+
+    S = [plaintext[i:i+8] for i in range(0, len(plaintext), 8)]
+    n = len(S)
+    s = 6 * (n - 1)
+    A = S[0]
+    R = deque(S[1:])
+
+    for t in range(1, s + 1):
+        t_64 = struct.pack('>Q', t)
+        ct = cipher.encrypt(A + R.popleft())
+        A = strxor(ct[:8], t_64)
+        R.append(ct[8:])
+
+    return A + b''.join(R)
+
+
+def W_inverse(cipher: ModuleType,
+              ciphertext: Union[bytes, bytearray]) -> bytes:
+
+    C = [ciphertext[i:i+8] for i in range(0, len(ciphertext), 8)]
+    n = len(C)
+    s = 6 * (n - 1)
+    A = C[0]
+    R = deque(C[1:])
+
+    for t in range(s, 0, -1):
+        t_64 = struct.pack('>Q', t)
+        pt = cipher.decrypt(strxor(A, t_64) + R.pop())
+        A = pt[:8]
+        R.appendleft(pt[8:])
+
+    return A + b''.join(R)
+
+
+class KWMode(object):
+    """Key Wrap (KW) mode.
+
+    This is a deterministic Authenticated Encryption (AE) mode
+    for protecting cryptographic keys. See `NIST SP800-38F`_.
+
+    It provides both confidentiality and authenticity, and it designed
+    so that any bit of the ciphertext depends on all bits of the plaintext.
+
+    This mode is only available for ciphers that operate on 128 bits blocks
+    (e.g., AES).
+
+    .. _`NIST SP800-38F`: http://csrc.nist.gov/publications/nistpubs/800-38F/SP-800-38F.pdf
+
+    :undocumented: __init__
+    """
+
+    def __init__(self,
+                 factory: ModuleType,
+                 key: Union[bytes, bytearray]):
+
+        self.block_size = factory.block_size
+        if self.block_size != 16:
+            raise ValueError("Key Wrap mode is only available for ciphers"
+                             " that operate on 128 bits blocks")
+
+        self._factory = factory
+        self._cipher = factory.new(key, factory.MODE_ECB)
+
+    def seal(self, plaintext: Union[bytes, bytearray]) -> bytes:
+        """Encrypt and authenticate (wrap) a cryptographic key.
+
+        Args:
+          plaintext:
+            The cryptographic key to wrap.
+            It must be at least 16 bytes long, and its length
+            must be a multiple of 8.
+
+        Returns:
+            The wrapped key.
+        """
+
+        if len(plaintext) % 8:
+            raise ValueError("The plaintext must have length multiple of 8 bytes")
+
+        if len(plaintext) < 16:
+            raise ValueError("The plaintext must be at least 16 bytes long")
+
+        res = W(self._cipher, b'\xA6\xA6\xA6\xA6\xA6\xA6\xA6\xA6' + plaintext)
+        return res
+
+    def unseal(self, ciphertext: Union[bytes, bytearray]) -> bytes:
+        """Decrypt and authenticate (unwrap) a cryptographic key.
+
+        Args:
+          ciphertext:
+            The cryptographic key to unwrap.
+            It must be at least 24 bytes long, and its length
+            must be a multiple of 8.
+
+        Returns:
+            The original key.
+
+        Raises: ValueError
+           If the ciphertext or the key are not valid.
+        """
+
+
+        if len(ciphertext) % 8:
+            raise ValueError("The ciphertext must have length multiple of 8 bytes")
+
+        if len(ciphertext) < 24:
+            raise ValueError("The ciphertext must be at least 24 bytes long")
+
+        pt = W_inverse(self._cipher, ciphertext)
+
+        if pt[:8] != b'\xA6\xA6\xA6\xA6\xA6\xA6\xA6\xA6':
+            raise ValueError("Incorrect integrity check value")
+
+        return pt[8:]
+
+
+def _create_kw_cipher(factory: ModuleType,
+                      **kwargs: Union[bytes, bytearray]) -> KWMode:
+    """Create a new block cipher in Key Wrap mode.
+
+    Args:
+      factory:
+        A block cipher module, taken from `Crypto.Cipher`.
+        The cipher must have block length of 16 bytes, such as AES.
+
+    Keywords:
+      key:
+        The secret key to use to seal or unseal.
+    """
+
+    try:
+        key = kwargs["key"]
+    except KeyError as e:
+        raise TypeError("Missing parameter:" + str(e))
+
+    return KWMode(factory, key)

lib/Crypto/SelfTest/Cipher/__init__.py

@@ -24,7 +24,8 @@
 
 """Self-test for cipher modules"""
 
-__revision__ = "$Id$"
+import sys
+
 
 def get_tests(config={}):
     tests = []
@@ -50,11 +51,14 @@ def get_tests(config={}):
     from Crypto.SelfTest.Cipher import test_EAX;        tests += test_EAX.get_tests(config=config)
     from Crypto.SelfTest.Cipher import test_GCM;        tests += test_GCM.get_tests(config=config)
     from Crypto.SelfTest.Cipher import test_SIV;        tests += test_SIV.get_tests(config=config)
+
+    if sys.version_info >= (3, 9):
+        from Crypto.SelfTest.Cipher import test_KW
+        tests += test_KW.get_tests(config=config)
+
     return tests
 
 if __name__ == '__main__':
     import unittest
     suite = lambda: unittest.TestSuite(get_tests())
     unittest.main(defaultTest='suite')
-
-# vim:set ts=4 sw=4 sts=4 expandtab:

lib/Crypto/SelfTest/Cipher/test_KW.py

@@ -0,0 +1,74 @@
+import unittest
+
+from Crypto.SelfTest.st_common import list_test_cases
+
+from Crypto.Cipher import AES
+
+
+class KW_Tests(unittest.TestCase):
+
+    # From RFC3394
+    tvs = [
+        ("000102030405060708090A0B0C0D0E0F",
+         "00112233445566778899AABBCCDDEEFF",
+         "1FA68B0A8112B447AEF34BD8FB5A7B829D3E862371D2CFE5"),
+        ("000102030405060708090A0B0C0D0E0F1011121314151617",
+         "00112233445566778899AABBCCDDEEFF",
+         "96778B25AE6CA435F92B5B97C050AED2468AB8A17AD84E5D"),
+        ("000102030405060708090A0B0C0D0E0F101112131415161718191A1B1C1D1E1F",
+         "00112233445566778899AABBCCDDEEFF",
+         "64E8C3F9CE0F5BA263E9777905818A2A93C8191E7D6E8AE7"),
+        ("000102030405060708090A0B0C0D0E0F1011121314151617",
+         "00112233445566778899AABBCCDDEEFF0001020304050607",
+         "031D33264E15D33268F24EC260743EDCE1C6C7DDEE725A936BA814915C6762D2"),
+        ("000102030405060708090A0B0C0D0E0F101112131415161718191A1B1C1D1E1F",
+         "00112233445566778899AABBCCDDEEFF0001020304050607",
+         "A8F9BC1612C68B3FF6E6F4FBE30E71E4769C8B80A32CB8958CD5D17D6B254DA1"),
+        ("000102030405060708090A0B0C0D0E0F101112131415161718191A1B1C1D1E1F",
+         "00112233445566778899AABBCCDDEEFF000102030405060708090A0B0C0D0E0F",
+         "28C9F404C4B810F4CBCCB35CFB87F8263F5786E2D80ED326CBC7F0E71A99F43BFB988B9B7A02DD21"),
+    ]
+
+    def test_rfc3394(self):
+        for tv in self.tvs:
+            kek, pt, ct = [bytes.fromhex(x) for x in tv]
+
+            cipher = AES.new(kek, AES.MODE_KW)
+            ct2 = cipher.seal(pt)
+
+            self.assertEqual(ct, ct2)
+
+            pt2 = cipher.unseal(ct)
+            self.assertEqual(pt, pt2)
+
+    def test_neg1(self):
+
+        cipher = AES.new(b'-' * 16, AES.MODE_KW)
+
+        with self.assertRaises(ValueError):
+            cipher.seal(b'')
+
+        with self.assertRaises(ValueError):
+            cipher.seal(b'8' * 17)
+
+    def test_neg2(self):
+
+        cipher = AES.new(b'-' * 16, AES.MODE_KW)
+        ct = bytearray(cipher.seal(b'7' * 16))
+        cipher.unseal(ct)
+
+        ct[0] ^= 0xFF
+        with self.assertRaises(ValueError):
+            cipher.unseal(ct)
+
+
+def get_tests(config={}):
+    tests = []
+    tests += list_test_cases(KW_Tests)
+    return tests
+
+
+if __name__ == '__main__':
+    def suite():
+        return unittest.TestSuite(get_tests())
+    unittest.main(defaultTest='suite')

lib/Crypto/SelfTest/Protocol/test_HPKE.py

@@ -474,10 +474,6 @@ def test_hpke_unseal(self):
             print(".", end="", flush=True)
 
 
-if __name__ == "__main__":
-    unittest.main()
-
-
 def get_tests(config={}):
 
     tests = []