#!/usr/bin/env python3 # Copyright 2020 The Chromium Authors # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. """Updates the Fuchsia images to the given revision. Should be used in a 'hooks_os' entry so that it only runs when .gclient's target_os includes 'fuchsia'. Note, for a smooth transition, this file automatically adds '-release' at the end of the image gcs file name to eliminate the difference between product bundle v2 and gce files.""" # TODO(crbug.com/1496426): Remove this file. import argparse import itertools import logging import os import re import subprocess import sys from typing import Dict, Optional sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), 'test'))) from common import DIR_SRC_ROOT, IMAGES_ROOT, get_host_os, \ make_clean_directory from gcs_download import DownloadAndUnpackFromCloudStorage from update_sdk import GetSDKOverrideGCSPath IMAGE_SIGNATURE_FILE = '.hash' # TODO(crbug.com/1138433): Investigate whether we can deprecate # use of sdk_bucket.txt. def GetOverrideCloudStorageBucket(): """Read bucket entry from sdk_bucket.txt""" return ReadFile('sdk-bucket.txt').strip() def ReadFile(filename): """Read a file in this directory.""" with open(os.path.join(os.path.dirname(__file__), filename), 'r') as f: return f.read() def StrExpansion(): return lambda str_value: str_value def VarLookup(local_scope): return lambda var_name: local_scope['vars'][var_name] def GetImageHashList(bucket): """Read filename entries from sdk-hash-files.list (one per line), substitute {platform} in each entry if present, and read from each filename.""" assert (get_host_os() == 'linux') filenames = [ line.strip() for line in ReadFile('sdk-hash-files.list').replace( '{platform}', 'linux_internal').splitlines() ] image_hashes = [ReadFile(filename).strip() for filename in filenames] return image_hashes def ParseDepsDict(deps_content): local_scope = {} global_scope = { 'Str': StrExpansion(), 'Var': VarLookup(local_scope), 'deps_os': {}, } exec(deps_content, global_scope, local_scope) return local_scope def ParseDepsFile(filename): with open(filename, 'rb') as f: deps_content = f.read() return ParseDepsDict(deps_content) def GetImageHash(bucket): """Gets the hash identifier of the newest generation of images.""" if bucket == 'fuchsia-sdk': hashes = GetImageHashList(bucket) return max(hashes) deps_file = os.path.join(DIR_SRC_ROOT, 'DEPS') return ParseDepsFile(deps_file)['vars']['fuchsia_version'].split(':')[1] def GetImageSignature(image_hash, boot_images): return 'gn:{image_hash}:{boot_images}:'.format(image_hash=image_hash, boot_images=boot_images) def GetAllImages(boot_image_names): if not boot_image_names: return all_device_types = ['generic', 'qemu'] all_archs = ['x64', 'arm64'] images_to_download = set() for boot_image in boot_image_names.split(','): components = boot_image.split('.') if len(components) != 2: continue device_type, arch = components device_images = all_device_types if device_type == '*' else [device_type] arch_images = all_archs if arch == '*' else [arch] images_to_download.update(itertools.product(device_images, arch_images)) return images_to_download def DownloadBootImages(bucket, image_hash, boot_image_names, image_root_dir): images_to_download = GetAllImages(boot_image_names) for image_to_download in images_to_download: device_type = image_to_download[0] arch = image_to_download[1] image_output_dir = os.path.join(image_root_dir, arch, device_type) if os.path.exists(image_output_dir): continue logging.info('Downloading Fuchsia boot images for %s.%s...', device_type, arch) # Legacy images use different naming conventions. See fxbug.dev/85552. legacy_delimiter_device_types = ['qemu', 'generic'] if bucket == 'fuchsia-sdk' or \ device_type not in legacy_delimiter_device_types: type_arch_connector = '.' else: type_arch_connector = '-' images_tarball_url = 'gs://{bucket}/development/{image_hash}/images/'\ '{device_type}{type_arch_connector}{arch}.tgz'.format( bucket=bucket, image_hash=image_hash, device_type=device_type, type_arch_connector=type_arch_connector, arch=arch) try: DownloadAndUnpackFromCloudStorage(images_tarball_url, image_output_dir) except subprocess.CalledProcessError as e: logging.exception('Failed to download image %s from URL: %s', image_to_download, images_tarball_url) raise e def _GetImageOverrideInfo() -> Optional[Dict[str, str]]: """Get the bucket location from sdk_override.txt.""" location = GetSDKOverrideGCSPath() if not location: return None m = re.match(r'gs://([^/]+)/development/([^/]+)/?(?:sdk)?', location) if not m: raise ValueError('Badly formatted image override location %s' % location) return { 'bucket': m.group(1), 'image_hash': m.group(2), } def GetImageLocationInfo(default_bucket: str, allow_override: bool = True) -> Dict[str, str]: """Figures out where to pull the image from. Defaults to the provided default bucket and generates the hash from defaults. If sdk_override.txt exists (and is allowed) it uses that bucket instead. Args: default_bucket: a given default for what bucket to use allow_override: allow SDK override to be used. Returns: A dictionary containing the bucket and image_hash """ # if sdk_override.txt exists (and is allowed) use the image from that bucket. if allow_override: override = _GetImageOverrideInfo() if override: return override # Use the bucket in sdk-bucket.txt if an entry exists. # Otherwise use the default bucket. bucket = GetOverrideCloudStorageBucket() or default_bucket return { 'bucket': bucket, 'image_hash': GetImageHash(bucket), } def main(): parser = argparse.ArgumentParser() parser.add_argument('--verbose', '-v', action='store_true', help='Enable debug-level logging.') parser.add_argument( '--boot-images', type=str, required=True, help='List of boot images to download, represented as a comma separated ' 'list. Wildcards are allowed. ') parser.add_argument( '--default-bucket', type=str, default='fuchsia', help='The Google Cloud Storage bucket in which the Fuchsia images are ' 'stored. Entry in sdk-bucket.txt will override this flag.') parser.add_argument( '--image-root-dir', default=IMAGES_ROOT, help='Specify the root directory of the downloaded images. Optional') parser.add_argument( '--allow-override', action='store_true', help='Whether sdk_override.txt can be used for fetching the image, if ' 'it exists.') args = parser.parse_args() logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO) # If no boot images need to be downloaded, exit. if not args.boot_images: return 0 for index, item in enumerate(args.boot_images): # The gclient configuration is in the src-internal and cannot be changed # atomically with the test script in src. So the update_images.py needs to # support both scenarios before being fully deprecated for older milestones. if not item.endswith('-release'): args.boot_images[index] = item + '-release' # Check whether there's Fuchsia support for this platform. get_host_os() image_info = GetImageLocationInfo(args.default_bucket, args.allow_override) bucket = image_info['bucket'] image_hash = image_info['image_hash'] if not image_hash: return 1 signature_filename = os.path.join(args.image_root_dir, IMAGE_SIGNATURE_FILE) current_signature = (open(signature_filename, 'r').read().strip() if os.path.exists(signature_filename) else '') new_signature = GetImageSignature(image_hash, args.boot_images) if current_signature != new_signature: logging.info('Downloading Fuchsia images %s from bucket %s...', image_hash, bucket) make_clean_directory(args.image_root_dir) try: DownloadBootImages(bucket, image_hash, args.boot_images, args.image_root_dir) with open(signature_filename, 'w') as f: f.write(new_signature) except subprocess.CalledProcessError as e: logging.exception("command '%s' failed with status %d.%s", ' '.join(e.cmd), e.returncode, ' Details: ' + e.output if e.output else '') raise e else: logging.info('Signatures matched! Got %s', new_signature) return 0 if __name__ == '__main__': sys.exit(main())