| # |
| # Copyright (c) 2013, Intel Corporation. |
| # |
| # SPDX-License-Identifier: GPL-2.0-only |
| # |
| # DESCRIPTION |
| |
| # This module implements the image creation engine used by 'wic' to |
| # create images. The engine parses through the OpenEmbedded kickstart |
| # (wks) file specified and generates images that can then be directly |
| # written onto media. |
| # |
| # AUTHORS |
| # Tom Zanussi <tom.zanussi (at] linux.intel.com> |
| # |
| |
| import logging |
| import os |
| import tempfile |
| import json |
| import subprocess |
| |
| from collections import namedtuple, OrderedDict |
| from distutils.spawn import find_executable |
| |
| from wic import WicError |
| from wic.filemap import sparse_copy |
| from wic.pluginbase import PluginMgr |
| from wic.misc import get_bitbake_var, exec_cmd |
| |
| logger = logging.getLogger('wic') |
| |
| def verify_build_env(): |
| """ |
| Verify that the build environment is sane. |
| |
| Returns True if it is, false otherwise |
| """ |
| if not os.environ.get("BUILDDIR"): |
| raise WicError("BUILDDIR not found, exiting. (Did you forget to source oe-init-build-env?)") |
| |
| return True |
| |
| |
| CANNED_IMAGE_DIR = "lib/wic/canned-wks" # relative to scripts |
| SCRIPTS_CANNED_IMAGE_DIR = "scripts/" + CANNED_IMAGE_DIR |
| WIC_DIR = "wic" |
| |
| def build_canned_image_list(path): |
| layers_path = get_bitbake_var("BBLAYERS") |
| canned_wks_layer_dirs = [] |
| |
| if layers_path is not None: |
| for layer_path in layers_path.split(): |
| for wks_path in (WIC_DIR, SCRIPTS_CANNED_IMAGE_DIR): |
| cpath = os.path.join(layer_path, wks_path) |
| if os.path.isdir(cpath): |
| canned_wks_layer_dirs.append(cpath) |
| |
| cpath = os.path.join(path, CANNED_IMAGE_DIR) |
| canned_wks_layer_dirs.append(cpath) |
| |
| return canned_wks_layer_dirs |
| |
| def find_canned_image(scripts_path, wks_file): |
| """ |
| Find a .wks file with the given name in the canned files dir. |
| |
| Return False if not found |
| """ |
| layers_canned_wks_dir = build_canned_image_list(scripts_path) |
| |
| for canned_wks_dir in layers_canned_wks_dir: |
| for root, dirs, files in os.walk(canned_wks_dir): |
| for fname in files: |
| if fname.endswith("~") or fname.endswith("#"): |
| continue |
| if ((fname.endswith(".wks") and wks_file + ".wks" == fname) or \ |
| (fname.endswith(".wks.in") and wks_file + ".wks.in" == fname)): |
| fullpath = os.path.join(canned_wks_dir, fname) |
| return fullpath |
| return None |
| |
| |
| def list_canned_images(scripts_path): |
| """ |
| List the .wks files in the canned image dir, minus the extension. |
| """ |
| layers_canned_wks_dir = build_canned_image_list(scripts_path) |
| |
| for canned_wks_dir in layers_canned_wks_dir: |
| for root, dirs, files in os.walk(canned_wks_dir): |
| for fname in files: |
| if fname.endswith("~") or fname.endswith("#"): |
| continue |
| if fname.endswith(".wks") or fname.endswith(".wks.in"): |
| fullpath = os.path.join(canned_wks_dir, fname) |
| with open(fullpath) as wks: |
| for line in wks: |
| desc = "" |
| idx = line.find("short-description:") |
| if idx != -1: |
| desc = line[idx + len("short-description:"):].strip() |
| break |
| basename = fname.split('.')[0] |
| print(" %s\t\t%s" % (basename.ljust(30), desc)) |
| |
| |
| def list_canned_image_help(scripts_path, fullpath): |
| """ |
| List the help and params in the specified canned image. |
| """ |
| found = False |
| with open(fullpath) as wks: |
| for line in wks: |
| if not found: |
| idx = line.find("long-description:") |
| if idx != -1: |
| print() |
| print(line[idx + len("long-description:"):].strip()) |
| found = True |
| continue |
| if not line.strip(): |
| break |
| idx = line.find("#") |
| if idx != -1: |
| print(line[idx + len("#:"):].rstrip()) |
| else: |
| break |
| |
| |
| def list_source_plugins(): |
| """ |
| List the available source plugins i.e. plugins available for --source. |
| """ |
| plugins = PluginMgr.get_plugins('source') |
| |
| for plugin in plugins: |
| print(" %s" % plugin) |
| |
| |
| def wic_create(wks_file, rootfs_dir, bootimg_dir, kernel_dir, |
| native_sysroot, options): |
| """ |
| Create image |
| |
| wks_file - user-defined OE kickstart file |
| rootfs_dir - absolute path to the build's /rootfs dir |
| bootimg_dir - absolute path to the build's boot artifacts directory |
| kernel_dir - absolute path to the build's kernel directory |
| native_sysroot - absolute path to the build's native sysroots dir |
| image_output_dir - dirname to create for image |
| options - wic command line options (debug, bmap, etc) |
| |
| Normally, the values for the build artifacts values are determined |
| by 'wic -e' from the output of the 'bitbake -e' command given an |
| image name e.g. 'core-image-minimal' and a given machine set in |
| local.conf. If that's the case, the variables get the following |
| values from the output of 'bitbake -e': |
| |
| rootfs_dir: IMAGE_ROOTFS |
| kernel_dir: DEPLOY_DIR_IMAGE |
| native_sysroot: STAGING_DIR_NATIVE |
| |
| In the above case, bootimg_dir remains unset and the |
| plugin-specific image creation code is responsible for finding the |
| bootimg artifacts. |
| |
| In the case where the values are passed in explicitly i.e 'wic -e' |
| is not used but rather the individual 'wic' options are used to |
| explicitly specify these values. |
| """ |
| try: |
| oe_builddir = os.environ["BUILDDIR"] |
| except KeyError: |
| raise WicError("BUILDDIR not found, exiting. (Did you forget to source oe-init-build-env?)") |
| |
| if not os.path.exists(options.outdir): |
| os.makedirs(options.outdir) |
| |
| pname = options.imager |
| plugin_class = PluginMgr.get_plugins('imager').get(pname) |
| if not plugin_class: |
| raise WicError('Unknown plugin: %s' % pname) |
| |
| plugin = plugin_class(wks_file, rootfs_dir, bootimg_dir, kernel_dir, |
| native_sysroot, oe_builddir, options) |
| |
| plugin.do_create() |
| |
| logger.info("The image(s) were created using OE kickstart file:\n %s", wks_file) |
| |
| |
| def wic_list(args, scripts_path): |
| """ |
| Print the list of images or source plugins. |
| """ |
| if args.list_type is None: |
| return False |
| |
| if args.list_type == "images": |
| |
| list_canned_images(scripts_path) |
| return True |
| elif args.list_type == "source-plugins": |
| list_source_plugins() |
| return True |
| elif len(args.help_for) == 1 and args.help_for[0] == 'help': |
| wks_file = args.list_type |
| fullpath = find_canned_image(scripts_path, wks_file) |
| if not fullpath: |
| raise WicError("No image named %s found, exiting. " |
| "(Use 'wic list images' to list available images, " |
| "or specify a fully-qualified OE kickstart (.wks) " |
| "filename)" % wks_file) |
| |
| list_canned_image_help(scripts_path, fullpath) |
| return True |
| |
| return False |
| |
| |
| class Disk: |
| def __init__(self, imagepath, native_sysroot, fstypes=('fat', 'ext')): |
| self.imagepath = imagepath |
| self.native_sysroot = native_sysroot |
| self.fstypes = fstypes |
| self._partitions = None |
| self._partimages = {} |
| self._lsector_size = None |
| self._psector_size = None |
| self._ptable_format = None |
| |
| # find parted |
| # read paths from $PATH environment variable |
| # if it fails, use hardcoded paths |
| pathlist = "/bin:/usr/bin:/usr/sbin:/sbin/" |
| try: |
| self.paths = os.environ['PATH'] + ":" + pathlist |
| except KeyError: |
| self.paths = pathlist |
| |
| if native_sysroot: |
| for path in pathlist.split(':'): |
| self.paths = "%s%s:%s" % (native_sysroot, path, self.paths) |
| |
| self.parted = find_executable("parted", self.paths) |
| if not self.parted: |
| raise WicError("Can't find executable parted") |
| |
| self.partitions = self.get_partitions() |
| |
| def __del__(self): |
| for path in self._partimages.values(): |
| os.unlink(path) |
| |
| def get_partitions(self): |
| if self._partitions is None: |
| self._partitions = OrderedDict() |
| out = exec_cmd("%s -sm %s unit B print" % (self.parted, self.imagepath)) |
| parttype = namedtuple("Part", "pnum start end size fstype") |
| splitted = out.splitlines() |
| # skip over possible errors in exec_cmd output |
| try: |
| idx =splitted.index("BYT;") |
| except ValueError: |
| raise WicError("Error getting partition information from %s" % (self.parted)) |
| lsector_size, psector_size, self._ptable_format = splitted[idx + 1].split(":")[3:6] |
| self._lsector_size = int(lsector_size) |
| self._psector_size = int(psector_size) |
| for line in splitted[idx + 2:]: |
| pnum, start, end, size, fstype = line.split(':')[:5] |
| partition = parttype(int(pnum), int(start[:-1]), int(end[:-1]), |
| int(size[:-1]), fstype) |
| self._partitions[pnum] = partition |
| |
| return self._partitions |
| |
| def __getattr__(self, name): |
| """Get path to the executable in a lazy way.""" |
| if name in ("mdir", "mcopy", "mdel", "mdeltree", "sfdisk", "e2fsck", |
| "resize2fs", "mkswap", "mkdosfs", "debugfs"): |
| aname = "_%s" % name |
| if aname not in self.__dict__: |
| setattr(self, aname, find_executable(name, self.paths)) |
| if aname not in self.__dict__ or self.__dict__[aname] is None: |
| raise WicError("Can't find executable '{}'".format(name)) |
| return self.__dict__[aname] |
| return self.__dict__[name] |
| |
| def _get_part_image(self, pnum): |
| if pnum not in self.partitions: |
| raise WicError("Partition %s is not in the image") |
| part = self.partitions[pnum] |
| # check if fstype is supported |
| for fstype in self.fstypes: |
| if part.fstype.startswith(fstype): |
| break |
| else: |
| raise WicError("Not supported fstype: {}".format(part.fstype)) |
| if pnum not in self._partimages: |
| tmpf = tempfile.NamedTemporaryFile(prefix="wic-part") |
| dst_fname = tmpf.name |
| tmpf.close() |
| sparse_copy(self.imagepath, dst_fname, skip=part.start, length=part.size) |
| self._partimages[pnum] = dst_fname |
| |
| return self._partimages[pnum] |
| |
| def _put_part_image(self, pnum): |
| """Put partition image into partitioned image.""" |
| sparse_copy(self._partimages[pnum], self.imagepath, |
| seek=self.partitions[pnum].start) |
| |
| def dir(self, pnum, path): |
| if self.partitions[pnum].fstype.startswith('ext'): |
| return exec_cmd("{} {} -R 'ls -l {}'".format(self.debugfs, |
| self._get_part_image(pnum), |
| path), as_shell=True) |
| else: # fat |
| return exec_cmd("{} -i {} ::{}".format(self.mdir, |
| self._get_part_image(pnum), |
| path)) |
| |
| def copy(self, src, pnum, path): |
| """Copy partition image into wic image.""" |
| if self.partitions[pnum].fstype.startswith('ext'): |
| cmd = "printf 'cd {}\nwrite {} {}\n' | {} -w {}".\ |
| format(path, src, os.path.basename(src), |
| self.debugfs, self._get_part_image(pnum)) |
| else: # fat |
| cmd = "{} -i {} -snop {} ::{}".format(self.mcopy, |
| self._get_part_image(pnum), |
| src, path) |
| exec_cmd(cmd, as_shell=True) |
| self._put_part_image(pnum) |
| |
| def remove(self, pnum, path): |
| """Remove files/dirs from the partition.""" |
| partimg = self._get_part_image(pnum) |
| if self.partitions[pnum].fstype.startswith('ext'): |
| cmd = "{} {} -wR 'rm {}'".format(self.debugfs, |
| self._get_part_image(pnum), |
| path) |
| out = exec_cmd(cmd , as_shell=True) |
| for line in out.splitlines(): |
| if line.startswith("rm:"): |
| if "file is a directory" in line: |
| # Try rmdir to see if this is an empty directory. This won't delete |
| # any non empty directory so let user know about any error that this might |
| # generate. |
| print(exec_cmd("{} {} -wR 'rmdir {}'".format(self.debugfs, |
| self._get_part_image(pnum), |
| path), as_shell=True)) |
| else: |
| raise WicError("Could not complete operation: wic %s" % str(line)) |
| else: # fat |
| cmd = "{} -i {} ::{}".format(self.mdel, partimg, path) |
| try: |
| exec_cmd(cmd) |
| except WicError as err: |
| if "not found" in str(err) or "non empty" in str(err): |
| # mdel outputs 'File ... not found' or 'directory .. non empty" |
| # try to use mdeltree as path could be a directory |
| cmd = "{} -i {} ::{}".format(self.mdeltree, |
| partimg, path) |
| exec_cmd(cmd) |
| else: |
| raise err |
| self._put_part_image(pnum) |
| |
| def write(self, target, expand): |
| """Write disk image to the media or file.""" |
| def write_sfdisk_script(outf, parts): |
| for key, val in parts['partitiontable'].items(): |
| if key in ("partitions", "device", "firstlba", "lastlba"): |
| continue |
| if key == "id": |
| key = "label-id" |
| outf.write("{}: {}\n".format(key, val)) |
| outf.write("\n") |
| for part in parts['partitiontable']['partitions']: |
| line = '' |
| for name in ('attrs', 'name', 'size', 'type', 'uuid'): |
| if name == 'size' and part['type'] == 'f': |
| # don't write size for extended partition |
| continue |
| val = part.get(name) |
| if val: |
| line += '{}={}, '.format(name, val) |
| if line: |
| line = line[:-2] # strip ', ' |
| if part.get('bootable'): |
| line += ' ,bootable' |
| outf.write("{}\n".format(line)) |
| outf.flush() |
| |
| def read_ptable(path): |
| out = exec_cmd("{} -dJ {}".format(self.sfdisk, path)) |
| return json.loads(out) |
| |
| def write_ptable(parts, target): |
| with tempfile.NamedTemporaryFile(prefix="wic-sfdisk-", mode='w') as outf: |
| write_sfdisk_script(outf, parts) |
| cmd = "{} --no-reread {} < {} ".format(self.sfdisk, target, outf.name) |
| exec_cmd(cmd, as_shell=True) |
| |
| if expand is None: |
| sparse_copy(self.imagepath, target) |
| else: |
| # copy first sectors that may contain bootloader |
| sparse_copy(self.imagepath, target, length=2048 * self._lsector_size) |
| |
| # copy source partition table to the target |
| parts = read_ptable(self.imagepath) |
| write_ptable(parts, target) |
| |
| # get size of unpartitioned space |
| free = None |
| for line in exec_cmd("{} -F {}".format(self.sfdisk, target)).splitlines(): |
| if line.startswith("Unpartitioned space ") and line.endswith("sectors"): |
| free = int(line.split()[-2]) |
| # Align free space to a 2048 sector boundary. YOCTO #12840. |
| free = free - (free % 2048) |
| if free is None: |
| raise WicError("Can't get size of unpartitioned space") |
| |
| # calculate expanded partitions sizes |
| sizes = {} |
| num_auto_resize = 0 |
| for num, part in enumerate(parts['partitiontable']['partitions'], 1): |
| if num in expand: |
| if expand[num] != 0: # don't resize partition if size is set to 0 |
| sectors = expand[num] // self._lsector_size |
| free -= sectors - part['size'] |
| part['size'] = sectors |
| sizes[num] = sectors |
| elif part['type'] != 'f': |
| sizes[num] = -1 |
| num_auto_resize += 1 |
| |
| for num, part in enumerate(parts['partitiontable']['partitions'], 1): |
| if sizes.get(num) == -1: |
| part['size'] += free // num_auto_resize |
| |
| # write resized partition table to the target |
| write_ptable(parts, target) |
| |
| # read resized partition table |
| parts = read_ptable(target) |
| |
| # copy partitions content |
| for num, part in enumerate(parts['partitiontable']['partitions'], 1): |
| pnum = str(num) |
| fstype = self.partitions[pnum].fstype |
| |
| # copy unchanged partition |
| if part['size'] == self.partitions[pnum].size // self._lsector_size: |
| logger.info("copying unchanged partition {}".format(pnum)) |
| sparse_copy(self._get_part_image(pnum), target, seek=part['start'] * self._lsector_size) |
| continue |
| |
| # resize or re-create partitions |
| if fstype.startswith('ext') or fstype.startswith('fat') or \ |
| fstype.startswith('linux-swap'): |
| |
| partfname = None |
| with tempfile.NamedTemporaryFile(prefix="wic-part{}-".format(pnum)) as partf: |
| partfname = partf.name |
| |
| if fstype.startswith('ext'): |
| logger.info("resizing ext partition {}".format(pnum)) |
| partimg = self._get_part_image(pnum) |
| sparse_copy(partimg, partfname) |
| exec_cmd("{} -pf {}".format(self.e2fsck, partfname)) |
| exec_cmd("{} {} {}s".format(\ |
| self.resize2fs, partfname, part['size'])) |
| elif fstype.startswith('fat'): |
| logger.info("copying content of the fat partition {}".format(pnum)) |
| with tempfile.TemporaryDirectory(prefix='wic-fatdir-') as tmpdir: |
| # copy content to the temporary directory |
| cmd = "{} -snompi {} :: {}".format(self.mcopy, |
| self._get_part_image(pnum), |
| tmpdir) |
| exec_cmd(cmd) |
| # create new msdos partition |
| label = part.get("name") |
| label_str = "-n {}".format(label) if label else '' |
| |
| cmd = "{} {} -C {} {}".format(self.mkdosfs, label_str, partfname, |
| part['size']) |
| exec_cmd(cmd) |
| # copy content from the temporary directory to the new partition |
| cmd = "{} -snompi {} {}/* ::".format(self.mcopy, partfname, tmpdir) |
| exec_cmd(cmd, as_shell=True) |
| elif fstype.startswith('linux-swap'): |
| logger.info("creating swap partition {}".format(pnum)) |
| label = part.get("name") |
| label_str = "-L {}".format(label) if label else '' |
| uuid = part.get("uuid") |
| uuid_str = "-U {}".format(uuid) if uuid else '' |
| with open(partfname, 'w') as sparse: |
| os.ftruncate(sparse.fileno(), part['size'] * self._lsector_size) |
| exec_cmd("{} {} {} {}".format(self.mkswap, label_str, uuid_str, partfname)) |
| sparse_copy(partfname, target, seek=part['start'] * self._lsector_size) |
| os.unlink(partfname) |
| elif part['type'] != 'f': |
| logger.warning("skipping partition {}: unsupported fstype {}".format(pnum, fstype)) |
| |
| def wic_ls(args, native_sysroot): |
| """List contents of partitioned image or vfat partition.""" |
| disk = Disk(args.path.image, native_sysroot) |
| if not args.path.part: |
| if disk.partitions: |
| print('Num Start End Size Fstype') |
| for part in disk.partitions.values(): |
| print("{:2d} {:12d} {:12d} {:12d} {}".format(\ |
| part.pnum, part.start, part.end, |
| part.size, part.fstype)) |
| else: |
| path = args.path.path or '/' |
| print(disk.dir(args.path.part, path)) |
| |
| def wic_cp(args, native_sysroot): |
| """ |
| Copy local file or directory to the vfat partition of |
| partitioned image. |
| """ |
| disk = Disk(args.dest.image, native_sysroot) |
| disk.copy(args.src, args.dest.part, args.dest.path) |
| |
| def wic_rm(args, native_sysroot): |
| """ |
| Remove files or directories from the vfat partition of |
| partitioned image. |
| """ |
| disk = Disk(args.path.image, native_sysroot) |
| disk.remove(args.path.part, args.path.path) |
| |
| def wic_write(args, native_sysroot): |
| """ |
| Write image to a target device. |
| """ |
| disk = Disk(args.image, native_sysroot, ('fat', 'ext', 'swap')) |
| disk.write(args.target, args.expand) |
| |
| def find_canned(scripts_path, file_name): |
| """ |
| Find a file either by its path or by name in the canned files dir. |
| |
| Return None if not found |
| """ |
| if os.path.exists(file_name): |
| return file_name |
| |
| layers_canned_wks_dir = build_canned_image_list(scripts_path) |
| for canned_wks_dir in layers_canned_wks_dir: |
| for root, dirs, files in os.walk(canned_wks_dir): |
| for fname in files: |
| if fname == file_name: |
| fullpath = os.path.join(canned_wks_dir, fname) |
| return fullpath |
| |
| def get_custom_config(boot_file): |
| """ |
| Get the custom configuration to be used for the bootloader. |
| |
| Return None if the file can't be found. |
| """ |
| # Get the scripts path of poky |
| scripts_path = os.path.abspath("%s/../.." % os.path.dirname(__file__)) |
| |
| cfg_file = find_canned(scripts_path, boot_file) |
| if cfg_file: |
| with open(cfg_file, "r") as f: |
| config = f.read() |
| return config |