7643853211

Committed 24 Jan 2024 05:20PM UTC coverage: 43.472%. Remained the same

Build # 7643853211

Build Type

Pull #1368

github

Committed by

web-flow

Commit Message

Merge 5a98881af into 071718649

Pull Request Pull Request #1368: s390: Escape volid before using it

Run Details

595 of 1582 branches covered (0.0%)

Branch coverage included in aggregate %.

1616 of 3504 relevant lines covered (46.12%)

0.46 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

39.43

/src/pylorax/treebuilder.py

# treebuilder.py - handle arch-specific tree building stuff using templates
#
# Copyright (C) 2011-2015 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
# Author(s):  Will Woods <wwoods@redhat.com>

import logging
logger = logging.getLogger("pylorax.treebuilder")

import os, re
from os.path import basename
from shutil import copytree, copy2
from subprocess import CalledProcessError
from pathlib import Path
import itertools
import libdnf5 as dnf5
from libdnf5.common import QueryCmp_EQ as EQ

from pylorax.sysutils import joinpaths, remove
from pylorax.base import DataHolder
from pylorax.ltmpl import LoraxTemplateRunner
import pylorax.imgutils as imgutils
from pylorax.imgutils import DracutChroot
from pylorax.executils import runcmd, runcmd_output, execWithCapture

templatemap = {
    'x86_64':  'x86.tmpl',
    'ppc64le': 'ppc64le.tmpl',
    's390x':   's390.tmpl',
    'aarch64': 'aarch64.tmpl',
}

def generate_module_info(moddir, outfile=None):
    def module_desc(mod):
        output = runcmd_output(["modinfo", "-F", "description", mod])
        return output.strip()
    def read_module_set(name):
        return set(l.strip() for l in open(joinpaths(moddir,name)) if ".ko" in l)
    modsets = {'scsi':read_module_set("modules.block"),
               'eth':read_module_set("modules.networking")}

    modinfo = list()
    for root, _dirs, files in os.walk(moddir):
        for modtype, modset in modsets.items():
            for mod in modset.intersection(files):  # modules in this dir
                (name, _ext) = os.path.splitext(mod) # foo.ko -> (foo, .ko)
                desc = module_desc(joinpaths(root,mod)) or "%s driver" % name
                modinfo.append(dict(name=name, type=modtype, desc=desc))

    out = open(outfile or joinpaths(moddir,"module-info"), "w")
    out.write("Version 0\n")
    for mod in sorted(modinfo, key=lambda m: m.get('name')):
        out.write('{name}\n\t{type}\n\t"{desc:.65}"\n'.format(**mod))

class RuntimeBuilder(object):
    '''Builds the anaconda runtime image.

    NOTE: dbo is optional, but if it is not included root must be set.
    '''
    def __init__(self, product, arch, dbo=None, templatedir=None,
                 installpkgs=None, excludepkgs=None,
                 add_templates=None,
                 add_template_vars=None,
                 skip_branding=False,
                 root=None):
        self.dbo = dbo
        if dbo:
            root = dbo.get_config().installroot

        if not root:
            raise RuntimeError("No root directory passed to RuntimeBuilder")

        self._runner = LoraxTemplateRunner(inroot=root, outroot=root,
                                           dbo=dbo, templatedir=templatedir,
                                           basearch=arch.basearch)
        self.add_templates = add_templates or []
        self.add_template_vars = add_template_vars or {}
        self._installpkgs = installpkgs or []
        self._excludepkgs = excludepkgs or []

        # use a copy of product so we can modify it locally
        product = product.copy()
        product.name = product.name.lower()
        self._branding = self.get_branding(skip_branding, product)
        self.vars = DataHolder(arch=arch, product=product, dbo=dbo, root=root,
                               basearch=arch.basearch, libdir=arch.libdir,
                               branding=self._branding)
        self._runner.defaults = self.vars

    def get_branding(self, skip, product):
        """Select the branding from the available 'system-release' packages
        The *best* way to control this is to have a single package in the repo provide 'system-release'
        When there are more than 1 package it will:
        - Make a list of the available packages
        - If variant is set look for a package ending with lower(variant) and use that
        - If there are one or more non-generic packages, use the first one after sorting

        Returns the package names of the system-release and release logos package
        """
        if skip:
            return DataHolder(release=None, logos=None)

        release = None
        query = dnf5.rpm.PackageQuery(self.dbo)
        query.filter_provides(["system-release"], EQ)
        pkgs = sorted([p for p in list(query)
                               if not p.get_name().startswith("generic")])
        if not pkgs:
            logger.error("No system-release packages found, could not get the release")
            return DataHolder(release=None, logos=None)

        logger.debug("system-release packages: %s", ",".join(p.get_name() for p in pkgs))
        if product.variant:
            variant = [p.get_name() for p in pkgs if p.get_name().endswith("-"+product.variant.lower())]
            if variant:
                release = variant[0]
        if not release:
            release = pkgs[0].get_name()

        # release
        logger.info('got release: %s', release)

        # logos uses the basename from release (fedora, redhat, centos, ...)
        logos, _suffix = release.split('-', 1)
        return DataHolder(release=release, logos=logos+"-logos")

    def install(self):
        '''Install packages and do initial setup with runtime-install.tmpl'''
        if self._branding.release:
            self._runner.installpkg(self._branding.release)
        if self._branding.logos:
            self._runner.installpkg(self._branding.logos)

        if len(self._installpkgs) > 0:
            self._runner.installpkg(*self._installpkgs)
        if len(self._excludepkgs) > 0:
            self._runner.removepkg(*self._excludepkgs)

        self._runner.run("runtime-install.tmpl")

        for tmpl in self.add_templates:
            self._runner.run(tmpl, **self.add_template_vars)

    def writepkglists(self, pkglistdir):
        '''debugging data: write out lists of package contents'''
        self._runner._writepkglists(pkglistdir)

    def writepkgsizes(self, pkgsizefile):
        '''debugging data: write a big list of pkg sizes'''
        self._runner._writepkgsizes(pkgsizefile)

    def postinstall(self):
        '''Do some post-install setup work with runtime-postinstall.tmpl'''
        # copy configdir into runtime root beforehand
        configdir = joinpaths(self._runner.templatedir,"config_files")
        configdir_path = "tmp/config_files"
        fullpath = joinpaths(self.vars.root, configdir_path)
        if os.path.exists(fullpath):
            remove(fullpath)
        copytree(configdir, fullpath)
        self._runner.run("runtime-postinstall.tmpl", configdir=configdir_path)

    def cleanup(self):
        '''Remove unneeded packages and files with runtime-cleanup.tmpl'''
        self._runner.run("runtime-cleanup.tmpl")

    def verify(self):
        '''Ensure that contents of the installroot can run'''
        status = True

        ELF_MAGIC = b'\x7fELF'

        # Iterate over all files in /usr/bin and /usr/sbin
        # For ELF files, gather them into a list and we'll check them all at
        # the end. For files with a #!, check them as we go
        elf_files = []
        usr_bin = Path(self.vars.root + '/usr/bin')
        usr_sbin = Path(self.vars.root + '/usr/sbin')
        for path in (str(x) for x in itertools.chain(usr_bin.iterdir(), usr_sbin.iterdir()) \
                     if x.is_file()):
            with open(path, "rb") as f:
                magic = f.read(4)
                if magic == ELF_MAGIC:
                    # Save the path, minus the chroot prefix
                    elf_files.append(path[len(self.vars.root):])
                elif magic[:2] == b'#!':
                    # Reopen the file as text and read the first line.
                    # Open as latin-1 so that stray 8-bit characters don't make
                    # things blow up. We only really care about ASCII parts.
                    with open(path, "rt", encoding="latin-1") as f_text:
                        # Remove the #!, split on space, and take the first part
                        shabang = f_text.readline()[2:].split()[0]

                    # Does the path exist?
                    if not os.path.exists(self.vars.root + shabang):
                        logger.error('%s, needed by %s, does not exist', shabang, path)
                        status = False

        # Now, run ldd on all the ELF files
        # Just run ldd once on everything so it isn't logged a million times.
        # At least one thing in the list isn't going to be a dynamic executable,
        # so use execWithCapture to ignore the exit code.
        filename = ''
        for line in execWithCapture('ldd', elf_files, root=self.vars.root,
                log_output=False, filter_stderr=True).split('\n'):
            if line and not line[0].isspace():
                # New filename header, strip the : at the end and save
                filename = line[:-1]
            elif 'not found' in line:
                logger.error('%s, needed by %s, not found', line.split()[0], filename)
                status = False

        return status

    def generate_module_data(self):
        root = self.vars.root
        moddir = joinpaths(root, "lib/modules/")
        for kernel in findkernels(root=root):
            ksyms = joinpaths(root, "boot/System.map-%s" % kernel.version)
            logger.info("doing depmod and module-info for %s", kernel.version)
            runcmd(["depmod", "-a", "-F", ksyms, "-b", root, kernel.version])
            generate_module_info(moddir+kernel.version, outfile=moddir+"module-info")

    def create_squashfs_runtime(self, outfile="/var/tmp/squashfs.img", compression="xz", compressargs=None, size=2):
        """Create a plain squashfs runtime"""
        compressargs = compressargs or []
        os.makedirs(os.path.dirname(outfile))

        # squash the rootfs
        return imgutils.mksquashfs(self.vars.root, outfile, compression, compressargs)

    def create_ext4_runtime(self, outfile="/var/tmp/squashfs.img", compression="xz", compressargs=None, size=2):
        """Create a squashfs compressed ext4 runtime"""
        # make live rootfs image - must be named "LiveOS/rootfs.img" for dracut
        compressargs = compressargs or []
        workdir = joinpaths(os.path.dirname(outfile), "runtime-workdir")
        os.makedirs(joinpaths(workdir, "LiveOS"))

        # Catch problems with the rootfs being too small and clearly log them
        try:
            imgutils.mkrootfsimg(self.vars.root, joinpaths(workdir, "LiveOS/rootfs.img"),
                                 "Anaconda", size=size)
        except CalledProcessError as e:
            if e.stdout and "No space left on device" in e.stdout:
                logger.error("The rootfs ran out of space with size=%d", size)
            raise

        # squash the live rootfs and clean up workdir
        rc = imgutils.mksquashfs(workdir, outfile, compression, compressargs)
        remove(workdir)
        return rc

    def finished(self):
        pass

class TreeBuilder(object):
    '''Builds the arch-specific boot images.
    inroot should be the installtree root (the newly-built runtime dir)'''
    def __init__(self, product, arch, inroot, outroot, runtime, isolabel, domacboot=True, doupgrade=True,
                 templatedir=None, add_templates=None, add_template_vars=None, workdir=None, extra_boot_args=""):

        # NOTE: if you pass an arg named "runtime" to a mako template it'll
        # clobber some mako internal variables - hence "runtime_img".
        self.vars = DataHolder(arch=arch, product=product, runtime_img=runtime,
                               runtime_base=basename(runtime),
                               inroot=inroot, outroot=outroot,
                               basearch=arch.basearch, libdir=arch.libdir,
                               isolabel=isolabel, udev=udev_escape, domacboot=domacboot, doupgrade=doupgrade,
                               workdir=workdir, lower=string_lower,
                               extra_boot_args=extra_boot_args)
        self._runner = LoraxTemplateRunner(inroot, outroot, templatedir=templatedir,
                                           basearch=arch.basearch)
        self._runner.defaults = self.vars
        self.add_templates = add_templates or []
        self.add_template_vars = add_template_vars or {}
        self.templatedir = templatedir
        self.treeinfo_data = None

    @property
    def kernels(self):
        return findkernels(root=self.vars.inroot)

    def rebuild_initrds(self, add_args=None, backup="", prefix=""):
        '''Rebuild all the initrds in the tree. If backup is specified, each
        initrd will be renamed with backup as a suffix before rebuilding.
        If backup is empty, the existing initrd files will be overwritten.
        If suffix is specified, the existing initrd is untouched and a new
        image is built with the filename "${prefix}-${kernel.version}.img"

        If the initrd doesn't exist its name will be created based on the
        name of the kernel.
        '''
        add_args = add_args or []
        args = ["--nomdadmconf", "--nolvmconf"] + add_args
        if not backup:
            args.append("--force")

        if not self.kernels:
            raise RuntimeError("No kernels found, cannot rebuild_initrds")

        with DracutChroot(self.vars.inroot) as dracut:
            for kernel in self.kernels:
                if prefix:
                    idir = os.path.dirname(kernel.path)
                    outfile = joinpaths(idir, prefix+'-'+kernel.version+'.img')
                elif hasattr(kernel, "initrd"):
                    # If there is an existing initrd, use that
                    outfile = kernel.initrd.path
                else:
                    # Construct an initrd from the kernel name
                    outfile = kernel.path.replace("vmlinuz-", "initrd-") + ".img"
                logger.info("rebuilding %s", outfile)

                if backup:
                    initrd = joinpaths(self.vars.inroot, outfile)
                    if os.path.exists(initrd):
                        os.rename(initrd, initrd + backup)
                dracut.Run(args + [outfile, kernel.version])

    def build(self):
        templatefile = templatemap[self.vars.arch.basearch]
        for tmpl in self.add_templates:
            self._runner.run(tmpl, **self.add_template_vars)
        self._runner.run(templatefile, kernels=self.kernels)
        self.treeinfo_data = self._runner.results.treeinfo
        self.implantisomd5()

    def implantisomd5(self):
        for _section, data in self.treeinfo_data.items():
            if 'boot.iso' in data:
                iso = joinpaths(self.vars.outroot, data['boot.iso'])
                runcmd(["implantisomd5", iso])

    @property
    def dracut_hooks_path(self):
        """ Return the path to the lorax dracut hooks scripts

            Use the configured share dir if it is setup,
            otherwise default to /usr/share/lorax/dracut_hooks
        """
        if self.templatedir:
            return joinpaths(self.templatedir, "dracut_hooks")
        else:
            return "/usr/share/lorax/dracut_hooks"

    def copy_dracut_hooks(self, hooks):
        """ Copy the hook scripts in hooks into the installroot's /tmp/
        and return a list of commands to pass to dracut when creating the
        initramfs

        hooks is a list of tuples with the name of the hook script and the
        target dracut hook directory
        (eg. [("99anaconda-copy-ks.sh", "/lib/dracut/hooks/pre-pivot")])
        """
        dracut_commands = []
        for hook_script, dracut_path in hooks:
            src = joinpaths(self.dracut_hooks_path, hook_script)
            if not os.path.exists(src):
                logger.error("Missing lorax dracut hook script %s", (src))
                continue
            dst = joinpaths(self.vars.inroot, "/tmp/", hook_script)
            copy2(src, dst)
            dracut_commands += ["--include", joinpaths("/tmp/", hook_script),
                                dracut_path]
        return dracut_commands

#### TreeBuilder helper functions

def findkernels(root="/", kdir="boot"):
    # To find possible flavors, awk '/BuildKernel/ { print $4 }' kernel.spec
    flavors = ('debug', 'PAE', 'PAEdebug', 'smp', 'xen', 'lpae')
    kre = re.compile(r"vmlinuz-(?P<version>.+?\.(?P<arch>[a-z0-9_]+)"
                     r"(.(?P<flavor>{0}))?)$".format("|".join(flavors)))
    kernels = []
    bootfiles = os.listdir(joinpaths(root, kdir))
    for f in bootfiles:
        match = kre.match(f)
        if match:
            kernel = DataHolder(path=joinpaths(kdir, f))
            kernel.update(match.groupdict()) # sets version, arch, flavor
            kernels.append(kernel)

    # look for associated initrd/initramfs/etc.
    for kernel in kernels:
        for f in bootfiles:
            if f.endswith('-'+kernel.version+'.img'):
                imgtype, _rest = f.split('-',1)
                # special backwards-compat case
                if imgtype == 'initramfs':
                    imgtype = 'initrd'
                kernel[imgtype] = DataHolder(path=joinpaths(kdir, f))

    logger.debug("kernels=%s", kernels)
    return kernels

# udev whitelist: 'a-zA-Z0-9#+.:=@_-' (see is_whitelisted in libudev-util.c)
udev_blacklist=' !"$%&\'()*,/;<>?[\\]^`{|}~' # ASCII printable, minus whitelist
udev_blacklist += ''.join(chr(i) for i in range(32)) # ASCII non-printable
def udev_escape(label):
    out = ''
    for ch in label:
        out += ch if ch not in udev_blacklist else '\\x%02x' % ord(ch)
    return out

def string_lower(string):
    """ Return a lowercase string.

    :param string: String to lowercase

    This is used as a filter in the templates.
    """
    return string.lower()

1	# treebuilder.py - handle arch-specific tree building stuff using templates
2	#
3	# Copyright (C) 2011-2015 Red Hat, Inc.
4	#
5	# This program is free software; you can redistribute it and/or modify
6	# it under the terms of the GNU General Public License as published by
7	# the Free Software Foundation; either version 2 of the License, or
8	# (at your option) any later version.
9	#
10	# This program is distributed in the hope that it will be useful,
11	# but WITHOUT ANY WARRANTY; without even the implied warranty of
12	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13	# GNU General Public License for more details.
14	#
15	# You should have received a copy of the GNU General Public License
16	# along with this program. If not, see <http://www.gnu.org/licenses/>.
17	#
18	# Author(s): Will Woods <wwoods@redhat.com>
19
20	import logging	1✔
21	logger = logging.getLogger("pylorax.treebuilder")	1✔
22
23	import os, re	1✔
24	from os.path import basename	1✔
25	from shutil import copytree, copy2	1✔
26	from subprocess import CalledProcessError	1✔
27	from pathlib import Path	1✔
28	import itertools	1✔
29	import libdnf5 as dnf5	1✔
30	from libdnf5.common import QueryCmp_EQ as EQ	1✔
31
32	from pylorax.sysutils import joinpaths, remove	1✔
33	from pylorax.base import DataHolder	1✔
34	from pylorax.ltmpl import LoraxTemplateRunner	1✔
35	import pylorax.imgutils as imgutils	1✔
36	from pylorax.imgutils import DracutChroot	1✔
37	from pylorax.executils import runcmd, runcmd_output, execWithCapture	1✔
38
39	templatemap = {	1✔
40	'x86_64': 'x86.tmpl',
41	'ppc64le': 'ppc64le.tmpl',
42	's390x': 's390.tmpl',
43	'aarch64': 'aarch64.tmpl',
44	}
45
46	def generate_module_info(moddir, outfile=None):	1✔
47	def module_desc(mod):	×
48	output = runcmd_output(["modinfo", "-F", "description", mod])	×
49	return output.strip()	×
50	def read_module_set(name):	×
51	return set(l.strip() for l in open(joinpaths(moddir,name)) if ".ko" in l)	×
52	modsets = {'scsi':read_module_set("modules.block"),	×
53	'eth':read_module_set("modules.networking")}
54
55	modinfo = list()	×
56	for root, _dirs, files in os.walk(moddir):	×
57	for modtype, modset in modsets.items():	×
58	for mod in modset.intersection(files): # modules in this dir	×
59	(name, _ext) = os.path.splitext(mod) # foo.ko -> (foo, .ko)	×
60	desc = module_desc(joinpaths(root,mod)) or "%s driver" % name	×
61	modinfo.append(dict(name=name, type=modtype, desc=desc))	×
62
63	out = open(outfile or joinpaths(moddir,"module-info"), "w")	×
64	out.write("Version 0\n")	×
65	for mod in sorted(modinfo, key=lambda m: m.get('name')):	×
66	out.write('{name}\n\t{type}\n\t"{desc:.65}"\n'.format(**mod))	×
67
68	class RuntimeBuilder(object):	1✔
69	'''Builds the anaconda runtime image.
70
71	NOTE: dbo is optional, but if it is not included root must be set.
72	'''
73	def __init__(self, product, arch, dbo=None, templatedir=None,	1✔
74	installpkgs=None, excludepkgs=None,
75	add_templates=None,
76	add_template_vars=None,
77	skip_branding=False,
78	root=None):
79	self.dbo = dbo	1✔
80	if dbo:	1✔
81	root = dbo.get_config().installroot	1✔
82
83	if not root:	1!
84	raise RuntimeError("No root directory passed to RuntimeBuilder")	×
85
86	self._runner = LoraxTemplateRunner(inroot=root, outroot=root,	1✔
87	dbo=dbo, templatedir=templatedir,
88	basearch=arch.basearch)
89	self.add_templates = add_templates or []	1✔
90	self.add_template_vars = add_template_vars or {}	1✔
91	self._installpkgs = installpkgs or []	1✔
92	self._excludepkgs = excludepkgs or []	1✔
93
94	# use a copy of product so we can modify it locally
95	product = product.copy()	1✔
96	product.name = product.name.lower()	1✔
97	self._branding = self.get_branding(skip_branding, product)	1✔
98	self.vars = DataHolder(arch=arch, product=product, dbo=dbo, root=root,	1✔
99	basearch=arch.basearch, libdir=arch.libdir,
100	branding=self._branding)
101	self._runner.defaults = self.vars	1✔
102
103	def get_branding(self, skip, product):	1✔
104	"""Select the branding from the available 'system-release' packages
105	The best way to control this is to have a single package in the repo provide 'system-release'
106	When there are more than 1 package it will:
107	- Make a list of the available packages
108	- If variant is set look for a package ending with lower(variant) and use that
109	- If there are one or more non-generic packages, use the first one after sorting
110
111	Returns the package names of the system-release and release logos package
112	"""
113	if skip:	1✔
114	return DataHolder(release=None, logos=None)	1✔
115
116	release = None	1✔
117	query = dnf5.rpm.PackageQuery(self.dbo)	1✔
118	query.filter_provides(["system-release"], EQ)	1✔
119	pkgs = sorted([p for p in list(query)	1!
120	if not p.get_name().startswith("generic")])
121	if not pkgs:	1✔
122	logger.error("No system-release packages found, could not get the release")	1✔
123	return DataHolder(release=None, logos=None)	1✔
124
125	logger.debug("system-release packages: %s", ",".join(p.get_name() for p in pkgs))	1✔
126	if product.variant:	1✔
127	variant = [p.get_name() for p in pkgs if p.get_name().endswith("-"+product.variant.lower())]	1!
128	if variant:	1✔
129	release = variant[0]	1✔
130	if not release:	1✔
131	release = pkgs[0].get_name()	1✔
132
133	# release
134	logger.info('got release: %s', release)	1✔
135
136	# logos uses the basename from release (fedora, redhat, centos, ...)
137	logos, _suffix = release.split('-', 1)	1✔
138	return DataHolder(release=release, logos=logos+"-logos")	1✔
139
140	def install(self):	1✔
141	'''Install packages and do initial setup with runtime-install.tmpl'''
142	if self._branding.release:	×
143	self._runner.installpkg(self._branding.release)	×
144	if self._branding.logos:	×
145	self._runner.installpkg(self._branding.logos)	×
146
147	if len(self._installpkgs) > 0:	×
148	self._runner.installpkg(*self._installpkgs)	×
149	if len(self._excludepkgs) > 0:	×
150	self._runner.removepkg(*self._excludepkgs)	×
151
152	self._runner.run("runtime-install.tmpl")	×
153
154	for tmpl in self.add_templates:	×
155	self._runner.run(tmpl, **self.add_template_vars)	×
156
157	def writepkglists(self, pkglistdir):	1✔
158	'''debugging data: write out lists of package contents'''
159	self._runner._writepkglists(pkglistdir)	×
160
161	def writepkgsizes(self, pkgsizefile):	1✔
162	'''debugging data: write a big list of pkg sizes'''
163	self._runner._writepkgsizes(pkgsizefile)	×
164
165	def postinstall(self):	1✔
166	'''Do some post-install setup work with runtime-postinstall.tmpl'''
167	# copy configdir into runtime root beforehand
168	configdir = joinpaths(self._runner.templatedir,"config_files")	×
169	configdir_path = "tmp/config_files"	×
170	fullpath = joinpaths(self.vars.root, configdir_path)	×
171	if os.path.exists(fullpath):	×
172	remove(fullpath)	×
173	copytree(configdir, fullpath)	×
174	self._runner.run("runtime-postinstall.tmpl", configdir=configdir_path)	×
175
176	def cleanup(self):	1✔
177	'''Remove unneeded packages and files with runtime-cleanup.tmpl'''
178	self._runner.run("runtime-cleanup.tmpl")	×
179
180	def verify(self):	1✔
181	'''Ensure that contents of the installroot can run'''
182	status = True	×
183
184	ELF_MAGIC = b'\x7fELF'	×
185
186	# Iterate over all files in /usr/bin and /usr/sbin
187	# For ELF files, gather them into a list and we'll check them all at
188	# the end. For files with a #!, check them as we go
189	elf_files = []	×
190	usr_bin = Path(self.vars.root + '/usr/bin')	×
191	usr_sbin = Path(self.vars.root + '/usr/sbin')	×
192	for path in (str(x) for x in itertools.chain(usr_bin.iterdir(), usr_sbin.iterdir()) \	×
193	if x.is_file()):
194	with open(path, "rb") as f:	×
195	magic = f.read(4)	×
196	if magic == ELF_MAGIC:	×
197	# Save the path, minus the chroot prefix
198	elf_files.append(path[len(self.vars.root):])	×
199	elif magic[:2] == b'#!':	×
200	# Reopen the file as text and read the first line.
201	# Open as latin-1 so that stray 8-bit characters don't make
202	# things blow up. We only really care about ASCII parts.
203	with open(path, "rt", encoding="latin-1") as f_text:	×
204	# Remove the #!, split on space, and take the first part
205	shabang = f_text.readline()[2:].split()[0]	×
206
207	# Does the path exist?
208	if not os.path.exists(self.vars.root + shabang):	×
209	logger.error('%s, needed by %s, does not exist', shabang, path)	×
210	status = False	×
211
212	# Now, run ldd on all the ELF files
213	# Just run ldd once on everything so it isn't logged a million times.
214	# At least one thing in the list isn't going to be a dynamic executable,
215	# so use execWithCapture to ignore the exit code.
216	filename = ''	×
217	for line in execWithCapture('ldd', elf_files, root=self.vars.root,	×
218	log_output=False, filter_stderr=True).split('\n'):
219	if line and not line[0].isspace():	×
220	# New filename header, strip the : at the end and save
221	filename = line[:-1]	×
222	elif 'not found' in line:	×
223	logger.error('%s, needed by %s, not found', line.split()[0], filename)	×
224	status = False	×
225
226	return status	×
227
228	def generate_module_data(self):	1✔
229	root = self.vars.root	×
230	moddir = joinpaths(root, "lib/modules/")	×
231	for kernel in findkernels(root=root):	×
232	ksyms = joinpaths(root, "boot/System.map-%s" % kernel.version)	×
233	logger.info("doing depmod and module-info for %s", kernel.version)	×
234	runcmd(["depmod", "-a", "-F", ksyms, "-b", root, kernel.version])	×
235	generate_module_info(moddir+kernel.version, outfile=moddir+"module-info")	×
236
237	def create_squashfs_runtime(self, outfile="/var/tmp/squashfs.img", compression="xz", compressargs=None, size=2):	1✔
238	"""Create a plain squashfs runtime"""
239	compressargs = compressargs or []	1✔
240	os.makedirs(os.path.dirname(outfile))	1✔
241
242	# squash the rootfs
243	return imgutils.mksquashfs(self.vars.root, outfile, compression, compressargs)	1✔
244
245	def create_ext4_runtime(self, outfile="/var/tmp/squashfs.img", compression="xz", compressargs=None, size=2):	1✔
246	"""Create a squashfs compressed ext4 runtime"""
247	# make live rootfs image - must be named "LiveOS/rootfs.img" for dracut
248	compressargs = compressargs or []	×
249	workdir = joinpaths(os.path.dirname(outfile), "runtime-workdir")	×
250	os.makedirs(joinpaths(workdir, "LiveOS"))	×
251
252	# Catch problems with the rootfs being too small and clearly log them
253	try:	×
254	imgutils.mkrootfsimg(self.vars.root, joinpaths(workdir, "LiveOS/rootfs.img"),	×
255	"Anaconda", size=size)
256	except CalledProcessError as e:	×
257	if e.stdout and "No space left on device" in e.stdout:	×
258	logger.error("The rootfs ran out of space with size=%d", size)	×
259	raise	×
260
261	# squash the live rootfs and clean up workdir
262	rc = imgutils.mksquashfs(workdir, outfile, compression, compressargs)	×
263	remove(workdir)	×
264	return rc	×
265
266	def finished(self):	1✔
267	pass	×
268
269	class TreeBuilder(object):	1✔
270	'''Builds the arch-specific boot images.
271	inroot should be the installtree root (the newly-built runtime dir)'''
272	def __init__(self, product, arch, inroot, outroot, runtime, isolabel, domacboot=True, doupgrade=True,	1✔
273	templatedir=None, add_templates=None, add_template_vars=None, workdir=None, extra_boot_args=""):
274
275	# NOTE: if you pass an arg named "runtime" to a mako template it'll
276	# clobber some mako internal variables - hence "runtime_img".
277	self.vars = DataHolder(arch=arch, product=product, runtime_img=runtime,	1✔
278	runtime_base=basename(runtime),
279	inroot=inroot, outroot=outroot,
280	basearch=arch.basearch, libdir=arch.libdir,
281	isolabel=isolabel, udev=udev_escape, domacboot=domacboot, doupgrade=doupgrade,
282	workdir=workdir, lower=string_lower,
283	extra_boot_args=extra_boot_args)
284	self._runner = LoraxTemplateRunner(inroot, outroot, templatedir=templatedir,	1✔
285	basearch=arch.basearch)
286	self._runner.defaults = self.vars	1✔
287	self.add_templates = add_templates or []	1✔
288	self.add_template_vars = add_template_vars or {}	1✔
289	self.templatedir = templatedir	1✔
290	self.treeinfo_data = None	1✔
291
292	@property	1✔
293	def kernels(self):	1✔
294	return findkernels(root=self.vars.inroot)	×
295
296	def rebuild_initrds(self, add_args=None, backup="", prefix=""):	1✔
297	'''Rebuild all the initrds in the tree. If backup is specified, each
298	initrd will be renamed with backup as a suffix before rebuilding.
299	If backup is empty, the existing initrd files will be overwritten.
300	If suffix is specified, the existing initrd is untouched and a new
301	image is built with the filename "${prefix}-${kernel.version}.img"
302
303	If the initrd doesn't exist its name will be created based on the
304	name of the kernel.
305	'''
306	add_args = add_args or []	×
307	args = ["--nomdadmconf", "--nolvmconf"] + add_args	×
308	if not backup:	×
309	args.append("--force")	×
310
311	if not self.kernels:	×
312	raise RuntimeError("No kernels found, cannot rebuild_initrds")	×
313
314	with DracutChroot(self.vars.inroot) as dracut:	×
315	for kernel in self.kernels:	×
316	if prefix:	×
317	idir = os.path.dirname(kernel.path)	×
318	outfile = joinpaths(idir, prefix+'-'+kernel.version+'.img')	×
319	elif hasattr(kernel, "initrd"):	×
320	# If there is an existing initrd, use that
321	outfile = kernel.initrd.path	×
322	else:
323	# Construct an initrd from the kernel name
324	outfile = kernel.path.replace("vmlinuz-", "initrd-") + ".img"	×
325	logger.info("rebuilding %s", outfile)	×
326
327	if backup:	×
328	initrd = joinpaths(self.vars.inroot, outfile)	×
329	if os.path.exists(initrd):	×
330	os.rename(initrd, initrd + backup)	×
331	dracut.Run(args + [outfile, kernel.version])	×
332
333	def build(self):	1✔
334	templatefile = templatemap[self.vars.arch.basearch]	×
335	for tmpl in self.add_templates:	×
336	self._runner.run(tmpl, **self.add_template_vars)	×
337	self._runner.run(templatefile, kernels=self.kernels)	×
338	self.treeinfo_data = self._runner.results.treeinfo	×
339	self.implantisomd5()	×
340
341	def implantisomd5(self):	1✔
342	for _section, data in self.treeinfo_data.items():	×
343	if 'boot.iso' in data:	×
344	iso = joinpaths(self.vars.outroot, data['boot.iso'])	×
345	runcmd(["implantisomd5", iso])	×
346
347	@property	1✔
348	def dracut_hooks_path(self):	1✔
349	""" Return the path to the lorax dracut hooks scripts
350
351	Use the configured share dir if it is setup,
352	otherwise default to /usr/share/lorax/dracut_hooks
353	"""
354	if self.templatedir:	×
355	return joinpaths(self.templatedir, "dracut_hooks")	×
356	else:
357	return "/usr/share/lorax/dracut_hooks"	×
358
359	def copy_dracut_hooks(self, hooks):	1✔
360	""" Copy the hook scripts in hooks into the installroot's /tmp/
361	and return a list of commands to pass to dracut when creating the
362	initramfs
363
364	hooks is a list of tuples with the name of the hook script and the
365	target dracut hook directory
366	(eg. [("99anaconda-copy-ks.sh", "/lib/dracut/hooks/pre-pivot")])
367	"""
368	dracut_commands = []	×
369	for hook_script, dracut_path in hooks:	×
370	src = joinpaths(self.dracut_hooks_path, hook_script)	×
371	if not os.path.exists(src):	×
372	logger.error("Missing lorax dracut hook script %s", (src))	×
373	continue	×
374	dst = joinpaths(self.vars.inroot, "/tmp/", hook_script)	×
375	copy2(src, dst)	×
376	dracut_commands += ["--include", joinpaths("/tmp/", hook_script),	×
377	dracut_path]
378	return dracut_commands	×
379
380	#### TreeBuilder helper functions
381
382	def findkernels(root="/", kdir="boot"):	1✔
383	# To find possible flavors, awk '/BuildKernel/ { print $4 }' kernel.spec
384	flavors = ('debug', 'PAE', 'PAEdebug', 'smp', 'xen', 'lpae')	1✔
385	kre = re.compile(r"vmlinuz-(?P<version>.+?\.(?P<arch>[a-z0-9_]+)"	1✔
386	r"(.(?P<flavor>{0}))?)$".format("\|".join(flavors)))
387	kernels = []	1✔
388	bootfiles = os.listdir(joinpaths(root, kdir))	1✔
389	for f in bootfiles:	1✔
390	match = kre.match(f)	1✔
391	if match:	1✔
392	kernel = DataHolder(path=joinpaths(kdir, f))	1✔
393	kernel.update(match.groupdict()) # sets version, arch, flavor	1✔
394	kernels.append(kernel)	1✔
395
396	# look for associated initrd/initramfs/etc.
397	for kernel in kernels:	1✔
398	for f in bootfiles:	1✔
399	if f.endswith('-'+kernel.version+'.img'):	1✔
400	imgtype, _rest = f.split('-',1)	1✔
401	# special backwards-compat case
402	if imgtype == 'initramfs':	1!
403	imgtype = 'initrd'	1✔
404	kernel[imgtype] = DataHolder(path=joinpaths(kdir, f))	1✔
405
406	logger.debug("kernels=%s", kernels)	1✔
407	return kernels	1✔
408
409	# udev whitelist: 'a-zA-Z0-9#+.:=@_-' (see is_whitelisted in libudev-util.c)
410	udev_blacklist=' !"$%&\'()*,/;<>?[\\]^`{\|}~' # ASCII printable, minus whitelist	1✔
411	udev_blacklist += ''.join(chr(i) for i in range(32)) # ASCII non-printable	1✔
412	def udev_escape(label):	1✔
413	out = ''	×
414	for ch in label:	×
415	out += ch if ch not in udev_blacklist else '\\x%02x' % ord(ch)	×
416	return out	×
417
418	def string_lower(string):	1✔
419	""" Return a lowercase string.
420
421	:param string: String to lowercase
422
423	This is used as a filter in the templates.
424	"""
425	return string.lower()	×

weldr / lorax / 7643853211

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous