# 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 = 'sha1'
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 dummy 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)
if type(copy_files) is list:
for file in copy_files:
pass # TODO(wad)
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, 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, 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()
# Example callback for mod_and_test that does nothing
def mod_nothing(self, run_count, backing_path, block_size, block_count):
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