| #!/usr/bin/python |
| # |
| # 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. |
| |
| """ |
| gooftool: Google Factory Tool, providing all Google Required Test |
| functionality. |
| """ |
| |
| |
| # TODO(tammo): Refactor to match Chromium style. |
| |
| |
| import copy |
| import glob |
| import optparse |
| import os |
| import re |
| import sys |
| import tempfile |
| |
| ######################################################################## |
| # cros/gft modules |
| |
| import crosfw |
| import gft_common |
| import gft_report |
| import gft_upload |
| import gft_vpd |
| import gft_wpfw |
| import gpio |
| import hwid_tool |
| import probe |
| |
| from common import Obj |
| from gft_common import WarningMsg, VerboseMsg, DebugMsg, ErrorMsg, ErrorDie |
| |
| |
| #######################################################################r |
| # global variables and cached objects |
| |
| g_options = None |
| g_args = None |
| |
| # list of commands |
| g_commands = [] |
| |
| # gooftool version |
| VERSION = 2 |
| |
| # cached objects |
| g_cached_device_data = None |
| |
| |
| ######################################################################## |
| # factory utilities |
| |
| |
| def GetPrimaryDevicePath(partition=None): |
| def isfixed(dev): |
| sysfs_path = '/sys/block/%s/removable' % dev |
| return (os.path.exists(sysfs_path) and |
| open(sysfs_path).read().strip() == '0') |
| alpha_re = re.compile(r'^/dev/([a-zA-Z]+)[0-9]+$') |
| alnum_re = re.compile(r'^/dev/([a-zA-Z]+[0-9]+)p[0-9]+$') |
| dev_set = set( |
| dev |
| for path in gft_common.SystemOutput('cgpt find -t rootfs').split() |
| for dev in alpha_re.findall(path) + alnum_re.findall(path) |
| if isfixed(dev)) |
| if len(dev_set) != 1: |
| ErrorDie('zero or multiple primary devs: %s' % dev_set) |
| dev_path = '/dev/' + dev_set.pop() |
| if partition is None: |
| return dev_path |
| fmt_str = '%sp%d' if alnum_re.match(dev_path) else '%s%d' |
| return fmt_str % (dev_path, partition) |
| |
| |
| def GetReleaseRootPartitionPath(): |
| return GetPrimaryDevicePath(5) |
| |
| |
| def GetReleaseKernelPartitionPath(): |
| return GetPrimaryDevicePath(4) |
| |
| |
| def VerifySwitch(name, value): |
| """ Verifies if hardware switches are in correct states. """ |
| my_gpio = gpio.Gpio() |
| my_gpio.setup() |
| try: |
| DebugMsg('reading GPIO %s (expected %s)' % (name, value)) |
| if value != my_gpio.read(name): |
| VerboseMsg('VerifySwitch: %s is not in proper state.' % name) |
| return False |
| except: |
| ErrorDie('VerifySwitch: cannot read: ' + name) |
| return True |
| |
| |
| def EnableWriteProtect(target, image): |
| """ Enables and verifies firmware write protection. """ |
| return gft_wpfw.EnableWriteProtect(target, image) |
| |
| |
| def VerboseSystemCommand(command, prompt=None): |
| """ Invokes command and put stdout/stderr messages into verbose log. """ |
| VerboseMsg(" # Executing command: %s" % command) |
| (_, stdout, stderr) = gft_common.ShellExecution( |
| command, show_progress=g_options.verbose, progress_message=prompt) |
| message = '\n'.join((stdout.strip(), stderr.strip())).strip() |
| if message: |
| VerboseMsg(message) |
| |
| |
| def GFTCommand(f): |
| """ Decorator for gooftool base commands. """ |
| def wrapper(*args, **kw): |
| try: |
| DebugMsg('@Starting to execute command: %s.' % f.__name__) |
| result = f(*args, **kw) |
| DebugMsg('@Executed command: %s. Result: %s' % |
| (f.__name__, "SUCCESS" |
| if result is not False else "NOT SUCCESS")) |
| return result |
| except gft_common.GFTError, e: |
| DebugMsg('@Executed command: %s. Result: ERROR - %s' % |
| (f.__name__, e.value)) |
| raise |
| except: |
| DebugMsg('@Executed command: %s. Result: FAILED.' % f.__name__) |
| raise |
| wrapper.__name__ = f.__name__ |
| wrapper.__doc__ = re.sub(r'\s+', ' ', f.__doc__).strip() |
| g_commands.append(wrapper) |
| return wrapper |
| |
| |
| def GFTCommandDependency(command_list): |
| """Decorator to define the depenedncy of a gooftool command. |
| |
| Args |
| command_list: a list or tuple of command functions to be executed in order. |
| |
| Raises |
| GFTError if any of the commands in dependency return False. |
| """ |
| def meta_wrapper(f): |
| |
| def wrapper(*args, **kargs): |
| for command in command_list: |
| if not command(): |
| ErrorDie('%s: failed in sub-command: %s.' % |
| (f.__name__, command.__name__)) |
| return f(*args, **kargs) |
| |
| wrapper.__name__ = f.__name__ |
| wrapper.__doc__ = ('Runs: %s. %s' % |
| (', '.join(['--%s' % command.__name__ |
| for command in command_list]), |
| f.__doc__)) |
| return wrapper |
| return meta_wrapper |
| |
| |
| ######################################################################## |
| # standard procedure |
| # (function names are used for self-reflection to command line parameters) |
| |
| # HWID, GBB, and hardware components validation |
| |
| |
| @GFTCommand |
| def write_hwid(): |
| """Write specified HWID value into the system GBB.""" |
| if not g_options.hwid: |
| ErrorDie('write_hwid: No HWID specified (must specify using --hwid arg).') |
| hwid = g_options.hwid |
| DebugMsg('write_hwid: using hwid string %r' % hwid) |
| main_fw_file = crosfw.LoadMainFirmware().path |
| cmd = ('gbb_utility --set --hwid="%s" "%s"' % (hwid, main_fw_file)) |
| gft_common.System(cmd) |
| flashrom = crosfw.Flashrom(crosfw.TARGET_MAIN) |
| flashrom.Write(filename=main_fw_file, sections=['GBB']) |
| return True |
| |
| |
| @GFTCommand |
| def verify_hwid(): |
| """Verify system HWID properties match probed device properties. |
| |
| First probe components, volatile and initial_config parameters for |
| the DUT. Then use the available device data to produce a list of |
| candidate HWIDs. Then verify the HWID from the DUT is present in |
| that list. Finally, verify that the DUT initial config values match |
| those specified for its HWID. |
| """ |
| main_fw_file = crosfw.LoadMainFirmware().path |
| gbb_result = gft_common.SystemOutput('gbb_utility -g --hwid %s' % |
| main_fw_file) |
| hwid = re.findall(r'hardware_id:(.*)', gbb_result)[0].strip() |
| DebugMsg('system HWID: %r\n' % hwid) |
| try: |
| data = hwid_tool.ReadDatastore(g_options.db_path) |
| hwid_properties = hwid_tool.LookupHwidProperties(data, hwid) |
| DebugMsg('expected system properties: %s' % hwid_properties.__dict__) |
| probe_results = probe.Probe(data.comp_db.component_registry) |
| DebugMsg('found system properties: %s' % probe_results.__dict__) |
| cooked_components = hwid_tool.CookComponentProbeResults( |
| data.comp_db, probe_results) |
| DebugMsg('cooked components: %s' % cooked_components.__dict__) |
| device = data.device_db[hwid_properties.board] |
| cooked_device_details = hwid_tool.CookDeviceProbeResults( |
| device, probe_results) |
| DebugMsg('cooked device details: %s' % cooked_device_details.__dict__) |
| global g_cached_device_data |
| g_cached_device_data = Obj( |
| hwid_properties=hwid_properties, |
| probe_results=probe_results, |
| cooked_components=cooked_components, |
| cooked_device_details=cooked_device_details) |
| except hwid_tool.Error, e: |
| ErrorDie('verify_hwid: FAILED.\n%s' % e) |
| match_errors = [] |
| for comp_class, comp_value in hwid_properties.component_map.items(): |
| if comp_value == 'ANY': |
| continue |
| found_value = cooked_components.known.get(comp_class, None) |
| if found_value != comp_value: |
| if found_value is None: |
| alt_value = cooked_components.unknown.get(comp_class, None) |
| found_value = 'UNKNOWN %r' % alt_value |
| match_errors.append(' %s component mismatch, expected %s, found %s' % |
| (comp_class, comp_value, found_value)) |
| if match_errors: |
| ErrorDie('verify_hwid: FAILED.\n%s' % '\n'.join(match_errors)) |
| if hwid_properties.volatile not in cooked_device_details.volatile_set: |
| msg = (' HWID specified volatile %s, but found match only for %s' % |
| (hwid_properties.volatile, |
| ', '.join(cooked_device_details.volatile_set))) |
| ErrorDie('verify_hwid: FAILED.\n%s' % msg) |
| if (hwid_properties.initial_config is not None and |
| hwid_properties.initial_config not in |
| cooked_device_details.initial_config_set): |
| msg = (' HWID specified initial_config %s, but found match only for %s' % |
| (hwid_properties.initial_config, |
| ', '.join(cooked_device_details.initial_config_set))) |
| ErrorDie('verify_hwid: FAILED.\n%s' % msg) |
| # TODO(tammo): Verify HWID status is supported (or deprecated for RMA). |
| # TODO(tammo): Verify release string. |
| # TODO(tammo): Verify VPD here, using hwid_properties.vpd_ro_field_list. |
| print 'Verified: %s' % hwid |
| return True |
| |
| |
| @GFTCommand |
| def verify_keys(): |
| """ Verifies if the keys in firmware and SSD are matched. """ |
| script = os.path.join(os.path.split(sys.argv[0])[0], 'gft_verify_keys.sh') |
| if not os.path.exists(script): |
| ErrorDie('verify_keys: cannot find script to verify.') |
| kernel_device = GetReleaseKernelPartitionPath() |
| main_fw_path = crosfw.LoadMainFirmware().path |
| command = '%s %s %s' % (script, kernel_device, main_fw_path) |
| try: |
| VerboseSystemCommand(command, prompt='verify_keys: ') |
| except gft_common.GFTError, e: |
| ErrorDie('verify_keys: FAILED.\n%s' % e.value) |
| return True |
| |
| |
| @GFTCommand |
| def verify_vpd(): |
| """ Verifies if the VPD contains valid required data. """ |
| try: |
| # TODO(tammo): Verify that all of the VPD fields specified for the |
| # board (using hwid properties) have values. |
| image_file = crosfw.LoadMainFirmware().path |
| gft_vpd.ValidateVpdData(image_file, g_options.verbose) |
| gft_vpd.SetFirmwareBitmapLocale(image_file) |
| except gft_common.GFTError, e: |
| ErrorDie('verify_vpd: FAILED.\n%s' % e.value) |
| return True |
| |
| |
| @GFTCommand |
| def verify_system_time(): |
| """Verifies if system time is later than release filesystem creation time.""" |
| script = os.path.join(os.path.split(sys.argv[0])[0], |
| 'gft_verify_system_time.sh') |
| if not os.path.exists(script): |
| ErrorDie('verify_system_time: cannot find script to verify.') |
| rootfs_device = GetReleaseRootPartitionPath() |
| command = '%s %s' % (script, rootfs_device) |
| try: |
| VerboseSystemCommand(command, prompt='verify_system_time: ') |
| except gft_common.GFTError, e: |
| ErrorDie('verify_system_time: FAILED.\n%s' % e.value) |
| return True |
| |
| |
| @GFTCommand |
| def verify_rootfs(): |
| """ Verifies if the rootfs on SSD is valid (by checking hash). """ |
| script = os.path.join(os.path.split(sys.argv[0])[0], 'gft_verify_rootfs.sh') |
| if not os.path.exists(script): |
| ErrorDie('verify_rootfs: cannot find script to verify.') |
| rootfs_device = GetReleaseRootPartitionPath() |
| command = '%s %s' % (script, rootfs_device) |
| try: |
| VerboseSystemCommand(command, prompt='verify_rootfs: ') |
| except gft_common.GFTError, e: |
| ErrorDie('verify_rootfs: FAILED.\n%s' % e.value) |
| return True |
| |
| |
| # Hardware Switches |
| |
| @GFTCommand |
| def verify_switch_wp(): |
| """ Verifies if hardware write protection switch is enabled. """ |
| if VerifySwitch('write_protect', 1): |
| return True |
| ErrorDie('verify_switch_wp: failed') |
| |
| |
| @GFTCommand |
| def verify_switch_dev(): |
| """ Verifies if developer switch is disabled. """ |
| if VerifySwitch('developer_switch', 0): |
| return True |
| ErrorDie('verify_switch_dev: failed') |
| |
| |
| # Report Creation |
| |
| @GFTCommand |
| def create_report(): |
| """ Creates the report (with vpd, hwid, date, ...) for uploading. """ |
| # decide log format |
| if g_options.report_format == 'gz': |
| text_armed = False |
| elif g_options.report_format == 'base64': |
| text_armed = True |
| else: |
| ErrorDie('create_report: invalid format: %s' % g_options.report_format) |
| |
| tag = g_options.report_tag if g_options.report_tag else '' |
| |
| if g_cached_device_data is None: |
| # TODO(tammo): Refactor verify_hwid to extract the probing and |
| # cooking from the actual validation, so that we can generate |
| # reports for machines where the hwid validation fails. |
| ErrorDie('create_report: must run verify_hwid first') |
| |
| vpd_source = crosfw.LoadMainFirmware().path |
| report = gft_report.CreateReport(sys.argv, |
| system_details=g_cached_device_data, |
| verbose_log_path=g_options.log_path, |
| vpd_source=vpd_source, |
| tag=tag, |
| verbose=g_options.debug) |
| |
| data = gft_report.EncodeReport(report, text_armed=text_armed) |
| if not g_options.report_path: |
| g_options.report_path = os.path.join(tempfile.gettempdir(), |
| gft_report.GetReportName(data)) |
| |
| gft_common.WriteFile(g_options.report_path, data) |
| report_saved_message = 'Report saved in: %s' % g_options.report_path |
| # To help scripting, the message must be printed to console and logged. |
| print report_saved_message |
| gft_common.Log(report_saved_message) |
| return True |
| |
| |
| @GFTCommand |
| def upload_report(): |
| """ Uploads a report (by --create_report) for tracking factory process. """ |
| report_path = g_options.report_path |
| if not (report_path and os.path.exists(report_path)): |
| ErrorDie('upload_report: need an existing report file ' |
| '(--report_path or --create_report)') |
| if not g_options.upload_method: |
| ErrorDie('upload_report: need proper --upload_method.') |
| |
| # Always report in standard naming scheme. |
| report_blob = gft_common.ReadBinaryFile(report_path) |
| report_name = gft_report.GetReportName(report_blob) |
| local_file = os.path.join(tempfile.gettempdir(), report_name) |
| if local_file != report_path: |
| gft_common.WriteBinaryFile(local_file, report_blob) |
| if not gft_upload.Upload(local_file, g_options.upload_method): |
| ErrorDie('upload_report: failed to upload.') |
| return True |
| |
| |
| @GFTCommand |
| def wpfw(): |
| """ Enables and verifies firmware write protection status. """ |
| if g_options.debug_dryrun_wpfw: |
| ErrorMsg('wpfw is by-passed. This device CANNOT be qualified.') |
| return True |
| read_binary = gft_common.ReadBinaryFile |
| ec_fw = crosfw.LoadEcFirmware() |
| main_fw = crosfw.LoadMainFirmware() |
| target_set = [('bios', main_fw.path)] |
| if ec_fw.chip_id is not None: |
| target_set.insert(0, ('ec', ec_fw.path)) |
| assert target_set[0][0] == 'ec' |
| else: |
| VerboseMsg('wpfw: current platform does not have EC. Ignored.') |
| assert len(target_set) > 0 |
| for target_name, loader_api in target_set: |
| if not EnableWriteProtect(target_name, read_binary(loader_api())): |
| ErrorDie('wpfw: failed to enable firmware write protection.') |
| return True |
| |
| |
| @GFTCommand |
| def clear_gbb_flags(): |
| """ Clears GBB Header Flags for shipping devices. """ |
| script = os.path.join(os.path.split(sys.argv[0])[0], |
| 'gft_clear_gbb_flags.sh') |
| command = '%s' % script |
| try: |
| VerboseSystemCommand(command, prompt='clear_gbb_flags: ') |
| except gft_common.GFTError, e: |
| ErrorDie('clear_gbb_flags: FAILED.\n%s' % e.value) |
| return True |
| |
| |
| @GFTCommand |
| def prepare_wipe(): |
| """ Prepares the system for transitioning to release state in next reboot. """ |
| if g_options.debug_dryrun_prepare_wipe: |
| ErrorMsg('prepare_wipe is by-passed. This device CANNOT be qualified.') |
| return True |
| wipe_script = os.path.join(os.path.split(sys.argv[0])[0], |
| 'gft_prepare_wipe.sh') |
| if not os.path.exists(wipe_script): |
| ErrorDie('prepare_wipe: cannot find script to prepare for wiping.') |
| wipe_method_map = { |
| 'fast': 'fast', |
| 'secure': '', |
| } |
| method = g_options.wipe_method |
| if method not in wipe_method_map: |
| ErrorDie('prepare_wipe: unknown wipe method: %s' % method) |
| tag = wipe_method_map[method] |
| rootfs_device = GetReleaseRootPartitionPath() |
| command = 'FACTORY_WIPE_TAGS=%s %s %s' % (tag, wipe_script, rootfs_device) |
| try: |
| VerboseSystemCommand(command, prompt='prepare_wipe: ') |
| except gft_common.GFTError, e: |
| ErrorDie('prepare_wipe: FAILED.\n%s' % e.value) |
| return True |
| |
| |
| @GFTCommand |
| @GFTCommandDependency((verify_switch_dev, |
| verify_switch_wp, |
| verify_hwid, |
| verify_vpd, |
| verify_system_time, |
| verify_keys, |
| verify_rootfs)) |
| def verify(): |
| """ Verifies if whole factory process is ready for finalization. """ |
| # See the dependency list for the real commands to be executed. |
| return True |
| |
| |
| @GFTCommand |
| @GFTCommandDependency((clear_gbb_flags, |
| verify, |
| wpfw, |
| create_report, |
| upload_report, |
| prepare_wipe)) |
| def finalize(): |
| """ Finalizes all factory tests and transit system into release state. """ |
| # See the dependency list for the real commands to be executed. |
| return True |
| |
| |
| @GFTCommand |
| @GFTCommandDependency((clear_gbb_flags, |
| verify_hwid, |
| verify_vpd, |
| verify_system_time, |
| verify_keys, |
| verify_rootfs, |
| create_report, |
| upload_report, |
| prepare_wipe)) |
| def developer_finalize(): |
| """ Finalizes tests and leave the system in developer-friendly mode. |
| |
| This is similar to 'finalize' command but not enforcing commands that would |
| prevent developers to restart factory process, for example write protection |
| (wpfw) and non-developer mode (verify_switch_dev). |
| """ |
| # See the dependency list for the real commands to be executed. |
| return True |
| |
| |
| ######################################################################## |
| # console command line options |
| |
| def ConsoleParser(): |
| """ command line option parser """ |
| parser = optparse.OptionParser() |
| |
| # Message control |
| parser.add_option('-d', '--debug', action='store_true', |
| help='provide debug messages') |
| parser.add_option('-v', '--verbose', action='store_true', default=False, |
| help='provide verbose messages') |
| parser.add_option('--log_path', metavar='PATH', |
| default=gft_common.DEFAULT_CONSOLE_LOG_PATH, |
| help='use the given path for verbose logs.') |
| # Debugging |
| parser.add_option('--debug_dryrun_wpfw', action='store_true', |
| help='make --wpfw in dry-run mode (debugging only)') |
| parser.add_option('--debug_dryrun_prepare_wipe', action='store_true', |
| help='make --prepare_wipe in dry-run mode (debugging only)') |
| parser.add_option('--ignore_factory_initial', action='store_true', |
| help='ignore components in config_factory_initial.') |
| # Configuration |
| parser.add_option('--config', metavar='PATH', |
| help='path to the config file (TBD)') |
| # TODO(tammo): Consider removing db_path option once the hwid_tool |
| # has meaningful default path searching. |
| parser.add_option('--db_path', metavar='PATH', |
| default=hwid_tool.DEFAULT_HWID_DATA_PATH, |
| help='path to the HWID component database') |
| # TODO(tammo): Add proper support for per-command required positional args. |
| parser.add_option('--hwid', metavar='HWID', help='HWID string') |
| # Reports |
| parser.add_option('--report_path', metavar='PATH', |
| help='override the path for report file') |
| parser.add_option('--report_format', metavar='FORMAT', default='gz', |
| help='format of the generated report (see gft_report.py)') |
| parser.add_option('--report_tag', metavar='TAG', default='', |
| help='tag to be included in report (see gft_report.py), ' |
| 'current possible values may include rma, mp, pvt.') |
| parser.add_option('--upload_method', metavar='METHOD:PARAM', |
| help='assign the upload method (see gft_upload).') |
| # Wiping |
| parser.add_option('--wipe_method', metavar='METHOD', default='secure', |
| help='assign the wipe method (secure/fast).') |
| |
| for command in g_commands: |
| parser.add_option('--%s' % command.__name__, |
| action='store_true', |
| help=command.__doc__) |
| return parser |
| |
| |
| ######################################################################## |
| # main entry |
| |
| |
| @gft_common.GFTConsole |
| def _main(): |
| """ main entry """ |
| global g_options, g_args |
| parser = ConsoleParser() |
| (g_options, g_args) = parser.parse_args() |
| if g_args: |
| parser.error('Un-expected parameter(s): %s\n' % ' '.join(g_args)) |
| |
| if g_options.debug: |
| g_options.verbose = True |
| gft_common.SetDebugLevel(g_options.debug) |
| gft_common.SetVerboseLevel(g_options.verbose, g_options.log_path) |
| DebugMsg('options: ' + repr(g_options) + '\nargs: ' + repr(g_args), log=False) |
| |
| # execute commands by order. |
| executed_commands = 0 |
| return_value = 0 |
| for command in g_commands: |
| command_name = command.__name__ |
| if not getattr(g_options, command_name): |
| continue |
| DebugMsg('COMMAND: ' + command_name, log=False) |
| if not command(): |
| return_value = 1 |
| executed_commands = executed_commands + 1 |
| |
| if executed_commands == 0: |
| parser.print_help() |
| return return_value |
| |
| |
| if __name__ == '__main__': |
| sys.exit(_main()) |