| #!/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 autotest_lib.client.common_lib.cros import chip_utils |
| |
| 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, fwid_name=None): |
| self._sig_name = sig_name |
| self._body_name = body_name |
| self._fwid_name = fwid_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, self._fwid_name) |
| |
| def get_sig_name(self): |
| return self._sig_name |
| |
| def get_body_name(self): |
| return self._body_name |
| |
| def get_fwid_name(self): |
| return self._fwid_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' |
| EC_EFS_KEY_FILE_NAME = 'key_ec_efs.vbprik2' |
| |
| def __init__(self): |
| # make sure it does not accidentally overwrite the image. |
| self.fum = None |
| self.os_if = None |
| self.image = '' |
| self.pub_key_file = '' |
| |
| def init(self, flashrom_util_module, |
| os_if, |
| pub_key_file=None, |
| dev_key_path='./', |
| target='bios'): |
| """Flashrom handler initializer. |
| |
| Args: |
| flashrom_util_module - a module providing flashrom access utilities. |
| os_if - a module providing interface to 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( |
| os_if, target_is_ec=False) |
| self.fv_sections = { |
| 'a': FvSection('VBOOTA', 'FVMAIN', 'RW_FWID_A'), |
| 'b': FvSection('VBOOTB', 'FVMAINB', 'RW_FWID_B'), |
| 'rec': FvSection(None, 'RECOVERY_MRC_CACHE'), |
| 'ec_a': FvSection(None, 'ECMAINA'), |
| 'ec_b': FvSection(None, 'ECMAINB'), |
| } |
| elif target == 'ec': |
| self.fum = flashrom_util_module.flashrom_util( |
| os_if, target_is_ec=True) |
| self.fv_sections = { |
| 'rw': FvSection(None, 'EC_RW', 'RW_FWID'), |
| 'rw_b': FvSection(None, 'EC_RW_B'), |
| } |
| else: |
| raise FlashromHandlerError("Invalid target.") |
| self.os_if = os_if |
| self.pub_key_file = pub_key_file |
| self.dev_key_path = dev_key_path |
| self.new_image() |
| |
| 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 os_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.os_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.os_if.retrieve_body_version(vb_section)) |
| section.set_flags(self.os_if.retrieve_preamble_flags(vb_section)) |
| section.set_datakey_version( |
| self.os_if.retrieve_datakey_version(vb_section)) |
| section.set_kernel_subkey_version( |
| self.os_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.os_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.os_if.state_dir_file(section.get_sig_name()), |
| self.pub_key_file, |
| self.os_if.state_dir_file(section.get_body_name())) |
| self.os_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())) |
| # If there is no "sig" subsection, skip copying signature. |
| if src_sect.get_sig_name() and dst_sect.get_sig_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())) |
| self.write_whole() |
| |
| 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 write_partial(self, subsection_name, blob=None, write_through=True): |
| """Write the subsection part into the flashrom. |
| |
| One can pass a blob to update the data of the subsection before write |
| it into the flashrom. |
| """ |
| |
| if not self.image: |
| raise FlashromHandlerError( |
| 'Attempt at using an uninitialized object') |
| |
| if blob is not None: |
| self.image = self.fum.put_section(self.image, subsection_name, blob) |
| |
| if write_through: |
| self.dump_partial(subsection_name, |
| self.os_if.state_dir_file(subsection_name)) |
| self.fum.write_partial(self.image, (subsection_name, )) |
| |
| 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 dump_section_body(self, section, filename): |
| """Write the body of a firmware section into a file""" |
| subsection_name = self.fv_sections[section].get_body_name() |
| self.dump_partial(subsection_name, filename) |
| |
| def get_section_hash(self, section): |
| """Retrieve the hash of the body of a firmware section""" |
| ecrw = chip_utils.ecrw() |
| with tempfile.NamedTemporaryFile(prefix=ecrw.chip_name) as f: |
| self.dump_section_body(section, f.name) |
| ecrw.set_from_file(f.name) |
| result = ecrw.compute_hash_bytes() |
| return result |
| |
| 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 set_gbb_flags(self, flags, write_through=False): |
| """Retrieve the GBB flags""" |
| gbb_header_format = '<L' |
| section_name = 'FV_GBB' |
| gbb_section = self.fum.get_section(self.image, section_name) |
| try: |
| formatted_flags = struct.pack(gbb_header_format, flags) |
| except struct.error, e: |
| raise FlashromHandlerError(e) |
| gbb_section = gbb_section[:12] + formatted_flags + gbb_section[16:] |
| self.write_partial(section_name, gbb_section, write_through) |
| |
| 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 has_section_body(self, section): |
| """Return True if the section body is in the image""" |
| return bool(self.get_section_body(section)) |
| |
| 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 get_section_fwid(self, section): |
| """Retrieve fwid blob of a firmware section""" |
| subsection_name = self.fv_sections[section].get_fwid_name() |
| blob = self.fum.get_section(self.image, subsection_name) |
| return blob |
| |
| 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.write_partial(subsection_name, blob, write_through) |
| |
| 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.write_partial(subsection_name, blob, write_through) |
| |
| def set_section_fwid(self, section, blob, write_through=False): |
| """Put the supplied blob to the fwid of the firmware section""" |
| subsection_name = self.fv_sections[section].get_fwid_name() |
| self.write_partial(subsection_name, blob, write_through) |
| |
| def resign_ec_rwsig(self): |
| """Resign the EC image using rwsig.""" |
| key_ec_efs = os.path.join(self.dev_key_path, self.EC_EFS_KEY_FILE_NAME) |
| # Dump whole EC image to a file and execute the sign command. |
| with tempfile.NamedTemporaryFile() as f: |
| self.dump_whole(f.name) |
| self.os_if.run_shell_command( |
| 'futility sign --type rwsig --prikey %s %s' % ( |
| key_ec_efs, f.name)) |
| self.new_image(f.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.os_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.os_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.os_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.write_partial(fv_section.get_sig_name(), new_sig, write_through) |
