image/image.py: Fix off-by-one error when creating boot partition in full image

chroot/abstract.py

@@ -60,8 +60,12 @@ 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
 
@@ -225,9 +229,12 @@ 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,
-        switch_user: Optional[str] = None,
+        stdin: Optional[FileDescriptor] = None,
+        stdin_input: Optional[str] = None,
+        check: Optional[bool] = 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`')
@@ -244,13 +251,23 @@ 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:
+        if switch_user and switch_user != 'root':
             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)
+        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,
+        )
 
     def mount_pkgbuilds(self, fail_if_mounted: bool = False) -> str:
         return self.mount(

chroot/cli.py

@@ -7,6 +7,7 @@ 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
@@ -30,7 +31,6 @@ 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,5 +1,6 @@
 import click
 import logging
+import os
 
 from copy import deepcopy
 from typing import Any, Callable, Iterable, Mapping, Optional, Union
@@ -260,6 +261,7 @@ 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:
@@ -282,6 +284,13 @@ 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,6 +15,8 @@ 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,3 +172,6 @@ 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,6 +97,9 @@ 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 = {}
@@ -111,6 +114,10 @@ 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

@@ -171,9 +171,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):
+def get_temp_dir(register_cleanup=True, mode: int = 0o0755, prefix='kupfertmp_'):
     "create a new tempdir and sanitize ownership so root can access user files as god intended"
-    t = mkdtemp()
+    t = mkdtemp(prefix=prefix)
     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_rootfs_image, get_image_path, dump_aboot, dump_lk2nd
+from .image import get_device_name, losetup_setup_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_rootfs_image(image_path, sector_size)
+            loop_device = losetup_setup_image(image_path, sector_size)
             if type == LK2ND:
                 path = dump_lk2nd(loop_device + 'p1')
             elif type == ABOOT:

image/cli.py

@@ -1,6 +1,248 @@
+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 cmd_image
+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"""
+
 
 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

@@ -0,0 +1,179 @@
+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,6 +5,7 @@ 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
@@ -14,7 +15,8 @@ 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_rootfs_image, partprobe, shrink_fs
+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
 
 ABOOT = FLASH_PARTS['ABOOT']
 LK2ND = FLASH_PARTS['LK2ND']
@@ -47,15 +49,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) -> str:
+def prepare_minimal_image(source_path: str, sector_size: int, encrypted: Optional[bool], encryption_password: Optional[str]) -> 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_rootfs_image(minimal_image_path, sector_size)
+    loop_device = losetup_setup_image(minimal_image_path, sector_size)
     partprobe(loop_device)
-    shrink_fs(loop_device, minimal_image_path, sector_size)
+    shrink_fs(loop_device, minimal_image_path, sector_size, encrypted, encryption_password)
     losetup_destroy(loop_device)
     return minimal_image_path
 
@@ -67,6 +69,7 @@ def prepare_minimal_image(source_path: str, sector_size: int) -> str:
 @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(
@@ -78,6 +81,7 @@ def cmd_flash(
     shrink: bool = True,
     sector_size: Optional[int] = None,
     confirm: bool = False,
+    encryption: Optional[bool] = None,
 ):
     """
     Flash a partition onto a device.
@@ -115,7 +119,12 @@ 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) if shrink else device_image_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
         if method == FASTBOOT:
             fastboot_flash(
                 partition=location,
@@ -136,7 +145,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_rootfs_image(device_image_path, sector_size)
+        loop_device = losetup_setup_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 config.state import config, Profile
+from chroot.build import BuildChroot, get_build_chroot
 from chroot.device import DeviceChroot, get_device_chroot
-from constants import Arch, BASE_LOCAL_PACKAGES, BASE_PACKAGES, POST_INSTALL_CMDS
+from config.state import config, Profile
+from constants import Arch, LUKS_MAPPER_DEFAULT, POST_INSTALL_CMDS
 from distro.distro import get_base_distro, get_kupfer_https
-from devices.device import Device, get_profile_device
+from devices.device import Device
 from exec.cmd import run_root_cmd, generate_cmd_su
-from exec.file import get_temp_dir, root_write_file, root_makedir, makedir
+from exec.file import get_temp_dir, root_write_file, root_makedir
-from flavours.flavour import Flavour, get_profile_flavour
+from flavours.flavour import Flavour
 from net.ssh import copy_ssh_keys
-from packages.build import build_enable_qemu_binfmt, build_packages, filter_pkgbuilds
+from utils import programs_available
-from wrapper import enforce_wrap
 
-# image files need to be slightly smaller than partitions to fit
+from .cryptsetup import is_luks, get_luks_offset, luks_close, luks_open
-IMG_FILE_ROOT_DEFAULT_SIZE = "1800M"
+
-IMG_FILE_BOOT_DEFAULT_SIZE = "90M"
+MAPPER_DIR = '/dev/mapper/'
+
+IMG_DEFAULT_SIZE_BOOT_MB = 200
 
 
 def dd_image(input: str, output: str, blocksize='1M') -> CompletedProcess:
@@ -76,24 +76,58 @@ def align_bytes(size_bytes: int, alignment: int = 4096) -> int:
     return size_bytes
 
 
-def shrink_fs(loop_device: str, file: str, sector_size: int):
+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,
+):
     partprobe(loop_device)
-    logging.debug(f"Checking filesystem at {loop_device}p2")
+    root_partition = f'{loop_device}p2'
-    result = run_root_cmd(['e2fsck', '-fy', 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])
     if result.returncode > 2:
         # https://man7.org/linux/man-pages/man8/e2fsck.8.html#EXIT_CODE
-        raise Exception(f'Failed to e2fsck {loop_device}p2 with exit code {result.returncode}')
+        raise Exception(f'Failed to e2fsck {root_partition_fs} with exit code {result.returncode}')
 
-    logging.info(f'Shrinking filesystem at {loop_device}p2')
+    logging.info(f'Shrinking filesystem at {root_partition_fs}')
-    result = run_root_cmd(['resize2fs', '-M', f'{loop_device}p2'])
+    result = run_root_cmd(['resize2fs', '-M', root_partition_fs])
     if result.returncode != 0:
-        raise Exception(f'Failed to resize2fs {loop_device}p2')
+        raise Exception(f'Failed to resize2fs {root_partition_fs}')
 
-    logging.debug(f'Reading size of shrunken filesystem on {loop_device}p2')
+    logging.debug(f'Reading size of shrunken filesystem on {root_partition_fs}')
-    fs_blocks, fs_block_size = get_fs_size(f'{loop_device}p2')
+    fs_blocks, fs_block_size = get_fs_size(root_partition_fs)
     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")
 
-    logging.info(f'Shrinking partition at {loop_device}p2 to {sectors} sectors ({sectors * sector_size} 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)')
     child_proccess = subprocess.Popen(
         generate_cmd_su(['fdisk', '-b', str(sector_size), loop_device], switch_user='root'),  # type: ignore
         stdin=subprocess.PIPE,
@@ -117,27 +151,18 @@ def shrink_fs(loop_device: str, file: str, sector_size: int):
         # 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 {loop_device}p2 with fdisk')
+        raise Exception(f'Failed to shrink partition size of {root_partition} with fdisk')
 
     partprobe(loop_device).check_returncode()
 
-    logging.debug(f'Finding end sector of partition at {loop_device}p2')
+    end_sector, _ = find_end_sector(loop_device, root_partition, sector_size)
-    result = run_root_cmd(['fdisk', '-b', str(sector_size), '-l', loop_device], capture_output=True)
+    if end_sector == -1:
-    if result.returncode != 0:
+        raise Exception(f'Failed to find end sector of {root_partition}')
-        print(result.stdout)
-        print(result.stderr)
-        raise Exception(f'Failed to fdisk -l {loop_device}')
 
-    end_sector = 0
+    if end_sector > image_size:
-    for line in result.stdout.decode('utf-8').split('\n'):
+        logging.warning(f"Clipping sectors ({end_sector}) to {image_size=}")
-        if line.startswith(f'{loop_device}p2'):
+        end_sector = image_size
-            parts = list(filter(lambda part: part != '', line.split(' ')))
+    end_size = align_bytes((end_sector + 1024) * sector_size, 4096)
-            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')
@@ -147,6 +172,26 @@ def shrink_fs(loop_device: str, file: str, sector_size: int):
     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(
@@ -177,7 +222,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_rootfs_image(image_path: str, sector_size: int) -> str:
+def losetup_setup_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',
@@ -214,16 +259,59 @@ def losetup_rootfs_image(image_path: str, sector_size: int) -> str:
     return loop_device
 
 
-def mount_chroot(rootfs_source: str, boot_src: str, chroot: DeviceChroot):
+def resolve_rootfs_crypt(
-    logging.debug(f'Mounting {rootfs_source} at {chroot.path}')
+    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
 
-    chroot.mount_rootfs(rootfs_source)
-    assert (os.path.ismount(chroot.path))
 
-    root_makedir(chroot.get_path('boot'))
+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,
+        )
 
-    logging.debug(f'Mounting {boot_src} at {chroot.path}/boot')
+    logging.debug(f'Mounting {rootfs_source} at {device_chroot.path}')
-    chroot.mount(boot_src, '/boot', options=['defaults'])
+
+    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'])
 
 
 def dump_file_from_image(image_path: str, file_path: str, target_path: Optional[str] = None):
@@ -266,11 +354,12 @@ def create_img_file(image_path: str, size_str: str):
     return image_path
 
 
-def partition_device(device: str):
+def partition_device(device: str, sector_size: int, boot_partition_size_mb: int = IMG_DEFAULT_SIZE_BOOT_MB):
-    boot_partition_size = '100MiB'
+    initial_offset = 1048576 // sector_size  # 2048 for 512, 256 for 4096
+    boot_partition_size: int = align_bytes((boot_partition_size_mb + 1) * 1024 * 1024, 4096)
     create_partition_table = ['mklabel', 'msdos']
-    create_boot_partition = ['mkpart', 'primary', 'ext2', '0%', boot_partition_size]
+    create_boot_partition = ['mkpart', 'primary', 'ext2', f'{initial_offset}s', f'{boot_partition_size}b']
-    create_root_partition = ['mkpart', 'primary', boot_partition_size, '100%']
+    create_root_partition = ['mkpart', 'primary', f'{bytes_to_sectors(boot_partition_size, sector_size) + initial_offset}s', '100%']
     enable_boot = ['set', '1', 'boot', 'on']
     result = run_root_cmd([
         'parted',
@@ -281,7 +370,7 @@ def partition_device(device: str):
         raise Exception(f'Failed to create partitions on {device}')
 
 
-def create_filesystem(device: str, blocksize: Optional[int], label=None, options=[], fstype='ext4'):
+def create_filesystem(device: str, blocksize: Optional[int] = None, label=None, options=[], fstype='ext4'):
     """Creates a new filesystem. Blocksize defaults"""
     labels = ['-L', label] if label else []
     cmd = [f'mkfs.{fstype}', '-F', *labels]
@@ -301,12 +390,12 @@ def create_filesystem(device: str, blocksize: Optional[int], label=None, options
         raise Exception(f'Failed to create {fstype} filesystem on {device} with CMD: {cmd}')
 
 
-def create_root_fs(device: str, blocksize: Optional[int]):
+def create_root_fs(device: str):
-    create_filesystem(device, blocksize=blocksize, label='kupfer_root', options=['-O', '^metadata_csum', '-N', '100000'])
+    create_filesystem(device, label='kupfer_root', options=['-O', '^metadata_csum', '-N', '100000'])
 
 
-def create_boot_fs(device: str, blocksize: Optional[int]):
+def create_boot_fs(device: str):
-    create_filesystem(device, blocksize=blocksize, label='kupfer_boot', fstype='ext2')
+    create_filesystem(device, label='kupfer_boot', fstype='ext2')
 
 
 def install_rootfs(
@@ -318,18 +407,28 @@ 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)
 
-    mount_chroot(rootfs_device, bootfs_device, chroot)
+    # 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
+    )
 
     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(
@@ -342,7 +441,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)
@@ -363,177 +462,3 @@ 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()