xref: /external/autotest/client/common_lib/cros/cr50_utils.py

# Copyright 2017 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.

import argparse
import logging
import os
import re

from autotest_lib.client.common_lib import error


RO = 'ro'
RW = 'rw'
CR50_FILE = '/opt/google/cr50/firmware/cr50.bin.prod'
CR50_STATE = '/var/cache/cr50*'
GET_CR50_VERSION = 'cat /var/cache/cr50-version'
GET_CR50_MESSAGES ='grep "cr50-.*\[" /var/log/messages'
UPDATE_FAILURE = 'unexpected cr50-update exit code'
DUMMY_VER = '-1.-1.-1'
# This dictionary is used to search the usb_updater output for the version
# strings. There are two usb_updater commands that will return versions:
# 'fwver' and 'binvers'.
#
# 'fwver'   is used to get the running RO and RW versions from cr50
# 'binvers'  gets the version strings for each RO and RW region in the given
#            file
#
# The value in the dictionary is the regular expression that can be used to
# find the version strings for each region.
VERSION_RE = {
    "--fwver" : '\nRO (?P<ro>\S+).*\nRW (?P<rw>\S+)',
    "--binvers" : 'RO_A:(?P<ro_a>\S+).*RW_A:(?P<rw_a>\S+).*' \
           'RO_B:(?P<ro_b>\S+).*RW_B:(?P<rw_b>\S+)',
}
UPDATE_TIMEOUT = 60
UPDATE_OK = 1

ERASED_BID_INT = 0xffffffff
# With an erased bid, the flags and board id will both be erased
ERASED_BID = (ERASED_BID_INT, ERASED_BID_INT)

usb_update = argparse.ArgumentParser()
# use /dev/tpm0 to send the command
usb_update.add_argument('-s', '--systemdev', dest='systemdev',
                        action='store_true')
# fwver, binver, and board id are used to get information about cr50 or an
# image.
usb_update.add_argument('-b', '--binvers', '-f', '--fwver', '-i', '--board_id',
                        dest='info_cmd', action='store_true')
# upstart and post_reset will post resets instead of rebooting immediately
usb_update.add_argument('-u', '--upstart', '-p', '--post_reset',
                        dest='post_reset', action='store_true')
usb_update.add_argument('extras', nargs=argparse.REMAINDER)


def AssertVersionsAreEqual(name_a, ver_a, name_b, ver_b):
    """Raise an error ver_a isn't the same as ver_b

    Args:
        name_a: the name of section a
        ver_a: the version string for section a
        name_b: the name of section b
        ver_b: the version string for section b

    Raises:
        AssertionError if ver_a is not equal to ver_b
    """
    assert ver_a == ver_b, ("Versions do not match: %s %s %s %s" %
                            (name_a, ver_a, name_b, ver_b))


def GetNewestVersion(ver_a, ver_b):
    """Compare the versions. Return the newest one. If they are the same return
    None."""
    a = [int(x) for x in ver_a.split('.')]
    b = [int(x) for x in ver_b.split('.')]

    if a > b:
        return ver_a
    if b > a:
        return ver_b
    return None


def GetVersion(versions, name):
    """Return the version string from the dictionary.

    Get the version for each key in the versions dictionary that contains the
    substring name. Make sure all of the versions match and return the version
    string. Raise an error if the versions don't match.

    Args:
        version: dictionary with the partition names as keys and the
                 partition version strings as values.
        name: the string used to find the relevant items in versions.

    Returns:
        the version from versions or "-1.-1.-1" if an invalid RO was detected.
    """
    ver = None
    key = None
    for k, v in versions.iteritems():
        if name in k:
            if v == DUMMY_VER:
                logging.info("Detected invalid %s %s", name, v)
                return v
            elif ver:
                AssertVersionsAreEqual(key, ver, k, v)
            else:
                ver = v
                key = k
    return ver


def FindVersion(output, arg):
    """Find the ro and rw versions.

    Args:
        output: The string to search
        arg: string representing the usb_updater option, either '--binvers' or
             '--fwver'

    Returns:
        a tuple of the ro and rw versions
    """
    versions = re.search(VERSION_RE[arg], output)
    versions = versions.groupdict()
    ro = GetVersion(versions, RO)
    rw = GetVersion(versions, RW)
    return ro, rw


def GetSavedVersion(client):
    """Return the saved version from /var/cache/cr50-version"""
    result = client.run(GET_CR50_VERSION).stdout.strip()
    return FindVersion(result, "--fwver")


def UsbUpdater(client, args):
    """Run usb_update with the given args.

    Args:
        client: the object to run commands on
        args: a list of strings that contiain the usb_updater args

    Returns:
        the result of usb_update
    """
    options = usb_update.parse_args(args)

    result = client.run("status trunksd")
    if options.systemdev and 'running' in result.stdout:
        client.run("stop trunksd")

    # If we are updating the cr50 image, usb_update will return a non-zero exit
    # status so we should ignore it.
    ignore_status = not options.info_cmd
    # immediate reboots are only honored if the command is sent using /dev/tpm0
    expect_reboot = (options.systemdev and not options.post_reset and
                     not options.info_cmd)

    result = client.run("usb_updater %s" % ' '.join(args),
                        ignore_status=ignore_status,
                        ignore_timeout=expect_reboot,
                        timeout=UPDATE_TIMEOUT)

    # After a posted reboot, the usb_update exit code should equal 1.
    if result.exit_status and result.exit_status != UPDATE_OK:
        logging.debug(result)
        raise error.TestFail("Unexpected usb_update exit code after %s %d" %
                             (' '.join(args), result.exit_status))
    return result


def GetVersionFromUpdater(client, args):
    """Return the version from usb_updater"""
    result = UsbUpdater(client, args).stdout.strip()
    return FindVersion(result, args[0])


def GetFwVersion(client):
    """Get the running version using 'usb_updater --fwver'"""
    return GetVersionFromUpdater(client, ['--fwver', '-s'])


def GetBinVersion(client, image=CR50_FILE):
    """Get the image version using 'usb_updater --binvers image'"""
    # TODO(mruthven) b/37958867: change to ["--binvers", image] when usb_updater
    # is fixed
    return GetVersionFromUpdater(client, ['--binvers', image, image, '-s'])


def GetVersionString(ver):
    return 'RO %s RW %s' % (ver[0], ver[1])


def GetRunningVersion(client):
    """Get the running Cr50 version.

    The version from usb_updater and /var/cache/cr50-version should be the
    same. Get both versions and make sure they match.

    Args:
        client: the object to run commands on

    Returns:
        running_ver: a tuple with the ro and rw version strings

    Raises:
        TestFail
        - If the version in /var/cache/cr50-version is not the same as the
          version from 'usb_updater --fwver'
    """
    running_ver = GetFwVersion(client)
    saved_ver = GetSavedVersion(client)

    AssertVersionsAreEqual("Running", GetVersionString(running_ver),
                           "Saved", GetVersionString(saved_ver))
    return running_ver


def CheckForFailures(client, last_message):
    """Check for any unexpected cr50-update exit codes.

    This only checks the cr50 update messages that have happened since
    last_message. If a unexpected exit code is detected it will raise an error>

    Args:
        client: the object to run commands on
        last_message: the last cr50 message from the last update run

    Returns:
        the last cr50 message in /var/log/messages

    Raises:
        TestFail
            - If there is a unexpected cr50-update exit code after last_message
              in /var/log/messages
    """
    messages = client.run(GET_CR50_MESSAGES).stdout.strip()
    if last_message:
        messages = messages.rsplit(last_message, 1)[-1].split('\n')
        failures = []
        for message in messages:
            if UPDATE_FAILURE in message:
                failures.append(message)
        if len(failures):
            logging.info(messages)
            raise error.TestFail("Detected unexpected exit code during update: "
                                 "%s" % failures)
    return messages[-1]


def VerifyUpdate(client, ver='', last_message=''):
    """Verify that the saved update state is correct and there were no
    unexpected cr50-update exit codes since the last update.

    Args:
        client: the object to run commands on
        ver: the expected version tuple (ro ver, rw ver)
        last_message: the last cr50 message from the last update run

    Returns:
        new_ver: a tuple containing the running ro and rw versions
        last_message: The last cr50 update message in /var/log/messages
    """
    # Check that there were no unexpected reboots from cr50-result
    last_message = CheckForFailures(client, last_message)
    logging.debug("last cr50 message %s", last_message)

    new_ver = GetRunningVersion(client)
    if ver != '':
        if DUMMY_VER != ver[0]:
            AssertVersionsAreEqual("Old RO", ver[0], "Updated RO", new_ver[0])
        AssertVersionsAreEqual("Old RW", ver[1], "Updated RW", new_ver[1])
    return new_ver, last_message


def ClearUpdateStateAndReboot(client):
    """Removes the cr50 status files in /var/cache and reboots the AP"""
    client.run("rm %s" % CR50_STATE)
    client.reboot()


def InstallImage(client, src, dest=CR50_FILE):
    """Copy the image at src to dest on the dut

    Args:
        client: the object to run commands on
        src: the image location of the server
        dest: the desired location on the dut

    Returns:
        The filename where the image was copied to on the dut, a tuple
        containing the RO and RW version of the file
    """
    # Send the file to the DUT
    client.send_file(src, dest)

    ver = GetBinVersion(client, dest)
    client.run("sync")
    return dest, ver


def GetSymbolicBoardId(symbolic_board_id):
    """Convert the symbolic board id str to an int

    Args:
        symbolic_board_id: a ASCII string. It can be up to 4 characters

    Returns:
        the symbolic board id string converted to an int
    """
    board_id = 0
    for c in symbolic_board_id:
        board_id = ord(c) | (board_id << 8)
    return board_id


def GetExpectedBoardId(board_id):
    """"Return the usb_updater interpretation of board_id

    Args:
        board_id: a int or string value of the board id

    Returns:
        a int representation of the board id
    """
    if type(board_id) == int:
        return board_id

    if len(board_id) <= 4:
        return GetSymbolicBoardId(board_id)

    return int(board_id, 16)


def GetExpectedFlags(flags):
    """If flags are not specified, usb_updater will set them to 0xff00

    Args:
        flags: The int value or None

    Returns:
        the original flags or 0xff00 if flags is None
    """
    return flags if flags != None else 0xff00


def GetBoardId(client):
    """Return the board id and flags

    Args:
        client: the object to run commands on

    Returns:
        a tuple with the hex value board id, flags

    Raises:
        TestFail if the second board id response field is not ~board_id
    """
    result = UsbUpdater(client, ["-i"]).stdout.strip()
    board_id_info = result.split("Board ID space: ")[-1].strip().split(":")
    board_id, board_id_inv, flags = [int(val, 16) for val in board_id_info]
    logging.info('BOARD_ID: %x:%x:%x', board_id, board_id_inv, flags)

    if board_id == board_id_inv == flags == ERASED_BID_INT:
        logging.info('board id is erased')
    elif board_id & board_id_inv:
        raise error.TestFail('board_id_inv should be ~board_id got %x %x' %
                             (board_id, board_id_inv))
    return board_id, flags


def CheckBoardId(client, board_id, flags):
    """Compare the given board_id and flags to the running board_id and flags

    Interpret board_id and flags how usb_updater would interpret them, then
    compare those interpreted values to the running board_id and flags.

    Args:
        client: the object to run commands on
        board_id: a hex, symbolic or int value for board_id
        flags: the int value of flags or None

    Raises:
        TestFail if the new board id info does not match
    """
    # Read back the board id and flags
    new_board_id, new_flags = GetBoardId(client)

    expected_board_id = GetExpectedBoardId(board_id)
    expected_flags = GetExpectedFlags(flags)

    if new_board_id != expected_board_id or new_flags != expected_flags:
        raise error.TestFail('Failed to set board id expected %x:%x, but got '
                             '%x:%x' % (expected_board_id, expected_flags,
                             new_board_id, new_flags))


def SetBoardId(client, board_id, flags=None):
    """Sets the board id and flags

    Args:
        client: the object to run commands on
        board_id: a string of the symbolic board id or board id hex value. If
                  the string is less than 4 characters long it will be
                  considered a symbolic value
        flags: the desired flag value. If board_id is a symbolic value, then
               this will be ignored.

    Raises:
        TestFail if we were unable to set the flags to the correct value
    """

    board_id_arg = board_id
    if flags != None:
        board_id_arg += ':' + hex(flags)

    # Set the board id using the given board id and flags
    result = UsbUpdater(client, ["-s", "-i", board_id_arg]).stdout.strip()

    CheckBoardId(client, board_id, flags)