image/image: tolerate pub-key copying to fail during image build

Authored-by: InsanePrawn <insane.prawny@gmail.com>

chroot/abstract.py

@@ -60,12 +60,8 @@ class AbstractChroot(Protocol):
         capture_output: bool,
         cwd: str,
         fail_inactive: bool,
-        switch_user: Optional[str],
         stdout: Optional[FileDescriptor],
         stderr: Optional[FileDescriptor],
-        stdin: Optional[FileDescriptor],
-        stdin_input: Optional[str],
-        check: Optional[bool],
     ):
         pass
 
@@ -229,12 +225,9 @@ class Chroot(AbstractChroot):
         capture_output: bool = False,
         cwd: Optional[str] = None,
         fail_inactive: bool = True,
-        switch_user: Optional[str] = None,
         stdout: Optional[FileDescriptor] = None,
         stderr: Optional[FileDescriptor] = None,
-        stdin: Optional[FileDescriptor] = None,
-        stdin_input: Optional[str] = None,
-        check: Optional[bool] = None,
+        switch_user: Optional[str] = None,
     ) -> Union[int, subprocess.CompletedProcess]:
         if not self.active and fail_inactive:
             raise Exception(f'Chroot {self.name} is inactive, not running command! Hint: pass `fail_inactive=False`')
@@ -251,23 +244,13 @@ class Chroot(AbstractChroot):
             script = flatten_shell_script(script, shell_quote_items=False, wrap_in_shell_quote=False)
         if cwd:
             script = f"cd {shell_quote(cwd)} && ( {script} )"
-        if switch_user and switch_user != 'root':
+        if switch_user:
             inner_cmd = generate_cmd_su(script, switch_user=switch_user, elevation_method='none', force_su=True)
         else:
             inner_cmd = wrap_in_bash(script, flatten_result=False)
         cmd = flatten_shell_script(['chroot', self.path] + env_cmd + inner_cmd, shell_quote_items=True)
 
-        return run_root_cmd(
-            cmd,
-            env=outer_env,
-            attach_tty=attach_tty,
-            capture_output=capture_output,
-            stdout=stdout,
-            stderr=stderr,
-            stdin=stdin,
-            stdin_input=stdin_input,
-            check=check,
-        )
+        return run_root_cmd(cmd, env=outer_env, attach_tty=attach_tty, capture_output=capture_output, stdout=stdout, stderr=stderr)
 
     def mount_pkgbuilds(self, fail_if_mounted: bool = False) -> str:
         return self.mount(

chroot/cli.py

@@ -7,7 +7,6 @@ from typing import Optional
 from config.state import config
 from wrapper import enforce_wrap
 from devices.device import get_profile_device
-from image.cli import cmd_inspect
 
 from .abstract import Chroot
 from .base import get_base_chroot
@@ -31,6 +30,7 @@ def cmd_chroot(ctx: click.Context, type: str = 'build', name: Optional[str] = No
         raise Exception(f'Unknown chroot type: "{type}"')
 
     if type == 'rootfs':
+        from image.image import cmd_inspect
         assert isinstance(cmd_inspect, click.Command)
         ctx.invoke(cmd_inspect, profile=name, shell=True)
         return

config/cli.py

@@ -1,6 +1,5 @@
 import click
 import logging
-import os
 
 from copy import deepcopy
 from typing import Any, Callable, Iterable, Mapping, Optional, Union
@@ -261,7 +260,6 @@ def cmd_config_init(
 ):
     """Initialize the config file"""
     if not non_interactive:
-        from packages.cli import cmd_init as cmd_init_pkgbuilds
         logging.info(CONFIG_MSG)
         results: dict[str, dict] = {}
         for section in sections:
@@ -284,13 +282,6 @@ def cmd_config_init(
                 config.write()
             else:
                 return
-            if not non_interactive and not os.path.exists(os.path.join(config.get_path('pkgbuilds'), '.git')):
-                extra_msg = " This way, we can give you a list of devices and flavours later" if 'profiles' in sections else ''
-                if click.confirm(
-                        f"It seems you don't have our PKGBUILDs checked out yet.\nWould you like KBS to fetch them?{extra_msg}",
-                        default=True,
-                ):
-                    execute_without_exit(click.Context(cmd_config).invoke, ['packages', 'init'], cmd_init_pkgbuilds)
         if 'profiles' in sections:
             print("Configuring profiles")
             current_profile = 'default' if 'current' not in config.file.profiles else config.file.profiles.current

config/profile.py

@@ -15,8 +15,6 @@ PROFILE_DEFAULTS_DICT = {
     'username': 'kupfer',
     'password': None,
     'size_extra_mb': "0",
-    'encryption': None,
-    'encryption_password': None,
 }
 PROFILE_DEFAULTS = Profile.fromDict(PROFILE_DEFAULTS_DICT)
 

config/scheme.py

@@ -17,20 +17,20 @@ class SparseProfile(DictScheme):
     username: Optional[str]
     password: Optional[str]
     size_extra_mb: Optional[Union[str, int]]
-    encryption: Optional[bool]
-    encryption_password: Optional[str]
 
     def __repr__(self):
         return f'{type(self)}{dict.__repr__(self.toDict())}'
 
 
 class Profile(SparseProfile):
+    parent: Optional[str]
     device: str
     flavour: str
     pkgs_include: list[str]
     pkgs_exclude: list[str]
     hostname: str
     username: str
+    password: Optional[str]
     size_extra_mb: Union[str, int]
 
 

constants.py

@@ -172,6 +172,3 @@ SRCINFO_TARBALL_URL = f'{KUPFER_HTTPS_BASE}/{SRCINFO_TARBALL_FILE}'
 
 FLAVOUR_INFO_FILE = 'flavourinfo.json'
 FLAVOUR_DESCRIPTION_PREFIX = 'kupfer flavour:'
-
-LUKS_LABEL_DEFAULT = 'kupfer_cryptroot'
-LUKS_MAPPER_DEFAULT = 'kupfer-crypt'

exec/cmd.py

@@ -97,9 +97,6 @@ def run_cmd(
     elevation_method: Optional[ElevationMethod] = None,
     stdout: Optional[FileDescriptor] = None,
     stderr: Optional[FileDescriptor] = None,
-    stdin: Optional[FileDescriptor] = None,
-    stdin_input: Optional[str] = None,
-    check: Optional[bool] = None,
 ) -> Union[CompletedProcess, int]:
     "execute `script` as `switch_user`, elevating and su'ing as necessary"
     kwargs: dict = {}
@@ -114,10 +111,6 @@ def run_cmd(
             for name, fd in {'stdout': stdout, 'stderr': stderr}.items():
                 if fd is not None:
                     kwargs[name] = fd
-        for name, value in {'stdin': stdin, 'input': stdin_input, 'check': check}.items():
-            if value is not None:
-                kwargs[name] = value
-
     script = flatten_shell_script(script)
     if cwd:
         kwargs['cwd'] = cwd

exec/file.py

@@ -144,7 +144,13 @@ def remove_file(path: str, recursive=False):
             raise Exception(f"Unable to remove {path}: cmd returned {rc}")
 
 
-def makedir(path, user: Optional[Union[str, int]] = None, group: Optional[Union[str, int]] = None, parents: bool = True):
+def makedir(
+    path,
+    user: Optional[Union[str, int]] = None,
+    group: Optional[Union[str, int]] = None,
+    parents: bool = True,
+    mode: Optional[Union[int, str]] = None,
+):
     if not root_check_exists(path):
         try:
             if parents:
@@ -153,6 +159,8 @@ def makedir(path, user: Optional[Union[str, int]] = None, group: Optional[Union[
                 os.mkdir(path)
         except:
             run_root_cmd(['mkdir'] + (['-p'] if parents else []) + [path])
+    if mode is not None:
+        chmod(path, mode=mode)
     chown(path, user, group)
 
 
@@ -171,9 +179,9 @@ def symlink(source, target):
             raise Exception(f'Symlink creation of {target} pointing at {source} failed')
 
 
-def get_temp_dir(register_cleanup=True, mode: int = 0o0755, prefix='kupfertmp_'):
+def get_temp_dir(register_cleanup=True, mode: int = 0o0755):
     "create a new tempdir and sanitize ownership so root can access user files as god intended"
-    t = mkdtemp(prefix=prefix)
+    t = mkdtemp()
     chmod(t, mode, privileged=False)
     if register_cleanup:
         atexit.register(remove_file, t, recursive=True)

image/boot.py

@@ -13,7 +13,7 @@ from flavours.cli import profile_option
 from wrapper import enforce_wrap
 
 from .fastboot import fastboot_boot, fastboot_erase
-from .image import get_device_name, losetup_setup_image, get_image_path, dump_aboot, dump_lk2nd
+from .image import get_device_name, losetup_rootfs_image, get_image_path, dump_aboot, dump_lk2nd
 
 LK2ND = FLASH_PARTS['LK2ND']
 ABOOT = FLASH_PARTS['ABOOT']
@@ -61,7 +61,7 @@ def cmd_boot(
             if not os.path.exists(path):
                 urllib.request.urlretrieve(f'https://github.com/dreemurrs-embedded/Jumpdrive/releases/download/{JUMPDRIVE_VERSION}/{file}', path)
         else:
-            loop_device = losetup_setup_image(image_path, sector_size)
+            loop_device = losetup_rootfs_image(image_path, sector_size)
             if type == LK2ND:
                 path = dump_lk2nd(loop_device + 'p1')
             elif type == ABOOT:

image/cli.py

@@ -1,248 +1,6 @@
-import click
-import logging
-import os
-
-from signal import pause
-from typing import Optional
-
-from config.state import config, Profile
-from constants import BASE_LOCAL_PACKAGES, BASE_PACKAGES, LUKS_MAPPER_DEFAULT
-from devices.device import get_profile_device
-from exec.file import makedir
-from flavours.flavour import get_profile_flavour
-from packages.build import build_enable_qemu_binfmt, build_packages, filter_pkgbuilds
-from wrapper import enforce_wrap
-
 from .boot import cmd_boot
-from .cryptsetup import encryption_option, get_cryptmapper_path, luks_close, luks_create, luks_open
 from .flash import cmd_flash
-from .image import (
-    IMG_DEFAULT_SIZE_BOOT_MB,
-    create_boot_fs,
-    create_img_file,
-    create_root_fs,
-    dd_image,
-    get_device_chroot,
-    get_image_path,
-    install_rootfs,
-    losetup_setup_image,
-    mount_chroot,
-    partprobe,
-    partition_device,
-)
-
-
-@click.group(name='image')
-def cmd_image():
-    """Build, flash and boot device images"""
-
+from .image import cmd_image
 
 for cmd in [cmd_boot, cmd_flash]:
     cmd_image.add_command(cmd)
-
-sectorsize_option = click.option(
-    '-b',
-    '--sector-size',
-    help="Override the device's sector size",
-    type=int,
-    default=None,
-)
-
-
-@cmd_image.command(name='build')
-@click.argument('profile_name', required=False)
-@click.option(
-    '--local-repos/--no-local-repos',
-    '-l/-L',
-    help='Whether to use local package repos at all or only use HTTPS repos.',
-    default=True,
-    show_default=True,
-    is_flag=True,
-)
-@click.option(
-    '--build-pkgs/--no-build-pkgs',
-    '-p/-P',
-    help='Whether to build missing/outdated local packages if local repos are enabled.',
-    default=True,
-    show_default=True,
-    is_flag=True,
-)
-@click.option(
-    '--no-download-pkgs',
-    help='Disable trying to download packages instead of building if building is enabled.',
-    default=False,
-    is_flag=True,
-)
-@click.option(
-    '--block-target',
-    help='Override the block device file to write the final image to',
-    type=click.Path(),
-    default=None,
-)
-@click.option(
-    '--skip-part-images',
-    help='Skip creating image files for the partitions and directly work on the target block device.',
-    default=False,
-    is_flag=True,
-)
-@encryption_option
-@sectorsize_option
-def cmd_build(
-    profile_name: Optional[str] = None,
-    local_repos: bool = True,
-    build_pkgs: bool = True,
-    no_download_pkgs=False,
-    block_target: Optional[str] = None,
-    sector_size: Optional[int] = None,
-    skip_part_images: bool = False,
-    encryption: Optional[bool] = None,
-    encryption_password: Optional[str] = None,
-    encryption_mapper: str = LUKS_MAPPER_DEFAULT,
-):
-    """
-    Build a device image.
-
-    Unless overriden, required packages will be built or preferably downloaded from HTTPS repos.
-    """
-
-    config.enforce_profile_device_set()
-    config.enforce_profile_flavour_set()
-    enforce_wrap()
-    device = get_profile_device(profile_name)
-    arch = device.arch
-    # check_programs_wrap(['makepkg', 'pacman', 'pacstrap'])
-    profile: Profile = config.get_profile(profile_name)
-    flavour = get_profile_flavour(profile_name)
-    rootfs_size_mb = flavour.parse_flavourinfo().rootfs_size * 1000 + int(profile.size_extra_mb)
-    bootfs_size_mb = IMG_DEFAULT_SIZE_BOOT_MB
-
-    if encryption is None:
-        encryption = profile.encryption
-
-    packages = BASE_LOCAL_PACKAGES + [device.package.name, flavour.pkgbuild.name]
-    packages_extra = BASE_PACKAGES + profile.pkgs_include
-
-    if encryption:
-        packages_extra += ['cryptsetup', 'util-linux']  # TODO: select osk-sdl here somehow
-
-    if arch != config.runtime.arch:
-        build_enable_qemu_binfmt(arch)
-
-    if local_repos and build_pkgs:
-        logging.info("Making sure all packages are built")
-        # enforce that local base packages are built
-        pkgbuilds = set(filter_pkgbuilds(packages, arch=arch, allow_empty_results=False, use_paths=False))
-        # extra packages might be a mix of package names that are in our PKGBUILDs and packages from the base distro
-        pkgbuilds |= set(filter_pkgbuilds(packages_extra, arch=arch, allow_empty_results=True, use_paths=False))
-        build_packages(pkgbuilds, arch, try_download=not no_download_pkgs)
-
-    sector_size = sector_size or device.get_image_sectorsize() or 512
-
-    image_path = block_target or get_image_path(device, flavour.name)
-
-    makedir(os.path.dirname(image_path))
-
-    logging.info(f'Creating new file at {image_path}')
-    create_img_file(image_path, f"{rootfs_size_mb + bootfs_size_mb}M")
-
-    loop_device = losetup_setup_image(image_path, sector_size or device.get_image_sectorsize_default())
-
-    partition_device(loop_device, sector_size=sector_size, boot_partition_size_mb=bootfs_size_mb)
-    partprobe(loop_device)
-
-    boot_dev: str
-    root_dev: str
-    root_dev_raw: str
-    loop_boot = loop_device + 'p1'
-    loop_root = loop_device + 'p2'
-    if skip_part_images:
-        boot_dev = loop_boot
-        root_dev = loop_root
-    else:
-        logging.info('Creating per-partition image files')
-        boot_dev = create_img_file(get_image_path(device, flavour, 'boot'), f'{bootfs_size_mb}M')
-        root_dev = create_img_file(get_image_path(device, flavour, 'root'), f'{rootfs_size_mb - 200}M')
-
-    root_dev_raw = root_dev
-
-    if encryption:
-        encryption_password = encryption_password or profile.encryption_password
-        if not encryption_password:
-            encryption_password = click.prompt(
-                "Please enter your encryption password (input hidden)",
-                hide_input=True,
-                confirmation_prompt=True,
-            )
-        luks_create(root_dev, password=encryption_password)
-        luks_open(root_dev, mapper_name=encryption_mapper, password=encryption_password)
-        root_dev = get_cryptmapper_path(encryption_mapper)
-
-    assert os.path.exists(root_dev)
-    create_root_fs(root_dev)
-    create_boot_fs(boot_dev)
-
-    install_rootfs(
-        root_dev,
-        boot_dev,
-        device,
-        flavour,
-        arch,
-        list(set(packages) | set(packages_extra)),
-        local_repos,
-        profile,
-        encrypted=bool(encryption),
-    )
-    if encryption:
-        luks_close(mapper_name=encryption_mapper)
-
-    if not skip_part_images:
-        logging.info('Copying partition image files into full image:')
-        logging.info(f'Block-copying /boot to {image_path}')
-        dd_image(input=boot_dev, output=loop_boot)
-        logging.info(f'Block-copying rootfs to {image_path}')
-        dd_image(input=root_dev_raw, output=loop_root)
-
-    logging.info(f'Done! Image saved to {image_path}')
-
-
-@cmd_image.command(name='inspect')
-@click.option('--shell', '-s', is_flag=True)
-@click.option('--use-local-repos', '-l', is_flag=True)
-@sectorsize_option
-@encryption_option
-@click.argument('profile', required=False)
-def cmd_inspect(
-    profile: Optional[str] = None,
-    shell: bool = False,
-    sector_size: Optional[int] = None,
-    use_local_repos: bool = False,
-    encryption: Optional[bool] = None,
-):
-    """Loop-mount the device image for inspection."""
-    config.enforce_profile_device_set()
-    config.enforce_profile_flavour_set()
-    enforce_wrap()
-    profile_conf = config.get_profile(profile)
-    device = get_profile_device(profile)
-    arch = device.arch
-    flavour = get_profile_flavour(profile).name
-    sector_size = sector_size or device.get_image_sectorsize_default()
-
-    chroot = get_device_chroot(device.name, flavour, arch, packages=[], use_local_repos=use_local_repos)
-    image_path = get_image_path(device, flavour)
-    loop_device = losetup_setup_image(image_path, sector_size)
-    partprobe(loop_device)
-    mount_chroot(loop_device + 'p2', loop_device + 'p1', chroot, password=profile_conf.encryption_password)
-
-    logging.info(f'Inspect the rootfs image at {chroot.path}')
-
-    if shell:
-        chroot.initialized = True
-        chroot.activate()
-        if arch != config.runtime.arch:
-            logging.info('Installing requisites for foreign-arch shell')
-            build_enable_qemu_binfmt(arch)
-        logging.info('Starting inspection shell')
-        chroot.run_cmd('/bin/bash')
-    else:
-        pause()

image/cryptsetup.py

@@ -1,179 +0,0 @@
-import atexit
-import click
-import logging
-import os
-
-from typing import Optional
-
-from constants import LUKS_LABEL_DEFAULT
-from chroot.build import BuildChroot
-from exec.cmd import run_cmd, CompletedProcess
-
-encryption_option = click.option(
-    '--encryption/--no-encryption',
-    help="Force applying/ignoring LUKS encryption when handling the device image."
-    "Defaults to using the Profile's setting.",
-    default=None,
-    is_flag=True,
-)
-
-
-def get_accessible_user(path):
-    return None if os.access(path, os.R_OK) else 'root'
-
-
-def check_dev_exists(device_path: str, verb: str = 'find'):
-    if not os.path.exists(device_path):
-        raise Exception(f"Can't {verb} LUKS on {device_path!r}: file does not exist")
-
-
-def get_cryptmapper_path(mapper_name: str) -> str:
-    return f'/dev/mapper/{mapper_name}'
-
-
-def mapper_exists(mapper_name: str, chroot: Optional[BuildChroot]) -> bool:
-    path = get_cryptmapper_path(mapper_name)
-    paths = [path]
-    if chroot:
-        paths.append(chroot.get_path(path))
-    for p in paths:
-        if os.path.exists(p):
-            return True
-    return False
-
-
-def get_password_io(password: Optional[str]) -> Optional[bytes]:
-    return password.encode() if password else None
-
-
-def is_luks(device_path: str, native_chroot: Optional[BuildChroot] = None) -> bool:
-    check_dev_exists(device_path, 'check')
-    run_func = native_chroot.run_cmd if native_chroot else run_cmd
-    user = get_accessible_user(device_path)
-    cmd = ["blkid", '--match-token', 'TYPE=crypto_LUKS', device_path]
-    result = run_func(cmd, capture_output=True, switch_user=user)  # type: ignore[operator]
-    assert isinstance(result, CompletedProcess)
-    return bool(result.stdout and result.stdout.strip())
-
-
-def get_luks_offset(
-    mapper_name: str,
-    native_chroot: Optional[BuildChroot] = None,
-) -> tuple[int, int]:
-    device_path = get_cryptmapper_path(mapper_name)
-    check_dev_exists(device_path, 'get offset of')
-    run_func = native_chroot.run_cmd if native_chroot else run_cmd
-    user = get_accessible_user(device_path)
-    stdout: str = ''
-    cmd = ['cryptsetup', 'status', mapper_name]
-    result = run_func(cmd, capture_output=True, switch_user=user)  # type: ignore[operator]
-    assert isinstance(result, CompletedProcess)
-    if not (result.stdout and (stdout := result.stdout.strip())):
-        raise Exception(f"Couldn't get LUKS offset for {mapper_name!r} from 'cryptsetup status': empty stdout: {stdout!r}")
-    markers = {'offset': -1, 'sector size': -1}
-    for line in stdout.decode().split('\n'):
-        line = line.strip()
-        for item in markers:
-            offset_marker = f'{item}:'
-            if line.startswith(offset_marker):
-                try:
-                    markers[item] = int(line.split(offset_marker)[-1].strip().split(' ')[0])
-                except Exception as ex:
-                    raise Exception(f"Couldn't get LUKS {item=} for {mapper_name!r} due to an exception parsing cryptsetup output: {ex}")
-    for i in markers.values():
-        if i != -1:
-            continue
-        logging.debug(f"Failed to find ':' in stdout: {stdout}")
-        raise Exception(f"Failed to find LUKS offset for {mapper_name!r}: Offset line not found")
-    return markers['offset'], markers['sector size']
-
-
-def luks_create(
-    backing_device: str,
-    label: str = LUKS_LABEL_DEFAULT,
-    native_chroot: Optional[BuildChroot] = None,
-    password: Optional[str] = None,
-    extra_opts: list[str] = [],
-    use_random: bool = True,
-    cipher: Optional[str] = None,
-    pbkdf: Optional[str] = None,
-    iter_time: Optional[int] = None,
-):
-    check_dev_exists(backing_device, 'create')
-    run_func = native_chroot.run_cmd if native_chroot else run_cmd
-    extra_opts = list(extra_opts)  # copy list before modification
-    if use_random:
-        extra_opts += ['--use-random']
-    if cipher:
-        extra_opts += ['--cipher', cipher]
-    if pbkdf:
-        extra_opts += ['--pbkdf', pbkdf]
-    if iter_time is not None:
-        extra_opts += ['--iter-time', str(iter_time)]
-    if label:
-        extra_opts += ['--label', label]
-    logging.info(f"Creating LUKS volume at {backing_device!r}{' (unattended)' if password else ''}")
-    result = run_func(  # type: ignore[operator]
-        ['cryptsetup', '-q', 'luksFormat', *extra_opts, backing_device],
-        switch_user=get_accessible_user(backing_device),
-        attach_tty=not password,
-        stdin_input=get_password_io(password),
-    )
-    rc = result if isinstance(result, int) else result.returncode
-    if rc:
-        raise Exception("Failed to format LUKS device: cryptsetup error^^^^")
-
-
-def luks_open(
-    backing_device: str,
-    mapper_name: str,
-    extra_opts: list[str] = [],
-    password: Optional[str] = None,
-    native_chroot: Optional[BuildChroot] = None,
-    schedule_close: bool = True,
-    idempotent: bool = False,
-):
-    check_dev_exists(backing_device, 'open')
-    run_func = native_chroot.run_cmd if native_chroot else run_cmd
-    if mapper_exists(mapper_name, native_chroot):
-        if idempotent:
-            logging.debug(f"LUKS mapper {mapper_name!r} already open")
-            return
-        raise Exception(f"Can't open LUKS for {backing_device!r} with mapper name {mapper_name!r}: "
-                        "mapper file already exists")
-    logging.info(f"Opening LUKS mapper {mapper_name!r} for {backing_device!r}")
-    result = run_func(  # type: ignore[operator]
-        ['cryptsetup', 'luksOpen', *extra_opts, backing_device, mapper_name],
-        switch_user='root',
-        attach_tty=not password,
-        stdin_input=get_password_io(password),
-    )
-    rc = result if isinstance(result, int) else result.returncode
-    if rc:
-        raise Exception("Failed to open LUKS device: cryptsetup error^^^^")
-    if schedule_close:
-        atexit.register(luks_close, mapper_name, native_chroot=native_chroot, idempotent=True)
-    logging.info(f"LUKS mapper {mapper_name!r} opened!")
-
-
-def luks_close(
-    mapper_name: str,
-    native_chroot: Optional[BuildChroot] = None,
-    extra_opts: list[str] = [],
-    idempotent: bool = False,
-):
-    run_func = native_chroot.run_cmd if native_chroot else run_cmd
-    if not mapper_exists(mapper_name, native_chroot):
-        if idempotent:
-            logging.debug(f"LUKS mapper {mapper_name!r} already closed")
-            return 0
-        raise Exception(f"Can't close LUKS mapper {mapper_name!r}: mapper doesn't exist")
-    logging.info(f"Closing LUKS mapper {mapper_name!r}")
-    result = run_func(  # type: ignore[operator]
-        ['cryptsetup', 'close', *extra_opts, mapper_name],
-        switch_user='root',
-    )
-    rc = result if isinstance(result, int) else result.returncode
-    if rc:
-        raise Exception("Failed to close LUKS device: cryptsetup error^^^^")
-    logging.info(f"LUKS mapper {mapper_name!r} closed.")

image/flash.py

@@ -5,7 +5,6 @@ import logging
 
 from typing import Optional
 
-from config.state import config
 from constants import FLASH_PARTS, LOCATIONS, FASTBOOT, JUMPDRIVE
 from exec.cmd import run_root_cmd
 from exec.file import get_temp_dir
@@ -15,8 +14,7 @@ from flavours.cli import profile_option
 from wrapper import enforce_wrap
 
 from .fastboot import fastboot_flash
-from .image import dd_image, dump_aboot, dump_lk2nd, dump_qhypstub, get_image_path, losetup_destroy, losetup_setup_image, partprobe, shrink_fs
-from .cryptsetup import encryption_option
+from .image import dd_image, dump_aboot, dump_lk2nd, dump_qhypstub, get_image_path, losetup_destroy, losetup_rootfs_image, partprobe, shrink_fs
 
 ABOOT = FLASH_PARTS['ABOOT']
 LK2ND = FLASH_PARTS['LK2ND']
@@ -49,15 +47,15 @@ def test_blockdev(path: str):
                         'microSD inserted or no microSD card slot installed in the device) or corrupt or defect')
 
 
-def prepare_minimal_image(source_path: str, sector_size: int, encrypted: Optional[bool], encryption_password: Optional[str]) -> str:
+def prepare_minimal_image(source_path: str, sector_size: int) -> str:
     minimal_image_dir = get_temp_dir(register_cleanup=True)
     minimal_image_path = os.path.join(minimal_image_dir, f'minimal-{os.path.basename(source_path)}')
     logging.info(f"Copying image {os.path.basename(source_path)} to {minimal_image_dir} for shrinking")
     shutil.copyfile(source_path, minimal_image_path)
 
-    loop_device = losetup_setup_image(minimal_image_path, sector_size)
+    loop_device = losetup_rootfs_image(minimal_image_path, sector_size)
     partprobe(loop_device)
-    shrink_fs(loop_device, minimal_image_path, sector_size, encrypted, encryption_password)
+    shrink_fs(loop_device, minimal_image_path, sector_size)
     losetup_destroy(loop_device)
     return minimal_image_path
 
@@ -69,7 +67,6 @@ def prepare_minimal_image(source_path: str, sector_size: int, encrypted: Optiona
 @click.option('--shrink/--no-shrink', is_flag=True, default=True, help="Copy and shrink the image file to minimal size")
 @click.option('-b', '--sector-size', type=int, help="Override the device's sector size", default=None)
 @click.option('--confirm', is_flag=True, help="Ask for confirmation before executing fastboot commands")
-@encryption_option
 @click.argument('what', type=click.Choice(list(FLASH_PARTS.values())))
 @click.argument('location', type=str, required=False)
 def cmd_flash(
@@ -81,7 +78,6 @@ def cmd_flash(
     shrink: bool = True,
     sector_size: Optional[int] = None,
     confirm: bool = False,
-    encryption: Optional[bool] = None,
 ):
     """
     Flash a partition onto a device.
@@ -119,12 +115,7 @@ def cmd_flash(
         if not location:
             raise Exception(f'You need to specify a location to flash {what} to')
         path = ''
-        image_path = prepare_minimal_image(
-            device_image_path,
-            sector_size,
-            encrypted=encryption,
-            encryption_password=config.get_profile(profile).encryption_password,
-        ) if shrink else device_image_path
+        image_path = prepare_minimal_image(device_image_path, sector_size) if shrink else device_image_path
         if method == FASTBOOT:
             fastboot_flash(
                 partition=location,
@@ -145,7 +136,7 @@ def cmd_flash(
     else:
         if method and method != FASTBOOT:
             raise Exception(f'Flashing "{what}" with method "{method}" not supported, try no parameter or "{FASTBOOT}"')
-        loop_device = losetup_setup_image(device_image_path, sector_size)
+        loop_device = losetup_rootfs_image(device_image_path, sector_size)
         if what == ABOOT:
             path = dump_aboot(f'{loop_device}p1')
             fastboot_flash(location or 'boot', path, confirm=confirm)

image/image.py

@@ -1,29 +1,29 @@
 import atexit
 import json
-import logging
 import os
 import re
 import subprocess
+import click
+import logging
+from signal import pause
 from subprocess import CompletedProcess
 from typing import Optional, Union
 
-from chroot.build import BuildChroot, get_build_chroot
-from chroot.device import DeviceChroot, get_device_chroot
 from config.state import config, Profile
-from constants import Arch, LUKS_MAPPER_DEFAULT, POST_INSTALL_CMDS
+from chroot.device import DeviceChroot, get_device_chroot
+from constants import Arch, BASE_LOCAL_PACKAGES, BASE_PACKAGES, POST_INSTALL_CMDS
 from distro.distro import get_base_distro, get_kupfer_https
-from devices.device import Device
+from devices.device import Device, get_profile_device
 from exec.cmd import run_root_cmd, generate_cmd_su
-from exec.file import get_temp_dir, root_write_file, root_makedir
-from flavours.flavour import Flavour
+from exec.file import get_temp_dir, root_write_file, root_makedir, makedir
+from flavours.flavour import Flavour, get_profile_flavour
 from net.ssh import copy_ssh_keys
-from utils import programs_available
+from packages.build import build_enable_qemu_binfmt, build_packages, filter_pkgbuilds
+from wrapper import enforce_wrap
 
-from .cryptsetup import is_luks, get_luks_offset, luks_close, luks_open
-
-MAPPER_DIR = '/dev/mapper/'
-
-IMG_DEFAULT_SIZE_BOOT_MB = 200
+# image files need to be slightly smaller than partitions to fit
+IMG_FILE_ROOT_DEFAULT_SIZE = "1800M"
+IMG_FILE_BOOT_DEFAULT_SIZE = "90M"
 
 
 def dd_image(input: str, output: str, blocksize='1M') -> CompletedProcess:
@@ -76,58 +76,24 @@ def align_bytes(size_bytes: int, alignment: int = 4096) -> int:
     return size_bytes
 
 
-def shrink_fs(
-    loop_device: str,
-    file: str,
-    sector_size: int,
-    encrypted: Optional[bool] = None,
-    encryption_password: Optional[str] = None,
-    crypt_mapper=LUKS_MAPPER_DEFAULT,
-):
+def shrink_fs(loop_device: str, file: str, sector_size: int):
     partprobe(loop_device)
-    root_partition = f'{loop_device}p2'
-    root_partition_fs = root_partition
-    if not (encrypted is False):
-        root_partition_fs, native_chroot, encrypted = resolve_rootfs_crypt(
-            root_partition,
-            fail_on_unencrypted=bool(encrypted),
-            crypt_mapper=crypt_mapper,
-            password=encryption_password,
-        )
-    logging.debug(f"Checking filesystem at {root_partition_fs}")
-    result = run_root_cmd(['e2fsck', '-fy', root_partition_fs])
+    logging.debug(f"Checking filesystem at {loop_device}p2")
+    result = run_root_cmd(['e2fsck', '-fy', f'{loop_device}p2'])
     if result.returncode > 2:
         # https://man7.org/linux/man-pages/man8/e2fsck.8.html#EXIT_CODE
-        raise Exception(f'Failed to e2fsck {root_partition_fs} with exit code {result.returncode}')
+        raise Exception(f'Failed to e2fsck {loop_device}p2 with exit code {result.returncode}')
 
-    logging.info(f'Shrinking filesystem at {root_partition_fs}')
-    result = run_root_cmd(['resize2fs', '-M', root_partition_fs])
+    logging.info(f'Shrinking filesystem at {loop_device}p2')
+    result = run_root_cmd(['resize2fs', '-M', f'{loop_device}p2'])
     if result.returncode != 0:
-        raise Exception(f'Failed to resize2fs {root_partition_fs}')
+        raise Exception(f'Failed to resize2fs {loop_device}p2')
 
-    logging.debug(f'Reading size of shrunken filesystem on {root_partition_fs}')
-    fs_blocks, fs_block_size = get_fs_size(root_partition_fs)
+    logging.debug(f'Reading size of shrunken filesystem on {loop_device}p2')
+    fs_blocks, fs_block_size = get_fs_size(f'{loop_device}p2')
     sectors = bytes_to_sectors(fs_blocks * fs_block_size, sector_size)
-    logging.debug(f"shrunken FS length is {fs_blocks} blocks * {fs_block_size} bytes = {sectors} bytes")
 
-    _, image_size = find_end_sector(loop_device, root_partition, sector_size)
-    if image_size == -1:
-        raise Exception(f'Failed to find pre-repartition size of {loop_device}')
-
-    if encrypted:
-        if sectors > image_size:
-            raise Exception("Shrunk FS size allegedly larger than the image itself; this is probably "
-                            f"a kupferbootstrap parsing bug. shrunk partition end={sectors}, image size={image_size}, {sector_size=}")
-        old_sectors = sectors
-        luks_offset, luks_sector_size = get_luks_offset(crypt_mapper, native_chroot)
-        #luks_offset_bytes = align_bytes((luks_offset + 1) * luks_sector_size, sector_size)
-        luks_offset_normalized = bytes_to_sectors(luks_offset * luks_sector_size, sector_size)
-        logging.debug(f"Discovered LUKS attrs: {luks_offset=}, {luks_sector_size=}, {luks_offset_normalized=}")
-        luks_close(crypt_mapper, native_chroot)
-        sectors += luks_offset_normalized + 1024
-        logging.debug(f"Increasing sectors from {old_sectors} to {sectors} ({sectors - old_sectors}) to leave space for the LUKS header")
-
-    logging.info(f'Shrinking partition at {root_partition} to {sectors} {sector_size}b sectors ({sectors * sector_size} bytes)')
+    logging.info(f'Shrinking partition at {loop_device}p2 to {sectors} sectors ({sectors * sector_size} bytes)')
     child_proccess = subprocess.Popen(
         generate_cmd_su(['fdisk', '-b', str(sector_size), loop_device], switch_user='root'),  # type: ignore
         stdin=subprocess.PIPE,
@@ -151,18 +117,27 @@ def shrink_fs(
         # For some reason re-reading the partition table fails, but that is not a problem
         partprobe(loop_device)
     if returncode > 1:
-        raise Exception(f'Failed to shrink partition size of {root_partition} with fdisk')
+        raise Exception(f'Failed to shrink partition size of {loop_device}p2 with fdisk')
 
     partprobe(loop_device).check_returncode()
 
-    end_sector, _ = find_end_sector(loop_device, root_partition, sector_size)
-    if end_sector == -1:
-        raise Exception(f'Failed to find end sector of {root_partition}')
+    logging.debug(f'Finding end sector of partition at {loop_device}p2')
+    result = run_root_cmd(['fdisk', '-b', str(sector_size), '-l', loop_device], capture_output=True)
+    if result.returncode != 0:
+        print(result.stdout)
+        print(result.stderr)
+        raise Exception(f'Failed to fdisk -l {loop_device}')
 
-    if end_sector > image_size:
-        logging.warning(f"Clipping sectors ({end_sector}) to {image_size=}")
-        end_sector = image_size
-    end_size = align_bytes((end_sector + 1024) * sector_size, 4096)
+    end_sector = 0
+    for line in result.stdout.decode('utf-8').split('\n'):
+        if line.startswith(f'{loop_device}p2'):
+            parts = list(filter(lambda part: part != '', line.split(' ')))
+            end_sector = int(parts[2])
+
+    if end_sector == 0:
+        raise Exception(f'Failed to find end sector of {loop_device}p2')
+
+    end_size = align_bytes((end_sector + 1) * sector_size, 4096)
 
     logging.debug(f'({end_sector} + 1) sectors * {sector_size} bytes/sector = {end_size} bytes')
     logging.info(f'Truncating {file} to {end_size} bytes')
@@ -172,26 +147,6 @@ def shrink_fs(
     partprobe(loop_device)
 
 
-def find_end_sector(device: str, partition: str, sector_size: int) -> tuple[int, int]:
-    """Return (last_sector_index, sector_count) of a partition on a device, returns (-1, -1) if not found"""
-    logging.debug(f'Finding end sector of partition at {partition}')
-    result = run_root_cmd(['fdisk', '-b', str(sector_size), '-l', device], capture_output=True)
-    if result.returncode != 0:
-        print(result.stdout)
-        print(result.stderr)
-        raise Exception(f'Failed to fdisk -l {device}')
-
-    end_sector = -1
-    num_sectors = -1
-    for line in result.stdout.decode('utf-8').split('\n'):
-        if line.startswith(partition):
-            parts = list(filter(lambda part: part != '', line.split(' ')))
-            end_sector = int(parts[2])
-            num_sectors = int(parts[3])
-
-    return end_sector, num_sectors
-
-
 def losetup_destroy(loop_device):
     logging.debug(f'Destroying loop device {loop_device}')
     run_root_cmd(
@@ -222,7 +177,7 @@ def get_image_path(device: Union[str, Device], flavour: Union[str, Flavour], img
     return os.path.join(config.get_path('images'), get_image_name(device, flavour, img_type))
 
 
-def losetup_setup_image(image_path: str, sector_size: int) -> str:
+def losetup_rootfs_image(image_path: str, sector_size: int) -> str:
     logging.debug(f'Creating loop device for {image_path} with sector size {sector_size}')
     result = run_root_cmd([
         'losetup',
@@ -259,59 +214,16 @@ def losetup_setup_image(image_path: str, sector_size: int) -> str:
     return loop_device
 
 
-def resolve_rootfs_crypt(
-    rootfs_source: str,
-    password: Optional[str] = None,
-    crypt_mapper: str = LUKS_MAPPER_DEFAULT,
-    native_chroot: Optional[BuildChroot] = None,
-    fail_on_unencrypted: bool = True,
-) -> tuple[str, Optional[BuildChroot], bool]:
-    assert config.runtime.arch
-    is_encrypted = False
-    if not (native_chroot or programs_available(['blkid'])):
-        native_chroot = get_build_chroot(config.runtime.arch, packages=['base', 'util-linux'])
-    if is_luks(rootfs_source, native_chroot=native_chroot):
-        if not (native_chroot or programs_available(['cryptsetup'])):
-            native_chroot = get_build_chroot(config.runtime.arch, packages=['base', 'cryptsetup', 'util-linux'])
-        luks_open(rootfs_source, crypt_mapper, password=password, native_chroot=native_chroot)
-        rootfs_source = f'{MAPPER_DIR}{crypt_mapper}'
-        is_encrypted = True
-    elif fail_on_unencrypted:
-        hint = ''
-        if rootfs_source.startswith(MAPPER_DIR):
-            hint = (f' HINT: path starts with {MAPPER_DIR!r}, probably already a decrypted volume.'
-                    ' This is likely a kupferbootstrap bug.')
-        raise Exception(f"Error: {rootfs_source!r} is not an encrypted LUKS volume.{hint}")
-    return rootfs_source, native_chroot, is_encrypted
+def mount_chroot(rootfs_source: str, boot_src: str, chroot: DeviceChroot):
+    logging.debug(f'Mounting {rootfs_source} at {chroot.path}')
 
+    chroot.mount_rootfs(rootfs_source)
+    assert (os.path.ismount(chroot.path))
 
-def mount_chroot(
-    rootfs_source: str,
-    boot_src: str,
-    device_chroot: DeviceChroot,
-    encrypted: Optional[bool] = None,
-    password: Optional[str] = None,
-    native_chroot: Optional[BuildChroot] = None,
-    crypt_mapper: str = LUKS_MAPPER_DEFAULT,
-):
-    if encrypted is not False:
-        rootfs_source, native_chroot, encrypted = resolve_rootfs_crypt(
-            rootfs_source,
-            native_chroot=native_chroot,
-            crypt_mapper=crypt_mapper,
-            fail_on_unencrypted=bool(encrypted),
-            password=password,
-        )
+    root_makedir(chroot.get_path('boot'))
 
-    logging.debug(f'Mounting {rootfs_source} at {device_chroot.path}')
-
-    device_chroot.mount_rootfs(rootfs_source)
-    assert (os.path.ismount(device_chroot.path))
-
-    root_makedir(device_chroot.get_path('boot'))
-
-    logging.debug(f'Mounting {boot_src} at {device_chroot.path}/boot')
-    device_chroot.mount(boot_src, '/boot', options=['defaults'])
+    logging.debug(f'Mounting {boot_src} at {chroot.path}/boot')
+    chroot.mount(boot_src, '/boot', options=['defaults'])
 
 
 def dump_file_from_image(image_path: str, file_path: str, target_path: Optional[str] = None):
@@ -354,12 +266,11 @@ def create_img_file(image_path: str, size_str: str):
     return image_path
 
 
-def partition_device(device: str, sector_size: int, boot_partition_size_mb: int = IMG_DEFAULT_SIZE_BOOT_MB):
-    initial_offset = 1048576 // sector_size  # 2048 for 512, 256 for 4096
-    boot_partition_size: int = align_bytes(boot_partition_size_mb * 1024 * 1024, 4096)
+def partition_device(device: str):
+    boot_partition_size = '100MiB'
     create_partition_table = ['mklabel', 'msdos']
-    create_boot_partition = ['mkpart', 'primary', 'ext2', f'{initial_offset}s', f'{boot_partition_size}b']
-    create_root_partition = ['mkpart', 'primary', f'{bytes_to_sectors(boot_partition_size, sector_size) + initial_offset}s', '100%']
+    create_boot_partition = ['mkpart', 'primary', 'ext2', '0%', boot_partition_size]
+    create_root_partition = ['mkpart', 'primary', boot_partition_size, '100%']
     enable_boot = ['set', '1', 'boot', 'on']
     result = run_root_cmd([
         'parted',
@@ -370,7 +281,7 @@ def partition_device(device: str, sector_size: int, boot_partition_size_mb: int
         raise Exception(f'Failed to create partitions on {device}')
 
 
-def create_filesystem(device: str, blocksize: Optional[int] = None, label=None, options=[], fstype='ext4'):
+def create_filesystem(device: str, blocksize: Optional[int], label=None, options=[], fstype='ext4'):
     """Creates a new filesystem. Blocksize defaults"""
     labels = ['-L', label] if label else []
     cmd = [f'mkfs.{fstype}', '-F', *labels]
@@ -390,12 +301,12 @@ def create_filesystem(device: str, blocksize: Optional[int] = None, label=None,
         raise Exception(f'Failed to create {fstype} filesystem on {device} with CMD: {cmd}')
 
 
-def create_root_fs(device: str):
-    create_filesystem(device, label='kupfer_root', options=['-O', '^metadata_csum', '-N', '100000'])
+def create_root_fs(device: str, blocksize: Optional[int]):
+    create_filesystem(device, blocksize=blocksize, label='kupfer_root', options=['-O', '^metadata_csum', '-N', '100000'])
 
 
-def create_boot_fs(device: str):
-    create_filesystem(device, label='kupfer_boot', fstype='ext2')
+def create_boot_fs(device: str, blocksize: Optional[int]):
+    create_filesystem(device, blocksize=blocksize, label='kupfer_boot', fstype='ext2')
 
 
 def install_rootfs(
@@ -407,33 +318,24 @@ def install_rootfs(
     packages: list[str],
     use_local_repos: bool,
     profile: Profile,
-    encrypted: bool,
 ):
-    user = profile.username or 'kupfer'
+    user = profile['username'] or 'kupfer'
     chroot = get_device_chroot(device=get_device_name(device), flavour=flavour.name, arch=arch, packages=packages, use_local_repos=use_local_repos)
 
-    # rootfs_device must be passed the crypt_mapper if encrypted is True
-    if encrypted:
-        assert rootfs_device.startswith(MAPPER_DIR)
-
-    mount_chroot(
-        rootfs_device,
-        bootfs_device,
-        chroot,
-        encrypted=False,  # rootfs_device is already the crypt_mapper
-    )
+    mount_chroot(rootfs_device, bootfs_device, chroot)
 
     chroot.mount_pacman_cache()
     chroot.initialize()
     chroot.activate()
     chroot.create_user(
         user=user,
-        password=profile.password,
+        password=profile['password'],
     )
     chroot.add_sudo_config(config_name='wheel', privilegee='%wheel', password_required=True)
     copy_ssh_keys(
-        chroot.path,
+        chroot,
         user=user,
+        allow_fail=True,
     )
     files = {
         'etc/pacman.conf': get_base_distro(arch).get_pacman_conf(
@@ -441,7 +343,7 @@ def install_rootfs(
             extra_repos=get_kupfer_https(arch).repos,
             in_chroot=True,
         ),
-        'etc/hostname': profile.hostname or 'kupfer',
+        'etc/hostname': profile['hostname'] or 'kupfer',
     }
     for target, content in files.items():
         root_write_file(os.path.join(chroot.path, target.lstrip('/')), content)
@@ -462,3 +364,177 @@ def install_rootfs(
     res = run_root_cmd(['umount', chroot.path])
     assert isinstance(res, CompletedProcess)
     logging.debug(f'rc: {res.returncode}')
+
+
+@click.group(name='image')
+def cmd_image():
+    """Build, flash and boot device images"""
+
+
+sectorsize_option = click.option(
+    '-b',
+    '--sector-size',
+    help="Override the device's sector size",
+    type=int,
+    default=None,
+)
+
+
+@cmd_image.command(name='build')
+@click.argument('profile_name', required=False)
+@click.option(
+    '--local-repos/--no-local-repos',
+    '-l/-L',
+    help='Whether to use local package repos at all or only use HTTPS repos.',
+    default=True,
+    show_default=True,
+    is_flag=True,
+)
+@click.option(
+    '--build-pkgs/--no-build-pkgs',
+    '-p/-P',
+    help='Whether to build missing/outdated local packages if local repos are enabled.',
+    default=True,
+    show_default=True,
+    is_flag=True,
+)
+@click.option(
+    '--no-download-pkgs',
+    help='Disable trying to download packages instead of building if building is enabled.',
+    default=False,
+    is_flag=True,
+)
+@click.option(
+    '--block-target',
+    help='Override the block device file to write the final image to',
+    type=click.Path(),
+    default=None,
+)
+@click.option(
+    '--skip-part-images',
+    help='Skip creating image files for the partitions and directly work on the target block device.',
+    default=False,
+    is_flag=True,
+)
+@sectorsize_option
+def cmd_build(
+    profile_name: Optional[str] = None,
+    local_repos: bool = True,
+    build_pkgs: bool = True,
+    no_download_pkgs=False,
+    block_target: Optional[str] = None,
+    sector_size: Optional[int] = None,
+    skip_part_images: bool = False,
+):
+    """
+    Build a device image.
+
+    Unless overriden, required packages will be built or preferably downloaded from HTTPS repos.
+    """
+
+    config.enforce_profile_device_set()
+    config.enforce_profile_flavour_set()
+    enforce_wrap()
+    device = get_profile_device(profile_name)
+    arch = device.arch
+    # check_programs_wrap(['makepkg', 'pacman', 'pacstrap'])
+    profile: Profile = config.get_profile(profile_name)
+    flavour = get_profile_flavour(profile_name)
+    rootfs_size_mb = flavour.parse_flavourinfo().rootfs_size * 1000 + int(profile.size_extra_mb)
+
+    packages = BASE_LOCAL_PACKAGES + [device.package.name, flavour.pkgbuild.name]
+    packages_extra = BASE_PACKAGES + profile.pkgs_include
+
+    if arch != config.runtime.arch:
+        build_enable_qemu_binfmt(arch)
+
+    if local_repos and build_pkgs:
+        logging.info("Making sure all packages are built")
+        # enforce that local base packages are built
+        pkgbuilds = set(filter_pkgbuilds(packages, arch=arch, allow_empty_results=False, use_paths=False))
+        # extra packages might be a mix of package names that are in our PKGBUILDs and packages from the base distro
+        pkgbuilds |= set(filter_pkgbuilds(packages_extra, arch=arch, allow_empty_results=True, use_paths=False))
+        build_packages(pkgbuilds, arch, try_download=not no_download_pkgs)
+
+    sector_size = sector_size or device.get_image_sectorsize()
+
+    image_path = block_target or get_image_path(device, flavour.name)
+
+    makedir(os.path.dirname(image_path))
+
+    logging.info(f'Creating new file at {image_path}')
+    create_img_file(image_path, f"{rootfs_size_mb}M")
+
+    loop_device = losetup_rootfs_image(image_path, sector_size or device.get_image_sectorsize_default())
+
+    partition_device(loop_device)
+    partprobe(loop_device)
+
+    boot_dev: str
+    root_dev: str
+    loop_boot = loop_device + 'p1'
+    loop_root = loop_device + 'p2'
+    if skip_part_images:
+        boot_dev = loop_boot
+        root_dev = loop_root
+    else:
+        logging.info('Creating per-partition image files')
+        boot_dev = create_img_file(get_image_path(device, flavour, 'boot'), IMG_FILE_BOOT_DEFAULT_SIZE)
+        root_dev = create_img_file(get_image_path(device, flavour, 'root'), f'{rootfs_size_mb - 200}M')
+
+    create_boot_fs(boot_dev, sector_size)
+    create_root_fs(root_dev, sector_size)
+
+    install_rootfs(
+        root_dev,
+        boot_dev,
+        device,
+        flavour,
+        arch,
+        list(set(packages) | set(packages_extra)),
+        local_repos,
+        profile,
+    )
+
+    if not skip_part_images:
+        logging.info('Copying partition image files into full image:')
+        logging.info(f'Block-copying /boot to {image_path}')
+        dd_image(input=boot_dev, output=loop_boot)
+        logging.info(f'Block-copying rootfs to {image_path}')
+        dd_image(input=root_dev, output=loop_root)
+
+    logging.info(f'Done! Image saved to {image_path}')
+
+
+@cmd_image.command(name='inspect')
+@click.option('--shell', '-s', is_flag=True)
+@sectorsize_option
+@click.argument('profile', required=False)
+def cmd_inspect(profile: Optional[str] = None, shell: bool = False, sector_size: Optional[int] = None):
+    """Loop-mount the device image for inspection."""
+    config.enforce_profile_device_set()
+    config.enforce_profile_flavour_set()
+    enforce_wrap()
+    device = get_profile_device(profile)
+    arch = device.arch
+    flavour = get_profile_flavour(profile).name
+    sector_size = sector_size or device.get_image_sectorsize_default()
+
+    chroot = get_device_chroot(device.name, flavour, arch)
+    image_path = get_image_path(device, flavour)
+    loop_device = losetup_rootfs_image(image_path, sector_size)
+    partprobe(loop_device)
+    mount_chroot(loop_device + 'p2', loop_device + 'p1', chroot)
+
+    logging.info(f'Inspect the rootfs image at {chroot.path}')
+
+    if shell:
+        chroot.initialized = True
+        chroot.activate()
+        if arch != config.runtime.arch:
+            logging.info('Installing requisites for foreign-arch shell')
+            build_enable_qemu_binfmt(arch)
+        logging.info('Starting inspection shell')
+        chroot.run_cmd('/bin/bash')
+    else:
+        pause()

integration_tests.py

@@ -37,6 +37,11 @@ def ctx() -> click.Context:
     return click.Context(click.Command('integration_tests'))
 
 
+def test_main_import():
+    from main import cli
+    assert cli
+
+
 def test_config_load(ctx: click.Context):
     path = config.runtime.config_file
     assert path

net/ssh.py

@@ -6,7 +6,9 @@ import click
 
 from config.state import config
 from constants import SSH_COMMON_OPTIONS, SSH_DEFAULT_HOST, SSH_DEFAULT_PORT
+from chroot.abstract import Chroot
 from exec.cmd import run_cmd
+from exec.file import write_file
 from wrapper import check_programs_wrap
 
 
@@ -83,21 +85,16 @@ def find_ssh_keys():
     return keys
 
 
-def copy_ssh_keys(root_dir: str, user: str):
+def copy_ssh_keys(chroot: Chroot, user: str, allow_fail: bool = False):
     check_programs_wrap(['ssh-keygen'])
-    authorized_keys_file = os.path.join(
-        root_dir,
-        'home',
-        user,
-        '.ssh',
-        'authorized_keys',
-    )
-    if os.path.exists(authorized_keys_file):
-        os.unlink(authorized_keys_file)
+    ssh_dir_relative = os.path.join('/home', user, '.ssh')
+    ssh_dir = chroot.get_path(ssh_dir_relative)
+    authorized_keys_file_rel = os.path.join(ssh_dir_relative, 'authorized_keys')
+    authorized_keys_file = chroot.get_path(authorized_keys_file_rel)
 
     keys = find_ssh_keys()
     if len(keys) == 0:
-        logging.info("Could not find any ssh key to copy")
+        logging.warning("Could not find any ssh key to copy")
         create = click.confirm("Do you want me to generate an ssh key for you?", True)
         if not create:
             return
@@ -116,15 +113,34 @@ def copy_ssh_keys(root_dir: str, user: str):
             logging.fatal("Failed to generate ssh key")
         keys = find_ssh_keys()
 
-    ssh_dir = os.path.join(root_dir, 'home', user, '.ssh')
-    if not os.path.exists(ssh_dir):
-        os.makedirs(ssh_dir, exist_ok=True, mode=0o700)
+    if not keys:
+        logging.warning("No SSH keys to be copied. Skipping.")
+        return
 
-    with open(authorized_keys_file, 'a') as authorized_keys:
-        for key in keys:
-            pub = f'{key}.pub'
-            if not os.path.exists(pub):
-                logging.debug(f'Skipping key {key}: {pub} not found')
-                continue
+    auth_key_lines = []
+    for key in keys:
+        pub = f'{key}.pub'
+        if not os.path.exists(pub):
+            logging.debug(f'Skipping key {key}: {pub} not found')
+            continue
+        try:
             with open(pub, 'r') as file:
-                authorized_keys.write(file.read())
+                contents = file.read()
+                if not contents.strip():
+                    continue
+                auth_key_lines.append(contents)
+        except Exception as ex:
+            logging.warning(f"Could not read ssh pub key {pub}", exc_info=ex)
+            continue
+
+    if not os.path.exists(ssh_dir):
+        logging.info(f"Creating {ssh_dir_relative!r} dir in chroot {chroot.path!r}")
+        chroot.run_cmd(["mkdir", "-p", "-m", "700", ssh_dir_relative], switch_user=user)
+    logging.info(f"Writing SSH pub keys to {authorized_keys_file}")
+    try:
+        write_file(authorized_keys_file, "\n".join(auth_key_lines), user=str(chroot.get_uid(user)), mode="644")
+    except Exception as ex:
+        logging.error(f"Failed to write SSH authorized_keys_file at {authorized_keys_file!r}:", exc_info=ex)
+        if allow_fail:
+            return
+        raise ex from ex

packages/build.py

@@ -575,7 +575,7 @@ def build_package(
         build_root = target_chroot
         makepkg_compile_opts += ['--nodeps' if package.nodeps else '--syncdeps']
         env = deepcopy(get_makepkg_env(arch))
-        if foreign_arch and enable_crossdirect and package.name not in CROSSDIRECT_PKGS:
+        if foreign_arch and package.crossdirect and enable_crossdirect and package.name not in CROSSDIRECT_PKGS:
             env['PATH'] = f"/native/usr/lib/crossdirect/{arch}:{env['PATH']}"
             target_chroot.mount_crossdirect(native_chroot)
         else:

packages/cli.py

@@ -313,7 +313,7 @@ def cmd_list():
     logging.info(f'Done! {len(packages)} Pkgbuilds:')
     for name in sorted(packages.keys()):
         p = packages[name]
-        print(f'name: {p.name}; ver: {p.version}; mode: {p.mode}; provides: {p.provides}; replaces: {p.replaces};'
+        print(f'name: {p.name}; ver: {p.version}; mode: {p.mode}; crossdirect: {p.crossdirect} provides: {p.provides}; replaces: {p.replaces};'
               f'local_depends: {p.local_depends}; depends: {p.depends}')
 
 
@@ -346,6 +346,7 @@ def cmd_check(paths):
 
         mode_key = '_mode'
         nodeps_key = '_nodeps'
+        crossdirect_key = '_crossdirect'
         pkgbase_key = 'pkgbase'
         pkgname_key = 'pkgname'
         arches_key = '_arches'
@@ -356,6 +357,7 @@ def cmd_check(paths):
         required = {
             mode_key: True,
             nodeps_key: False,
+            crossdirect_key: False,
             pkgbase_key: False,
             pkgname_key: True,
             'pkgdesc': False,

packages/pkgbuild.py

@@ -156,6 +156,7 @@ class Pkgbuild(PackageInfo):
     repo: str
     mode: str
     nodeps: bool
+    crossdirect: bool
     path: str
     pkgver: str
     pkgrel: str
@@ -190,6 +191,7 @@ class Pkgbuild(PackageInfo):
         self.repo = repo or ''
         self.mode = ''
         self.nodeps = False
+        self.crossdirect = True
         self.path = relative_path
         self.pkgver = ''
         self.pkgrel = ''
@@ -223,6 +225,7 @@ class Pkgbuild(PackageInfo):
         self.repo = pkg.repo
         self.mode = pkg.mode
         self.nodeps = pkg.nodeps
+        self.crossdirect = pkg.crossdirect
         self.path = pkg.path
         self.pkgver = pkg.pkgver
         self.pkgrel = pkg.pkgrel
@@ -357,7 +360,11 @@ def parse_pkgbuild(
         else:
             raise Exception(msg)
 
+    # if _crossdirect is unset (None), it defaults to True
+    crossdirect_enabled = srcinfo_cache.build_crossdirect in (None, True)
+
     base_package = Pkgbase(relative_pkg_dir, sources_refreshed=sources_refreshed, srcinfo_cache=srcinfo_cache)
+    base_package.crossdirect = crossdirect_enabled
     base_package.mode = mode
     base_package.nodeps = nodeps
     base_package.repo = relative_pkg_dir.split('/')[0]
@@ -396,7 +403,7 @@ def parse_pkgbuild(
         elif splits[0] in ['depends', 'makedepends', 'checkdepends', 'optdepends']:
             spec = splits[1].split(': ', 1)[0]
             if not current.depends:
-                current.depends = {}
+                current.depends = (base_package.makedepends or {}).copy()
             current.depends = get_version_specs(spec, current.depends)
             if splits[0] == 'makedepends':
                 if not current.makedepends:

packages/srcinfo_cache.py

@@ -68,11 +68,19 @@ class SrcInitialisedFile(JsonFile):
                 raise ex
 
 
+srcinfo_meta_defaults = {
+    'build_mode': None,
+    "build_nodeps": None,
+    "build_crossdirect": None,
+}
+
+
 class SrcinfoMetaFile(JsonFile):
 
     checksums: dict[str, str]
     build_mode: Optional[str]
     build_nodeps: Optional[bool]
+    build_crossdirect: Optional[bool]
 
     _changed: bool
     _filename: ClassVar[str] = SRCINFO_METADATA_FILE
@@ -92,9 +100,8 @@ class SrcinfoMetaFile(JsonFile):
         s = SrcinfoMetaFile({
             '_relative_path': relative_pkg_dir,
             '_changed': True,
-            'build_mode': '',
-            'build_nodeps': None,
             'checksums': {},
+            **srcinfo_meta_defaults,
         })
         return s, s.refresh_all()
 
@@ -120,9 +127,11 @@ class SrcinfoMetaFile(JsonFile):
         if not force_refresh:
             logging.debug(f'{metadata._relative_path}: srcinfo checksums match!')
             lines = lines or metadata.read_srcinfo_file()
-            for build_field in ['build_mode', 'build_nodeps']:
+            for build_field in srcinfo_meta_defaults.keys():
                 if build_field not in metadata:
                     metadata.refresh_build_fields()
+                    if write:
+                        metadata.write()
                     break
         else:
             lines = metadata.refresh_all(write=write)
@@ -143,8 +152,7 @@ class SrcinfoMetaFile(JsonFile):
             self._changed = True
 
     def refresh_build_fields(self):
-        self['build_mode'] = None
-        self['build_nodeps'] = None
+        self.update(srcinfo_meta_defaults)
         with open(os.path.join(config.get_path('pkgbuilds'), self._relative_path, 'PKGBUILD'), 'r') as file:
             lines = file.read().split('\n')
         for line in lines:
@@ -156,6 +164,8 @@ class SrcinfoMetaFile(JsonFile):
                 self.build_mode = val
             elif key == '_nodeps':
                 self.build_nodeps = val.lower() == 'true'
+            elif key == '_crossdirect':
+                self.build_crossdirect = val.lower() == 'true'
             else:
                 continue