Binman requires various tools to actually work, such as 'lz4' to compress data and 'futility' to sign Chrome OS firmware. At present these are handled in an ad-hoc manner and there is no easy way to find out what tools are needd to build an image, nor where to obtain them.
Add an implementation of 'bintool', a base class which implements this functionality. When a bintool is required, it can be requested from this module, then executed. When the tool is missing, it can provide a way to obtain it.
Note that this uses Command directly, not the tools.Run() function. This allows proper handling of missing tools and avoids needing to catch and re-raise exceptions.
Signed-off-by: Simon Glass sjg@chromium.org ---
(no changes since v1)
tools/binman/bintool.py | 421 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 421 insertions(+) create mode 100644 tools/binman/bintool.py
diff --git a/tools/binman/bintool.py b/tools/binman/bintool.py new file mode 100644 index 00000000000..34102dafa2a --- /dev/null +++ b/tools/binman/bintool.py @@ -0,0 +1,421 @@ +# SPDX-License-Identifier: GPL-2.0+ +# Copyright 2022 Google LLC +# +"""Base class for all bintools + +This defines the common functionality for all bintools, including running +the tool, checking its version and fetching it if needed. +""" + +import collections +import glob +import importlib +import multiprocessing +import os +import shutil +import tempfile +import urllib.error + +from patman import command +from patman import terminal +from patman import tools +from patman import tout + +BINMAN_DIR = os.path.dirname(os.path.realpath(__file__)) + +# Format string for listing bintools, see also the header in list_all() +FORMAT = '%-16.16s %-12.12s %-26.26s %s' + +# List of known modules, to avoid importing the module multiple times +modules = {} + +# Possible ways of fetching a tool (FETCH_COUNT is number of ways) +FETCH_ANY, FETCH_BIN, FETCH_BUILD, FETCH_COUNT = range(4) + +FETCH_NAMES = { + FETCH_ANY: 'any method', + FETCH_BIN: 'binary download', + FETCH_BUILD: 'build from source' + } + +# Status of tool fetching +FETCHED, FAIL, PRESENT, STATUS_COUNT = range(4) + +DOWNLOAD_DESTDIR = os.path.join(os.getenv('HOME'), 'bin') + +class Bintool: + """Tool which operates on binaries to help produce entry contents + + This is the base class for all bintools + """ + # List of bintools to regard as missing + missing_list = [] + + def __init__(self, name, desc): + self.name = name + self.desc = desc + + @staticmethod + def find_bintool_class(btype): + """Look up the bintool class for bintool + + Args: + byte: Bintool to use, e.g. 'mkimage' + + Returns: + The bintool class object if found, else a tuple: + module name that could not be found + exception received + """ + # Convert something like 'u-boot' to 'u_boot' since we are only + # interested in the type. + module_name = btype.replace('-', '_') + module = modules.get(module_name) + + # Import the module if we have not already done so + if not module: + try: + module = importlib.import_module('binman.btool.' + module_name) + except ImportError as exc: + return module_name, exc + modules[module_name] = module + + # Look up the expected class name + return getattr(module, 'Bintool%s' % module_name) + + @staticmethod + def create(name): + """Create a new bintool object + + Args: + name (str): Bintool to create, e.g. 'mkimage' + + Returns: + A new object of the correct type (a subclass of Binutil) + """ + cls = Bintool.find_bintool_class(name) + if isinstance(cls, tuple): + raise ValueError("Cannot import bintool module '%s': %s" % cls) + + # Call its constructor to get the object we want. + obj = cls(name) + return obj + + def show(self): + """Show a line of information about a bintool""" + if self.is_present(): + version = self.version() + else: + version = '-' + print(FORMAT % (self.name, version, self.desc, + self.get_path() or '(not found)')) + + @classmethod + def set_missing_list(cls, missing_list): + cls.missing_list = missing_list or [] + + @staticmethod + def get_tool_list(include_testing=False): + """Get a list of the known tools + + Returns: + list of str: names of all tools known to binman + """ + files = glob.glob(os.path.join(BINMAN_DIR, 'btool/*')) + names = [os.path.splitext(os.path.basename(fname))[0] + for fname in files] + names = [name for name in names if name[0] != '_'] + if include_testing: + names.append('_testing') + return sorted(names) + + @staticmethod + def list_all(): + """List all the bintools known to binman""" + names = Bintool.get_tool_list() + print(FORMAT % ('Name', 'Version', 'Description', 'Path')) + print(FORMAT % ('-' * 15,'-' * 11, '-' * 25, '-' * 30)) + for name in names: + btool = Bintool.create(name) + btool.show() + + def is_present(self): + """Check if a bintool is available on the system + + Returns: + bool: True if available, False if not + """ + if self.name in self.missing_list: + return False + return bool(self.get_path()) + + def get_path(self): + """Get the path of a bintool + + Returns: + str: Path to the tool, if available, else None + """ + return tools.tool_find(self.name) + + def fetch_tool(self, method, col, skip_present): + """Fetch a single tool + + Args: + method (FETCH_...): Method to use + col (terminal.Color): Color terminal object + skip_present (boo;): Skip fetching if it is already present + + Returns: + int: Result of fetch either FETCHED, FAIL, PRESENT + """ + def try_fetch(meth): + res = None + try: + res = self.fetch(meth) + except urllib.error.URLError as uerr: + message = uerr.reason + print(col.Color(col.RED, f'- {message}')) + + except ValueError as exc: + print(f'Exception: {exc}') + return res + + if skip_present and self.is_present(): + return PRESENT + print(col.Color(col.YELLOW, 'Fetch: %s' % self.name)) + if method == FETCH_ANY: + for try_method in range(1, FETCH_COUNT): + print(f'- trying method: {FETCH_NAMES[try_method]}') + result = try_fetch(try_method) + if result: + break + else: + result = try_fetch(method) + if not result: + return FAIL + if result is not True: + fname, tmpdir = result + dest = os.path.join(DOWNLOAD_DESTDIR, self.name) + print(f"- writing to '{dest}'") + shutil.move(fname, dest) + if tmpdir: + shutil.rmtree(tmpdir) + return FETCHED + + @staticmethod + def fetch_tools(method, names_to_fetch): + """Fetch bintools from a suitable place + + This fetches or builds the requested bintools so that they can be used + by binman + + Args: + names_to_fetch (list of str): names of bintools to fetch + + Returns: + True on success, False on failure + """ + def show_status(color, prompt, names): + print(col.Color( + color, f'{prompt}:%s{len(names):2}: %s' % + (' ' * (16 - len(prompt)), ' '.join(names)))) + + col = terminal.Color() + skip_present = False + name_list = names_to_fetch + if len(names_to_fetch) == 1 and names_to_fetch[0] in ['all', 'missing']: + name_list = Bintool.get_tool_list() + if names_to_fetch[0] == 'missing': + skip_present = True + print(col.Color(col.YELLOW, + 'Fetching tools: %s' % ' '.join(name_list))) + status = collections.defaultdict(list) + for name in name_list: + btool = Bintool.create(name) + result = btool.fetch_tool(method, col, skip_present) + status[result].append(name) + if result == FAIL: + if method == FETCH_ANY: + print('- failed to fetch with all methods') + else: + print(f"- method '{FETCH_NAMES[method]}' is not supported") + + if len(name_list) > 1: + if skip_present: + show_status(col.GREEN, 'Already present', status[PRESENT]) + show_status(col.GREEN, 'Tools fetched', status[FETCHED]) + if status[FAIL]: + show_status(col.RED, 'Failures', status[FAIL]) + return not status[FAIL] + + def run_cmd_result(self, *args, binary=False, raise_on_error=True): + """Run the bintool using command-line arguments + + Args: + args (list of str): Arguments to provide, in addition to the bintool + name + binary (bool): True to return output as bytes instead of str + raise_on_error (bool): True to raise a ValueError exception if the + tool returns a non-zero return code + + Returns: + CommandResult: Resulting output from the bintool, or None if the + tool is not present + """ + if self.name in self.missing_list: + return None + name = os.path.expanduser(self.name) # Expand paths containing ~ + all_args = (name,) + args + env = tools.get_env_with_path() + tout.Detail(f"bintool: {' '.join(all_args)}") + result = command.RunPipe( + [all_args], capture=True, capture_stderr=True, env=env, + raise_on_error=False, binary=binary) + + if result.return_code: + # Return None if the tool was not found. In this case there is no + # output from the tool and it does not appear on the path. We still + # try to run it (as above) since RunPipe() allows faking the tool's + # output + if not any([result.stdout, result.stderr, tools.tool_find(name)]): + tout.Info(f"bintool '{name}' not found") + return None + if raise_on_error: + tout.Info(f"bintool '{name}' failed") + raise ValueError("Error %d running '%s': %s" % + (result.return_code, ' '.join(all_args), + result.stderr or result.stdout)) + if result.stdout: + tout.Debug(result.stdout) + if result.stderr: + tout.Debug(result.stderr) + return result + + def run_cmd(self, *args, binary=False): + """Run the bintool using command-line arguments + + Args: + args (list of str): Arguments to provide, in addition to the bintool + name + binary (bool): True to return output as bytes instead of str + + Returns: + str or bytes: Resulting stdout from the bintool + """ + result = self.run_cmd_result(*args, binary=binary) + if result: + return result.stdout + + @classmethod + def build_from_git(cls, git_repo, make_target, bintool_path): + """Build a bintool from a git repo + + This clones the repo in a temporary directory, builds it with 'make', + then returns the filename of the resulting executable bintool + + Args: + git_repo (str): URL of git repo + make_target (str): Target to pass to 'make' to build the tool + bintool_path (str): Relative path of the tool in the repo, after + build is complete + + Returns: + tuple: + str: Filename of fetched file to copy to a suitable directory + str: Name of temp directory to remove, or None + or None on error + """ + tmpdir = tempfile.mkdtemp(prefix='binmanf.') + print(f"- clone git repo '{git_repo}' to '{tmpdir}'") + tools.Run('git', 'clone', '--depth', '1', git_repo, tmpdir) + print(f"- build target '{make_target}'") + tools.Run('make', '-C', tmpdir, '-j', f'{multiprocessing.cpu_count()}', + make_target) + fname = os.path.join(tmpdir, bintool_path) + if not os.path.exists(fname): + print(f"- File '{fname}' was not produced") + return None + return fname, tmpdir + + @classmethod + def fetch_from_url(cls, url): + """Fetch a bintool from a URL + + Args: + url (str): URL to fetch from + + Returns: + tuple: + str: Filename of fetched file to copy to a suitable directory + str: Name of temp directory to remove, or None + """ + fname, tmpdir = tools.Download(url) + tools.Run('chmod', 'a+x', fname) + return fname, tmpdir + + @classmethod + def fetch_from_drive(cls, drive_id): + """Fetch a bintool from Google drive + + Args: + drive_id (str): ID of file to fetch. For a URL of the form + 'https://drive.google.com/file/d/xxx/view?usp=sharing' the value + passed here should be 'xxx' + + Returns: + tuple: + str: Filename of fetched file to copy to a suitable directory + str: Name of temp directory to remove, or None + """ + url = f'https://drive.google.com/uc?export=download&id=%7Bdrive_id%7D' + return cls.fetch_from_url(url) + + @classmethod + def apt_install(cls, package): + """Install a bintool using the 'aot' tool + + This requires use of servo so may request a password + + Args: + package (str): Name of package to install + + Returns: + True, assuming it completes without error + """ + args = ['sudo', 'apt', 'install', '-y', package] + print('- %s' % ' '.join(args)) + tools.Run(*args) + return True + + # pylint: disable=W0613 + def fetch(self, method): + """Fetch handler for a bintool + + This should be implemented by the base class + + Args: + method (FETCH_...): Method to use + + Returns: + tuple: + str: Filename of fetched file to copy to a suitable directory + str: Name of temp directory to remove, or None + or True if the file was fetched and already installed + or None if no fetch() implementation is available + + Raises: + Valuerror: Fetching could not be completed + """ + print(f"No method to fetch bintool '{self.name}'") + return False + + # pylint: disable=R0201 + def version(self): + """Version handler for a bintool + + This should be implemented by the base class + + Returns: + str: Version string for this bintool + """ + return 'unknown'