fork_and_wait,
run,
spawn,
+ which,
)
from mkosi.state import MkosiState
from mkosi.types import PathString
if not any(gen_kernel_images(state)) and state.config.bootable == ConfigFeature.auto:
return
- if not shutil.which("bootctl"):
+ if not which("bootctl", tools=state.config.tools_tree):
if state.config.bootable == ConfigFeature.enabled:
die("A bootable image was requested but bootctl was not found")
return
if (state.config.secure_boot_sign_tool == SecureBootSignTool.sbsign or
state.config.secure_boot_sign_tool == SecureBootSignTool.auto and
- shutil.which("sbsign") is not None):
+ which("sbsign", state.config.tools_tree) is not None):
bwrap(["sbsign",
"--key", state.config.secure_boot_key,
"--cert", state.config.secure_boot_certificate,
tools=state.config.tools_tree)
elif (state.config.secure_boot_sign_tool == SecureBootSignTool.pesign or
state.config.secure_boot_sign_tool == SecureBootSignTool.auto and
- shutil.which("pesign") is not None):
+ which("pesign", tools=state.config.tools_tree) is not None):
pesign_prepare(state)
bwrap(["pesign",
"--certdir", state.workspace / "pesign",
copy_path(state.install_dir, state.root, tools=state.config.tools_tree)
-def gzip_binary() -> str:
- return "pigz" if shutil.which("pigz") else "gzip"
+def gzip_binary(config: MkosiConfig) -> str:
+ return "pigz" if which("pigz", tools=config.tools_tree) else "gzip"
-def tar_binary() -> str:
+def tar_binary(config: MkosiConfig) -> str:
# Some distros (Mandriva) install BSD tar as "tar", hence prefer
# "gtar" if it exists, which should be GNU tar wherever it exists.
# We are interested in exposing same behaviour everywhere hence
# everywhere. In particular given the limited/different SELinux
# support in BSD tar and the different command line syntax
# compared to GNU tar.
- return "gtar" if shutil.which("gtar") else "tar"
+ return "gtar" if which("gtar", tools=config.tools_tree) else "tar"
def make_tar(state: MkosiState) -> None:
return
cmd: list[PathString] = [
- tar_binary(),
+ tar_binary(state.config),
"-C", state.root,
"-c", "--xattrs",
"--xattrs-include=*",
die(f"sd-stub not found at /{stub.relative_to(state.root)} in the image")
cmd: list[PathString] = [
- shutil.which("ukify") or "/usr/lib/systemd/ukify",
+ which("ukify", tools=state.config.tools_tree) or "/usr/lib/systemd/ukify",
"--cmdline", f"@{state.workspace / 'cmdline'}",
"--os-release", f"@{state.root / 'usr/lib/os-release'}",
"--stub", stub,
sign_expected_pcr = (state.config.sign_expected_pcr == ConfigFeature.enabled or
(state.config.sign_expected_pcr == ConfigFeature.auto and
- shutil.which("systemd-measure") is not None))
+ which("systemd-measure", tools=state.config.tools_tree) is not None))
if sign_expected_pcr:
cmd += [
die("A bootable image was requested but no kernel was found")
-def compressor_command(compression: Compression) -> list[PathString]:
+def compressor_command(config: MkosiConfig, compression: Compression) -> list[PathString]:
"""Returns a command suitable for compressing archives."""
if compression == Compression.gz:
- return [gzip_binary(), "--fast", "--stdout", "-"]
+ return [gzip_binary(config), "--fast", "--stdout", "-"]
elif compression == Compression.xz:
return ["xz", "--check=crc32", "--fast", "-T0", "--stdout", "-"]
elif compression == Compression.zst:
src.unlink() # if src == dst, make sure dst doesn't truncate the src file but creates a new file.
with dst.open("wb") as o:
- bwrap(compressor_command(compression), stdin=i, stdout=o, tools=state.config.tools_tree)
+ bwrap(compressor_command(state.config, compression), stdin=i, stdout=o, tools=state.config.tools_tree)
os.chown(dst, uid=state.uid, gid=state.gid)
# SPDX-License-Identifier: LGPL-2.1+
-import shutil
import subprocess
from pathlib import Path
from typing import cast
from mkosi.config import ConfigFeature, MkosiConfig
from mkosi.install import copy_path
from mkosi.log import die
-from mkosi.run import bwrap
+from mkosi.run import bwrap, which
def statfs(config: MkosiConfig, path: Path) -> str:
def btrfs_maybe_make_subvolume(config: MkosiConfig, path: Path, mode: int) -> None:
- if config.use_subvolumes == ConfigFeature.enabled and not shutil.which("btrfs"):
+ if config.use_subvolumes == ConfigFeature.enabled and not which("btrfs", tools=config.tools_tree):
die("Subvolumes requested but the btrfs command was not found")
if statfs(config, path.parent) != "btrfs":
path.mkdir(mode)
return
- if config.use_subvolumes != ConfigFeature.disabled and shutil.which("btrfs") is not None:
+ if config.use_subvolumes != ConfigFeature.disabled and which("btrfs", tools=config.tools_tree) is not None:
result = bwrap(["btrfs", "subvolume", "create", path],
check=config.use_subvolumes == ConfigFeature.enabled,
tools=config.tools_tree).returncode
def btrfs_maybe_snapshot_subvolume(config: MkosiConfig, src: Path, dst: Path) -> None:
subvolume = (config.use_subvolumes == ConfigFeature.enabled or
- config.use_subvolumes == ConfigFeature.auto and shutil.which("btrfs") is not None)
+ config.use_subvolumes == ConfigFeature.auto and which("btrfs", tools=config.tools_tree) is not None)
- if config.use_subvolumes == ConfigFeature.enabled and not shutil.which("btrfs"):
+ if config.use_subvolumes == ConfigFeature.enabled and not which("btrfs", tools=config.tools_tree):
die("Subvolumes requested but the btrfs command was not found")
# Subvolumes always have inode 256 so we can use that to check if a directory is a subvolume.
if dst.exists():
dst.rmdir()
- if shutil.which("btrfs"):
+ if which("btrfs", config.tools_tree):
result = bwrap(["btrfs", "subvolume", "snapshot", src, dst],
check=config.use_subvolumes == ConfigFeature.enabled,
tools=config.tools_tree).returncode
# SPDX-License-Identifier: LGPL-2.1+
-import shutil
import tempfile
from collections.abc import Sequence
from pathlib import Path
from mkosi.architecture import Architecture
from mkosi.distributions import DistributionInstaller
from mkosi.log import die
-from mkosi.run import bwrap
+from mkosi.run import bwrap, which
from mkosi.state import MkosiState
from mkosi.types import CompletedProcess, PathString
"-o", f"Dir::Etc::trusted={trustedkeys}",
"-o", f"Dir::Etc::trustedparts={trustedkeys_dir}",
"-o", f"Dir::Log={state.pkgmngr / 'var/log/apt'}",
- "-o", f"Dir::Bin::dpkg={shutil.which('dpkg')}",
+ "-o", f"Dir::Bin::dpkg={which('dpkg', tools=state.config.tools_tree)}",
"-o", "Debug::NoLocking=true",
"-o", f"DPkg::Options::=--root={state.root}",
"-o", f"DPkg::Options::=--log={state.pkgmngr / 'var/log/apt/dpkg.log'}",
from mkosi.distributions import DistributionInstaller
from mkosi.log import die
from mkosi.remove import unlink_try_hard
-from mkosi.run import bwrap
+from mkosi.run import bwrap, which
from mkosi.state import MkosiState
from mkosi.util import Distribution, detect_distribution, sort_packages
state.pkgmngr.joinpath("var/lib/dnf").mkdir(exist_ok=True, parents=True)
# dnf5 does not support building for foreign architectures yet (missing --forcearch)
- dnf = shutil.which("dnf5") if state.config.architecture.is_native() else None
- dnf = dnf or shutil.which("dnf") or "yum"
+ dnf = which("dnf5", tools=state.config.tools_tree) if state.config.architecture.is_native() else None
+ dnf = dnf or which("dnf", tools=state.config.tools_tree) or "yum"
cmdline = [
dnf,
# SPDX-License-Identifier: LGPL-2.1+
-import shutil
import textwrap
import urllib.request
import xml.etree.ElementTree as ElementTree
from mkosi.distributions import DistributionInstaller
from mkosi.distributions.fedora import Repo, fixup_rpmdb_location, invoke_dnf, setup_dnf
from mkosi.log import die
-from mkosi.run import bwrap
+from mkosi.run import bwrap, which
from mkosi.state import MkosiState
release_url = f"{state.config.mirror}/distribution/leap/{release}/repo/oss/"
updates_url = f"{state.config.mirror}/update/leap/{release}/oss/"
- zypper = shutil.which("zypper")
+ zypper = which("zypper", tools=state.config.tools_tree)
# If we need to use a local mirror, create a temporary repository definition
# that doesn't get in the image, as it is valid only at image build time.
@classmethod
def remove_packages(cls, state: MkosiState, packages: Sequence[str]) -> None:
- if shutil.which("zypper"):
+ if which("zypper", tools=state.config.tools_tree):
invoke_zypper(state, "remove", packages, ["--clean-deps"])
else:
invoke_dnf(state, "remove", packages)
from mkosi.config import ConfigFeature, MkosiArgs, MkosiConfig
from mkosi.log import die
from mkosi.remove import unlink_try_hard
-from mkosi.run import MkosiAsyncioThread, bwrap, bwrap_cmd, spawn
+from mkosi.run import MkosiAsyncioThread, bwrap, bwrap_cmd, spawn, which
from mkosi.types import PathString
from mkosi.util import (
Distribution,
def find_qemu_binary(config: MkosiConfig) -> str:
binaries = ["qemu", "qemu-kvm", f"qemu-system-{config.architecture.to_qemu()}"]
for binary in binaries:
- if shutil.which(binary) is not None:
+ if which(binary, tools=config.tools_tree) is not None:
return binary
die("Couldn't find QEMU/KVM binary")
"-device", "virtio-scsi-pci,id=scsi",
"-device", "scsi-hd,drive=hd,bootindex=1"]
- if config.qemu_swtpm != ConfigFeature.disabled and shutil.which("swtpm") is not None:
+ if config.qemu_swtpm != ConfigFeature.disabled and which("swtpm", tools=config.tools_tree) is not None:
sock = stack.enter_context(start_swtpm(config))
cmdline += ["-chardev", f"socket,id=chrtpm,path={sock}",
"-tpmdev", "emulator,id=tpm0,chardev=chrtpm"]
import pwd
import queue
import shlex
+import shutil
import signal
import subprocess
import sys
return cmdline
+def which(program: str, tools: Optional[Path]) -> Optional[str]:
+ return shutil.which(program, path=f"{tools}/usr/bin:{tools}/usr/sbin" if tools else None)
+
+
class MkosiAsyncioThread(threading.Thread):
"""
The default threading.Thread() is not interruptable, so we make our own version by using the concurrency
projects / thirdparty / mkosi.git / commitdiff
