xref: /aosp_15_r20/external/autotest/client/cros/verity_utils.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

# Lint as: python2, python3
# 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, mmap, os, time

import common
from autotest_lib.client.bin import os_dep, test
from autotest_lib.client.common_lib import error, logging_manager, utils

""" a wrapper for using verity/dm-verity with a test backing store """

# enum for the 3 possible values of the module parameter.
ERROR_BEHAVIOR_ERROR = 'eio'
ERROR_BEHAVIOR_REBOOT = 'panic'
ERROR_BEHAVIOR_IGNORE = 'none'
ERROR_BEHAVIOR_NOTIFIER = 'notify'  # for platform specific behavior.

# Default configuration for verity_image
DEFAULT_TARGET_NAME = 'verity_image'
DEFAULT_ALG = 'sha256'
DEFAULT_IMAGE_SIZE_IN_BLOCKS = 100
DEFAULT_ERROR_BEHAVIOR = ERROR_BEHAVIOR_ERROR
# TODO(wad) make this configurable when dm-verity doesn't hard-code 4096.
BLOCK_SIZE = 4096

def system(command, timeout=None):
    """Delegate to utils.system to run |command|, logs stderr only on fail.

    Runs |command|, captures stdout and stderr.  Logs stdout to the DEBUG
    log no matter what, logs stderr only if the command actually fails.
    Will time the command out after |timeout|.
    """
    utils.run(command, timeout=timeout, ignore_status=False,
              stdout_tee=utils.TEE_TO_LOGS, stderr_tee=utils.TEE_TO_LOGS,
              stderr_is_expected=True)

class verity_image(object):
    """ a helper for creating dm-verity targets for testing.

        To use,
          vi = verity_image()
          vi.initialize(self.tmpdir, "dmveritytesta")
          # Create a 409600 byte image with /bin/ls on it
          # The size in bytes is returned.
          backing_path = vi.create_backing_image(100, copy_files=['/bin/ls'])
          # Performs hashing of the backing_path and sets up a device.
          loop_dev = vi.prepare_backing_device()
          # Sets up the mapped device and returns the path:
          # E.g., /dev/mapper/autotest_dmveritytesta
          dev = vi.create_verity_device()
          # Access the mapped device using the returned string.

       TODO(wad) add direct verified and backing store access functions
                 to make writing modifiers easier (e.g., mmap).
    """
    # Define the command template constants.
    verity_cmd = \
        'verity mode=create alg=%s payload=%s payload_blocks=%d hashtree=%s'
    dd_cmd = 'dd if=/dev/zero of=%s bs=4096 count=0 seek=%d'
    mkfs_cmd = 'mkfs.ext3 -b 4096 -F %s'
    dmsetup_cmd = "dmsetup -r create autotest_%s --table '%s'"

    def _device_release(self, cmd, device):
        if utils.system(cmd, ignore_status=True) == 0:
            return
        logging.warning("Could not release %s. Retrying...", device)
        # Other things (like cros-disks) may have the device open briefly,
        # so if we initially fail, try again and attempt to gather details
        # on who else is using the device.
        fuser = utils.system_output('fuser -v %s' % (device),
                                    retain_output=True,
                                    ignore_status=True)
        lsblk = utils.system_output('lsblk %s' % (device),
                                    retain_output=True,
                                    ignore_status=True)
        time.sleep(1)
        if utils.system(cmd, ignore_status=True) == 0:
            return
        raise error.TestFail('"%s" failed: %s\n%s' % (cmd, fuser, lsblk))

    def reset(self):
        """Idempotent call which will free any claimed system resources"""
        # Pre-initialize these values to None
        for attr in ['mountpoint', 'device', 'loop', 'file', 'hash_file']:
            if not hasattr(self, attr):
                setattr(self, attr, None)
        logging.info("verity_image is being reset")

        if self.mountpoint is not None:
            system('umount %s' % self.mountpoint)
            self.mountpoint = None

        if self.device is not None:
            self._device_release('dmsetup remove %s' % (self.device),
                                 self.device)
            self.device = None

        if self.loop is not None:
            self._device_release('losetup -d %s' % (self.loop), self.loop)
            self.loop = None

        if self.file is not None:
            os.remove(self.file)
            self.file = None

        if self.hash_file is not None:
            os.remove(self.hash_file)
            self.hash_file = None

        self.alg = DEFAULT_ALG
        self.error_behavior = DEFAULT_ERROR_BEHAVIOR
        self.blocks = DEFAULT_IMAGE_SIZE_IN_BLOCKS
        self.file = None
        self.has_fs = False
        self.hash_file = None
        self.table = None
        self.target_name = DEFAULT_TARGET_NAME

        self.__initialized = False

    def __init__(self):
        """Sets up the defaults for the object and then calls reset()
        """
        self.reset()

    def __del__(self):
        # Release any and all system resources.
        self.reset()

    def _create_image(self):
        """Creates a placeholder file."""
        # TODO(wad) replace with python
        utils.system_output(self.dd_cmd % (self.file, self.blocks))

    def _create_fs(self, copy_files):
        """sets up ext3 on the image"""
        self.has_fs = True
        system(self.mkfs_cmd % self.file)

    def _hash_image(self):
        """runs verity over the image and saves the device mapper table"""
        self.table = utils.system_output(self.verity_cmd % (self.alg,
                                                            self.file,
                                                            self.blocks,
                                                            self.hash_file))
        # The verity tool doesn't include a templated error value.
        # For now, we add one.
        self.table += " error_behavior=ERROR_BEHAVIOR"
        logging.info("table is %s", self.table)

    def _append_hash(self):
        f = open(self.file, 'ab')
        f.write(utils.read_file(self.hash_file))
        f.close()

    def _setup_loop(self):
        # Setup a loop device
        self.loop = utils.system_output('losetup -f --show %s' % (self.file))

    def _setup_target(self):
        # Update the table with the loop dev
        self.table = self.table.replace('HASH_DEV', self.loop)
        self.table = self.table.replace('ROOT_DEV', self.loop)
        self.table = self.table.replace('ERROR_BEHAVIOR', self.error_behavior)

        system(self.dmsetup_cmd % (self.target_name, self.table))
        self.device = "/dev/mapper/autotest_%s" % self.target_name

    def initialize(self,
                   tmpdir,
                   target_name,
                   alg=DEFAULT_ALG,
                   size_in_blocks=DEFAULT_IMAGE_SIZE_IN_BLOCKS,
                   error_behavior=DEFAULT_ERROR_BEHAVIOR):
        """Performs any required system-level initialization before use.
        """
        try:
            os_dep.commands('losetup', 'mkfs.ext3', 'dmsetup', 'verity', 'dd',
                            'dumpe2fs')
        except ValueError as e:
            logging.error('verity_image cannot be used without: %s', e)
            return False

        # Used for the mapper device name and the tmpfile names.
        self.target_name = target_name

        # Reserve some files to use.
        self.file = os.tempnam(tmpdir, '%s.img.' % self.target_name)
        self.hash_file = os.tempnam(tmpdir, '%s.hash.' % self.target_name)

        # Set up the configurable bits.
        self.alg = alg
        self.error_behavior = error_behavior
        self.blocks = size_in_blocks

        self.__initialized = True
        return True

    def create_backing_image(self, size_in_blocks, with_fs=True,
                             copy_files=None):
        """Creates an image file of the given number of blocks and if specified
           will create a filesystem and copy any files in a copy_files list to
           the fs.
        """
        self.blocks = size_in_blocks
        self._create_image()

        if with_fs is True:
            self._create_fs(copy_files)
        else:
            if type(copy_files) is list and len(copy_files) != 0:
                logging.warning("verity_image.initialize called with " \
                             "files to copy but no fs")

        return self.file

    def prepare_backing_device(self):
        """Hashes the backing image, appends it to the backing image, points
           a loop device at it and returns the path to the loop."""
        self._hash_image()
        self._append_hash()
        self._setup_loop()
        return self.loop

    def create_verity_device(self):
        """Sets up the device mapper node and returns its path"""
        self._setup_target()
        return self.device

    def verifiable(self):
        """Returns True if the dm-verity device does not throw any errors
           when being walked completely or False if it does."""
        try:
            if self.has_fs is True:
                system('dumpe2fs %s' % self.device)
            # TODO(wad) replace with mmap.mmap-based access
            system('dd if=%s of=/dev/null bs=4096' % self.device)
            return True
        except error.CmdError as e:
            return False


class VerityImageTest(test.test):
    """VerityImageTest provides a base class for verity_image tests
       to be derived from.  It sets up a verity_image object for use
       and provides the function mod_and_test() to wrap simple test
       cases for verity_images.

       See platform_DMVerityCorruption as an example usage.
    """
    version = 1
    image_blocks = DEFAULT_IMAGE_SIZE_IN_BLOCKS

    def initialize(self):
        """Overrides test.initialize() to setup a verity_image"""
        self.verity = verity_image()

    def mod_nothing(self, run_count, backing_path, block_size, block_count):
        """Example callback for mod_and_test that does nothing."""
        pass

    def mod_and_test(self, modifier, count, expected):
        """Takes in a callback |modifier| and runs it |count| times over
           the verified image checking for |expected| out of verity.verifiable()
        """
        tries = 0
        while tries < count:
            # Start fresh then modify each block in the image.
            self.verity.reset()
            self.verity.initialize(self.tmpdir, self.__class__.__name__)
            backing_path = self.verity.create_backing_image(self.image_blocks)
            loop_dev = self.verity.prepare_backing_device()

            modifier(tries,
                     backing_path,
                     BLOCK_SIZE,
                     self.image_blocks)

            mapped_dev = self.verity.create_verity_device()

            # Now check for failure.
            if self.verity.verifiable() is not expected:
                raise error.TestFail(
                    '%s: verity.verifiable() not as expected (%s)' %
                    (modifier.__name__, expected))
            tries += 1