Introduce SAFT TPM testing support.

saft_utility.py

 #!/usr/bin/python
 # Copyright (c) 2010 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.
 
 '''Test utility for verifying ChromeOS firmware.'''
 
 import datetime
 import functools
 import getopt
 import os
 import re
 import shutil
 import sys
 import tempfile
 import traceback
 
 import cgpt_handler
 import chromeos_interface
 import flashrom_handler
 import kernel_handler
 import saft_flashrom_util
+import tpm_handler
 
 #
 # We need to know the names of two files:
 #
 # - the SAFT shell script used by the upstart script (/etc/init/saft.conf) to
 #   determine if SAFT is in progress
 #
 # - the name of the log file where the upstart script redirects its output.
 #
 # These file names are defined in the upstart script as <NAME>=<VALUE>, one
@@ skipping 33 matching lines @@
     mkdir "${mount_point}"
     mount "${saft_dev}" "${mount_point}"
 fi
 
 cd "${mount_point}%s"
 
 # Start our own X server to make it easier to control the screen
 X_PORT=1
 X :"${X_PORT}" -config saft.xorg.conf vt02 > /dev/null 2>&1 &
 xpid="$!"
+initctl stop update-engine  # Prevent messing with cgpt attributes
 export DISPLAY=":${X_PORT}.0"
 ./%s --next_step
 kill "${xpid}"
 '''
 
 # Subdirectory to keep the state of this test over reboots. Created in the
 # /var on the USB flash drive the test is running from.
 STATE_SUBDIR = '.fw_test'
 
 # Files storing SAFT state over reboots, located in state_dir defined below.
@@ skipping 68 matching lines @@
     def __init__(self):
         '''Object initializer, does nothing to make mocking easier.'''
         self.mydir = None
         self.base_partition = None
         self.chros_if = None
         self.progname = None
         self.test_state_sequence = None
         self.kern_handler = None
         self.step_failed = False
         self.window = None
+        self.tpm_handler = None
 
     def _verify_fw_id(self, compare_to_file):
         '''Verify if the current firmware ID matches the contents a file.
 
         compare_to_file - a string, name of the file in the state directory.
         '''
         old_fwid = open(
             self.chros_if.state_dir_file(compare_to_file), 'r').read()
         now_fwid = open(self.chros_if.acpi_file('FWID'), 'r').read()
         return old_fwid == now_fwid
@@ skipping 92 matching lines @@
                 mount_point = line.split()[2]
                 state_root = mount_point + env_root
                 break
         else:
             tmp_dir = tempfile.mkdtemp()
             self.chros_if.run_shell_command('mount %s %s' % (state_fs, tmp_dir))
             state_root = '%s%s' % (tmp_dir, env_root)
 
         self.chros_if.init(os.path.join(state_root, STATE_SUBDIR), LOG_FILE)
 
-    def init(self, progname, chros_if, kern_handler=None, state_sequence=None):
+    def init(self, progname, chros_if,
+             kern_handler=None, state_sequence=None, tpm_handler=None):
         '''Initialize the Firmware self test instance.
 
         progname - a string, name of this program as it was invoked.
 
         chros_if - an object of ChromeOSInterface type to be initialized and
                    used by this instance of FirmwareTest.
 
         kern_handler - an object providing SAFT with services manipulating
                    kernel images. Unittest does not provide this object.
 
         test_state_sequence - a tuple of three-tuples driving test execution,
                    see description below. Unittest does not provide this
                    sequence.
         '''
         real_name = os.path.realpath(progname)
         self.mydir = os.path.dirname(real_name)
         self.progname = os.path.basename(real_name)
         self.chros_if = chros_if
         self._check_runtime_env()
         self.test_state_sequence = state_sequence
         if kern_handler:
             self.kern_handler = kern_handler
             self.kern_handler.init(self.chros_if)
+        if tpm_handler:
+            self.tpm_handler = tpm_handler
+            self.tpm_handler.init(self.chros_if)
+
+        if self.kern_handler and self.tpm_handler:
+            if not self.chros_if.is_removable_device(self.base_partition):
+                # On each non USB flash rooted start confirm that kernel
+                # versions and TPM setting match.
+                vers_a = self.kern_handler.get_version('a')
+                vers_b = self.kern_handler.get_version('b')
+                if not self.tpm_handler.kernel_version_good(vers_a, vers_b):
+                    raise FwError('TPM kernel version mismatch (%d %d %s)' % (
+                            vers_a, vers_b, self.base_partition))
 
     def set_try_fw_b(self):
         '''Request running firmware B on the next restart.'''
         self.chros_if.log('Requesting restart with FW B')
         self.chros_if.run_shell_command('reboot_mode --try_firmware_b=1')
 
+    def request_recovery_boot(self):
+        '''Request running in recovery mode on the restart.'''
+        self.chros_if.log('Requesting restart in recovery mode')
+        self.chros_if.run_shell_command('reboot_mode --recovery=1')
+
     @allow_multiple_section_input
     def restore_firmware(self, section):
         '''Restore the requested firmware section (previously corrupted).'''
         self.chros_if.log('Restoring firmware %s' % section)
         FLASHROM_HANDLER.new_image()
         FLASHROM_HANDLER.restore_firmware(section)
 
     @allow_multiple_section_input
     def corrupt_firmware(self, section):
         '''Corrupt the requested firmware section.'''
@@ skipping 40 matching lines @@
         for part, state in expected_cgpt.iteritems():
             props = cgpth.get_partition(BASE_STORAGE_DEVICE, part)
             for prop, value in state.iteritems():
                 if value != props[prop]:
                     self.chros_if.log('wrong partition %s value' % part)
                     self.chros_if.log(cgpth.dump_partition(BASE_STORAGE_DEVICE,
                                                            part))
                     self.step_failed = True
                     break
 
+    def move_kernel_backward(self, section):
+        '''Decrement kernel version for the requested section.'''
+        new_version = self.kern_handler.get_version(section) - 1
+        self.chros_if.log(
+            'setting section %s version to %d' % (section, new_version))
+        self.kern_handler.set_version(section, new_version)
+
+    def jump_kernels_forward(self):
+        '''Add two to both kernels' versions.
+
+        This compensates for the previous decrement and increases the version
+        number by one as compared to the original state (at the start of the
+        test).
+        '''
+        for section in ('a', 'b'):
+            new_version = self.kern_handler.get_version(section) + 2
+            self.chros_if.log(
+                'setting section %s version to %d' % (section, new_version))
+            self.kern_handler.set_version(section, new_version)
+
+    def terminate_tpm_tests(self):
+        '''Restore TPM and kernel states.
+
+        Move kernel versions to where they were before SAFT started, set the
+        TPM kernel version to the biggest of the two kernels and prevent
+        future TPM write accesses by restoring the upstart file.
+        '''
+        new_tpm_version  = 0
+        for section in ('a', 'b'):
+            new_version = self.kern_handler.get_version(section) - 1
+            self.chros_if.log(
+                'setting section %s version to %d' % (section, new_version))
+            self.kern_handler.set_version(section, new_version)
+            new_tpm_version = max(new_version, new_tpm_version)
+        self.tpm_handler.set_kernel_version(new_tpm_version)
+        self.tpm_handler.disable_write_access()
+
     def revert_firmware(self):
         '''Restore firmware to the image backed up when SAFT started.'''
         self.chros_if.log('restoring original firmware image')
         self.chros_if.run_shell_command(
             'flashrom -w %s' % self.chros_if.state_dir_file(FW_BACKUP_FILE))
 
+    def new_fw_image(self, image_file = None):
+        FLASHROM_HANDLER.new_image(image_file)
+        FLASHROM_HANDLER.verify_image()
+        if self.tpm_handler:
+            vers_a = FLASHROM_HANDLER.get_section_version('a')
+            vers_b = FLASHROM_HANDLER.get_section_version('b')
+            if not self.tpm_handler.fw_version_good(vers_a, vers_b):
+                raise FwError('TPM firmware version mismatch')
+
+    def prepare_tpm_tests(self):
+        '''Prepare TPM for testing.
+
+        Enable write access on the next reboot, and roll back the kernel we
+        are running now (the other kernel is expected to be used on the next
+        restart).
+        '''
+        mount_point = self.chros_if.run_shell_command_get_output(
+            'df %s' % self.mydir)[-1].split()[-1]
+        cfg_file = os.path.join(mount_point, 'etc/init/tcsd.conf')
+        self.tpm_handler.enable_write_access(cfg_file)
+        self.move_kernel_backward('a')
+
     def init_fw_test(self, opt_dictionary, chros_if):
         '''Prepare firmware test context.
 
       This function tries creating the state directory for the fw test and
       initializes the test state machine.
 
       Return
         True on success
         False on any failure or if the directory already exists
       '''
         chros_if.init_environment()
         chros_if.log('Automated firmware test log generated on %s' % (
-                datetime.datetime.strftime(datetime.datetime.now(),
-                                               '%b %d %Y')))
+                datetime.datetime.strftime(
+                    datetime.datetime.now(), '%b %d %Y')))
         chros_if.log('Original boot state %s' % chros_if.boot_state_vector())
         self.chros_if = chros_if
         fw_image = opt_dictionary['image_file']
-        FLASHROM_HANDLER.new_image()
-        FLASHROM_HANDLER.verify_image()
+        self.new_fw_image()
         FLASHROM_HANDLER.dump_whole(
             self.chros_if.state_dir_file(FW_BACKUP_FILE))
-        FLASHROM_HANDLER.new_image(fw_image)
-        FLASHROM_HANDLER.verify_image()
+        self.new_fw_image(fw_image)
         self._handle_saft_script(True)
         shutil.copyfile(self.chros_if.acpi_file('FWID'),
                         self.chros_if.state_dir_file(FWID_BACKUP_FILE))
         shutil.copyfile(fw_image, self.chros_if.state_dir_file(FW_COPY_FILE))
+
         self._set_step(0)
 
     def next_step(self):
         '''Function to execute a single SAFT step.
 
       This function is running after each reboot. It determines the current
       step the SAFT is on, executes the appropriate action, increments the
       step value and then restats the machine.
       '''
 
@@ skipping 11 matching lines @@
 
             if self._verify_fw_id(FWID_BACKUP_FILE):
           # we expected FWID to change, but it did not - have the firmware
           # been even replaced?
                 self.chros_if.log('New firmware - old FWID')
                 self.finish_saft(False)
         test_state_tuple = self.test_state_sequence[this_step]
         expected_vector = test_state_tuple[0]
         action = test_state_tuple[1]
         boot_vector = self.chros_if.boot_state_vector()
-        self.chros_if.log('Rebooted into state %s on step %d' % (
+        self.chros_if.log('\nRebooted into state %s on step %d' % (
                 boot_vector, this_step))
 
         conf_log_file = RetriveSaftConfDefinion(CONF_LOG_VAR_NAME)
 
         if os.path.exists(conf_log_file):
             contents = open(conf_log_file).read().rstrip()
             if len(contents):
                 self.chros_if.log('startup log contents:')
                 self.chros_if.log(contents)
 
@@ skipping 61 matching lines @@
 #
 # The second element of the component tuples is the action to take to
 # advance the test. The action is a function to call. The last line has
 # action set to None, which indicates to the state machine that the test
 # is over.
 #
 # The third component, if present, is the parameter to pass to the action
 # function.
 
 TEST_STATE_SEQUENCE = (
-    ('1:1:1:0:3', FST.set_try_fw_b),
+    ('1:1:1:0:3', FST.set_try_fw_b),                   # Step 0
     ('1:2:1:0:3', None),
     ('1:1:1:0:3', FST.corrupt_firmware, 'a'),
     ('1:2:1:0:3', FST.restore_firmware, 'a'),
     ('1:1:1:0:3', FST.corrupt_firmware, ('a', 'b')),
-    ('5:0:1:1:3', FST.restore_firmware, ('a', 'b')),
+    ('5:0:1:1:3', FST.restore_firmware, ('a', 'b')),   # Step 5
     ('1:1:1:0:3', FST.corrupt_kernel, 'a'),
     ('1:1:1:0:5', FST.corrupt_kernel, 'b'),
     ('6:0:1:1:3', FST.restore_kernel, ('a', 'b')),
-    ('1:1:1:0:3', FST.cgpt_test_start),
+    ('1:1:1:0:3', FST.request_recovery_boot),
+    ('8:0:1:1:3', FST.prepare_tpm_tests),              # Step 10
+    ('1:1:1:0:5', FST.move_kernel_backward, 'b'),
+    ('6:0:1:1:3', FST.jump_kernels_forward),
+    ('1:1:1:0:3', FST.request_recovery_boot),
+    ('8:0:1:1:3', FST.terminate_tpm_tests),
+    ('1:1:1:0:3', FST.cgpt_test_start),                # Step 15
     ('1:1:1:0:3', FST.cgpt_test_1),
     ('1:1:1:0:3', FST.revert_firmware),
     ('1:1:1:0:3', None),
     )
 
 
 # The string below serves two purposes:
 #
 # - spell out the usage string for this program
 #
@@ skipping 63 matching lines @@
     '''Process command line options and invoke the proper test entry point.'''
     (opts, params) = getopt.gnu_getopt(argv[1:], '', get_options_set())
     if params:
         raise FwError('unrecognized parameters: %s' % ' '.join(params))
 
     opt_dictionary = {}
     for (name, value) in opts:
         opt_dictionary[name.lstrip('-')] = value
 
     FST.init(argv[0], CHROS_IF,
-             kernel_handler.KernelHandler(), TEST_STATE_SEQUENCE)
+             kernel_handler.KernelHandler(), TEST_STATE_SEQUENCE,
+             tpm_handler.TpmHandler())
 
     FLASHROM_HANDLER.init(saft_flashrom_util, CHROS_IF,
                          opt_dictionary.get('pub_key'))
     if 'next_step' in opt_dictionary:
         if len(opt_dictionary) != 1:
             usage('--next_step (when specified) must be the only parameter', 1)
         try:
             FST.next_step()
         except SystemExit:
             pass  # No real exception, this is just an exit (for whatever
@@ skipping 35 matching lines @@
         main(sys.argv)
     except (getopt.GetoptError, ImportError):
         usage(sys.exc_info()[1], 1)
     except (FwError, flashrom_handler.FlashromHandlerError):
         MSG = 'Error: %s' % str(sys.exc_info()[1])
         print MSG
         CHROS_IF.log(MSG)
         sys.exit(1)
 
     sys.exit(0)