from mkosi.mounts import finalize_crypto_mounts, finalize_source_mounts, mount_overlay
from mkosi.pager import page
from mkosi.partition import Partition, finalize_root, finalize_roothash
-from mkosi.qemu import KernelType, copy_ephemeral, run_qemu, run_ssh, start_journal_remote
+from mkosi.qemu import (
+ KernelType,
+ copy_ephemeral,
+ finalize_credentials,
+ finalize_kernel_command_line_extra,
+ run_qemu,
+ run_ssh,
+ start_journal_remote,
+)
from mkosi.run import (
apivfs_options,
chroot_cmd,
name = config.machine_or_name().replace("_", "-")
cmdline += ["--machine", name]
- for k, v in config.credentials.items():
+ for k, v in finalize_credentials(config).items():
cmdline += [f"--set-credential={k}:{v}"]
with contextlib.ExitStack() as stack:
# When invoked by the kernel, all unknown arguments are passed as environment variables
# to pid1. Let's mimic the same behavior when we invoke nspawn as a container.
- for arg in itertools.chain(config.kernel_command_line, config.kernel_command_line_extra):
+ for arg in itertools.chain(
+ config.kernel_command_line,
+ finalize_kernel_command_line_extra(config),
+ ):
name, sep, value = arg.partition("=")
# If there's a '.' in the argument name, it's not considered an environment
import platform
import re
import shlex
-import shutil
import string
import subprocess
import sys
return args, tuple([load_config(ns) for ns in images] + [load_config(config)])
-def load_credentials(args: argparse.Namespace) -> dict[str, str]:
- creds = {
- "firstboot.locale": "C.UTF-8",
- **args.credentials,
- }
-
- if "firstboot.timezone" not in creds:
- if find_binary("timedatectl"):
- tz = run(
- ["timedatectl", "show", "-p", "Timezone", "--value"],
- stdout=subprocess.PIPE,
- check=False,
- ).stdout.strip()
- else:
- tz = "UTC"
-
- creds["firstboot.timezone"] = tz
-
- if "ssh.authorized_keys.root" not in creds:
- if args.ssh_certificate:
- pubkey = run(
- ["openssl", "x509", "-in", args.ssh_certificate, "-pubkey", "-noout"],
- stdout=subprocess.PIPE,
- env=dict(OPENSSL_CONF="/dev/null"),
- ).stdout.strip()
- sshpubkey = run(
- ["ssh-keygen", "-f", "/dev/stdin", "-i", "-m", "PKCS8"], input=pubkey, stdout=subprocess.PIPE
- ).stdout.strip()
- creds["ssh.authorized_keys.root"] = sshpubkey
- elif args.ssh:
- die(
- "Ssh= is enabled but no SSH certificate was found",
- hint="Run 'mkosi genkey' to automatically create one",
- )
-
- return creds
-
-
def finalize_term() -> str:
term = os.getenv("TERM", "unknown")
if term == "unknown":
return term if sys.stderr.isatty() else "dumb"
-def load_kernel_command_line_extra(args: argparse.Namespace) -> list[str]:
- columns, lines = shutil.get_terminal_size()
- term = finalize_term()
-
- cmdline = [
- "rw",
- # Make sure we set up networking in the VM/container.
- "systemd.wants=network.target",
- # Make sure we don't load vmw_vmci which messes with virtio vsock.
- "module_blacklist=vmw_vmci",
- f"systemd.tty.term.hvc0={term}",
- f"systemd.tty.columns.hvc0={columns}",
- f"systemd.tty.rows.hvc0={lines}",
- ]
-
- if not any(s.startswith("ip=") for s in args.kernel_command_line_extra):
- cmdline += ["ip=enc0:any", "ip=enp0s1:any", "ip=enp0s2:any", "ip=host0:any", "ip=none"]
-
- if not any(s.startswith("loglevel=") for s in args.kernel_command_line_extra):
- cmdline += ["loglevel=4"]
-
- if not any(s.startswith("SYSTEMD_SULOGIN_FORCE=") for s in args.kernel_command_line_extra):
- cmdline += ["SYSTEMD_SULOGIN_FORCE=1"]
-
- if not any(s.startswith("systemd.hostname=") for s in args.kernel_command_line_extra) and args.machine:
- cmdline += [f"systemd.hostname={args.machine}"]
-
- if args.qemu_cdrom:
- # CD-ROMs are read-only so tell systemd to boot in volatile mode.
- cmdline += ["systemd.volatile=yes"]
-
- if not args.qemu_gui:
- cmdline += [
- f"systemd.tty.term.console={term}",
- f"systemd.tty.columns.console={columns}",
- f"systemd.tty.rows.console={lines}",
- "console=hvc0",
- f"TERM={term}",
- ]
-
- for s in args.kernel_command_line_extra:
- key, sep, value = s.partition("=")
- if " " in value:
- value = f'"{value}"'
- cmdline += [key if not sep else f"{key}={value}"]
-
- return cmdline
-
-
def load_environment(args: argparse.Namespace) -> dict[str, str]:
env = {
"SYSTEMD_TMPFILES_FORCE_SUBVOL": "0",
if config.sign:
config.checksum = True
- if not config.image:
- config.credentials = load_credentials(config)
- config.kernel_command_line_extra = load_kernel_command_line_extra(config)
-
config.environment = load_environment(config)
if config.overlay and not config.base_trees:
QemuDrive,
QemuFirmware,
QemuVsockCID,
+ finalize_term,
format_bytes,
systemd_tool_version,
want_selinux_relabel,
p.unlink(missing_ok=True)
+def finalize_kernel_command_line_extra(config: Config) -> list[str]:
+ columns, lines = shutil.get_terminal_size()
+ term = finalize_term()
+
+ cmdline = [
+ "rw",
+ # Make sure we set up networking in the VM/container.
+ "systemd.wants=network.target",
+ # Make sure we don't load vmw_vmci which messes with virtio vsock.
+ "module_blacklist=vmw_vmci",
+ f"systemd.tty.term.hvc0={term}",
+ f"systemd.tty.columns.hvc0={columns}",
+ f"systemd.tty.rows.hvc0={lines}",
+ ]
+
+ if not any(s.startswith("ip=") for s in config.kernel_command_line_extra):
+ cmdline += ["ip=enc0:any", "ip=enp0s1:any", "ip=enp0s2:any", "ip=host0:any", "ip=none"]
+
+ if not any(s.startswith("loglevel=") for s in config.kernel_command_line_extra):
+ cmdline += ["loglevel=4"]
+
+ if not any(s.startswith("SYSTEMD_SULOGIN_FORCE=") for s in config.kernel_command_line_extra):
+ cmdline += ["SYSTEMD_SULOGIN_FORCE=1"]
+
+ if (
+ not any(s.startswith("systemd.hostname=") for s in config.kernel_command_line_extra)
+ and config.machine
+ ):
+ cmdline += [f"systemd.hostname={config.machine}"]
+
+ if config.qemu_cdrom:
+ # CD-ROMs are read-only so tell systemd to boot in volatile mode.
+ cmdline += ["systemd.volatile=yes"]
+
+ if not config.qemu_gui:
+ cmdline += [
+ f"systemd.tty.term.console={term}",
+ f"systemd.tty.columns.console={columns}",
+ f"systemd.tty.rows.console={lines}",
+ "console=hvc0",
+ f"TERM={term}",
+ ]
+
+ for s in config.kernel_command_line_extra:
+ key, sep, value = s.partition("=")
+ if " " in value:
+ value = f'"{value}"'
+ cmdline += [key if not sep else f"{key}={value}"]
+
+ return cmdline
+
+
+def finalize_credentials(config: Config) -> dict[str, str]:
+ creds = {
+ "firstboot.locale": "C.UTF-8",
+ **config.credentials,
+ }
+
+ if "firstboot.timezone" not in creds:
+ if find_binary("timedatectl"):
+ tz = run(
+ ["timedatectl", "show", "-p", "Timezone", "--value"],
+ stdout=subprocess.PIPE,
+ check=False,
+ ).stdout.strip()
+ else:
+ tz = "UTC"
+
+ creds["firstboot.timezone"] = tz
+
+ if "ssh.authorized_keys.root" not in creds:
+ if config.ssh_certificate:
+ pubkey = run(
+ ["openssl", "x509", "-in", config.ssh_certificate, "-pubkey", "-noout"],
+ stdout=subprocess.PIPE,
+ env=dict(OPENSSL_CONF="/dev/null"),
+ ).stdout.strip()
+ sshpubkey = run(
+ ["ssh-keygen", "-f", "/dev/stdin", "-i", "-m", "PKCS8"], input=pubkey, stdout=subprocess.PIPE
+ ).stdout.strip()
+ creds["ssh.authorized_keys.root"] = sshpubkey
+ elif config.ssh:
+ die(
+ "Ssh= is enabled but no SSH certificate was found",
+ hint="Run 'mkosi genkey' to automatically create one",
+ )
+
+ return creds
+
+
def scope_cmd(
name: str,
description: str,
KernelType.identify(config, kernel) != KernelType.uki
or not config.architecture.supports_smbios(firmware)
):
- kcl = config.kernel_command_line + config.kernel_command_line_extra
+ kcl = config.kernel_command_line + finalize_kernel_command_line_extra(config)
else:
- kcl = config.kernel_command_line_extra
+ kcl = finalize_kernel_command_line_extra(config)
if kernel:
cmdline += ["-kernel", kernel]
] # fmt: skip
kcl += ["root=root", "rootfstype=virtiofs"]
- credentials = dict(config.credentials)
+ credentials = finalize_credentials(config)
def add_virtiofs_mount(
sock: Path, dst: PathString, cmdline: list[PathString], credentials: dict[str, str], *, tag: str
from mkosi.qemu import (
apply_runtime_size,
copy_ephemeral,
+ finalize_credentials,
+ finalize_kernel_command_line_extra,
finalize_qemu_firmware,
)
from mkosi.run import run
if config.qemu_gui:
cmdline += ["--console=gui"]
- cmdline += [f"--set-credential={k}:{v}" for k, v in config.credentials.items()]
+ cmdline += [f"--set-credential={k}:{v}" for k, v in finalize_credentials(config).items()]
with contextlib.ExitStack() as stack:
fname = stack.enter_context(copy_ephemeral(config, config.output_dir_or_cwd() / config.output))
if config.forward_journal:
cmdline += ["--forward-journal", config.forward_journal]
- cmdline += [*args.cmdline, *config.kernel_command_line_extra]
+ cmdline += [*args.cmdline, *finalize_kernel_command_line_extra(config)]
run(
cmdline,
projects / thirdparty / mkosi.git / commitdiff
