handle bad randomness during IV creation

Since IVs must be unique we rely on /dev/urandom to yield a
different sequence of bytes when requesting a new fixed part.
In the unlikely event that a new fixed part has already been
used earlier, repeat it for number of times.

Abort if no unique IV could be generated this way since it
most likely indicates a faulty RNG.

diff --git a/deltatar/crypto.py b/deltatar/crypto.py
index df353db..f0af890 100755 (executable)
--- a/deltatar/crypto.py
+++ b/deltatar/crypto.py
@@ -196,8 +196,13 @@ class InvalidIVFixedPart (Exception):
     pass
 
 
+class IVFixedPartError (Exception):
+    """Error creating a unique IV fixed part."""
+    pass
+
+
 class InvalidFileCounter (Exception):
-    """IV fixed part not in supplied list."""
+    """File counter out of range."""
     pass
 
 
@@ -300,6 +305,10 @@ AES_GCM_IV_CNT_DATA         = AES_GCM_IV_CNT_INDEX    + 1 # also for multivolume
 AES_GCM_IV_CNT_MAX_DEFAULT  = 0xffFFffFF
 AES_GCM_IV_CNT_MAX          = AES_GCM_IV_CNT_MAX_DEFAULT
 
+# IV structure and generation
+PDTCRYPT_IV_GEN_MAX_RETRIES = 10 # ×
+PDTCRYPT_IV_FIXEDPART_SIZE  =  8 # B
+PDTCRYPT_IV_COUNTER_SIZE    =  4 # B
 
 ###############################################################################
 ## header, trailer
@@ -641,6 +650,7 @@ class Crypto (object):
 
 
     def next_fixed (self):
+        # NOP for decryption
         pass
 
 
@@ -692,15 +702,12 @@ class Crypto (object):
 
 
     def set_parameters (self, password=None, key=None, paramversion=None,
-                        nacl=None, counter=None, next_fixed=None,
-                        strict_ivs=False):
+                        nacl=None, counter=None, strict_ivs=False):
         """
         Configure the internal state of a crypto context. Not intended for
         external use.
         """
-        if next_fixed is not None:
-            self.next_fixed = next_fixed
-            self.next_fixed ()
+        self.next_fixed ()
         self.set_object_counter (counter)
 
         self.strict_ivs = strict_ivs
@@ -852,10 +859,34 @@ class Encrypt (Crypto):
         self.paramenc     = ENCRYPTION_PARAMETERS.get (paramversion) ["enc"]
 
         super().__init__ (password, key, paramversion, nacl, counter=counter,
-                          next_fixed=lambda: self.fixed.append (os.urandom(8)),
                           strict_ivs=strict_ivs)
 
 
+    def next_fixed (self, retries=PDTCRYPT_IV_GEN_MAX_RETRIES):
+        """
+        Generate the next IV fixed part by reading eight bytes from
+        ``/dev/urandom``. The buffer so obtained is tested against the fixed
+        parts used so far to prevent accidental reuse of IVs. After a
+        configurable number of attempts to create a unique fixed part, it will
+        refuse to continue with an ``IVFixedPartError``. This is unlikely to
+        ever happen on a normal system but may detect an issue with the random
+        generator.
+
+        The list of fixed parts that were used by the context at hand can be
+        accessed through the ``.fixed`` list. Its last element is the fixed
+        part currently in use.
+        """
+        i = 0
+        while i < retries:
+            fp = os.urandom (PDTCRYPT_IV_FIXEDPART_SIZE)
+            if fp not in self.fixed:
+                self.fixed.append (fp)
+                return
+            i += 1
+        raise IVFixedPartError ("error obtaining a unique IV fixed part from "
+                                "/dev/urandom; giving up after %d tries" % i)
+
+
     def iv_make (self):
         """
         Construct a 12-bytes IV from the current fixed part and the object

diff --git a/testing/test_crypto.py b/testing/test_crypto.py
index 8e5d210..0b04c2b 100644 (file)
--- a/testing/test_crypto.py
+++ b/testing/test_crypto.py
@@ -64,12 +64,15 @@ class CryptoLayerTest (unittest.TestCase):
 
 class AESGCMTest (CryptoLayerTest):
 
+    os_urandom = os.urandom
+
     def tearDown (self):
         """Reset globals altered for testing."""
         _ = crypto._testing_set_AES_GCM_IV_CNT_MAX \
                   ("I am fully aware that this will void my warranty.")
         _ = crypto._testing_set_PDTCRYPT_MAX_OBJ_SIZE \
                   ("I am fully aware that this will void my warranty.")
+        os.urandom = self.os_urandom
 
     def test_crypto_aes_gcm_enc_ctor (self):
         password   = str (os.urandom (42))
@@ -302,6 +305,8 @@ class AESGCMTest (CryptoLayerTest):
 
         for i in range (5): addobj (i)
 
+        assert len (encryptor.fixed) == 1
+
 
     def test_crypto_aes_gcm_enc_multiobj_strict_ivs (self):
         cnksiz    = 1 << 10
@@ -311,6 +316,7 @@ class AESGCMTest (CryptoLayerTest):
                                     password=password,
                                     nacl=TEST_STATIC_NACL,
                                     strict_ivs=True)
+        curfixed  = None # must remain constant after first
 
         def addobj (i):
             pt           = fill_mod (1 << 14, off=i)
@@ -324,12 +330,19 @@ class AESGCMTest (CryptoLayerTest):
                 ct += cnk
                 off += cnksiz
             cnk, header, fixed = encryptor.done (header_dummy)
+            nonlocal curfixed
+            if curfixed is None:
+                curfixed = fixed
+            else:
+                assert fixed == curfixed
             ct += cnk
 
             assert len (pt) == len (ct)
 
         for i in range (5): addobj (i)
 
+        assert len (encryptor.fixed) == 1
+
 
     def test_crypto_aes_gcm_enc_multiobj_cnt_wrap (self):
         """
@@ -390,6 +403,47 @@ class AESGCMTest (CryptoLayerTest):
         for j in range (i + 2, i + new_max - 1): addobj (j) # counter range: [4, 8]
         addobj (j + 1, True) # counter wraps to 3 again
 
+        assert len (encryptor.fixed) == 3
+
+
+    def test_crypto_aes_gcm_enc_multiobj_cnt_wrap_badfixed (self):
+        """
+        Test behavior when the file counter tops out and the transition to
+        the next IV fixed part fails on account of a bad random generator.
+
+        Replaces the ``urandom`` reference in ``os`` with a deterministic
+        function. The encryptor context must communicate this condition with an
+        ``IVFixedPartError``.
+        """
+        minimum = 3
+        new_max = 8
+        crypto._testing_set_AES_GCM_IV_CNT_MAX \
+                ("I am fully aware that this will void my warranty.", new_max)
+        cnksiz    = 1 << 10
+        os.urandom = lambda n: bytes (bytearray ([n % 256] * n))
+        password  = str (os.urandom (42))
+        encryptor = crypto.Encrypt (TEST_VERSION,
+                                    TEST_PARAMVERSION,
+                                    password=password,
+                                    nacl=TEST_STATIC_NACL,
+                                    strict_ivs=True)
+
+        def addobj (i):
+            pt = fill_mod (1 << 14, off=i)
+            header_dummy = encryptor.next ("%s_%d" % (TEST_DUMMY_FILENAME, i))
+
+            off = 0
+            while off < len (pt):
+                upto = min (off + cnksiz, len (pt))
+                _, cnk = encryptor.process (pt [off:upto])
+                off += cnksiz
+
+        for i in range (minimum, new_max): addobj (42 + i)
+
+        with self.assertRaises (crypto.IVFixedPartError):
+            addobj (42 + i)
+
+
 
     def test_crypto_aes_gcm_enc_length_cap (self):
         """