allow checking PDTCRYPT archives for IV integrity with crypto.py

This allows checking the IVs of objects in the given file for
uniqueness and consecutiveness without decrypting them.

Example:

    $ crypto.py ivcheck -i - <bfull-2020-04-20-1133-001.tar.gz.pdtcrypt
    PDT: Successfully traversed 411 encrypted objects in input.
    PDT:
    PDT: All IVs consecutive and unique.

diff --git a/deltatar/crypto.py b/deltatar/crypto.py
index 178a469..431466e 100755 (executable)
--- a/deltatar/crypto.py
+++ b/deltatar/crypto.py
@@ -34,7 +34,9 @@ Errors fall into roughly three categories:
         - ``InvalidGCMTag`` (decryption failed on account of an invalid GCM
           tag),
         - ``InvalidIVFixedPart`` (IV fixed part of object not found in list),
-        - ``DuplicateIV`` (the IV of an encrypted object already occurred),
+        - ``DuplicateIV`` (the IV of an object encrypted earlier was reused),
+        - ``NonConsecutiveIV`` (IVs of two encrypted objects are not
+          consecutive),
         - ``DecryptionError`` (used in CLI decryption for presenting error
           conditions to the user).
 
@@ -534,7 +536,7 @@ IV_FMT = "((f %s) (c %d))"
 def iv_fmt (iv):
     """Format the two components of an IV in a readable fashion."""
     fixed, cnt = struct.unpack (FMT_I2N_IV, iv)
-    return IV_FMT % (binascii.hexlify (fixed), cnt)
+    return IV_FMT % (binascii.hexlify (fixed).decode (), cnt)
 
 
 ###############################################################################
@@ -1607,11 +1609,13 @@ def open2_dump_file (fname, dir_fd, force=False):
 PDTCRYPT_SUB_PROCESS = 0
 PDTCRYPT_SUB_SCRYPT  = 1
 PDTCRYPT_SUB_SCAN    = 2
+PDTCRYPT_SUB_IVCHECK = 3
 
 PDTCRYPT_SUB = \
         { "process" : PDTCRYPT_SUB_PROCESS
         , "scrypt"  : PDTCRYPT_SUB_SCRYPT
-        , "scan"    : PDTCRYPT_SUB_SCAN }
+        , "scan"    : PDTCRYPT_SUB_SCAN
+        , "ivcheck" : PDTCRYPT_SUB_IVCHECK }
 
 PDTCRYPT_DECRYPT   = 1 << 0 # decrypt archive with password
 PDTCRYPT_SPLIT     = 1 << 1 # split archive into individual objects
@@ -1679,6 +1683,53 @@ class PassthroughDecryptor (object):
         return d
 
 
+def check_ivs (ifs):
+    """
+    Walk the objects in the given reader, validating uniqueness and
+    consecutiveness of the IVs in the object headers.
+
+    As the IVs are metadata this does not require decryption.
+    """
+    objs = 0
+    seen  = set ()
+    last = None
+
+    while True:
+        try:
+            hdr = hdr_read_stream (ifs)
+        except EndOfFile as exn:
+            break # done
+
+        objs += 1
+        cur = hdr ["iv"]
+
+        fixed, cnt = struct.unpack (FMT_I2N_IV, cur)
+
+        if PDTCRYPT_VERBOSE is True:
+            noise ("PDT: obj %d, iv %s" % (objs, iv_fmt (cur)))
+
+        if last is not None:
+            if fixed != last [0]:
+                noise ("PDT: obj %d, fixed part changed last: %s → this: %s"
+                       % (obj,
+                          binascii.hexlify (last [0]),
+                          binascii.hexlify (fixed)))
+            if cnt != last [1] + 1:
+                raise NonConsecutiveIV ("iv %s counter not successor of "
+                                        "last object (expected %d, found %d)"
+                                        % (iv_fmt (cur), last [1] + 1, cnt))
+
+        if cur in seen:
+            raise DuplicateIV ("iv %s was reused" % iv_fmt (cur))
+
+        seen.add (cur)
+        last = (fixed, cnt)
+
+        ifs.read (hdr ["ctsize"])
+
+    return objs
+
+
 def depdtcrypt (mode, secret, ins, outs):
     """
     Remove PDTCRYPT layer from all objects encrypted with the secret. Used on a
@@ -1998,6 +2049,30 @@ def find_overlaps (slices):
     return [ slices [i] for i in ovrlp ]
 
 
+def mode_ivcheck (ifd):
+    total_obj = 0
+    try:
+        total_obj = check_ivs (ifd)
+    except (NonConsecutiveIV, DuplicateIV) as exn:
+        noise ("PDT: Detected inconsistent initialization vectors")
+        noise ("PDT:")
+        noise ("PDT:    “%s”" % exn)
+        noise ("PDT:")
+        noise ("")
+        return 1
+    except Exception as exn:
+        noise ("PDT: Hit an error unrelated to checking IVs")
+        noise ("PDT:")
+        noise ("PDT:    “%s”" % exn)
+        noise ("PDT:")
+        return 1
+
+    noise ("PDT: Successfully traversed %d encrypted objects in input."
+           % total_obj)
+    noise ("PDT:")
+    noise ("PDT: All IVs consecutive and unique.")
+
+
 def mode_scan (secret, fname, outs=None, nacl=None):
     """
     Dissect a binary file, looking for PDTCRYPT headers and objects.
@@ -2089,6 +2164,7 @@ def mode_scan (secret, fname, outs=None, nacl=None):
         for slice in overlap:
             noise ("PDT:    × %d→%d" % (slice [0], slice [1]))
 
+
 def usage (err=False):
     out = print
     if err is True:
@@ -2103,10 +2179,12 @@ def usage (err=False):
     out ("       %s              [ -f | --format ]" % indent)
     out ("")
     out ("\twhere")
-    out ("\t\tSUBCOMMAND      main mode: { process | scrypt }")
+    out ("\t\tSUBCOMMAND      main mode: { process | scrypt | scan | ivcheck }")
     out ("\t\t                where:")
     out ("\t\t                   process: extract objects from PDT archive")
     out ("\t\t                   scrypt:  calculate hash from password and first object")
+    out ("\t\t                   scan:    scan input for PDTCRYPT headers")
+    out ("\t\t                   ivcheck: check whether IVs are consecutive")
     out ("\t\t-p PASSWORD     password to derive the encryption key from")
     out ("\t\t-k KEY          encryption key as 16 bytes in hexadecimal notation")
     out ("\t\t-s              enforce strict handling of initialization vectors")
@@ -2243,7 +2321,9 @@ def parse_argv (argv):
             checked_secret (make_secret (key=ek.strip ()))
 
     if secret is None:
-        if subcommand == PDTCRYPT_SUB_SCRYPT:
+        if subcommand == PDTCRYPT_SUB_IVCHECK:
+            pass
+        elif subcommand == PDTCRYPT_SUB_SCRYPT:
             bail ("ERROR: scrypt hash mode requested but no password given")
         elif mode & PDTCRYPT_DECRYPT:
             bail ("ERROR: decryption requested but no password given")
@@ -2280,6 +2360,10 @@ def parse_argv (argv):
             bail ("ERROR: input must be seekable; please specify a file")
         return True, partial (mode_scan, secret, insspec, outs, nacl=nacl)
 
+    if subcommand == PDTCRYPT_SUB_IVCHECK:
+        if insspec is None:
+            bail ("ERROR: please supply an input file for checking ivs")
+
     if subcommand == PDTCRYPT_SUB_SCRYPT:
         if secret [0] == PDTCRYPT_SECRET_KEY:
             bail ("ERROR: scrypt mode requires a password")
@@ -2293,6 +2377,9 @@ def parse_argv (argv):
     if insspec is not None or subcommand != PDTCRYPT_SUB_SCRYPT:
         ins = deptdcrypt_mk_stream (PDTCRYPT_SOURCE, insspec or "-")
 
+    if subcommand == PDTCRYPT_SUB_IVCHECK:
+        return True, partial (mode_ivcheck, ins)
+
     if subcommand == PDTCRYPT_SUB_SCRYPT:
         return True, partial (mode_scrypt, secret [1].encode (), ins, nacl,
                               fmt=scrypt_format)

diff --git a/testing/test_recover.py b/testing/test_recover.py
index 0a47d21..ff88148 100644 (file)
--- a/testing/test_recover.py
+++ b/testing/test_recover.py
@@ -31,8 +31,8 @@ communicated upward by throwing.
 
     - corrupt_entire_header ():
       Invert all bits of the first object header (PDTCRYPT, gzip, tar) without
-      affecting the payload. This renders the object unreadable; the file will
-      be resemble one with arbitrary leading data but all the remaining object
+      affecting the payload. This renders the object unreadable; the result will
+      resemble a file with arbitrary leading data but all the remaining object
       offsets intact, so the contents can still be extracted with index based
       recovery.
 
@@ -1068,7 +1068,7 @@ class GenIndexCorruptHoleGZAESTest (GenIndexCorruptHoleBaseTest):
 
 class GenIndexCorruptEntireHeaderBaseTest (GenIndexTest):
     """
-    Recreate index from file with hole.
+    Recreate index from file with defective headers.
     """
     COMPRESSION = None
     PASSWORD    = None