[python-delta-tar] / testing / test_recover.py

import logging
import os
import shutil
import stat

import deltatar.deltatar as deltatar
import deltatar.crypto   as crypto
import deltatar.tarfile  as tarfile

from . import BaseTest

TEST_PASSWORD     = "test1234"
TEST_VOLSIZ       = 2 # MB
TEST_FILESPERVOL  = 3
VOLUME_OVERHEAD   = 1.4 # account for tar overhead when fitting files into
                        # volumes; this is black magic
TEST_BLOCKSIZE    = 4096

###############################################################################
## helpers                                                                   ##
###############################################################################

def flip_bits (fname, off, b=0x01, n=1):
    """
    Open file *fname* at offset *off*, replacing the next *n* bytes with
    their values xor’ed with *b*.
    """
    fd = os.open (fname, os.O_RDWR)

    try:
        pos = os.lseek (fd, off, os.SEEK_SET)
        assert pos == off
        chunk = os.read (fd, n)
        chunk = bytes (map (lambda v: v ^ b, chunk))
        pos = os.lseek (fd, off, os.SEEK_SET)
        assert pos == off
        os.write (fd, chunk)
    finally:
        os.close (fd)


def gz_header_size (fname, off=0):
    """
    Determine the length of the gzip header starting at *off* in file fname.

    The header is variable length because it may contain the filename as NUL
    terminated bytes.
    """
    # length so we need to determine where the actual payload starts
    off = tarfile.GZ_HEADER_SIZE
    fd = os.open (fname, os.O_RDONLY)

    try:
        pos = os.lseek (fd, off, os.SEEK_SET)
        assert pos == off
        while os.read (fd, 1)[0] != 0:
            off += 1
            pos = os.lseek (fd, off, os.SEEK_SET)
            assert pos == off
    finally:
        os.close (fd)

    return off


def is_pdt_encrypted (fname):
    """
    Returns true if the file contains at least one PDT header plus enough
    space for the object.
    """
    try:
        with open (fname, "rb") as st:
            hdr = crypto.hdr_read_stream (st)
            siz = hdr ["ctsize"]
            assert (len (st.read (siz)) == siz)
    except Exception as exn:
        return False
    return True


###############################################################################
## corruption simulators                                                     ##
###############################################################################

def corrupt_header (_, fname, compress, encrypt):
    """
    Modify a significant byte in the object header of the format.
    """
    if encrypt is True: # damage GCM tag
        flip_bits (fname, crypto.HDR_OFF_TAG + 1)
    elif compress is True: # invalidate magic
        flip_bits (fname, 1)
    else: # Fudge checksum. From tar(5):
        #
        #       struct header_gnu_tar {
        #               char name[100];
        #               char mode[8];
        #               char uid[8];
        #               char gid[8];
        #               char size[12];
        #               char mtime[12];
        #               char checksum[8];
        #               …
        flip_bits (fname, 100 + 8 + 8 + 8 + 12 + 12 + 1)


def corrupt_entire_header (_, fname, compress, encrypt):
    """
    Flip all bits in the first object header.
    """
    if encrypt is True:
        flip_bits (fname, 0, 0xff, crypto.PDTCRYPT_HDR_SIZE)
    elif compress is True: # invalidate magic
        flip_bits (fname, 0, 0xff, gz_header_size (fname))
    else:
        flip_bits (fname, 0, 0xff, tarfile.BLOCKSIZE)


def corrupt_payload_start (_, fname, compress, encrypt):
    """
    Modify the byte following the object header structure of the format.
    """
    if encrypt is True:
        flip_bits (fname, crypto.PDTCRYPT_HDR_SIZE + 1)
    elif compress is True:
        flip_bits (fname, gz_header_size (fname) + 1)
    else:
        flip_bits (fname, tarfile.BLOCKSIZE + 1)


def corrupt_trailing_data (_, fname, compress, encrypt):
    """
    Modify the byte following the object header structure of the format.
    """
    junk = os.urandom (42)
    fd = os.open (fname, os.O_WRONLY | os.O_APPEND)
    os.write (fd, junk)
    os.close (fd)


def corrupt_volume (_, fname, compress, encrypt):
    """
    Zero out an entire volume.
    """
    fd = os.open (fname, os.O_WRONLY)
    size = os.lseek (fd, 0, os.SEEK_END)
    assert os.lseek (fd, 0, os.SEEK_SET) == 0
    zeros = bytes (b'\x00' * TEST_BLOCKSIZE)
    while size > 0:
        todo = min (size, TEST_BLOCKSIZE)
        os.write (fd, zeros [:todo])
        size -= todo
    os.close (fd)


def corrupt_hole (_, fname, compress, encrypt):
    """
    Cut file in three pieces, reassemble without the middle one.
    """
    aname = os.path.abspath (fname)
    infd = os.open (fname, os.O_RDONLY)
    size = os.lseek (infd, 0, os.SEEK_END)
    assert os.lseek (infd, 0, os.SEEK_SET) == 0
    assert size > 3 * TEST_BLOCKSIZE
    hole = (size / 3, size * 2 / 3)
    outfd = os.open (os.path.dirname (aname), os.O_WRONLY | os.O_TMPFILE,
                     stat.S_IRUSR | stat.S_IWUSR)
    
    zeros = bytes (b'\x00' * TEST_BLOCKSIZE)
    done = 0
    while done < size:
        data = os.read (infd, TEST_BLOCKSIZE)
        if done < hole [0] or hole [1] < done:
            # only copy from outside hole
            os.write (outfd, data)
        done += len (data)

    os.close (infd)
    os.unlink (fname)

    path = "/proc/self/fd/%d" % outfd
    os.link (path, aname, src_dir_fd=0, follow_symlinks=True)
    os.close (outfd)


###############################################################################
## tests                                                                     ##
###############################################################################

class DefectiveTest (BaseTest):
    """
    Disaster recovery: restore corrupt backups.
    """

    COMPRESSION = None
    PASSWORD    = None
    FAILURES    = 0     # files that could not be restored
    MISMATCHES  = 0     # files that were restored but corrupted
    CORRUPT     = corrupt_payload_start
    VOLUMES     = 1
    MISSING     = None  # normally the number of failures


    def setUp(self):
        '''
        Create base test data
        '''
        self.pwd      = os.getcwd()
        self.dst_path = "source_dir"
        self.src_path = "%s2" % self.dst_path
        self.hash     = dict()

        os.system('rm -rf target_dir source_dir* backup_dir* huge')
        os.makedirs (self.src_path)

        for i in range (5):
            f = "dummy_%d" % i
            self.hash [f] = self.create_file ("%s/%s"
                                              % (self.src_path, f), 5 + i)


    def tearDown(self):
        '''
        Remove temporal files created by unit tests and reset globals.
        '''
        os.chdir(self.pwd)
        os.system("rm -rf source_dir source_dir2 backup_dir*")


class RecoverTest (DefectiveTest):
    """
    Recover: restore corrupt backups from index file information.
    """

    def test_recover_corrupt (self):
        """
        Perform various damaging actions that cause unreadable objects.

        Expects the extraction to fail in normal mode. With disaster recovery,
        extraction must succeed, and exactly one file must be missing.
        """
        mode           = self.COMPRESSION or "#"
        bak_path       = "backup_dir"
        backup_file    = "the_full_backup_%0.2d.tar"
        backup_full    = ("%s/%s" % (bak_path, backup_file)) % 0
        index_file     = "the_full_index"

        if self.COMPRESSION is not None:
            backup_file += ".gz"
            backup_full += ".gz"
            index_file  += ".gz"

        if self.PASSWORD is not None:
            backup_file = "%s.%s" % (backup_file, deltatar.PDTCRYPT_EXTENSION)
            backup_full = "%s.%s" % (backup_full, deltatar.PDTCRYPT_EXTENSION)
            index_file  = "%s.%s" % (index_file , deltatar.PDTCRYPT_EXTENSION)

        if self.VOLUMES > 1:
            # add n files for one nth the volume size each, corrected
            # for metadata and tar block overhead
            fsiz = int (  (  TEST_VOLSIZ
                           / (TEST_FILESPERVOL * VOLUME_OVERHEAD))
                        * 1024 * 1024)
            fcnt = (self.VOLUMES - 1) * TEST_FILESPERVOL
            for i in range (fcnt):
                nvol, invol = divmod(i, TEST_FILESPERVOL)
                f = "dummy_vol_%d_n_%0.2d" % (nvol, invol)
                self.hash [f] = self.create_file ("%s/%s"
                                                  % (self.src_path, f),
                                                  fsiz,
                                                  random=True)

        def vname (_x, _y, n, *a, **kwa):
            return backup_file % n

        dtar = deltatar.DeltaTar (mode=mode,
                                  logger=None,
                                  password=self.PASSWORD,
                                  index_name_func=lambda _: index_file,
                                  volume_name_func=vname)

        dtar.create_full_backup \
            (source_path=self.src_path, backup_path=bak_path,
             max_volume_size=1)

        if self.PASSWORD is not None:
            # ensure all files are at least superficially in PDT format
            for f in os.listdir (bak_path):
                assert is_pdt_encrypted ("%s/%s" % (bak_path, f))

        # first restore must succeed
        dtar.restore_backup(target_path=self.dst_path,
                            backup_indexes_paths=[
                                "%s/%s" % (bak_path, index_file)
                            ])
        for key, value in self.hash.items ():
            f = "%s/%s" % (self.dst_path, key)
            assert os.path.exists (f)
            assert value == self.md5sum (f)
        shutil.rmtree (self.dst_path)
        shutil.rmtree (self.src_path)

        self.CORRUPT (backup_full,
                      self.COMPRESSION is not None,
                      self.PASSWORD    is not None)

        # normal restore must fail
        try:
            dtar.restore_backup(target_path=self.dst_path,
                                backup_tar_path=backup_full)
        except tarfile.CompressionError:
            if self.PASSWORD is not None or self.COMPRESSION is not None:
                pass
            else:
                raise
        except tarfile.ReadError:
            # can happen with all three modes
            pass
        except tarfile.DecryptionError:
            if self.PASSWORD is not None:
                pass
            else:
                raise

        os.chdir (self.pwd) # not restored due to the error above
        # but recover will succeed
        failed = dtar.recover_backup(target_path=self.dst_path,
                                     backup_indexes_paths=[
                                         "%s/%s" % (bak_path, index_file)
                                     ])

        assert len (failed) == self.FAILURES

        # with one file missing
        missing  = []
        mismatch = []
        for key, value in self.hash.items ():
            kkey = "%s/%s" % (self.dst_path, key)
            if os.path.exists (kkey):
                if value != self.md5sum (kkey):
                    mismatch.append (key)
            else:
                missing.append (key)

        # usually, an object whose extraction fails will not be found on
        # disk afterwards so the number of failures equals that of missing
        # files. however, some modes will create partial files for objects
        # spanning multiple volumes that contain the parts whose checksums
        # were valid.
        assert len (missing)  == (self.MISSING if self.MISSING is not None
                                               else self.FAILURES)
        assert len (mismatch) == self.MISMATCHES

        shutil.rmtree (self.dst_path)


class RescueTest (DefectiveTest):
    """
    Rescue: restore corrupt backups from backup set that is damaged to a degree
    that the index file is worthless.
    """

    def test_rescue_corrupt (self):
        """
        Perform various damaging actions that cause unreadable objects, then
        attempt to extract objects regardless.
        """
        mode           = self.COMPRESSION or "#"
        bak_path       = "backup_dir"
        backup_file    = "the_full_backup_%0.2d.tar"
        backup_full    = ("%s/%s" % (bak_path, backup_file)) % 0
        index_file     = "the_full_index"

        if self.COMPRESSION is not None:
            backup_file += ".gz"
            backup_full += ".gz"
            index_file  += ".gz"

        if self.PASSWORD is not None:
            backup_file = "%s.%s" % (backup_file, deltatar.PDTCRYPT_EXTENSION)
            backup_full = "%s.%s" % (backup_full, deltatar.PDTCRYPT_EXTENSION)
            index_file  = "%s.%s" % (index_file , deltatar.PDTCRYPT_EXTENSION)

        if self.VOLUMES > 1:
            # add n files for one nth the volume size each, corrected
            # for metadata and tar block overhead
            fsiz = int (  (  TEST_VOLSIZ
                           / (TEST_FILESPERVOL * VOLUME_OVERHEAD))
                        * 1024 * 1024)
            fcnt = (self.VOLUMES - 1) * TEST_FILESPERVOL
            for i in range (fcnt):
                nvol, invol = divmod(i, TEST_FILESPERVOL)
                f = "dummy_vol_%d_n_%0.2d" % (nvol, invol)
                self.hash [f] = self.create_file ("%s/%s"
                                                  % (self.src_path, f),
                                                  fsiz,
                                                  random=True)

        def vname (_x, _y, n, *a, **kwa):
            return backup_file % n

        dtar = deltatar.DeltaTar (mode=mode,
                                  logger=None,
                                  password=self.PASSWORD,
                                  index_name_func=lambda _: index_file,
                                  volume_name_func=vname)

        dtar.create_full_backup \
            (source_path=self.src_path, backup_path=bak_path,
             max_volume_size=1)

        if self.PASSWORD is not None:
            # ensure all files are at least superficially in PDT format
            for f in os.listdir (bak_path):
                assert is_pdt_encrypted ("%s/%s" % (bak_path, f))

        # first restore must succeed
        dtar.restore_backup(target_path=self.dst_path,
                            backup_indexes_paths=[
                                "%s/%s" % (bak_path, index_file)
                            ])
        for key, value in self.hash.items ():
            f = "%s/%s" % (self.dst_path, key)
            assert os.path.exists (f)
            assert value == self.md5sum (f)
        shutil.rmtree (self.dst_path)
        shutil.rmtree (self.src_path)

        self.CORRUPT (backup_full,
                      self.COMPRESSION is not None,
                      self.PASSWORD    is not None)

        # normal restore must fail
        try:
            dtar.restore_backup(target_path=self.dst_path,
                                backup_tar_path=backup_full)
        except tarfile.CompressionError:
            if self.PASSWORD is not None or self.COMPRESSION is not None:
                pass
            else:
                raise
        except tarfile.ReadError:
            # can happen with all three modes
            pass
        except tarfile.DecryptionError:
            if self.PASSWORD is not None:
                pass
            else:
                raise

        os.chdir (self.pwd) # not restored due to the error above
        # but recover will succeed
        failed = dtar.rescue_backup(target_path=self.dst_path,
                                    backup_indexes_paths=[
                                        "%s/%s" % (bak_path, index_file)
                                    ])

        assert len (failed) == self.FAILURES

        # with one file missing
        missing  = []
        mismatch = []
        for key, value in self.hash.items ():
            kkey = "%s/%s" % (self.dst_path, key)
            if os.path.exists (kkey):
                if value != self.md5sum (kkey):
                    mismatch.append (key)
            else:
                missing.append (key)

        ssert len (missing)  == (self.MISSING if self.MISSING is not None
                                              else self.FAILURES)
        assert len (mismatch) == self.MISMATCHES

        shutil.rmtree (self.dst_path)


class RecoverCorruptPayloadTestBase (RecoverTest):
    COMPRESSION = None
    PASSWORD    = None
    FAILURES    = 0 # tarfile will restore but corrupted, as
    MISMATCHES  = 1 # revealed by the hash

class RecoverCorruptPayloadSingleTest (RecoverCorruptPayloadTestBase):
    VOLUMES     = 1

class RecoverCorruptPayloadMultiTest (RecoverCorruptPayloadTestBase):
    VOLUMES     = 3


class RecoverCorruptPayloadGZTestBase (RecoverTest):
    COMPRESSION = "#gz"
    PASSWORD    = None
    FAILURES    = 1
    MISMATCHES  = 0

class RecoverCorruptPayloadGZSingleTest (RecoverCorruptPayloadGZTestBase):
    VOLUMES     = 1

class RecoverCorruptPayloadGZMultiTest (RecoverCorruptPayloadGZTestBase):
    VOLUMES     = 3


class RecoverCorruptPayloadGZAESTestBase (RecoverTest):
    COMPRESSION = "#gz"
    PASSWORD    = TEST_PASSWORD
    FAILURES    = 1
    MISMATCHES  = 0

class RecoverCorruptPayloadGZAESSingleTest (RecoverCorruptPayloadGZAESTestBase):
    VOLUMES     = 1

class RecoverCorruptPayloadGZAESMultiTest (RecoverCorruptPayloadGZAESTestBase):
    VOLUMES     = 3


class RecoverCorruptHeaderTestBase (RecoverTest):
    COMPRESSION = None
    PASSWORD    = None
    FAILURES    = 1
    CORRUPT     = corrupt_header
    MISMATCHES  = 0

class RecoverCorruptHeaderSingleTest (RecoverCorruptHeaderTestBase):
    VOLUMES     = 1

class RecoverCorruptHeaderMultiTest (RecoverCorruptHeaderTestBase):
    VOLUMES     = 3


class RecoverCorruptHeaderGZTestBase (RecoverTest):
    COMPRESSION = "#gz"
    PASSWORD    = None
    FAILURES    = 1
    CORRUPT     = corrupt_header
    MISMATCHES  = 0

class RecoverCorruptHeaderGZSingleTest (RecoverCorruptHeaderGZTestBase):
    VOLUMES     = 1

class RecoverCorruptHeaderGZMultiTest (RecoverCorruptHeaderGZTestBase):
    VOLUMES     = 3


class RecoverCorruptHeaderGZAESTestBase (RecoverTest):
    COMPRESSION = "#gz"
    PASSWORD    = TEST_PASSWORD
    FAILURES    = 1
    CORRUPT     = corrupt_header
    MISMATCHES  = 0

class RecoverCorruptHeaderGZAESSingleTest (RecoverCorruptHeaderGZAESTestBase):
    VOLUMES     = 1

class RecoverCorruptHeaderGZAESMultiTest (RecoverCorruptHeaderGZAESTestBase):
    VOLUMES     = 3


class RecoverCorruptEntireHeaderTestBase (RecoverTest):
    COMPRESSION = None
    PASSWORD    = None
    FAILURES    = 1
    CORRUPT     = corrupt_entire_header
    MISMATCHES  = 0

class RecoverCorruptEntireHeaderSingleTest (RecoverCorruptEntireHeaderTestBase):
    VOLUMES     = 1

class RecoverCorruptEntireHeaderMultiTest (RecoverCorruptEntireHeaderTestBase):
    VOLUMES     = 3


class RecoverCorruptEntireHeaderGZTestBase (RecoverTest):
    COMPRESSION = "#gz"
    PASSWORD    = None
    FAILURES    = 1
    CORRUPT     = corrupt_entire_header
    MISMATCHES  = 0

class RecoverCorruptEntireHeaderGZSingleTest (RecoverCorruptEntireHeaderGZTestBase):
    VOLUMES     = 1

class RecoverCorruptEntireHeaderGZMultiTest (RecoverCorruptEntireHeaderGZTestBase):
    VOLUMES     = 3


class RecoverCorruptEntireHeaderGZAESTestBase (RecoverTest):
    COMPRESSION = "#gz"
    PASSWORD    = TEST_PASSWORD
    FAILURES    = 1
    CORRUPT     = corrupt_entire_header
    MISMATCHES  = 0

class RecoverCorruptEntireHeaderGZAESSingleTest (RecoverCorruptEntireHeaderGZAESTestBase):
    VOLUMES     = 1

class RecoverCorruptEntireHeaderGZAESMultiTest (RecoverCorruptEntireHeaderGZAESTestBase):
    VOLUMES     = 3


class RecoverCorruptTrailingDataTestBase (RecoverTest):
    # plain Tar is indifferent against traling data and the results
    # are consistent
    COMPRESSION = None
    PASSWORD    = None
    FAILURES    = 0
    CORRUPT     = corrupt_trailing_data
    MISMATCHES  = 0

class RecoverCorruptTrailingDataSingleTest (RecoverCorruptTrailingDataTestBase):
    VOLUMES     = 1

class RecoverCorruptTrailingDataMultiTest (RecoverCorruptTrailingDataTestBase):
    # the last object in first archive has extra bytes somewhere in the
    # middle because tar itself performs no data checksumming.
    MISMATCHES  = 1
    VOLUMES     = 3


class RecoverCorruptTrailingDataGZTestBase (RecoverTest):
    # reading past the final object will cause decompression failure;
    # all objects except for the last survive unharmed though
    COMPRESSION = "#gz"
    PASSWORD    = None
    FAILURES    = 1
    CORRUPT     = corrupt_trailing_data
    MISMATCHES  = 0

class RecoverCorruptTrailingDataGZSingleTest (RecoverCorruptTrailingDataGZTestBase):
    VOLUMES     = 1

class RecoverCorruptTrailingDataGZMultiTest (RecoverCorruptTrailingDataGZTestBase):
    VOLUMES     = 3
    # the last file of the first volume will only contain the data of the
    # second part which is contained in the second volume. this happens
    # because the CRC32 is wrong for the first part so it gets discarded, then
    # the object is recreated from the first header of the second volume,
    # containing only the remainder of the data.
    MISMATCHES  = 1
    MISSING     = 0


class RecoverCorruptTrailingDataGZAESTestBase (RecoverTest):
    COMPRESSION = "#gz"
    PASSWORD    = TEST_PASSWORD
    FAILURES    = 0
    CORRUPT     = corrupt_trailing_data
    MISMATCHES  = 0

class RecoverCorruptTrailingDataGZAESSingleTest (RecoverCorruptTrailingDataGZAESTestBase):
    VOLUMES     = 1

class RecoverCorruptTrailingDataGZAESMultiTest (RecoverCorruptTrailingDataGZAESTestBase):
    VOLUMES     = 3


class RecoverCorruptVolumeBaseTest (RecoverTest):
    COMPRESSION = None
    PASSWORD    = None
    FAILURES    = 8
    CORRUPT     = corrupt_volume
    VOLUMES     = 3

class RecoverCorruptVolumeTest (RecoverCorruptVolumeBaseTest):
    pass

class RecoverCorruptVolumeGZTest (RecoverCorruptVolumeBaseTest):
    COMPRESSION = "#gz"

class RecoverCorruptVolumeGZAESTest (RecoverCorruptVolumeBaseTest):
    COMPRESSION = "#gz"
    PASSWORD    = TEST_PASSWORD


class RecoverCorruptHoleBaseTest (RecoverTest):
    """
    Cut bytes from the middle of a volume.

    Index-based recovery works only up to the hole.
    """
    COMPRESSION = None
    PASSWORD    = None
    FAILURES    = 3
    CORRUPT     = corrupt_hole
    VOLUMES     = 2 # request two vols to swell up the first one
    MISMATCHES  = 1

class RecoverCorruptHoleTest (RecoverCorruptHoleBaseTest):
    FAILURES    = 2

class RecoverCorruptHoleGZTest (RecoverCorruptHoleBaseTest):
    COMPRESSION = "#gz"
    MISSING     = 2

class RecoverCorruptHoleGZAESTest (RecoverCorruptHoleBaseTest):
    COMPRESSION = "#gz"
    PASSWORD    = TEST_PASSWORD
    MISSING     = 2