client/cros/saft/flashrom_handler.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.

"""A module to support automated testing of ChromeOS firmware.

Utilizes services provided by saft_flashrom_util.py read/write the
flashrom chip and to parse the flash rom image.

See docstring for FlashromHandler class below.
"""

import hashlib
import os
import struct
import tempfile
from chromeos_interface import ChromeOSInterfaceError

class FvSection(object):
    """An object to hold information about a firmware section.

    This includes file names for the signature header and the body, and the
    version number.
    """

    def __init__(self, sig_name, body_name):
        self._sig_name = sig_name
        self._body_name = body_name
        self._version = -1  # Is not set on construction.
        self._flags = 0  # Is not set on construction.
        self._sha = None  # Is not set on construction.
        self._sig_sha = None # Is not set on construction.
        self._datakey_version = -1 # Is not set on construction.
        self._kernel_subkey_version = -1 # Is not set on construction.

    def names(self):
        return (self._sig_name, self._body_name)

    def get_sig_name(self):
        return self._sig_name

    def get_body_name(self):
        return self._body_name

    def get_version(self):
        return self._version

    def get_flags(self):
        return self._flags

    def get_sha(self):
        return self._sha

    def get_sig_sha(self):
        return self._sig_sha

    def get_datakey_version(self):
        return self._datakey_version

    def get_kernel_subkey_version(self):
        return self._kernel_subkey_version

    def set_version(self, version):
        self._version = version

    def set_flags(self, flags):
        self._flags = flags

    def set_sha(self, sha):
        self._sha = sha

    def set_sig_sha(self, sha):
        self._sig_sha = sha

    def set_datakey_version(self, version):
        self._datakey_version = version

    def set_kernel_subkey_version(self, version):
        self._kernel_subkey_version = version

class FlashromHandlerError(Exception):
    pass


class FlashromHandler(object):
    """An object to provide logical services for automated flashrom testing."""

    DELTA = 1  # value to add to a byte to corrupt a section contents

    # File in the state directory to store public root key.
    PUB_KEY_FILE_NAME = 'root.pubkey'
    FW_KEYBLOCK_FILE_NAME = 'firmware.keyblock'
    FW_PRIV_DATA_KEY_FILE_NAME = 'firmware_data_key.vbprivk'
    KERNEL_SUBKEY_FILE_NAME = 'kernel_subkey.vbpubk'

    def __init__(self):
    # make sure it does not accidentally overwrite the image.
        self.fum = None
        self.chros_if = None
        self.image = ''
        self.pub_key_file = ''

    def init(self, flashrom_util_module,
             chros_if,
             pub_key_file=None,
             dev_key_path='./',
             target='bios'):
        """Flashrom handler initializer.

        Args:
          flashrom_util_module - a module providing flashrom access utilities.
          chros_if - a module providing interface to Chromium OS services
          pub_key_file - a string, name of the file contaning a public key to
                         use for verifying both existing and new firmware.
        """
        if target == 'bios':
            self.fum = flashrom_util_module.flashrom_util(target_is_ec=False)
            self.fv_sections = {
                'a': FvSection('VBOOTA', 'FVMAIN'),
                'b': FvSection('VBOOTB', 'FVMAINB'),
                'ec_a': FvSection(None, 'ECMAINA'),
                'ec_b': FvSection(None, 'ECMAINB'),
                }
        elif target == 'ec':
            self.fum = flashrom_util_module.flashrom_util(target_is_ec=True)
            self.fv_sections = {
                'rw': FvSection(None, 'EC_RW'),
                }
        else:
            raise FlashromHandlerError("Invalid target.")
        self.chros_if = chros_if
        self.pub_key_file = pub_key_file
        self.dev_key_path = dev_key_path

    def new_image(self, image_file=None):
        """Parse the full flashrom image and store sections into files.

        Args:
          image_file - a string, the name of the file contaning full ChromeOS
                       flashrom image. If not passed in or empty - the actual
                       flashrom is read and its contents are saved into a
                       temporary file which is used instead.

        The input file is parsed and the sections of importance (as defined in
        self.fv_sections) are saved in separate files in the state directory
        as defined in the chros_if object.
        """

        if image_file:
            self.image = open(image_file, 'rb').read()
            self.fum.set_firmware_layout(image_file)
        else:
            self.image = self.fum.read_whole()

        for section in self.fv_sections.itervalues():
            for subsection_name in section.names():
                if not subsection_name:
                    continue
                blob = self.fum.get_section(self.image, subsection_name)
                if blob:
                    f = open(self.chros_if.state_dir_file(subsection_name),
                             'wb')
                    f.write(blob)
                    f.close()

            blob = self.fum.get_section(self.image, section.get_body_name())
            if blob:
                s = hashlib.sha1()
                s.update(blob)
                section.set_sha(s.hexdigest())

            # If there is no "sig" subsection, skip reading version and flags.
            if not section.get_sig_name():
                continue

            # Now determine this section's version number.
            vb_section = self.fum.get_section(
                self.image, section.get_sig_name())

            section.set_version(self.chros_if.retrieve_body_version(vb_section))
            section.set_flags(self.chros_if.retrieve_preamble_flags(vb_section))
            section.set_datakey_version(
                self.chros_if.retrieve_datakey_version(vb_section))
            section.set_kernel_subkey_version(
                self.chros_if.retrieve_kernel_subkey_version(vb_section))

            s = hashlib.sha1()
            s.update(self.fum.get_section(self.image, section.get_sig_name()))
            section.set_sig_sha(s.hexdigest())

        if not self.pub_key_file:
            self._retrieve_pub_key()

    def _retrieve_pub_key(self):
        """Retrieve root public key from the firmware GBB section."""

        gbb_header_format = '<4s20s2I'
        pubk_header_format = '<2Q'

        gbb_section = self.fum.get_section(self.image, 'FV_GBB')

        # do some sanity checks
        try:
            sig, _, rootk_offs, rootk_size = struct.unpack_from(
                gbb_header_format, gbb_section)
        except struct.error, e:
            raise FlashromHandlerError(e)

        if sig != '$GBB' or (rootk_offs + rootk_size) > len(gbb_section):
            raise FlashromHandlerError('Bad gbb header')

        key_body_offset, key_body_size = struct.unpack_from(
            pubk_header_format, gbb_section, rootk_offs)

        # Generally speaking the offset field can be anything, but in case of
        # GBB section the key is stored as a standalone entity, so the offset
        # of the key body is expected to be equal to the key header size of
        # 0x20.
        # Should this convention change, the check below would fail, which
        # would be a good prompt for revisiting this test's behavior and
        # algorithms.
        if key_body_offset != 0x20 or key_body_size > rootk_size:
            raise FlashromHandlerError('Bad public key format')

        # All checks passed, let's store the key in a file.
        self.pub_key_file = self.chros_if.state_dir_file(self.PUB_KEY_FILE_NAME)
        keyf = open(self.pub_key_file, 'w')
        key = gbb_section[
            rootk_offs:rootk_offs + key_body_offset + key_body_size]
        keyf.write(key)
        keyf.close()

    def verify_image(self):
        """Confirm the image's validity.

        Using the file supplied to init() as the public key container verify
        the two sections' (FirmwareA and FirmwareB) integrity. The contents of
        the sections is taken from the files created by new_image()

        In case there is an integrity error raises FlashromHandlerError
        exception with the appropriate error message text.
        """

        for section in self.fv_sections.itervalues():
            if section.get_sig_name():
                cmd = 'vbutil_firmware --verify %s --signpubkey %s  --fv %s' % (
                    self.chros_if.state_dir_file(section.get_sig_name()),
                    self.pub_key_file,
                    self.chros_if.state_dir_file(section.get_body_name()))
                self.chros_if.run_shell_command(cmd)

    def _modify_section(self, section, delta, body_or_sig=False,
                        corrupt_all=False):
        """Modify a firmware section inside the image, either body or signature.

        If corrupt_all is set, the passed in delta is added to all bytes in the
        section. Otherwise, the delta is added to the value located at 2% offset
        into the section blob, either body or signature.

        Calling this function again for the same section the complimentary
        delta value would restore the section contents.
        """

        if not self.image:
            raise FlashromHandlerError(
                'Attempt at using an uninitialized object')
        if section not in self.fv_sections:
            raise FlashromHandlerError('Unknown FW section %s'
                                       % section)

        # Get the appropriate section of the image.
        if body_or_sig:
            subsection_name = self.fv_sections[section].get_body_name()
        else:
            subsection_name = self.fv_sections[section].get_sig_name()
        blob = self.fum.get_section(self.image, subsection_name)

        # Modify the byte in it within 2% of the section blob.
        modified_index = len(blob) / 50
        if corrupt_all:
            blob_list = [('%c' % ((ord(x) + delta) % 0x100)) for x in blob]
        else:
            blob_list = list(blob)
            blob_list[modified_index] = ('%c' %
                    ((ord(blob[modified_index]) + delta) % 0x100))
        self.image = self.fum.put_section(self.image,
                                          subsection_name, ''.join(blob_list))

        return subsection_name

    def corrupt_section(self, section, corrupt_all=False):
        """Corrupt a section signature of the image"""

        return self._modify_section(section, self.DELTA, body_or_sig=False,
                                    corrupt_all=corrupt_all)

    def corrupt_section_body(self, section, corrupt_all=False):
        """Corrupt a section body of the image"""

        return self._modify_section(section, self.DELTA, body_or_sig=True,
                                    corrupt_all=corrupt_all)

    def restore_section(self, section, restore_all=False):
        """Restore a previously corrupted section signature of the image."""

        return self._modify_section(section, -self.DELTA, body_or_sig=False,
                                    corrupt_all=restore_all)

    def restore_section_body(self, section, restore_all=False):
        """Restore a previously corrupted section body of the image."""

        return self._modify_section(section, -self.DELTA, body_or_sig=True,
                                    corrupt_all=restore_all)

    def corrupt_firmware(self, section, corrupt_all=False):
        """Corrupt a section signature in the FLASHROM!!!"""

        subsection_name = self.corrupt_section(section, corrupt_all=corrupt_all)
        self.fum.write_partial(self.image, (subsection_name, ))

    def corrupt_firmware_body(self, section, corrupt_all=False):
        """Corrupt a section body in the FLASHROM!!!"""

        subsection_name = self.corrupt_section_body(section,
                                                    corrupt_all=corrupt_all)
        self.fum.write_partial(self.image, (subsection_name, ))

    def restore_firmware(self, section, restore_all=False):
        """Restore the previously corrupted section sig in the FLASHROM!!!"""

        subsection_name = self.restore_section(section, restore_all=restore_all)
        self.fum.write_partial(self.image, (subsection_name, ))

    def restore_firmware_body(self, section, restore_all=False):
        """Restore the previously corrupted section body in the FLASHROM!!!"""

        subsection_name = self.restore_section_body(section,
                                                    restore_all=False)
        self.fum.write_partial(self.image, (subsection_name, ))

    def firmware_sections_equal(self):
        """Check if firmware sections A and B are equal.

        This function presumes that the entire BIOS image integrity has been
        verified, so different signature sections mean different images and
        vice versa.
        """
        sig_a = self.fum.get_section(self.image,
                                      self.fv_sections['a'].get_sig_name())
        sig_b = self.fum.get_section(self.image,
                                      self.fv_sections['b'].get_sig_name())
        return sig_a == sig_b

    def copy_from_to(self, src, dst):
        """Copy one firmware image section to another.

        This function copies both signature and body of one firmware section
        into another. After this function runs both sections are identical.
        """
        src_sect = self.fv_sections[src]
        dst_sect = self.fv_sections[dst]
        self.image = self.fum.put_section(
            self.image,
            dst_sect.get_body_name(),
            self.fum.get_section(self.image, src_sect.get_body_name()))
        self.image = self.fum.put_section(
            self.image,
            dst_sect.get_sig_name(),
            self.fum.get_section(self.image, src_sect.get_sig_name()))

    def write_whole(self):
        """Write the whole image into the flashrom."""

        if not self.image:
            raise FlashromHandlerError(
                'Attempt at using an uninitialized object')
        self.fum.write_whole(self.image)

    def dump_whole(self, filename):
        """Write the whole image into a file."""

        if not self.image:
            raise FlashromHandlerError(
                'Attempt at using an uninitialized object')
        open(filename, 'w').write(self.image)

    def dump_partial(self, subsection_name, filename):
        """Write the subsection part into a file."""

        if not self.image:
            raise FlashromHandlerError(
                'Attempt at using an uninitialized object')
        blob = self.fum.get_section(self.image, subsection_name)
        open(filename, 'w').write(blob)

    def get_gbb_flags(self):
        """Retrieve the GBB flags"""
        gbb_header_format = '<12sL'
        gbb_section = self.fum.get_section(self.image, 'FV_GBB')
        try:
            _, gbb_flags = struct.unpack_from(gbb_header_format, gbb_section)
        except struct.error, e:
            raise FlashromHandlerError(e)
        return gbb_flags

    def enable_write_protect(self):
        """Enable write protect of the flash chip"""
        self.fum.enable_write_protect()

    def disable_write_protect(self):
        """Disable write protect of the flash chip"""
        self.fum.disable_write_protect()

    def get_section_sig_sha(self, section):
        """Retrieve SHA1 hash of a firmware vblock section"""
        return self.fv_sections[section].get_sig_sha()

    def get_section_sha(self, section):
        """Retrieve SHA1 hash of a firmware body section"""
        return self.fv_sections[section].get_sha()

    def get_section_version(self, section):
        """Retrieve version number of a firmware section"""
        return self.fv_sections[section].get_version()

    def get_section_flags(self, section):
        """Retrieve preamble flags of a firmware section"""
        return self.fv_sections[section].get_flags()

    def get_section_datakey_version(self, section):
        """Retrieve data key version number of a firmware section"""
        return self.fv_sections[section].get_datakey_version()

    def get_section_kernel_subkey_version(self, section):
        """Retrieve kernel subkey version number of a firmware section"""
        return self.fv_sections[section].get_kernel_subkey_version()

    def get_section_body(self, section):
        """Retrieve body of a firmware section"""
        subsection_name = self.fv_sections[section].get_body_name()
        blob = self.fum.get_section(self.image, subsection_name)
        return blob

    def get_section_sig(self, section):
        """Retrieve vblock of a firmware section"""
        subsection_name = self.fv_sections[section].get_sig_name()
        blob = self.fum.get_section(self.image, subsection_name)
        return blob

    def _find_dtb_offset(self, blob):
        """Return the offset of DTB blob from the given firmware blob"""
        # The RW firmware is concatenated from u-boot, dtb, and ecbin.
        # Search the magic of dtb to locate the dtb bloc.
        return blob.find("\xD0\x0D\xFE\xED\x00")

    def _find_ecbin_offset(self, blob):
        """Return the offset of EC binary from the given firmware blob"""
        dtb_offset = self._find_dtb_offset(blob)
        # If the device tree was found, use it. Otherwise assume an index
        # structure.
        if dtb_offset != -1:
            # The dtb size is a 32-bit integer which follows the magic.
            _, dtb_size = struct.unpack_from(">2L", blob, dtb_offset)
            # The ecbin part is aligned to 4-byte.
            return (dtb_offset + dtb_size + 3) & ~3
        else:
            # The index will have a count, an offset and size for the system
            # firmware, and then an offset and size for the EC binary.
            return struct.unpack_from("<I", blob, 3 * 4)[0]

    def get_section_ecbin(self, section):
        """Retrieve EC binary of a firmware section"""
        blob = self.get_section_body(section)
        ecbin_offset = self._find_ecbin_offset(blob)
        # Remove the pads of ecbin.
        pad = blob[-1]
        ecbin = blob[ecbin_offset :].rstrip(pad)
        return ecbin

    def set_section_body(self, section, blob, write_through=False):
        """Put the supplied blob to the body of the firmware section"""
        subsection_name = self.fv_sections[section].get_body_name()
        self.image = self.fum.put_section(self.image, subsection_name, blob)

        if write_through:
            self.dump_partial(subsection_name,
                              self.chros_if.state_dir_file(subsection_name))
            self.fum.write_partial(self.image, (subsection_name, ))

    def set_section_sig(self, section, blob, write_through=False):
        """Put the supplied blob to the vblock of the firmware section"""
        subsection_name = self.fv_sections[section].get_sig_name()
        self.image = self.fum.put_section(self.image, subsection_name, blob)

        if write_through:
            self.dump_partial(subsection_name,
                              self.chros_if.state_dir_file(subsection_name))
            self.fum.write_partial(self.image, (subsection_name, ))

    def set_section_ecbin(self, section, ecbin, write_through=False):
        """Put the supplied EC binary to the firwmare section.

        Note that the updated firmware image is not signed yet. Should call
        set_section_version() afterward.
        """
        # Remove unncessary padding bytes.
        pad = '\xff'
        align = 4
        ecbin = ecbin.rstrip(pad)
        ecbin += pad * (-len(ecbin) % align)
        ecbin_size = len(ecbin)

        # Get the original main firmware body.
        main_blob = self.get_section_body(section)

        # Compute the hash of the ecbin.
        hasher = hashlib.sha256()
        hasher.update(ecbin)
        echash = hasher.digest()

        # Depthcharge firmware doesn't save the EC size on the DTB.
        is_depthcharge = (self._find_dtb_offset(main_blob) == -1)

        if is_depthcharge:
            # The Depthcharge index header in main firmware section is like:
            #  count = 2,
            #  offset and size of main firmware,
            #  offset and size of EC binary hash.
            echash_offset = struct.unpack_from('<I', main_blob, 3 * 4)[0]

            # Update the EC binary hash.
            main_blob = main_blob[0 : echash_offset] + echash

            # Construct the EC firmware section, which is composed of:
            #  count = 1,
            #  offset and size of EC firmware,
            #  EC firmware.
            ecbin_blob = struct.pack('<III', 1, 3 * 4, ecbin_size) + ecbin

            # Write the EC firmware section back.
            ec_section = 'ec_' + section
            self.set_section_body(ec_section, ecbin_blob)
            if write_through:
                ec_fmap = self.fv_sections[ec_section].get_body_name()
                self.dump_partial(ec_fmap,
                                  self.chros_if.state_dir_file(ec_fmap))
                self.fum.write_partial(self.image, (ec_fmap, ))
        else:
            # Update the size and the hash of the EC binary on the DTB.
            dtb_offset = self._find_dtb_offset(main_blob)
            ecbin_offset = self._find_ecbin_offset(main_blob)
            dtb_blob = main_blob[dtb_offset : ecbin_offset]
            dtb_size = ecbin_offset - dtb_offset
            with tempfile.NamedTemporaryFile() as dtb_file:
                dtb_file.write(dtb_blob)
                dtb_file.flush()

                cmd = ["fdtput", "-t lu", dtb_file.name,
                        "/flash/rw-a-boot/ecbin",
                        "reg", str(ecbin_offset), str(ecbin_size)]
                self.chros_if.run_shell_command(' '.join(cmd))

                # The old version of firmware doesn't save the EC hash in DTB.
                # So check the EC hash first before updating.
                try:
                    cmd = ["fdtget", "-t bu", dtb_file.name,
                            "/flash/rw-a-boot/ecbin", "hash"]
                    self.chros_if.run_shell_command(' '.join(cmd))
                except ChromeOSInterfaceError:
                    self.chros_if.log(
                            "Skip updating EC hash on the old firmware.")
                else:
                    cmd = ["fdtput", "-t bu", dtb_file.name,
                            "/flash/rw-a-boot/ecbin", "hash"]
                    cmd += [str(ord(c)) for c in echash]
                    self.chros_if.run_shell_command(' '.join(cmd))

                dtb_file.seek(0)
                dtb_blob = dtb_file.read()
                dtb_blob += pad * (-len(dtb_blob) % align)
                assert len(dtb_blob) == dtb_size, (
                        'Updating DTB should not change its size.')

            pad = main_blob[-1]
            pad_size = len(main_blob) - ecbin_offset - ecbin_size
            main_blob = (main_blob[0 : dtb_offset] +
                        dtb_blob + ecbin + pad * pad_size)

        # Write the main firmware section back.
        self.set_section_body(section, main_blob)
        if write_through:
            subsection_name = self.fv_sections[section].get_body_name()
            self.dump_partial(subsection_name,
                              self.chros_if.state_dir_file(subsection_name))
            self.fum.write_partial(self.image, (subsection_name, ))

    def set_section_version(self, section, version, flags,
                            write_through=False):
        """
        Re-sign the firmware section using the supplied version number and
        flag.
        """
        if (self.get_section_version(section) == version and
            self.get_section_flags(section) == flags):
            return  # No version or flag change, nothing to do.
        if version < 0:
            raise FlashromHandlerError(
                'Attempt to set version %d on section %s' % (version, section))
        fv_section = self.fv_sections[section]
        sig_name = self.chros_if.state_dir_file(fv_section.get_sig_name())
        sig_size = os.path.getsize(sig_name)

        # Construct the command line
        args = ['--vblock %s' % sig_name]
        args.append('--keyblock %s' % os.path.join(
                self.dev_key_path, self.FW_KEYBLOCK_FILE_NAME))
        args.append('--fv %s' % self.chros_if.state_dir_file(
                fv_section.get_body_name()))
        args.append('--version %d' % version)
        args.append('--kernelkey %s' % os.path.join(
                self.dev_key_path, self.KERNEL_SUBKEY_FILE_NAME))
        args.append('--signprivate %s' % os.path.join(
                self.dev_key_path, self.FW_PRIV_DATA_KEY_FILE_NAME))
        args.append('--flags %d' % flags)
        cmd = 'vbutil_firmware %s' % ' '.join(args)
        self.chros_if.run_shell_command(cmd)

        #  Pad the new signature.
        new_sig = open(sig_name, 'a')
        pad = ('%c' % 0) * (sig_size - os.path.getsize(sig_name))
        new_sig.write(pad)
        new_sig.close()

        # Inject the new signature block into the image
        new_sig = open(sig_name, 'r').read()
        self.image = self.fum.put_section(
            self.image, fv_section.get_sig_name(), new_sig)
        if write_through:
            self.fum.write_partial(self.image, (fv_section.get_sig_name(), ))