# Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import logging, os, tempfile, shutil, stat, time, posix
from autotest_lib.client.bin import test, utils
from autotest_lib.client.common_lib import error

# TODO:
#  - mock out TPM and check all error conditions
#  - test failure when things aren't mounted correctly

class test_checker(object):
    def __init__(self):
        logging.info("test_checker.__init__")
        # Empty failure list means test passes.
        self._failures = []

    def _passed(self, msg):
        logging.info('ok: %s' % (msg))

    def _failed(self, msg):
        logging.error('FAIL: %s' % (msg))
        self._failures.append(msg)

    def _fatal(self, msg):
        logging.error('FATAL: %s' % (msg))
        raise error.TestError(msg)

    def check(self, boolean, msg, fatal=False):
        if boolean == True:
            self._passed(msg)
        else:
            msg = "could not satisfy '%s'" % (msg)
            if fatal:
                self._fatal(msg)
            else:
                self._failed(msg)

    def test_raise(self):
        # Raise a failure if anything unexpected was seen.
        if len(self._failures):
            raise error.TestFail((", ".join(self._failures)))

chk = test_checker()


class EncryptedStateful(object):
    def _prepare_simulated_root(self):
        os.makedirs(self.var)
        os.makedirs(self.chronos)
        os.makedirs(self.stateful)

        # Build fake stateful block device (emulate 10G sda1).
        self.stateful_block = os.path.join(self.root, 'stateful.block')
        utils.system("truncate -s 10G %s" % (self.stateful_block))
        utils.system("mkfs.ext4 -F %s" % (self.stateful_block))
        utils.system("mount -n -t ext4 -o loop,noatime,commit=600 %s %s" %
                     (self.stateful_block, self.stateful))

    def __init__(self, root=None):
        if root == None:
            self.root = tempfile.mkdtemp(dir='/mnt/stateful_partition',
                                         prefix='.test-enc-stateful-')
            self.simulated = True
        else:
            self.root = root
            self.simulated = False

        self.var = os.path.join(self.root, 'var')
        self.chronos = os.path.join(self.root, 'home', 'chronos')
        self.stateful = os.path.join(self.root, 'mnt', 'stateful_partition')
        self.mount_log = os.path.join(self.stateful, 'mount.log')
        self.key = os.path.join(self.stateful, 'encrypted.key')
        self.needs_finalization = os.path.join(self.stateful,
                                               'encrypted.needs-finalization')
        self.block = os.path.join(self.stateful, 'encrypted.block')
        self.encrypted = os.path.join(self.stateful, 'encrypted')

        if self.simulated:
            try:
                self._prepare_simulated_root()
            except:
                shutil.rmtree(self.root)
                raise

        self.mounted = not self.simulated

    def mount(self, args=""):
        if self.mounted or not self.simulated:
            return
        # TODO(keescook): figure out what is killing the resizer and
        # remove the explicit use of "tee" here.
        # Without the pipe to "tee", mount-encrypted's forked resizing
        # process gets killed, even though it is using daemon(). (Is
        # autotest doing something odd here?) This leaves the filesystem
        # unresized. It would be better to have the resizer running in
        # the background, as it is designed, so we can examine its behavior
        # during testing (e.g. "does the filesystem actually grow?").
        utils.system("MOUNT_ENCRYPTED_ROOT=%s mount-encrypted %s 2>&1 "
                     "| tee %s" % (self.root, args, self.mount_log))
        self.mounted = True

    def umount(self):
        if not self.mounted or not self.simulated:
            return
        utils.system("MOUNT_ENCRYPTED_ROOT=%s mount-encrypted umount" %
                         (self.root))
        self.mounted = False

    # Clean up when destroyed.
    def __del__(self):
        if self.simulated:
            self.umount()
            utils.system("umount -n %s" % (self.stateful))
            shutil.rmtree(self.root)

    # Perform common post-mount size/owner checks on the filesystem and
    # backing files.
    def check_sizes(self, finalized=True):
        # Do we have the expected backing files?
        chk.check(os.path.exists(self.block), "%s exists" % (self.block))
        if finalized:
            keyfile = self.key
            other = self.needs_finalization
        else:
            keyfile = self.needs_finalization
            other = self.key
        chk.check(os.path.exists(keyfile), "%s exists" % (keyfile))
        chk.check(not os.path.exists(other), "%s does not exist" % (other))

        # Sanity check the key file stat.
        info = os.stat(keyfile)
        chk.check(stat.S_ISREG(info.st_mode),
                  "%s is regular file" % (keyfile))
        chk.check(info.st_uid == 0, "%s is owned by root" % (keyfile))
        chk.check(info.st_gid == 0, "%s has group root" % (keyfile))
        chk.check(stat.S_IMODE(info.st_mode) == (stat.S_IRUSR | stat.S_IWUSR),
                  "%s is S_IRUSR | S_IWUSR" % (keyfile))
        chk.check(info.st_size == 48, "%s is 48 bytes" % (keyfile))

        # Sanity check the block file stat.
        info = os.stat(self.block)
        chk.check(stat.S_ISREG(info.st_mode),
                  "%s is regular file" % (self.block))
        chk.check(info.st_uid == 0, "%s is owned by root" % (self.block))
        chk.check(info.st_gid == 0, "%s has group root" % (self.block))
        chk.check(stat.S_IMODE(info.st_mode) == (stat.S_IRUSR | stat.S_IWUSR),
                  "%s is S_IRUSR | S_IWUSR" % (self.block))
        # Make sure block file is roughly a third of the size of the root
        # filesystem (within 5%).
        top = os.statvfs(self.stateful)
        backing_size = float(info.st_size)
        third = top.f_blocks * top.f_frsize * .3
        chk.check(backing_size > (third * .95)
                  and backing_size < (third * 1.05),
                  "%s is near %d bytes (was %d)" % (self.block, third,
                                                    info.st_size))

        # Wait for resize manager task to finish.
        utils.poll_for_condition(lambda: utils.system("pgrep mount-encrypted",
                                                      ignore_status=True) != 0,
                                 error.TestError('resizer still running'))

        # Verify filesystem is within 5% of the block file size.
        info = os.statvfs(self.encrypted)
        encrypted_size = float(info.f_frsize) * float(info.f_blocks)
        chk.check(encrypted_size / backing_size > 0.95,
                  "%s fs (%d) is nearly the backing device size (%d)" %
                  (self.encrypted, encrypted_size, backing_size))
        # Verify there is a reasonable number of inodes in the encrypted
        # filesystem (near 25% inodes-to-blocks ratio).
        inode_ratio = float(info.f_files) / float(info.f_blocks)
        chk.check(inode_ratio > 0.20 and inode_ratio < 0.30,
                  "%s has close to 25%% ratio of inodes-to-blocks (%.2f%%)" %
                  (self.encrypted, inode_ratio*100))

        # Raise non-fatal failures now, if they were encountered.
        chk.test_raise()

    # Wait for kernel background writing to finish.
    def _backing_stabilize(self):
        start = None
        size = 0
        while True:
            k = long(utils.system_output("du -sk %s" % (self.block),
                                         retain_output = True).split()[0])
            if start == None:
                start = k
            if k == size:
                # Backing file has remained the same size for 10 seconds.
                # Assume the kernel is done with background initialization.
                break
            time.sleep(10)
            utils.system("sync")
            size = k
        logging.info("%s stabilized at %dK (was %dK)" %
                     (self.block, size, start))

    # Check that the backing file reclaims space when filesystem contents
    # are deleted.
    def check_reclamation(self):
        # This test is sensitive to other things happening on the filesystem,
        # so we must wait for background initialization to finish first.
        self._backing_stabilize()

        megs = 200
        data = os.path.join(self.var, "check_reclamation")
        orig = os.statvfs(self.stateful)

        # 200M file added to encrypted filesystem.
        utils.system("dd if=/dev/zero of=%s bs=1M count=%s; sync" % (data,
                                                                     megs))
        # Wait for background allocations to finish.
        self._backing_stabilize()
        filled = os.statvfs(self.stateful)

        # 200M file removed from encrypted filesystem.
        utils.system("rm %s; sync" % (data))
        # Wait for background hole-punching to finish.
        self._backing_stabilize()
        done = os.statvfs(self.stateful)

        # Did the underlying filesystem grow by the size of the test file?
        file_blocks_used = float((megs * 1024 * 1024) / orig.f_frsize)
        fs_blocks_used = float(orig.f_bfree - filled.f_bfree)
        chk.check(file_blocks_used / fs_blocks_used > 0.95,
                  "%d file blocks account for most of %d fs blocks" %
                  (file_blocks_used, fs_blocks_used))

        # Did the underlying filesystem shrink on removal?
        fs_blocks_done = float(orig.f_bfree - done.f_bfree)
        chk.check(fs_blocks_done / file_blocks_used < 0.05,
                  "most of %d fs blocks reclaimed (%d fs blocks left over, "
                  "free: %d -> %d -> %d)" %
                  (fs_blocks_used, fs_blocks_done,
                   orig.f_bfree, filled.f_bfree, done.f_bfree))

        # Raise non-fatal failures now, if they were encountered.
        chk.test_raise()


class platform_EncryptedStateful(test.test):
    version = 1

    def existing_partition(self):
        # Examine the existing encrypted partition.
        encstate = EncryptedStateful("/")

        # Perform post-mount sanity checks (and handle unfinalized devices).
        encstate.check_sizes(finalized=os.path.exists(encstate.key))

    def factory_key(self):
        # Create test root directory.
        encstate = EncryptedStateful()

        # Make sure we haven't run here before.
        chk.check(not os.path.exists(encstate.key),
                  "%s does not exist" % (encstate.key))
        chk.check(not os.path.exists(encstate.block),
                  "%s does not exist" % (encstate.block))

        # Mount a fresh encrypted stateful, with factory static key.
        encstate.mount("factory")

        # Perform post-mount sanity checks.
        encstate.check_sizes()

        # Check disk reclamation.
        encstate.check_reclamation()

        # Check explicit umount.
        encstate.umount()

    def no_tpm(self):
        encstate = EncryptedStateful()

        # Relocate the TPM device during mount.
        tpm = "/dev/tpm0"
        off = "%s.off" % (tpm)
        try:
            if os.path.exists(tpm):
                utils.system("mv %s %s" % (tpm, off))
            # Mount without a TPM.
            encstate.mount()
        finally:
            if os.path.exists(off):
                utils.system("mv %s %s" % (off, tpm))

        # Perform post-mount sanity checks.
        encstate.check_sizes(finalized=False)

    def run_once(self):
        # Do a no-write test of system's existing encrypted partition.
        self.existing_partition()

        # Do a no-write, no-TPM test with sanity checks.
        self.no_tpm()

        # There is no interactively controllable TPM mock yet for
        # mount-encrypted, so we can only test the static key currently.
        self.factory_key()