blob: 985e7381b9488a2952fe322a1c9ebd520cd396f0 [file] [log] [blame]
#!/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())