FreedomBox/container

#!/usr/bin/python3
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Script to manage systemd-nspawn container for FreedomBox development.

This script creates a simple container using systemd-nspawn for developing
FreedomBox. It has many advantages over running a VM using Vagrant. RAM is
allocated to processes in the container as needed without any fixed limit. Also
RAM does not have to be statically allocated so it is typically much lighter
than running an VM. There is no hardware emulation when running a container
with same architecture, so processes run as fast as they would on the host
machine.

Environment: The script will run only run on hosts having systemd-nspawn and
network-manager installed, typical GNU/Linux distributions. It has been
primarily developed and tested on Debian Buster but should work on most modern
GNU/Linux distributions.

Disk image: systemd-nspawn accepts not only a directory for starting a
container but also a disk image. This disk image is loop-back mounted and
container is started from that mounted directory. The partition to use is
determined by looking at the boot flag in the partition table. This happens to
work well with all existing FreedomBox images. In future, we may be able to run
different architectures in this manner.

After downloading, the disk image is expanded along with the partition and file
system inside so that development can be done without running into disk space
issues. Expanding the disk does not immediately consume disk space because it
will be a sparse file. As data is written to the disk, it will occupy more and
more space but the upper limit is the size to which disk has been expanded.

Downloading images: Images are downloaded from FreedomBox download server using
fixed URLs for each distribution. Signature is verified for the download
images. The fingerprint of the allowed signing key is hard-coded in this
script. Downloaded images are kept even after destroying the extracted raw
image along with container. This allows for quickly resetting the container
without downloading again.

Booting: systemd-nspawn is run in 'boot' mode. This means that init process
(happens to be systemd) is started inside the container. It then spawns all the
other necessary daemons including openssh-server, firewalld and
network-manager. A login terminal can be opened using 'machinectl login'
because container is running systemd. SSH into the container is possible
because network is up, configured by network-manager, and openssh server is
running.

Shared folder: Using systemd-nspawn, the project directory is mounted as
/freedombox inside the container. The project directory is determined as
directory in which this script resides. The project folder from the container
point of view will be read-only. Container should be able to write various
files such as build files into the /freedombox folder. To enable writing, an
additional read-write folder is overlayed onto /freedombox folder in the
container. This directory can't be created under the project folder and is
created instead in $XDG_DATA_HOME/freedombox-container/overlay/$DISTRIBUTION.
If XDG_DATA_HOME is not set, it is assumed to be $HOME/.local/shared/. Whenever
data is written into /freedombox directory inside the container, this directory
on the host receives the changes. See documentation for Overlay filesystem for
further details. When container is destroyed, this overlay folder is destroyed
to ensure clean state after bringing up the container again.

Users: PrivateUsers configuration flag for systemd-nspawn is currently off.
This means that each user's UID on the host is also the same UID in the
container as along as there is an entry in the container's password database.
In future, we may explore using private users inside the container.

'fbx' is the development user and its UID is changed during setup phase to
10000 hoping it would not match anything on the host system. 'fbx' user has
full sudo access inside the container without needing a password. Password for
this user is not set by default, but can be set if needed. If there is no
access to the container in any way, one can run 'sudo machinectl shell' and
then run 'passwd fbx' to set the password for the 'fbx' user.

'plinth' user's UID in the container is also changed and set to the UID of
whichever user owns the project directory. This allows the files to written by
'plinth' container user in the project directory because UID of the owner of
the directory is same as the 'plinth' user's UID in container.

Network: A private network is created inside the container using systemd-nspawn
feature. Network interfaces from the host are not available inside the
container. A new network interface called 'host0' is configured inside the
container which is automatically configured by network-manager. On the host a
new network interface is created. This script creates configuration for a
'shared' network using network-manager. When bringing up the container, this
network connection is also brought up. A DHCP server and a DNS server are
started network-manager on the host side so that DHCP and DNS client functions
work inside the container. Traffic from the container is also masqueraded so
that Internet connectivity inside the container works if the host has one.

If necessary, the network interface on host side can be differently configured.
For example, it can be bridged with another interface to expose the container
on a network that the host machine participates in.

The network IP address inside the container can be queried using machinectl.
This script queries that IP address and presents the address in its console
messages. All ports in the container can be reached from the host using this IP
address as long as the firewall inside the container allows it. There is no
need to perform port forwarding or mapping.

SSH: It is assumed that openssh-server is installed inside the container. SSH
server keys in the container are created if missing. Client side keys are
created in .container/ssh directory and the public key is installed in the
authorized keys file of the 'fbx' user. The 'ssh' sub-command to this script is
simply a convenience mechanism for quick launch of ssh with the right IP
address, user name and identity file.

Role of machinectl: Most of the work is done by systemd-nspawn. machinectl is
useful for running systemd-nspawn in the background and querying its current
state. It also helps with providing the IP address of the container. machinectl
is made to recognize the container by creating a link in /var/lib/machines/ to
the image file. systemd-nspawn options are added by creating a temporary file
in /run/systemd/nspawn. All machinectl commands should work.

"""

import argparse
import datetime
import ipaddress
import itertools
import json
import logging
import os
import pathlib
import platform
import re
import shlex
import shutil
import subprocess
import sys
import tempfile
import time
import urllib.parse
from typing import Callable
from urllib.request import urlopen

URLS_AMD64 = {
    'stable': 'https://ftp.freedombox.org/pub/freedombox/hardware/'
              'amd64/bookworm/freedombox-bookworm_all-amd64.img.xz',
    'testing': 'https://ftp.freedombox.org/pub/freedombox/hardware/'
               'amd64/testing/freedombox-testing_dev_all-amd64.img.xz',
    'unstable': 'https://ftp.freedombox.org/pub/freedombox/hardware/'
                'amd64/nightly/freedombox-unstable_dev_all-amd64.img.xz',
}

URLS_ARM64 = {
    'stable': 'https://ftp.freedombox.org/pub/freedombox/hardware/'
              'arm64/bookworm/freedombox-bookworm_all-arm64.img.xz',
    'testing': 'https://ftp.freedombox.org/pub/freedombox/hardware/'
               'arm64/testing/freedombox-testing_dev_all-arm64.img.xz',
    'unstable': 'https://ftp.freedombox.org/pub/freedombox/hardware/'
                'arm64/nightly/freedombox-unstable_dev_all-arm64.img.xz',
}

URLS = URLS_AMD64

TRUSTED_KEYS = ['D4B069124FCF43AA1FCD7FBC2ACFC1E15AF82D8C']
KEY_SERVER = 'keyserver.ubuntu.com'
KEY_SERVER_HTTPS_API = \
    'https://keyserver.ubuntu.com/pks/lookup?op=get&search=0x{key_id}'

PROVISION_SCRIPT = '''
set -xe pipefail

cd /freedombox/

sudo apt-get -y install make
sudo make provision-dev

echo 'alias freedombox-develop="cd /freedombox; sudo -u plinth /freedombox/run --develop"' \
  >> /home/fbx/.bashrc

# Make some pytest related files and directories writable to the fbx user
sudo touch geckodriver.log
sudo chmod a+rw geckodriver.log
sudo mkdir -p .pytest_cache/
sudo chmod --recursive a+rw .pytest_cache/
sudo chmod a+w /freedombox
sudo chmod --recursive --silent a+w htmlcov || true
sudo chmod --silent a+w .coverage || true

exit 0
''' # noqa

SETUP_AND_RUN_TESTS_SCRIPT = '''
set -x

BACKPORTS_SOURCES_LIST=/etc/apt/sources.list.d/freedombox2.list
LDAPSCRIPTS_CONF=/etc/ldapscripts/freedombox-ldapscripts.conf

# Remount /freedoombox to be up to date
echo "> In container: Remounting /freedombox"
mount -o remount /freedombox

# Activate backports if Debian stable
if [[ "{distribution}" == "stable"  && ! -e $BACKPORTS_SOURCES_LIST ]]
then
    echo "> In container: Enable backports"
    /freedombox/actions/actions upgrades activate_backports --no-args
fi

echo "> In container: Upgrade packages"
apt-get update
DEBIAN_FRONTEND=noninteractive apt-get -yq --with-new-pkgs upgrade

# Install requirements for tests if not already installed as root
if ! [[ -e /usr/local/bin/geckodriver ]]
then
    /freedombox/plinth/tests/functional/install.sh
fi

# Run the plinth server if functional tests are requested
if [[ "{pytest_command}" =~ "--include-functional" ]]
then
    is_plinth_running=0
    ps -ax -o cmd | grep -q "^sudo -u plinth /freedombox/run" && \
    is_plinth_running=1
    ps -ax -o cmd | grep -q "^/usr/bin/python3 /usr/bin/plinth" && \
    is_plinth_running=1

    if [[ $is_plinth_running -eq 1 ]]
    then
        echo "> In container: Plinth is already running"
    else
        echo -n "> In container: Starting plinth ... "
        sudo -u plinth /freedombox/run --develop > plinth.log 2>&1 &
        while ! grep -q "Setup thread finished" plinth.log
        do
            sleep 1
            echo -n .
        done
        echo
    fi

    if [[ "{pytest_command}" != *"--splinter-headless"* ]]
    then
        # Use the X11 authority file from the fbx user to run GUI programs
        xauth merge /home/fbx/.Xauthority
    fi
fi

# Run pytest
cd /freedombox
export FREEDOMBOX_URL=https://localhost
export FREEDOMBOX_SSH_PORT=22
export FREEDOMBOX_SAMBA_PORT=445
{pytest_command}

# Make pytest cache files writable to the fbx user
chmod --recursive --silent a+rw .pytest_cache/
chmod --recursive --silent a+w htmlcov
chmod --silent a+w .coverage

exit 0
'''

logger = logging.getLogger(__name__)


def parse_arguments() -> argparse.Namespace:
    """Return parsed command line arguments as dictionary."""
    parser = argparse.ArgumentParser(
        formatter_class=argparse.ArgumentDefaultsHelpFormatter)
    subparsers = parser.add_subparsers(dest='subcommand', help='Sub command')

    distributions = list(URLS.keys())

    distribution = os.environ.get('DISTRIBUTION')

    formatter_class = argparse.ArgumentDefaultsHelpFormatter
    default_distribution = 'testing'
    if distribution and distribution in distributions:
        default_distribution = distribution

    def _add_common_args(subparser):
        subparser.add_argument('--distribution', choices=distributions,
                               default=default_distribution,
                               help='Distribution to work with')
        subparser.add_argument('--machine-type', choices=('container', ),
                               default='container')

    # Up
    subparser = subparsers.add_parser('up', help='Bring up the container',
                                      formatter_class=formatter_class)
    _add_common_args(subparser)
    subparser.add_argument('--image-size', default='16G',
                           help='Disk image size to resize to after download')
    subparser.add_argument('--hkp-client', choices=('gpg', 'wget'),
                           default='gpg', help='Client for key retrieval')

    # Print IP address
    subparser = subparsers.add_parser(
        'ip', help='Print the IP address of the container.',
        formatter_class=formatter_class)
    _add_common_args(subparser)

    # ssh
    subparser = subparsers.add_parser('ssh', help='SSH into the container',
                                      formatter_class=formatter_class)
    _add_common_args(subparser)

    # Run tests
    subparser = subparsers.add_parser('run-tests',
                                      help='Run tests in the container',
                                      formatter_class=formatter_class)
    _add_common_args(subparser)
    subparser.add_argument(
        '--pytest-args', nargs='...',
        help='Additional arguments to pass to the pytest command')

    # Stop
    subparser = subparsers.add_parser('stop', help='Stop the container',
                                      formatter_class=formatter_class)
    _add_common_args(subparser)

    # Destroy
    subparser = subparsers.add_parser('destroy',
                                      help='Destroy the container image',
                                      formatter_class=formatter_class)
    _add_common_args(subparser)

    # Update
    subparser = subparsers.add_parser(
        'update', help='Update the container image to the latest version',
        formatter_class=formatter_class)
    _add_common_args(subparser)
    subparser.add_argument('--hkp-client', choices=('gpg', 'wget'),
                           default='gpg', help='Client for key retrieval')

    # Display help message when no args are passed
    if len(sys.argv) == 1:
        parser.print_help()
        sys.exit()

    return parser.parse_args()


def _get_work_directory() -> pathlib.Path:
    """Return script's working directory."""
    return pathlib.Path(__file__).parent / '.container'


def _check_command(command) -> bool:
    """Check that a command is present in PATH."""
    if command == 'sudo':
        which = ['which', command]
    else:
        which = ['sudo', 'which', command]

    process = subprocess.run(which, stdout=subprocess.DEVNULL, check=False)
    return process.returncode == 0


def _verify_dependencies() -> None:
    """Ensure that all the dependencies are present."""
    if not _check_command('sudo'):
        logger.info('sudo command is needed. Run `apt install sudo`.')
        sys.exit(1)

    # Old versions of machinectl can't modprobe loop driver and immediately use
    # it leading to machinectl start failing first time after boot. See
    # https://github.com/systemd/systemd/issues/13130 . Workaround for old
    # versions of machinectl. Ignore errors.
    subprocess.run(['sudo', 'modprobe', '--all', '--quiet', 'loop'],
                   check=False)

    dependencies = {
        'systemd-nspawn': 'systemd-container',
        'machinectl': 'systemd-container',
        'kpartx': 'kpartx',
        'wget': 'wget',
        'gpg': 'gpg',
        'xz': 'xz-utils',
        'parted': 'parted',
        'sgdisk': 'gdisk',
        'btrfs': 'btrfs-progs',
        'nmcli': 'network-manager',
        'dnsmasq': 'dnsmasq',
        'ssh': 'openssh-client',
        'ssh-keygen': 'openssh-client',
    }
    missing_commands = []
    missing_packages = []
    for command, package in dependencies.items():
        if not _check_command(command):
            missing_commands.append(command)
            missing_packages.append(package)

    if not missing_commands:
        return

    logger.info('You need to install the following commands: %s',
                ' '.join(missing_commands))

    process = subprocess.run(['lsb_release', '--id', '--short'],
                             stdout=subprocess.PIPE, check=False)
    if process.stdout.decode().strip() != 'Debian':
        sys.exit(1)

    logger.info('Running apt for missing packages: %s',
                ' '.join(missing_packages))
    subprocess.run(['sudo', 'apt', 'install'] + missing_packages, check=True)


def _download_file(url: str, target_file: pathlib.Path, force: bool = False):
    """Download a file from remote URL."""
    if target_file.exists():
        if force:
            os.remove(target_file)
        else:
            return

    partial_file = target_file.with_suffix(target_file.suffix + '.partial')

    logger.info('Downloading %s', target_file)
    subprocess.run([
        'wget', '--quiet', '--show-progress', '--continue',
        '--output-document',
        str(partial_file), url
    ], check=True)
    partial_file.rename(target_file)


def _receive_keys_with_gpg(gpg_home: pathlib.Path, key_ids: list[str]):
    """Use gpg to retrieve and import a list of keys."""
    subprocess.run([
        'gpg', '--quiet', '--homedir',
        str(gpg_home), '--keyserver', KEY_SERVER, '--recv-keys'
    ] + key_ids, check=True)


def _receive_keys_with_wget(gpg_home: pathlib.Path, key_ids: list[str]):
    """Use wget to retrieve a list of keys and import them."""
    for key_id in key_ids:
        # Download public key
        logger.info('Getting public key %s', key_id)
        url = KEY_SERVER_HTTPS_API.format(key_id=key_id)
        wget_result = subprocess.run(
            ['wget', '--quiet', '--output-document=-', url], check=True,
            capture_output=True)
        public_key = wget_result.stdout
        subprocess.run(
            ['gpg', '--quiet', '--homedir',
             str(gpg_home), '--import=-'], input=public_key, check=True)


def _verify_signature(hkp_client: str, data_file: pathlib.Path,
                      signature_file: pathlib.Path):
    """Verify the detached signature on a file using GPG."""
    verified_file = signature_file.with_suffix(signature_file.suffix +
                                               '.verified')
    if verified_file.exists():
        return

    gpg_home = _get_work_directory() / 'gpg'
    gpg_home.mkdir(exist_ok=True)
    gpg_home.chmod(0o700)

    logger.info('Receiving GPG keys')
    if hkp_client == 'wget':
        _receive_keys_with_wget(gpg_home, TRUSTED_KEYS)
    else:
        _receive_keys_with_gpg(gpg_home, TRUSTED_KEYS)

    process = subprocess.run(
        ['gpg', '--quiet', '--homedir',
         str(gpg_home), '--armor', '--export'] + TRUSTED_KEYS, check=True,
        stdout=subprocess.PIPE)
    exported_keys = process.stdout
    trusted_keys = gpg_home / 'trustedkeys.gpg'
    subprocess.run([
        'gpg', '--quiet', '--homedir',
        str(gpg_home), '--no-default-keyring', '--keyring',
        str(trusted_keys), '--import', '-'
    ], check=True, input=exported_keys)

    logger.info('Verify GPG signature')
    subprocess.run([
        'gpgv', '--quiet', '--homedir',
        str(gpg_home),
        str(signature_file),
        str(data_file)
    ], check=True)

    verified_file.touch()


def _extract_image(compressed_file: pathlib.Path):
    """Extract the image file."""
    decompressed_file = compressed_file.with_suffix('')
    if decompressed_file.exists():
        return decompressed_file

    logger.info('Decompressing file %s', compressed_file)
    partial_file = compressed_file.with_suffix('.partial')
    with partial_file.open('w', encoding='utf-8') as file_handle:
        subprocess.run([
            'xz', '--verbose', '--decompress', '--keep', '--to-stdout',
            str(compressed_file)
        ], check=True, stdout=file_handle)
    partial_file.rename(decompressed_file)
    return decompressed_file


def _get_compressed_image_path(distribution: str) -> pathlib.Path:
    """Return the path of the compressed image."""
    url = URLS[distribution]
    result = urllib.parse.urlparse(url)

    return _get_work_directory() / pathlib.Path(result.path).name


def _get_project_folder() -> pathlib.Path:
    """Return the read-only folder that should be exposed into container."""
    return _get_work_directory().parent.resolve()


def _get_overlay_folder(distribution: str) -> pathlib.Path:
    """Return the writable folder that should be exposed into container."""
    default_data_dir = pathlib.Path.home() / '.local' / 'share'
    data_folder = os.environ.get('XDG_DATA_HOME', default_data_dir)
    folder = pathlib.Path(data_folder) / \
        'freedombox-container' / 'overlay' / distribution
    return folder.resolve()


def _download_disk_image(distribution: str, hkp_client: str,
                         force: bool = False) -> pathlib.Path:
    """Download and unpack FreedomBox disk image."""
    _get_work_directory().mkdir(exist_ok=True)

    url = URLS[distribution]

    target_file = _get_compressed_image_path(distribution)
    _download_file(url, target_file, force=force)

    signature_file = target_file.with_suffix(target_file.suffix + '.sig')
    _download_file(url + '.sig', signature_file, force=force)

    _verify_signature(hkp_client, target_file, signature_file)

    return _extract_image(target_file)


def _get_partition_info(image_file: pathlib.Path) -> tuple[str, str]:
    """Return the number of the final partition in the image file."""
    process = subprocess.run(
        ['sudo', 'parted', '--script',
         str(image_file), 'print'], stdout=subprocess.PIPE, check=True)
    lines = process.stdout.decode().splitlines()
    last_partition_number = lines[-2].split()[0]
    partition_table_type = 'msdos'
    for line in lines:
        if line.startswith('Partition Table:'):
            partition_table_type = line.partition(': ')[2]

    logger.info('Main partition: %s', last_partition_number)
    return partition_table_type, last_partition_number


def _resize_disk_image(image_file: pathlib.Path, new_size: str):
    """Resize the disk image if has not already been."""
    if new_size[-1] != 'G':
        raise ValueError(f'Invalid size: {new_size}')

    new_size_bytes = int(new_size.strip('G')) * 1024 * 1024 * 1024
    image_size = image_file.stat().st_size
    if image_size >= new_size_bytes:
        return

    logger.info('Resizing disk image to %s', new_size)

    disk_free = shutil.disk_usage(_get_work_directory()).free
    if disk_free < new_size_bytes - image_size:
        raise ValueError(f'Not enough free space on disk: {disk_free} bytes')

    subprocess.run(
        ['truncate', '--size',
         str(new_size_bytes),
         str(image_file)], check=True)
    partition_table_type, last_partition_number = _get_partition_info(
        image_file)
    if partition_table_type == 'gpt':
        subprocess.run(
            ['sudo', 'sgdisk', '--move-second-header',
             str(image_file)], check=True)

    subprocess.run([
        'sudo', 'parted', '--align=optimal', '--script',
        str(image_file), 'resizepart', last_partition_number, '100%'
    ], check=True)
    process = subprocess.run(
        ['sudo', 'kpartx', '-avs', str(image_file)], stdout=subprocess.PIPE,
        check=True)
    last_line = process.stdout.decode().splitlines()[-1]
    partition = '/dev/mapper/' + last_line.split()[2]
    subprocess.run(['sudo', 'btrfstune', '-uf', partition], check=True)

    with tempfile.TemporaryDirectory(
            dir=_get_work_directory().resolve()) as mount_point:
        subprocess.run(['sudo', 'mount', partition, mount_point], check=True)
        subprocess.run(
            ['sudo', 'btrfs', 'filesystem', 'resize', 'max', mount_point],
            check=True, stdout=subprocess.DEVNULL)
        subprocess.run(['sudo', 'umount', mount_point], check=True)

    subprocess.run(
        ['sudo', 'kpartx', '-ds', str(image_file)], check=True,
        stderr=subprocess.DEVNULL)


def _runc(image_file: pathlib.Path, command: list[str], **kwargs):
    """Run a command inside the container."""
    subprocess.run([
        'sudo', 'systemd-nspawn', '--image',
        str(image_file), '--console=autopipe', '--quiet'
    ] + command, check=True, **kwargs)


def _setup_users(image_file: pathlib.Path):
    """Change UID, GID and password of fbx user in container."""
    folder = _get_project_folder()

    uid = 10000
    gid = 10000
    logger.info('In container: Setting UID/GID of "fbx" to %s/%s', uid, gid)
    _runc(image_file, ['groupmod', '--gid', str(gid), 'fbx'])
    _runc(image_file, ['usermod', '--uid',
                       str(uid), '--gid',
                       str(gid), 'fbx'], stdout=subprocess.DEVNULL)

    uid = folder.stat().st_uid
    gid = folder.stat().st_gid
    logger.info(
        'In container: Setting UID/GID of "plinth" to %s/%s to match %s/%s',
        uid, gid, folder.owner(), folder.group())
    _runc(image_file, ['groupmod', '--gid', str(gid), 'plinth'])
    _runc(image_file,
          ['usermod', '--uid',
           str(uid), '--gid',
           str(gid), 'plinth'], stdout=subprocess.DEVNULL)

    logger.info('In container: Setting up sudo for users "fbx" and "plinth"')
    sudo_config = 'Cmnd_Alias FREEDOMBOX_ACTION_DEV = /usr/share/plinth/' \
        'actions/actions, /freedombox/actions/actions\n' \
        'Defaults!FREEDOMBOX_ACTION_DEV closefrom_override\n' \
        'plinth    ALL=(ALL:ALL) NOPASSWD:SETENV : FREEDOMBOX_ACTION_DEV\n' \
        'fbx    ALL=(ALL:ALL) NOPASSWD : ALL\n'
    _runc(image_file, ['tee', '/etc/sudoers.d/01-freedombox-development'],
          input=sudo_config.encode(), stdout=subprocess.DEVNULL)


def _setup_ssh(image_file: pathlib.Path):
    """Setup SSH server and client keys."""
    logger.info('In container: Generating SSH server keys')
    _runc(image_file, ['dpkg-reconfigure', 'openssh-server'],
          stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

    ssh_directory = _get_work_directory() / 'ssh'
    ssh_directory.mkdir(exist_ok=True)
    ssh_directory.chmod(0o700)

    key_file = ssh_directory / 'id_ed25519'
    if not key_file.exists():
        logger.info('Generating SSH client key %s', key_file)
        subprocess.run(
            ['ssh-keygen', '-t', 'ed25519', '-N', '', '-f',
             str(key_file)], stdout=subprocess.DEVNULL, check=True)

    public_key_file = key_file.with_suffix('.pub')
    public_key = public_key_file.read_bytes()
    if not public_key:
        logger.error('Unable to read SSH public key from %s', public_key_file)
        sys.exit(1)

    logger.info(
        'Adding SSH client key to "fbx" user authorized keys in container')
    _runc(image_file, ['mkdir', '--parents', '/home/fbx/.ssh'])
    _runc(image_file, ['tee', '--append', '/home/fbx/.ssh/authorized_keys'],
          input=public_key, stdout=subprocess.DEVNULL)
    _runc(image_file, ['chmod', '600', '/home/fbx/.ssh/authorized_keys'])
    _runc(image_file, ['chown', 'fbx:fbx', '/home/fbx/.ssh/authorized_keys'])


def _setup_image(image_file: pathlib.Path):
    """Prepare the image for execution."""
    setup_file = image_file.with_suffix(image_file.suffix + '.setup')
    if setup_file.exists():
        return

    logger.info('In container: Disabling automatic updates temporarily')
    contents = 'APT::Periodic::Update-Package-Lists "0";\n' \
        'APT::Periodic::Unattended-Upgrade "0";\n'
    _runc(image_file, ['tee', '/etc/apt/apt.conf.d/20auto-upgrades'],
          input=contents.encode())

    logger.info('In container: Disabling FreedomBox service')
    _runc(image_file, ['systemctl', 'disable', 'plinth'],
          stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

    _setup_ssh(image_file)

    _setup_users(image_file)

    setup_file.touch()
    logger.info('Setup completed')


def _destroy_image(distribution: str):
    """Remove all traces of the machine and its image."""
    compressed_image = _get_compressed_image_path(distribution)
    image_file = compressed_image.with_suffix('')
    logger.info('Removing image file %s', image_file)
    try:
        image_file.with_suffix(image_file.suffix + '.provisioned').unlink()
    except FileNotFoundError:
        pass

    try:
        image_file.with_suffix(image_file.suffix + '.setup').unlink()
    except FileNotFoundError:
        pass

    try:
        image_file.unlink()
    except FileNotFoundError:
        pass

    logger.info('Keeping downloaded image: %s',
                _get_compressed_image_path(distribution))


def _is_provisioned(distribution: str) -> bool:
    """Return the container has been provisioned fully."""
    compressed_image = _get_compressed_image_path(distribution)
    image_file = compressed_image.with_suffix('')
    provision_file = image_file.with_suffix(image_file.suffix + '.provisioned')
    return provision_file.exists()


def _provision(image_file: pathlib.Path, machine_type: str, distribution: str):
    """Run app setup inside the container."""
    if _is_provisioned(distribution):
        return

    machine = Machine.get_instance(machine_type, distribution)
    ssh_command = machine.get_ssh_command()
    subprocess.run(ssh_command + ['bash'], check=True,
                   input=PROVISION_SCRIPT.encode())

    provision_file = image_file.with_suffix(image_file.suffix + '.provisioned')
    provision_file.touch()
    logger.info('Provision completed')


def _print_banner(machine_type: str, distribution: str):
    """Print a friendly message on how to use."""
    machine = Machine.get_instance(machine_type, distribution)
    ip_address = _wait_for(lambda: machine.get_ip_address())

    _get_work_directory().owner()
    script = sys.argv[0]
    options = ''
    if distribution != 'testing':
        options = f'--distribution={distribution}'

    project_folder = _get_project_folder()
    overlay_folder = _get_overlay_folder(distribution)
    message = f'''==== Ready ====

Development user      : fbx
Plinth user           : (host){project_folder.owner()} -> (container)plinth
Folder overlay        :    (host, read-only){project_folder}
                        +  (host, read-write){overlay_folder}
                        -> (container)/freedombox

SSH easily            : {script} ssh {options}
Run tests             : {script} run-tests {options} [ --pytest-args ... ]
Run FreedomBox inside : freedombox-develop

Web access            : https://{ip_address}/
Ports access          : Any port on {ip_address}

Terminal login        : sudo machinectl login fbx-{distribution}
Open a root shell     : sudo machinectl shell fbx-{distribution}
Shutdown              : {script} stop {options}
Destroy               : {script} destroy {options}
Reset                 : {script} destroy {options}; {script} up {options}'''
    logger.info(message)


def _wait_for(method: Callable[[], None | object]):
    """Wait until a condition is satisfied or finally give up."""
    for _ in range(60):
        return_value = method()
        if return_value:
            return return_value

        logger.info('Waiting for container')
        time.sleep(3)

    logger.error('Failed')
    sys.exit(1)


def _get_latest_image_timestamp(distribution: str) -> float:
    """Get the timestamp of the latest available image."""
    url = URLS[distribution]
    response = urlopen(url[0:url.rindex('/')])
    page_contents = response.read().decode()
    str_time = re.findall(r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}', page_contents)[0]
    return datetime.datetime.strptime(str_time, '%Y-%m-%d %H:%M').timestamp()


def _is_update_required(distribution: str) -> bool:
    """Compare local image timestamp against the latest image timestamp."""
    file_path = _get_work_directory() / URLS[distribution].split('/')[-1]

    if not file_path.exists():
        return True

    local_image_timestamp = os.path.getmtime(file_path)
    one_day = datetime.timedelta(days=1).total_seconds()
    latest_image_timestamp = _get_latest_image_timestamp(distribution)
    return latest_image_timestamp - local_image_timestamp > one_day


class Machine:
    """Base class for different types of machines."""

    def __init__(self, distribution: str):
        """Initialize the machine object."""
        machine_name = f'fbx-{distribution}'
        self.machine_name = machine_name
        self.distribution = distribution

    @staticmethod
    def get_instance(machine_type: str, distribution: str):
        """Return instance of a concrete machine class based on type."""
        if machine_type == 'container':
            return Container(distribution)

        raise ValueError('Unknown machine type')

    def get_status(self) -> bool:
        """Return whether the machine is currently running."""
        return False

    def setup(self) -> None:
        """Setup the infrastructure needed for the machine."""

    def launch(self) -> None:
        """Start the machine."""

    def stop(self) -> None:
        """Stop the machine."""

    def terminate(self) -> None:
        """Terminate, i.e., force stop the machine."""

    def destory(self) -> None:
        """Remove all traces of the machine from the host."""

    def get_ip_address(self) -> str | None:
        """Return the IP address assigned to the machine."""

    def get_ssh_command(self) -> list[str]:
        """Return the SSH command to execute for the machine."""
        return []


class Container(Machine):
    """Handle container specific operations."""

    def get_status(self) -> bool:
        """Return the running status of a container."""
        process = subprocess.run(
            ['sudo', 'machinectl', 'status', self.machine_name],
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)
        return process.returncode == 0

    def setup(self) -> None:
        """Setup the infrastructure needed for the container."""
        self._setup_nm_connection()

    def launch(self) -> None:
        """Start the container."""
        self._create_nspawn_machine()

        logger.info('Running `machinectl start %s`', self.machine_name)
        subprocess.run(['sudo', 'machinectl', 'start', self.machine_name],
                       check=True)

        logger.info('Bringing up host network connection: %s',
                    f'fbx-{self.distribution}-shared')
        # This command requires dnsmasq
        subprocess.run([
            'sudo', 'nmcli', 'connection', 'up',
            f'fbx-{self.distribution}-shared'
        ], stdout=subprocess.DEVNULL, check=True)

    def stop(self) -> None:
        """Stop the container."""
        logger.info('Running `machinectl stop %s`', self.machine_name)
        subprocess.run(['sudo', 'machinectl', 'stop', self.machine_name],
                       check=False)

    def terminate(self) -> None:
        """Terminate, i.e., force stop the container."""
        logger.info('Running `machinectl terminate %s`', self.machine_name)
        subprocess.run(['sudo', 'machinectl', 'terminate', self.machine_name],
                       check=False)

    def destory(self) -> None:
        """Remove all traces of the machine from the host."""
        image_link = pathlib.Path(f'/var/lib/machines/{self.machine_name}.raw')

        logger.info('Removing link to systemd-nspawn image %s', image_link)
        subprocess.run(['sudo', 'rm', '--force', str(image_link)], check=False)
        nspawn_file = f'/run/systemd/nspawn/{self.machine_name}.nspawn'

        logger.info('Removing systemd-nspawn configuration: %s', nspawn_file)
        subprocess.run(['sudo', 'rm', '--force', nspawn_file], check=False)

        overlay_folder = _get_overlay_folder(self.distribution)
        logger.info('Removing overlay folder with container written data: %s',
                    overlay_folder)
        subprocess.run(['sudo', 'rm', '-r', '--force', overlay_folder],
                       check=False)

        connection_name = f'fbx-{self.distribution}-shared'
        logger.info('Removing Network Manager connection %s', connection_name)
        result = subprocess.run(
            ['sudo', 'nmcli', 'connection', 'delete', connection_name],
            capture_output=True, check=False)
        if result.returncode not in (0, 10):
            # nmcli failed and not due to 'Connection, device, or access point
            # does not exist.' See
            # https://developer-old.gnome.org/NetworkManager/stable/nmcli.html
            logger.error('nmcli returned code %d', result.returncode)
            logger.error('Error message:\n%s', result.stderr.decode())
            logger.error('Output:\n%s', result.stdout.decode())

    def get_ip_address(self) -> str | None:
        """Return the IPv4 address assigned to the container or None."""
        process = subprocess.run(
            ['machinectl', 'list', '--output=json', '--max-addresses=all'],
            stdout=subprocess.PIPE, check=True)
        output = process.stdout.decode()
        if not output:
            return None

        machines = json.loads(output)
        for machine in machines:
            if (machine['machine'] == f'fbx-{self.distribution}'
                    and machine['addresses'] not in (None, '-')):
                for address in machine['addresses'].split():
                    if ipaddress.ip_address(address).version == 4:
                        return address

        return None

    def get_ssh_command(self) -> list[str]:
        """Return the SSH command to execute for the container."""
        ip_address = _wait_for(lambda: self.get_ip_address())
        public_key = _get_work_directory() / 'ssh' / 'id_ed25519'
        if ipaddress.ip_address(ip_address).is_link_local:
            ip_address = f'{ip_address}%' + self._get_interface_name()

        return [
            'ssh', '-Y', '-C', '-t', '-i',
            str(public_key), '-o', 'LogLevel=error', '-o',
            'StrictHostKeyChecking=no', '-o', 'UserKnownHostsFile=/dev/null',
            '-o', 'IdentitiesOnly=yes', f'fbx@{ip_address}'
        ]

    def _get_interface_name(self) -> str:
        """Return the name of the interface."""
        return f've-fbx-{self.distribution}'

    def _setup_nm_connection(self) -> None:
        """Create a NM conn. on host for DHCP/DNS with container."""
        connection_name = f'fbx-{self.distribution}-shared'
        process = subprocess.run(
            ['sudo', 'nmcli', 'connection', 'show', connection_name],
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)
        if not process.returncode:
            return

        logger.info('Creating Network Manager connection in host: %s',
                    connection_name)
        properties = {
            'connection.id': connection_name,
            'connection.type': '802-3-ethernet',
            'connection.interface-name': self._get_interface_name(),
            'connection.autoconnect': 'yes',
            'connection.zone': 'trusted',
            'ipv4.method': 'shared',
        }
        subprocess.run(['sudo', 'nmcli', 'connection', 'add'] +
                       list(itertools.chain(*properties.items())), check=True)
        subprocess.run(['sudo', 'nmcli', 'connection', 'up', connection_name],
                       stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                       check=False)

    def _create_nspawn_machine(self) -> None:
        """Create systemd-nspawn options/image used by machinectl."""
        compressed_image = _get_compressed_image_path(self.distribution)
        image_file = compressed_image.with_suffix('')

        overlay_folder = _get_overlay_folder(self.distribution)
        logger.info('Creating overlay folder %s', overlay_folder)
        overlay_folder.mkdir(parents=True, exist_ok=True)

        nspawn_dir = pathlib.Path('/run/systemd/nspawn/')
        if not nspawn_dir.exists():
            subprocess.run(['sudo', 'mkdir', '--mode=700',
                            str(nspawn_dir)], check=True)

        nspawn_options = f'''[Exec]
Boot=yes
PrivateUsers=no
# Allow all system calls to enable podman containers inside the nspawn
# container.
SystemCallFilter=@known

[Files]
Overlay={_get_project_folder()}:{overlay_folder}:/freedombox

[Network]
VirtualEthernet=yes
'''
        nspawn_file = f'/run/systemd/nspawn/{self.machine_name}.nspawn'
        logger.info('Creating systemd-nspawn configuration: %s', nspawn_file)
        subprocess.run(['sudo', 'rm', '--force', nspawn_file], check=False)
        subprocess.run(['sudo', 'tee', nspawn_file],
                       input=nspawn_options.encode(),
                       stdout=subprocess.DEVNULL, check=True)

        image_link = pathlib.Path(f'/var/lib/machines/{self.machine_name}.raw')
        logger.info('Linking systemd-nspawn image %s -> %s', image_link,
                    image_file)
        result = subprocess.run(
            ['sudo', 'test', '-e', str(image_link)], check=False)
        if not result.returncode:
            result = subprocess.run(
                ['sudo', 'test', '-L', str(image_link)], check=False)
            if result.returncode:
                raise Exception(f'Image file {image_link} is not a symlink.')

            subprocess.run(['sudo', 'rm', '--force',
                            str(image_link)], check=False)

        subprocess.run([
            'sudo', 'ln', '--symbolic',
            str(image_file.resolve()),
            str(image_link)
        ], check=False)


def subcommand_up(arguments: argparse.Namespace):
    """Download, setup and bring up the container."""
    machine = Machine.get_instance(arguments.machine_type,
                                   arguments.distribution)
    if machine.get_status() and _is_provisioned(arguments.distribution):
        logger.info('Container is already running')
        _print_banner(arguments.machine_type, arguments.distribution)
        return

    _verify_dependencies()
    image_file = _download_disk_image(arguments.distribution,
                                      arguments.hkp_client)
    _resize_disk_image(image_file, arguments.image_size)
    _setup_image(image_file)
    machine.setup()
    machine.launch()
    _provision(image_file, arguments.machine_type, arguments.distribution)
    _print_banner(arguments.machine_type, arguments.distribution)


def subcommand_ip(arguments: argparse.Namespace):
    """Print the IP address of the container."""
    machine = Machine.get_instance(arguments.machine_type,
                                   arguments.distribution)
    print(machine.get_ip_address() or '')


def subcommand_ssh(arguments: argparse.Namespace):
    """Open an SSH shell into the container."""
    machine = Machine.get_instance(arguments.machine_type,
                                   arguments.distribution)
    command = machine.get_ssh_command()
    logger.info('Running command: %s', ' '.join(command))
    os.execlp('ssh', *command)


def subcommand_run_tests(arguments: argparse.Namespace):
    """Run tests in the container."""
    distribution = arguments.distribution
    machine = Machine.get_instance(arguments.machine_type, distribution)
    pytest_args_list = arguments.pytest_args or []
    pytest_args = ' '.join((shlex.quote(arg) for arg in pytest_args_list))

    ssh_command = machine.get_ssh_command()

    pytest_command = f'py.test-3 --color=yes {pytest_args}'
    logger.info('Pytest command: %s', pytest_command)

    test_script = SETUP_AND_RUN_TESTS_SCRIPT.format(
        pytest_command=pytest_command, distribution=distribution)
    command = ssh_command + ['sudo', 'bash']
    process = subprocess.run(command, input=test_script.encode(), check=False)
    sys.exit(process.returncode)


def subcommand_stop(arguments: argparse.Namespace):
    """Stop the container."""
    machine = Machine.get_instance(arguments.machine_type,
                                   arguments.distribution)
    machine.stop()
    _wait_for(lambda: not machine.get_status())


def subcommand_destroy(arguments: argparse.Namespace):
    """Destroy the disk image."""
    machine = Machine.get_instance(arguments.machine_type,
                                   arguments.distribution)
    machine.terminate()
    _wait_for(lambda: not machine.get_status())

    machine.destory()
    _destroy_image(arguments.distribution)


def subcommand_update(arguments: argparse.Namespace):
    """Update the disk image."""
    if _is_update_required(arguments.distribution):
        logger.info('Updating...')
        _download_disk_image(arguments.distribution, arguments.hkp_client,
                             force=True)
    else:
        logger.info('Already using the latest image')


def set_URLs() -> None:
    global URLS
    arch = platform.machine()
    if arch == 'x86_64' or arch == 'amd64':
        URLS = URLS_AMD64
    elif arch == 'aarch64' or arch == 'arm64':
        URLS = URLS_ARM64
    else:
        logger.error('Unsupported architecture:', arch)
        sys.exit(1)


def main():
    """Parse arguments and perform operations."""
    set_URLs()

    logging.basicConfig(level='INFO', format='> %(message)s')
    arguments = parse_arguments()

    subcommand = arguments.subcommand.replace('-', '_')
    subcommand_method = globals()['subcommand_' + subcommand]
    subcommand_method(arguments)


if __name__ == '__main__':
    main()