cmtadm

#! /bin/sh
# vim: ts=4 filetype=python expandtab shiftwidth=4 softtabstop=4 syntax=python
''''eval version=$( ls /usr/bin/python3.* | \
    grep '.*[0-9]$' | sort -nr -k2 -t. | head -n1 ) && \
    version=${version##/usr/bin/python3.} && [ ${version} ] && \
    [ ${version} -ge 9 ] && exec /usr/bin/python3.${version} "$0" "$@" || \
    exec /usr/bin/env python3 "$0" "$@"' #'''
# The above hack is to handle distros where /usr/bin/python3
# doesn't point to the latest version of python3 they provide

# Requires: python3 (>= 3.9)
# Requires: python3-natsort
# Requires: python3-paramiko
#
# Copyright the Cluster Management Toolkit for Kubernetes contributors.
# SPDX-License-Identifier: MIT

# pylint: disable=too-many-lines

import errno
from getpass import getuser
from glob import glob
import grp
import os
from pathlib import Path
import pwd
import re
import shutil
import socket
from subprocess import CalledProcessError  # nosec
import sys
import tempfile
from typing import Any, cast, Optional, Union
from collections.abc import Callable

try:
    from natsort import natsorted
except ModuleNotFoundError:  # pragma: no cover
    sys.exit("ModuleNotFoundError: Could not import natsort; "
             "you may need to (re-)run `cmt-install` or `pip3 install natsort`; aborting.")

from clustermanagementtoolkit.cluster_actions import create_vm_hosts, join_nodes, get_crio_version
from clustermanagementtoolkit.cluster_actions import prepare_nodes, prepare_vm_template

from clustermanagementtoolkit.cmttypes import deep_get, DictPath, FilePath, FilePathAuditError
from clustermanagementtoolkit.cmttypes import SecurityChecks, SecurityPolicy, SecurityStatus

from clustermanagementtoolkit.cmtpaths import BASH_COMPLETION_BASE_DIR, BASH_COMPLETION_DIR
from clustermanagementtoolkit.cmtpaths import BINDIR, HOMEDIR
from clustermanagementtoolkit.cmtpaths import SSH_DIR, SSH_KEYGEN_BIN_PATH, SSH_KEYGEN_ARGS
from clustermanagementtoolkit.cmtpaths import DEPLOYMENT_DIR
from clustermanagementtoolkit.cmtpaths import CMT_PRE_PREPARE_DIR, CMT_POST_PREPARE_DIR
from clustermanagementtoolkit.cmtpaths import CMT_PRE_SETUP_DIR, CMT_POST_SETUP_DIR
from clustermanagementtoolkit.cmtpaths import CMT_PRE_UPGRADE_DIR, CMT_POST_UPGRADE_DIR
from clustermanagementtoolkit.cmtpaths import CMT_PRE_TEARDOWN_DIR, CMT_POST_TEARDOWN_DIR
from clustermanagementtoolkit.cmtpaths import CMT_PRE_PURGE_DIR, CMT_POST_PURGE_DIR
from clustermanagementtoolkit.cmtpaths import ANSIBLE_INVENTORY
from clustermanagementtoolkit.cmtpaths import DEFAULT_THEME_FILE
from clustermanagementtoolkit.cmtpaths import CMT_CONFIG_FILE, CMT_INSTALLATION_INFO_FILE
from clustermanagementtoolkit.cmtpaths import KUBE_CONFIG_DIR, KUBE_CONFIG_FILE

from clustermanagementtoolkit.commandparser import parse_commandline

from clustermanagementtoolkit.ansible_helper import ansible_configuration
from clustermanagementtoolkit.ansible_helper import ansible_get_inventory_dict, ansible_set_vars
from clustermanagementtoolkit.ansible_helper import ansible_run_playbook_on_selection
from clustermanagementtoolkit.ansible_helper import ansible_add_hosts
from clustermanagementtoolkit.ansible_helper import ansible_get_hosts_by_group, get_playbook_path
from clustermanagementtoolkit.ansible_helper import ansible_print_play_results
from clustermanagementtoolkit.ansible_helper import populate_playbooks_from_filenames

from clustermanagementtoolkit import cmtio
from clustermanagementtoolkit.cmtio import check_path
from clustermanagementtoolkit.cmtio import execute_command, execute_command_with_response
from clustermanagementtoolkit.cmtio import secure_mkdir, secure_read_string, secure_rm
from clustermanagementtoolkit.cmtio import secure_which, secure_write_string

from clustermanagementtoolkit.cmtio_yaml import secure_read_yaml, secure_write_yaml

from clustermanagementtoolkit.networkio import download_files
from clustermanagementtoolkit.networkio import scan_and_add_ssh_keys, update_version_cache

from clustermanagementtoolkit import cmtlib
from clustermanagementtoolkit.cmtlib import check_versions_apt
from clustermanagementtoolkit.cmtlib import check_versions_yum, check_versions_zypper
from clustermanagementtoolkit.cmtlib import get_latest_upstream_version, get_cluster_name
from clustermanagementtoolkit.cmtlib import identify_distro, read_cmtconfig
from clustermanagementtoolkit.cmtlib import substitute_list, substitute_string

from clustermanagementtoolkit.cni_data import cni_data

from clustermanagementtoolkit import kubernetes_helper
from clustermanagementtoolkit.kubernetes_helper import get_node_roles

from clustermanagementtoolkit.ansithemeprint import ANSIThemeStr, ansithemeinput
from clustermanagementtoolkit.ansithemeprint import ansithemeinput_password, ansithemeprint
from clustermanagementtoolkit.ansithemeprint import ansithemestr_join_list
from clustermanagementtoolkit.ansithemeprint import themearray_override_formatting

from clustermanagementtoolkit import fieldgetters

from clustermanagementtoolkit import checks

from clustermanagementtoolkit import about

PROGRAMDESCRIPTION = "Setup or teardown a Kubernetes cluster"
PROGRAMAUTHORS = "Written by David Weinehall."

DEFAULT_CNI = "cilium"
DEFAULT_POD_NETWORK_CIDR = "10.244.0.0/16"

no_password = False  # pylint: disable=invalid-name

kh: kubernetes_helper.KubernetesHelper = None  # type: ignore

cri_data: dict = {
    "containerd": {
        "socket": "unix:///run/containerd/containerd.sock",
    },
    "cri-o": {
        "socket": "unix:///run/crio/crio.sock",
    },
}

prepare_targets: dict = {
    "kubeadm": {
        "pretty_name": [("kubeadm", "programname"), (" (default)", "default")],
        "playbooks": [
            FilePath("prepare_passwordless_ansible.yaml"),
            FilePath("install_packages.yaml"),
            FilePath("prepare_control_plane.yaml"),
            FilePath("add_kubernetes_repo.yaml"),
        ],
        "deb_packages": [
        ],
        "fedora_packages": [
        ],
        "suse_packages": [
        ],
    },
    "localhost": {
        "pretty_name": [("host system", "programname")],
        "playbooks": [
            FilePath("prepare_passwordless_ansible.yaml"),
            FilePath("add_kubernetes_repo.yaml"),
            FilePath("install_packages.yaml"),
        ],
        "deb_packages": [
            "ansible",
        ],
        "fedora_packages": [
            "ansible-core",
        ],
        "suse_packages": [
            "ansible",
        ],
    },
    "rke2": {
        "pretty_name": [("kubeadm", "programname"), (" (default)", "default")],
        "playbooks": [
            FilePath("prepare_passwordless_ansible.yaml"),
            FilePath("install_packages.yaml"),
            FilePath("prepare_control_plane.yaml"),
        ],
        "deb_packages": [
        ],
        "fedora_packages": [
        ],
        "suse_packages": [
        ],
    },
}

setup_control_plane_targets: dict = {
    "kubeadm": {
        "pretty_name": [("kubeadm", "programname"), (" (default)", "default")],
        "playbooks": [
            FilePath("install_packages.yaml"),
            FilePath("kubeadm_setup_control_plane.yaml"),
            FilePath("fetch_kube_config.yaml"),
        ],
        "deb_packages": [
            "kubeadm",
            "kubectl",
            "kubelet",
            "kubernetes-cni",
        ],
        "deb_packages_held": [
            "kubeadm",
            "kubectl",
            "kubelet",
        ],
        "fedora_packages": [
            "kubeadm",
            "kubectl",
            "kubelet",
        ],
        "fedora_packages_held": [
            "kubeadm",
            "kubectl",
            "kubelet",
        ],
        "suse_packages": [
            "kubectl",
            # "kubernetes{major.minor}-client",
            "kubeadm",
            # "kubernetes{major.minor}-kubeadm",
            "kubelet",
            # "kubernetes{major.minor}-kubelet",
        ],
        "suse_packages_held": [
            "kubectl",
            # "kubernetes{major.minor}-client",
            "kubeadm",
            # "kubernetes{major.minor}-kubeadm",
            "kubelet",
            # "kubernetes{major.minor}-kubelet",
        ],
        "extra_values": {},
    },
    "localhost": {
        "pretty_name": [("host system", "programname")],
        "playbooks": [
            FilePath("install_packages.yaml"),
        ],
        "deb_packages": [
            "kubectl",
        ],
        "deb_packages_held": [
            "kubectl",
        ],
        "fedora_packages": [
            "kubectl",
        ],
        "fedora_packages_held": [
            "kubectl",
        ],
        "suse_packages": [
            "kubectl",
            # "kubernetes{major.minor}-client",
        ],
        "suse_packages_held": [
            "kubectl",
            # "kubernetes{major.minor}-client",
        ],
    },
    "rke2": {
        "pretty_name": [("RKE2", "programname")],
        "playbooks": [
            FilePath("rke2_setup_control_plane.yaml"),
            FilePath("fetch_kube_config.yaml"),
        ],
        "extra_values": {},
    },
}

upgrade_control_plane_targets: dict = {
    "kubeadm": {
        "pretty_name": [("kubeadm", "programname"), (" (default)", "default")],
        "playbooks": [
            FilePath("kubeadm_upgrade_control_plane.yaml"),
        ],
        "extra_values": {},
    },
    "localhost": {
        "pretty_name": [("host system", "programname")],
        "playbooks": [
            FilePath("install_packages.yaml"),
        ],
        "deb_packages": [
            "kubectl",
        ],
        "deb_packages_held": [
            "kubectl",
        ],
        "fedora_packages": [
            "kubectl",
        ],
        "fedora_packages_held": [
            "kubectl",
        ],
    },
    "rke2": {
        "pretty_name": [("RKE2", "programname")],
        "playbooks": [
            FilePath("rke2_upgrade_control_plane.yaml"),
        ],
        "extra_values": {},
    },
}

teardown_control_plane_targets: dict = {
    "kubeadm": {
        "pretty_name": [("kubeadm", "programname"), (" (default)", "default")],
        "playbooks": [
            FilePath("teardown_cni.yaml"),
            FilePath("kubeadm_teardown_control_plane.yaml"),
        ],
    },
    "rke2": {
        "pretty_name": [("RKE2", "programname")],
        "playbooks": [
            FilePath("rke2_teardown_control_plane.yaml"),
        ],
    },
}

purge_control_plane_targets: dict = {
    "kubeadm": {
        "pretty_name": [("kubeadm", "programname"), (" (default)", "default")],
        "playbooks": [
            FilePath("kubeadm_purge.yaml"),
        ],
        "deb_packages": [
            "kubeadm",
            "kubectl",
            "kubelet",
            "kubernetes-cni",
        ],
        "deb_packages_held": [
            "kubeadm",
            "kubectl",
            "kubelet",
        ],
        "fedora_packages": [
            "kubeadm",
            "kubectl",
            "kubelet",
            "kubernetes-cni",
        ],
        "fedora_packages_held": [
            "kubeadm",
            "kubectl",
            "kubelet",
        ],
    },
    "rke2": {
        "pretty_name": [("rke2", "programname")],
        "playbooks": [
            FilePath("rke2_purge.yaml"),
        ],
    },
}


# pylint: disable-next=too-many-locals,too-many-branches
def get_control_plane_version(controlplane: str, k8s_distro: str) -> str:
    """
    Return the Kubernetes version used by the control plane.

        Parameters:
            controlplane (str): Name of the node to check the version from
            k8s_distro (str): The Kubernetes distro used
        Returns:
            (str): The Kubernetes version if discernible, the empty string if not
    """
    global kh  # pylint: disable=global-statement

    version = None

    if k8s_distro == "rke2":
        # This will only work for running clusters
        if kh is None:
            kh = kubernetes_helper.KubernetesHelper(about.PROGRAM_SUITE_NAME,
                                                    about.PROGRAM_SUITE_VERSION, None)

        vlist, status = kh.get_list_by_kind_namespace(("Node", ""), "")
        if status != 200:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": API-server returned ", "default"),
                            ANSIThemeStr(f"{status}", "errorvalue"),
                            ANSIThemeStr("; aborting.", "default")], stderr=True)
            sys.exit(errno.EINVAL)
        if vlist is None:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": API-server did not return any data",
                                         "default")], stderr=True)
            sys.exit(errno.EINVAL)

        for node in vlist:
            name = deep_get(node, DictPath("metadata#name"))
            if name == controlplane:
                version = deep_get(node, DictPath("status#nodeInfo#kubeletVersion"))
                break
    else:
        # For kubeadm we try to use the package version
        get_versions_path = get_playbook_path(FilePath("get_versions.yaml"))
        retval, ansible_results = \
            ansible_run_playbook_on_selection(get_versions_path,
                                              selection=[controlplane], quiet=False)

        if not ansible_results:
            raise ValueError(f"Error: Failed to get package versions from {controlplane} "
                             f"(retval: {retval}); aborting.")

        k8s_distro = "<unknown>"
        version = "<unknown>"

        for result in deep_get(ansible_results, DictPath(controlplane), []):
            if deep_get(result, DictPath("task"), "") == "Package versions":
                tmp = deep_get(result, DictPath("msg_lines"), [])
                break

        if len(tmp) == 0:
            raise ValueError(f"Error: Received empty version data from {controlplane} "
                             f"(retval: {retval}); aborting.")

        package_version_regex: re.Pattern[str] = re.compile(r"^(.*?): (.*)")

        for line in tmp:
            tmp2 = package_version_regex.match(line)
            if tmp2 is None:
                continue
            package = tmp2[1]
            package_version = tmp2[2]
            if package == "kubeadm":
                version = package_version
                break

    if version is None:
        version = ""

    return version


# pylint: disable-next=too-many-locals,too-many-branches,too-many-statements
def rebuild_installation_info(state: Optional[str] = None) -> None:
    """
    If the installation info file does not exist, but a cluster already exists
    and is part of the inventory, this function will try to rebuild the installation info file.

        Parameters:
            state (str): The installation state
    """
    global kh  # pylint: disable=global-statement

    k8s_distro = None
    controlplane = None
    cri = "<none>"

    # This will only work for running clusters
    if kh is None:
        kh = kubernetes_helper.KubernetesHelper(about.PROGRAM_SUITE_NAME,
                                                about.PROGRAM_SUITE_VERSION, None)

    vlist, status = kh.get_list_by_kind_namespace(("Node", ""), "")
    if status != 200:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": API-server returned ", "default"),
                        ANSIThemeStr(f"{status}", "errorvalue"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)
    if vlist is None:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": API-server did not return any data", "default")],
                       stderr=True)
        sys.exit(errno.EINVAL)

    for node in vlist:
        name = deep_get(node, DictPath("metadata#name"))
        node_roles = get_node_roles(cast(dict, node))
        if "control-plane" in node_roles or "master" in node_roles:
            controlplane = name
            cri = deep_get(node, DictPath("status#nodeInfo#containerRuntimeVersion"), "")
            if cri is not None:
                cri = cri.split(":")[0]
            tmp_k8s_distro = None
            minikube_name = deep_get(node, DictPath("metadata#labels#minikube.k8s.io/name"), "")
            labels = deep_get(node, DictPath("metadata#labels"), {})
            images = deep_get(node, DictPath("status#images"), [])
            for image in images:
                names = deep_get(image, DictPath("names"), [])
                for name in names:
                    if "openshift-crc-cluster" in name:
                        tmp_k8s_distro = "crc"
                        break
                if tmp_k8s_distro is not None:
                    break
            if minikube_name != "":
                tmp_k8s_distro = "minikube"
            elif deep_get(labels, DictPath("microk8s.io/cluster"), False):
                tmp_k8s_distro = "microk8s"
            elif deep_get(node, DictPath("spec#providerID"), "").startswith("kind://"):
                tmp_k8s_distro = "kind"
            else:
                managed_fields = deep_get(node, DictPath("metadata#managedFields"), [])
                for managed_field in managed_fields:
                    manager = deep_get(managed_field, DictPath("manager"), "")
                    if manager == "rke2":
                        tmp_k8s_distro = "rke2"
                        break
                    if manager == "k0s":
                        tmp_k8s_distro = "k0s"
                        break
                    if manager.startswith("deploy@k3d"):
                        tmp_k8s_distro = "k3d"
                        break
                    if manager == "k3s":
                        tmp_k8s_distro = "k3s"
                        break
                    if manager == "kubeadm":
                        tmp_k8s_distro = "kubeadm"
                        break
            if tmp_k8s_distro is not None:
                if k8s_distro is not None:
                    ansithemeprint([ANSIThemeStr("Critical", "critical"),
                                    ANSIThemeStr(": The control planes are reporting conflicting "
                                                 "Kubernetes distros; aborting.",
                                                 "default")], stderr=True)
                    sys.exit(errno.EINVAL)
                else:
                    k8s_distro = tmp_k8s_distro

    if controlplane is None:
        ansithemeprint([ANSIThemeStr("", "default")])
        ansithemeprint([ANSIThemeStr("Critical", "critical"),
                        ANSIThemeStr(": Could not identify a control plane for the cluster; "
                                     "aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    if k8s_distro is None:
        ansithemeprint([ANSIThemeStr("Critical", "critical"),
                        ANSIThemeStr(": Unknown Kubernetes distro; cannot determine Kubernetes "
                                     "version.", "default")], stderr=True)
        sys.exit(errno.EINVAL)
    version = get_control_plane_version(controlplane=controlplane, k8s_distro=k8s_distro)

    if version is None or not version:
        ansithemeprint([ANSIThemeStr("", "default")])
        ansithemeprint([ANSIThemeStr("Critical", "critical"),
                        ANSIThemeStr(": Failed to get Kubernetes version; currently only kubeadm "
                                     "and RKE2 are supported. Aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    cluster_name = get_cluster_name()
    pod_network_cidr = kh.get_pod_network_cidr()
    cnis = kh.identify_cni()
    if len(cnis) == 1:
        cni = cnis[0][0]
    else:
        cni = "<unknown>"

    update_installation_info(installation_target=cluster_name, cluster_name=cluster_name,
                             distro=k8s_distro, version=version, requested_version="<none>",
                             state=state, phase="<none>", phase_skiplist=[], cni=cni, cri=cri,
                             pod_network_cidr=pod_network_cidr)


def get_installation_info(cluster_name: Optional[str] = None,
                          ignore_non_existing: bool = True) -> dict:
    """
    Return installation info for a cluster, or prepares a new entry if no entry exists yet.

        Parameters:
            cluster_name (str): The name of the cluster to get information for
            ignore_non_existing (bool): Ignore and recreate installation_info.yaml
                                        if the file doesn't exist
        Returns:
            (dict): A dictionary with information about a cluster
    """
    info = None

    # We are OK with the file not existing
    security_checks = [
        SecurityChecks.PARENT_RESOLVES_TO_SELF,
        SecurityChecks.OWNER_IN_ALLOWLIST,
        SecurityChecks.PARENT_OWNER_IN_ALLOWLIST,
        SecurityChecks.PERMISSIONS,
        SecurityChecks.PARENT_PERMISSIONS,
        SecurityChecks.IS_FILE,
    ]

    try:
        info = secure_read_yaml(CMT_INSTALLATION_INFO_FILE, checks=security_checks)
    except FileNotFoundError:
        if not ignore_non_existing:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": Could not open ", "default"),
                            ANSIThemeStr(f"{CMT_INSTALLATION_INFO_FILE}", "path"),
                            ANSIThemeStr("; aborting.", "default")], stderr=True)
            sys.exit(errno.ENOENT)

    if info is None or info.get("installation_target") is None \
            or (info.get("installation_target") is not None
                and cluster_name is not None and cluster_name not in info):
        if info is None:
            info = {}
        info["installation_target"] = cluster_name
        info[cluster_name] = {
            "distro": "<none>",
            "version": "<none>",
            "requested_version": "<none>",
            "state": "<none>",
            "phase": "<none>",
            "phase_skiplist": [],
            "cni": "<none>",
            "pod_network_cidr": "<none>",
            "cri": "<none>",
        }
    elif info.get("installation_target") is None and cluster_name is not None:
        # Old format file; transition it
        tmpinfo = info.copy()
        tmpinfo.pop("cluster_name")
        info = {}
        info["installation_target"] = cluster_name
        info[cluster_name] = tmpinfo

    return info


# pylint: disable-next=too-many-branches
def update_installation_info(**kwargs: Any) -> dict:
    """
    Update installation info for a cluster.

        Parameters:
            **kwargs (dict[str, Any]): Keyword arguments
                installation_target (str): The installation info entry to use
                cluster_name (str): The name of the cluster to update information for
                distro (str): The distribution used during installation
                              (currently the only supported distro is kubeadm)
                version (str): The current version of Kubernetes
                requested_version (str): The requested version of Kubernetes
                state (str): The installation state
                phase (str): The installation phase
                phase_skiplist (list[str]): A list of phases to skip
                cni (str): The CNI to use
                pod_network_cidr (str): The CIDR to use for the pod network
                cri (str): The CRI to use
                control_planes ([str]): A list of the control planes
                nodes ([str]): A list of the nodes
        Returns:
            (dict): The updated installation info
    """
    installation_target: Optional[str] = deep_get(kwargs, DictPath("installation_target"))
    cluster_name: Optional[str] = deep_get(kwargs, DictPath("cluster_name"))
    distro: Optional[str] = deep_get(kwargs, DictPath("distro"))
    version: Optional[str] = deep_get(kwargs, DictPath("version"))
    requested_version: Optional[str] = deep_get(kwargs, DictPath("requested_version"))
    state: Optional[str] = deep_get(kwargs, DictPath("state"))
    phase: Optional[Union[int, str]] = deep_get(kwargs, DictPath("phase"))
    phase_skiplist: Optional[list[str]] = deep_get(kwargs, DictPath("phase_skiplist"))
    cni: Optional[str] = deep_get(kwargs, DictPath("cni"))
    pod_network_cidr: Optional[str] = deep_get(kwargs, DictPath("pod_network_cidr"))
    cri: Optional[str] = deep_get(kwargs, DictPath("cri"))
    control_planes: Optional[list[str]] = deep_get(kwargs, DictPath("control_planes"))
    nodes: Optional[list[str]] = deep_get(kwargs, DictPath("nodes"))

    info = get_installation_info(cluster_name=cluster_name)

    if cluster_name is None:
        cluster_name = info.get("installation_target")

    if installation_target is not None:
        info["installation_target"] = installation_target
    elif info.get("installation_target") is None:
        info["installation_target"] = cluster_name

    if distro is not None:
        info[cluster_name]["distro"] = distro
    if version is not None:
        info[cluster_name]["version"] = version
    if requested_version is not None:
        info[cluster_name]["requested_version"] = requested_version
    if state is not None:
        info[cluster_name]["state"] = state
    if phase is not None:
        info[cluster_name]["phase"] = phase
    if phase_skiplist is not None:
        info[cluster_name]["phase_skiplist"] = phase_skiplist
    if cni is not None:
        info[cluster_name]["cni"] = cni
    if pod_network_cidr is not None:
        info[cluster_name]["pod_network_cidr"] = pod_network_cidr
    if cri is not None:
        info[cluster_name]["cri"] = cri
    if control_planes is not None:
        info[cluster_name]["control_planes"] = control_planes
    if nodes is not None:
        info[cluster_name]["nodes"] = nodes

    secure_write_yaml(CMT_INSTALLATION_INFO_FILE, info, sort_keys=False)

    return info


def check_and_print_status(retval: bool) -> None:
    """
    A wrapper that prints OK if retval is True
    and NOT OK and aborts if retval is False.

        Parameters:
            retval (bool): True on success, False on failure
    """
    if retval:
        ansithemeprint([ANSIThemeStr("OK", "ok")])
    else:
        ansithemeprint([ANSIThemeStr("NOT OK", "notok"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)


def check_version_from_url(url: str, version_regex: str) -> list[str]:
    """
    Given a URL download a text file and treat the first line
    that matches version_regex as a version number.

        Parameters:
            url (str): A URL
            version_regex (str): A regex
        Returns:
            (str): The version number, or None in case of failure
    """
    version: list[str] = []

    if url is not None:
        with tempfile.TemporaryDirectory() as td:
            check_and_print_status(download_files(td, [(url, "version.txt", None, None)],
                                                  permissions=0o600))
            tmp = secure_read_string(FilePath(f"{td}/version.txt"))
            versionoutput = tmp.splitlines()
            _version_regex: re.Pattern[str] = re.compile(version_regex)
            for line in versionoutput:
                tmp_match = _version_regex.match(line)
                if tmp_match is not None:
                    version = list(tmp_match.groups())
                    break
    return version


def check_version_from_executable(command: FilePath, args: list[str],
                                  version_regex: str) -> list[str]:
    """
    Given a path to an executable, the arguments needed to show version information,
    and a version_regex, return the executable version.

        Parameters:
            command (str): A path to an executable
            args ([str]): A list of arguments necessary to show version information
            version_regex (str): A regex
        Returns:
            ([str]): A list of version number elements, or None in case of failure
    """
    version: list[str] = []

    security_policy = SecurityPolicy.ALLOWLIST_RELAXED
    fallback_allowlist = ["/bin", "/sbin", "/usr/bin", "/usr/sbin",
                          "/usr/local/bin", "/usr/local/sbin", f"{HOMEDIR}/bin"]

    try:
        cpath = cmtio.secure_which(command, fallback_allowlist=fallback_allowlist,
                                   security_policy=security_policy)
    except FileNotFoundError:
        cpath = None

    if cpath is not None:
        result = execute_command_with_response([cpath] + args)

        if result is not None:
            versionoutput = result.splitlines()
            _version_regex: re.Pattern[str] = re.compile(version_regex)
            for line in versionoutput:
                tmp = _version_regex.match(line)
                if tmp is not None:
                    version = list(tmp.groups())
                    break
    return version


# pylint: disable-next=too-many-locals,too-many-branches,too-many-statements
def __upgrade_cni(cni: str, upgradetype: str, context: str,
                  pod_network_cidr: str, **kwargs: Any) -> None:
    """
    A helper that is used when upgrading a CNI;
    it can either upgrade the CNI itself or a helper executable.

        Parameters:
            cni (str): The CNI to upgrade
            upgradetype (str): Valid options CNI, executable
            context (str): The cluster context
            pod_network_cidr (str): The CIDR of the pod network
            cni_version (str): A hardcoded CNI version
    """
    verbose = deep_get(kwargs, DictPath("verbose"), False)
    cni_version = deep_get(kwargs, DictPath("cni_version"))

    if upgradetype not in ("CNI", "executable"):
        raise ValueError(f"Unknown upgradetype {upgradetype}; this is a programming error.")

    # FIXME: for now we hardcode this
    arch = "amd64"
    version_command = deep_get(cni_data, DictPath(f"{cni}#{upgradetype}#version_command"))
    version_command_regex = deep_get(cni_data, DictPath(f"{cni}#{upgradetype}#version_regex"))
    candidate_version_url = \
        deep_get(cni_data, DictPath(f"{cni}#{upgradetype}#candidate_version_url"))
    candidate_version_command = \
        deep_get(cni_data, DictPath(f"{cni}#{upgradetype}#candidate_version_command"))
    candidate_version_function = \
        deep_get(cni_data, DictPath(f"{cni}#{upgradetype}#candidate_version_function"))
    manual_candidate_version_regex = \
        deep_get(cni_data, DictPath(f"{cni}#{upgradetype}#manual_candidate_version_regex"))
    candidate_version_regex = \
        deep_get(cni_data, DictPath(f"{cni}#{upgradetype}#candidate_version_regex"))
    install_command = deep_get(cni_data, DictPath(f"{cni}#{upgradetype}#install"))
    upgrade_command = deep_get(cni_data, DictPath(f"{cni}#{upgradetype}#upgrade"))

    version_substitutions = {
        "<<<arch>>>": arch,
        "<<<context>>>": context,
    }

    version = []
    if version_command is not None:
        ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                        ANSIThemeStr(f"Checking {upgradetype} version", "action")])
        version = \
            check_version_from_executable(version_command[0],
                                          substitute_list(version_command[1:],
                                                          version_substitutions),
                                          version_command_regex)
    candidate_version = []
    if cni_version is not None:
        if (tmp_candidate_version := re.match(manual_candidate_version_regex, cni_version)) is None:
            check_and_print_status(False)
        # The type ignore below is necessary because mypy doesn't recognise that
        # check_and_print_status always aborts with sys.exit() when called with False
        candidate_version = list(tmp_candidate_version.groups())  # type: ignore
    elif candidate_version_url is not None:
        ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                        ANSIThemeStr(f"Checking {upgradetype} candidate version", "action")])
        if candidate_version_function is not None:
            tmp_candidate_version_fun = candidate_version_function(candidate_version_url,
                                                                   candidate_version_regex)
            if tmp_candidate_version_fun is None:
                check_and_print_status(False)
            candidate_version, _release_data, _body = tmp_candidate_version_fun
        else:
            candidate_version = \
                check_version_from_url(substitute_string(candidate_version_url,
                                                         version_substitutions),
                                       candidate_version_regex)
    elif candidate_version_command is not None:
        ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                        ANSIThemeStr(f"Checking {upgradetype} candidate version", "action")])
        candidate_version = \
            check_version_from_executable(candidate_version_command[0],
                                          substitute_list(candidate_version_command[1:],
                                                          version_substitutions),
                                          candidate_version_regex)

    if "urls" in deep_get(cni_data, DictPath(f"{cni}#{upgradetype}"), {}):
        # We should probably use semver here instead
        if version is not None and candidate_version is not None:
            is_new_version = None
            for v1, v2 in zip(version, candidate_version):
                if v1 != v2:
                    if v1.isnumeric() and v2.isnumeric():
                        is_new_version = int(v1) < int(v2)
                    else:
                        is_new_version = v1 < v2
                    break
            if is_new_version is None:
                is_new_version = len(candidate_version) > len(version)

        # pylint: disable-next=too-many-boolean-expressions
        if (version is None or not version) \
                or version is not None and (candidate_version is None or not candidate_version) \
                or is_new_version:
            new_version = "".join(candidate_version)

            ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                            ANSIThemeStr(f"Downloading {upgradetype} ", "action"),
                            ANSIThemeStr(f"{new_version}", "version")])

            substitutions = {
                "<<<version>>>": new_version,
                "<<<arch>>>": arch,
                "<<<context>>>": context,
            }

            kubectl_path = secure_which(FilePath("/usr/bin/kubectl"),
                                        fallback_allowlist=["/etc/alternatives"],
                                        security_policy=SecurityPolicy.ALLOWLIST_RELAXED)
            if upgradetype == "CNI":
                secure_mkdir(DEPLOYMENT_DIR)
                directory = DEPLOYMENT_DIR.joinpath("cni")
                permissions = 0o644
            elif upgradetype == "executable":
                directory = BINDIR
                permissions = 0o755
            else:
                raise ValueError(f"Unknown upgradetype {upgradetype}; this is a programming error.")

            secure_mkdir(directory)

            for url in deep_get(cni_data, DictPath(f"{cni}#{upgradetype}#urls"), []):
                download_url = deep_get(url, DictPath("url"), "")
                checksum_url = deep_get(url, DictPath("checksum_url"), None)
                checksum_type = deep_get(url, DictPath("checksum_type"), None)
                filename = deep_get(url, DictPath("filename"), "")

                download_url = substitute_string(download_url, substitutions)
                checksum_url = substitute_string(checksum_url, substitutions)
                filename = substitute_string(filename, substitutions)

                ansithemeprint([ANSIThemeStr("\n  • ", "separator"),
                                ANSIThemeStr("Downloading ", "subaction"),
                                ANSIThemeStr(f"{filename}", "version")])
                check_and_print_status(download_files(directory,
                                       [(download_url, filename, checksum_url, checksum_type)],
                                       permissions=permissions))

                if upgradetype == "CNI":
                    new_path = FilePath(directory).joinpath(filename)
                    args = [str(kubectl_path)]
                    kubectl_major_version, kubectl_minor_version, _kubectl_git_version, \
                        _server_major_version, _server_minor_version, _server_git_version = \
                        kubernetes_helper.kubectl_get_version()
                    if kubectl_major_version is None or kubectl_minor_version is None:
                        ansithemeprint([ANSIThemeStr("Critical", "critical"),
                                        ANSIThemeStr(": Could not extract ", "default"),
                                        ANSIThemeStr("kubectl", "programname"),
                                        ANSIThemeStr(" version; aborting.",
                                                     "default")], stderr=True)
                        sys.exit(errno.ENOENT)
                    if kubectl_major_version <= 1 and kubectl_minor_version < 22:
                        args += ["create"]
                    else:
                        args += ["apply", "--server-side"]
                    args += ["-f", new_path]

                    patch_cni_call = deep_get(url, DictPath("patch"))
                    if patch_cni_call is not None:
                        ansithemeprint([ANSIThemeStr("\n  • ", "separator"),
                                        ANSIThemeStr("Patching ", "subaction"),
                                        ANSIThemeStr(f"{filename}", "version")])
                        check_and_print_status(patch_cni_call(new_path, pod_network_cidr))
                    ansithemeprint([ANSIThemeStr("\n  • ", "separator"),
                                    ANSIThemeStr("Applying ", "subaction"),
                                    ANSIThemeStr(f"{filename}", "version")])
                    if verbose:
                        # This should check whether kubectl is >= 1.22;
                        # otherwise we need to use create
                        check_and_print_status(execute_command(args))
                    else:
                        # This should check whether kubectl is >= 1.22;
                        # otherwise we need to use create
                        execute_command_with_response(args)
        else:
            ansithemeprint([ANSIThemeStr("No newer version available.", "default")])
    elif install_command is not None and (version is None or len(version) == 0):
        new_version = "".join(candidate_version)

        ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                        ANSIThemeStr(f"Installing {upgradetype} ", "action"),
                        ANSIThemeStr(f"{new_version}", "version")])

        substitutions = {
            "<<<version>>>": new_version,
            "<<<arch>>>": arch,
            "<<<context>>>": context,
        }
        check_and_print_status(execute_command(substitute_list(install_command, substitutions)))
    elif upgrade_command is not None:
        if version is None or version < candidate_version:
            new_version = "".join(candidate_version)

            ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                            ANSIThemeStr(f"Upgrading {upgradetype} to ", "action"),
                            ANSIThemeStr(f"{new_version}", "version")])

            substitutions = {
                "<<<version>>>": new_version,
                "<<<arch>>>": arch,
                "<<<context>>>": context,
            }
            check_and_print_status(execute_command(substitute_list(upgrade_command, substitutions)))
        else:
            ansithemeprint([ANSIThemeStr("No newer version available.", "default")])


# pylint: disable-next=too-many-branches
def setup_cni(options: list[tuple[str, str]], args: list[str]) -> None:
    """
    Install and configure the specified CNI.

        Parameters:
            options ([(opt, optarg)]): Options to use when executing this action
            args ([str]): The CNI to install and configure
                          (optional; if not specified the default CNI will be used)
    """
    # default options
    confirm = True
    reinstall = False
    verbose = False

    for opt, _optarg in options:
        if opt == "--reinstall":
            reinstall = True
        elif opt == "-Y":
            confirm = False
        elif opt == "--verbose":
            verbose = True

    if not Path(CMT_INSTALLATION_INFO_FILE).is_file():
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": Could not find reliable installation information. "
                                     "Aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)

    installation_info = get_installation_info()
    cluster_name = installation_info["installation_target"]
    cni = deep_get(installation_info, DictPath(f"{cluster_name}#cni"))
    admin_name = f"kubernetes-admin+{cluster_name}"
    context_name = f"{admin_name}@{cluster_name}"

    if cni not in ("", "none", "<none>", "<unknown>") and not reinstall:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": The cluster already has a CNI installed; aborting.",
                                     "default")], stderr=True)
        sys.exit(errno.EEXIST)

    if not args:
        if not reinstall:
            ansithemeprint([ANSIThemeStr("Note", "note"),
                            ANSIThemeStr(": No CNI specified; defaulting to ", "default"),
                            ANSIThemeStr(f"{DEFAULT_CNI}", "programname"),
                            ANSIThemeStr(".", "default")], stderr=True)
    else:
        if reinstall and cni != args[0]:
            ansithemeprint([ANSIThemeStr("Error", "note"),
                            ANSIThemeStr("--reinstall", "option"),
                            ANSIThemeStr(" cannot be used to change CNI; aborting.",
                                         "default")], stderr=True)
            sys.exit(errno.EINVAL)
        cni = args[0]

    if confirm:
        input_retval = \
            ansithemeinput([ANSIThemeStr("\nSetup CNI (", "default"),
                            ANSIThemeStr(f"{cni}", "argument"),
                            ANSIThemeStr(")? [y/", "default"),
                            ANSIThemeStr("N", "emphasis"),
                            ANSIThemeStr("]: ", "default")])
        if input_retval.lower() not in ("y", "yes"):
            ansithemeprint([ANSIThemeStr("\nAborting:", "error"),
                            ANSIThemeStr(" User stopped CNI setup.", "default")], stderr=True)
            sys.exit(errno.EINTR)

    ansithemeprint([ANSIThemeStr("\n[Installing and configuring CNI]", "phase")])
    result = __setup_cni(cni, context_name, cluster_name, verbose=verbose)
    if result:
        update_installation_info(cluster_name=cluster_name, cni=cni)


# pylint: disable-next=unused-argument,too-many-locals,too-many-branches,too-many-statements
def teardown_cni(options: list[tuple[str, str]], args: list[str]) -> None:
    """
    Teardown the CNI.

        Parameters:
            options ([(opt, optarg)]): Options to use when executing this action
            args ([str]): Unused
    """
    global kh  # pylint: disable=global-statement

    confirm = True
    cni_version = None

    for opt, optarg in options:
        if opt == "--cni-version":
            cni_version = optarg
        elif opt == "-Y":
            confirm = False

    if not Path(CMT_INSTALLATION_INFO_FILE).is_file():
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": Could not find reliable installation information. "
                                     "Aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)

    installation_info = get_installation_info()
    cluster_name = installation_info["installation_target"]
    # We should get this using kubernetes_helper (and allow for manual override)
    kubectl_path = secure_which(FilePath("/usr/bin/kubectl"),
                                fallback_allowlist=["/etc/alternatives"],
                                security_policy=SecurityPolicy.ALLOWLIST_RELAXED)
    context = execute_command_with_response([kubectl_path,
                                             "config", "current-context"]).splitlines()[0]
    cni = installation_info[cluster_name]["cni"]
    if cni in ("", "<none>", "<unknown>"):
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": No CNI installed; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    directory = DEPLOYMENT_DIR.joinpath("cni")

    # This will only work for running clusters
    if kh is None:
        kh = kubernetes_helper.KubernetesHelper(about.PROGRAM_SUITE_NAME,
                                                about.PROGRAM_SUITE_VERSION, None)

    cnis = kh.identify_cni()
    if len(cnis) == 1:
        installed_cni = cnis[0][0]
        installed_cni_version = cnis[0][1]
    else:
        installed_cni = "<unknown>"
        installed_cni_version = "<unknown>"

    if installed_cni not in (cni, "<unknown>"):
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": ", "default"),
                        ANSIThemeStr(f"{CMT_INSTALLATION_INFO_FILE}", "path"),
                        ANSIThemeStr(" disagrees with the cluster about what CNI is installed",
                                     "default"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    if installed_cni_version == "<unknown>":
        if cni_version is None:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": the CNI version could not be identified; use ",
                                         "default"),
                            ANSIThemeStr("--cni-version", "option"),
                            ANSIThemeStr(" to specify it; aborting.", "default")], stderr=True)
            sys.exit(errno.ENOENT)
    else:
        if cni_version is None:
            cni_version = installed_cni_version

    cni_filenames = []
    uninstall_command = None

    urls = deep_get(cni_data, DictPath(f"{cni}#CNI#urls"), [])
    # We need to remove the files in reverse order
    for url in reversed(urls):
        filename = deep_get(url, DictPath("filename"), "")
        substitutions = {
            "<<<version>>>": cni_version,
        }
        if "<<<version>>>" in filename:
            filename = substitute_string(filename, substitutions)
        cni_filenames.append(filename)
    # We might use an uninstall command rather than files
    uninstall_command = deep_get(cni_data, DictPath(f"{cni}#CNI#uninstall"))

    if not cni_filenames and uninstall_command is None:
        ansithemeprint([ANSIThemeStr("Critical", "critical"),
                        ANSIThemeStr(": No data found for CNI. This is most likely a programming "
                                     "error; aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)

    for cni_filename in cni_filenames:
        cni_path = FilePath(directory).joinpath(cni_filename)

        security_violations = check_path(cni_path)

        if security_violations != [SecurityStatus.OK]:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(f": The CNI deployment file {cni_filename} is either "
                                         "missing or unsafe to use; aborting.",
                                         "default")], stderr=True)
            sys.exit(errno.EINVAL)

    all_nodes_in_cluster = ansible_get_hosts_by_group(ANSIBLE_INVENTORY, cluster_name)
    if not all_nodes_in_cluster:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": The inventory for this cluster does not contain any "
                                     "hosts; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    if confirm:
        input_retval = \
            ansithemeinput([ANSIThemeStr("\nUninstall CNI (", "default"),
                            ANSIThemeStr(f"{cni}", "argument"),
                            ANSIThemeStr(")? [y/", "default"),
                            ANSIThemeStr("N", "emphasis"),
                            ANSIThemeStr("]: ", "default")])
        if input_retval.lower() not in ("y", "yes"):
            ansithemeprint([ANSIThemeStr("\nAborting:", "error"),
                            ANSIThemeStr(" User stopped CNI removal.", "default")], stderr=True)
            sys.exit(errno.EINTR)

    ansithemeprint([ANSIThemeStr("\n[Removing CNI and configuration]", "phase")])

    for cni_filename in cni_filenames:
        cni_path = FilePath(directory).joinpath(cni_filename)

        # OK, we have paths to the CNI files; this should be enough to remove it
        kubectl_path = secure_which(FilePath("/usr/bin/kubectl"),
                                    fallback_allowlist=["/etc/alternatives"],
                                    security_policy=SecurityPolicy.ALLOWLIST_RELAXED)
        execute_command_with_response([kubectl_path, "delete", "-f", cni_path])

    if uninstall_command is not None:
        # In case we have an uninstall command we run that
        substitutions = {
            "<<<context>>>": context,
        }
        check_and_print_status(execute_command(substitute_list(uninstall_command, substitutions)))

    uninstall_cni_playbooks = [
        FilePath("teardown_cni.yaml"),
    ]
    playbooks = populate_playbooks_from_filenames(uninstall_cni_playbooks)

    retval = run_playbooks(playbooks=playbooks, hosts=all_nodes_in_cluster)
    if retval:
        ansithemeprint([ANSIThemeStr("OK", "ok")])
        update_installation_info(cluster_name=cluster_name, cni="<none>")
        ansithemeprint([ANSIThemeStr("\nNote", "note"),
                        ANSIThemeStr(": It is recommended to reboot the control plane(s) "
                                     "and all nodes after uninstalling the CNI\n", "default")])
    else:
        ansithemeprint([ANSIThemeStr("NOT OK", "notok"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)


# pylint: disable-next=too-many-branches,too-many-statements,too-many-locals
def upgrade_cni(options: list[tuple[str, str]], args: list[str]) -> None:
    """
    Upgrade the specified CNI to the latest version.

        Parameters:
            options ([(str, str)]): Options to use when executing this action
            args ([str]): The CNI to upgrade (optional; if not specified the CNI
                                              will be taken from installation_info.yaml)
    """
    global kh  # pylint: disable=global-statement

    pod_network_cidr = None
    verbose = False
    cni_version = None
    cni_executable_version = None

    if args:
        cni = args[0]
    else:
        installation_info = get_installation_info()
        cluster_name = installation_info["installation_target"]
        cni = deep_get(installation_info, DictPath(f"{cluster_name}#cni"))
        pod_network_cidr = deep_get(installation_info, DictPath(f"{cluster_name}#pod_network_cidr"))

    if pod_network_cidr is None:
        # If we have a running cluster and we are upgrading by CNI rather than through
        # installation info we need to get the Pod CIDR
        # by some other means; try this. If this fails we give up.
        if kh is None:
            kh = kubernetes_helper.KubernetesHelper(about.PROGRAM_SUITE_NAME,
                                                    about.PROGRAM_SUITE_VERSION, None)
        pod_network_cidr = kh.get_pod_network_cidr()
        if pod_network_cidr is None:
            installation_info = get_installation_info()
            cluster_name = installation_info["installation_target"]
            pod_network_cidr = \
                deep_get(installation_info, DictPath(f"{cluster_name}#pod_network_cidr"))
        if pod_network_cidr is None:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": Could not identify Pod network CIDR; aborting.",
                                         "default")], stderr=True)
            sys.exit(errno.ENOTSUP)

    if cni in ("", "<none>", "<unknown>"):
        if kh is None:
            kh = kubernetes_helper.KubernetesHelper(about.PROGRAM_SUITE_NAME,
                                                    about.PROGRAM_SUITE_VERSION, None)
        if not (tmp_cni := kh.identify_cni()):
            cni = "<unknown>"
        elif len(tmp_cni) > 1:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": Could not uniquely identify the CNI; multiple "
                                         "potential candidates identified; aborting.",
                                         "default")], stderr=True)
            sys.exit(errno.EINVAL)
        else:
            cni = tmp_cni[0][0]

    install_cni = False
    action_str = "Upgrading"
    for opt, optarg in options:
        if opt == "install":
            install_cni = True
            action_str = "Installing"
        elif opt == "--verbose":
            verbose = True
        elif opt == "--cni-version":
            cni_version = optarg
        elif opt == "--cni-executable-version":
            cni_executable_version = optarg

    if cni not in cni_data:
        ansithemeprint([ANSIThemeStr(f"{action_str} ", "default"),
                        ANSIThemeStr(f"{cni}", "command"),
                        ANSIThemeStr(" is currently not supported; exiting.", "default")])
        sys.exit(errno.ENOTSUP)

    kubectl_path = \
        secure_which(FilePath("/usr/bin/kubectl"),
                     fallback_allowlist=["/etc/alternatives"],
                     security_policy=SecurityPolicy.ALLOWLIST_RELAXED)
    context_name = \
        execute_command_with_response([kubectl_path, "config", "current-context"]).splitlines()[0]

    if install_cni:
        ansithemeprint([ANSIThemeStr("\n[Installing CNI]", "phase")])
    else:
        ansithemeprint([ANSIThemeStr("\n[Upgrading CNI]", "phase")])

    __upgrade_cni(cni, "executable", context_name, pod_network_cidr,
                  cni_version=cni_executable_version, verbose=verbose)
    __upgrade_cni(cni, "CNI", context_name, pod_network_cidr,
                  cni_version=cni_version, verbose=verbose)

    if install_cni:
        ansithemeprint([ANSIThemeStr("\nCNI installation successful", "success")])
    else:
        ansithemeprint([ANSIThemeStr("\nCNI upgrade successful", "success")])


def __setup_cni(cni: str, context_name: str, cluster_name: str, **kwargs: Any) -> bool:
    """
    Setup a CNI.

        Parameters:
            cni (str): The CNI to configure and install
            context_name (str): The name of the cluster context
            cluster_name (str): The name of the cluster
            cni_version (str): A hardcoded CNI version
            cni_executable_version (str): A hardcoded CNI executable version
        Returns:
            result (bool): True on success, False on failure
    """
    global kh  # pylint: disable=global-statement

    pass_cni = []
    verbose = deep_get(kwargs, DictPath("verbose"), False)
    cni_version = deep_get(kwargs, DictPath("cni_version"))
    cni_executable_version = deep_get(kwargs, DictPath("cni_executable_version"))

    installation_info = get_installation_info()
    cluster_name = deep_get(installation_info, DictPath("installation_target"))
    installation_info_cni = deep_get(installation_info, DictPath(f"{cluster_name}#cni"))
    if installation_info_cni is None or installation_info_cni in ("none", "<none>"):
        pass_cni = [cni]

    if cni == "none":
        ansithemeprint([ANSIThemeStr("Note", "note"),
                        ANSIThemeStr(": ", "default"),
                        ANSIThemeStr("none", "argument"),
                        ANSIThemeStr(" specified; skipping CNI installation.", "default")],
                       stderr=True)
        return True

    if cni not in cni_data:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": ", "default"),
                        ANSIThemeStr(f"{cni}", "argument"),
                        ANSIThemeStr(" is not a valid/supported CNI; aborting.", "default")],
                       stderr=True)
        sys.exit(errno.EINVAL)

    if kh is None:
        kh = kubernetes_helper.KubernetesHelper(about.PROGRAM_SUITE_NAME,
                                                about.PROGRAM_SUITE_VERSION, None)

    # upgrade_cni doesn't take a context, so we need to set the current context before calling it,
    # otherwise we might end up performing the changes on the wrong cluster
    if not kh.set_context(name=context_name, unchanged_is_success=True):
        ansithemeprint([ANSIThemeStr("Warning", "warning"),
                        ANSIThemeStr(": Failed to change to context ", "default"),
                        ANSIThemeStr(f"{context_name}", "hostname"),
                        ANSIThemeStr(" for cluster ", "default"),
                        ANSIThemeStr(f"{cluster_name}", "hostname"),
                        ANSIThemeStr("; cannot install/upgrade CNI; skipping.", "default")])
        return False

    # Don't pass CNI as an argument here; installation_info contains that information
    # during installation. upgrade_cni() will assume that the cluster is functional
    # if run with a CNI, and thus tries to get all information from the cluster.
    # This will most likely fail during installation, since the cluster
    # hasn't started fully yet and thus cannot provide the necessary information
    upgrade_cni(options=[("install", ""),
                         ("--verbose", verbose),
                         ("--cni-version", cni_version),
                         ("--cni-executable-version", cni_executable_version)], args=pass_cni)
    # FIXME
    return True


def check_for_ssh_key() -> bool:
    """
    Check whether there's an existing public ssh key on the system already.

        Returns:
            retval (bool): True if a key exists, False if no key exists
    """
    retval = False
    files = [file for file in glob(os.path.join(HOMEDIR, ".ssh", "*.pub")) if os.path.isfile(file)]
    for file in files:
        if os.path.isfile(file) and os.path.isfile(file.removesuffix(".pub")):
            retval = True
            break
    return retval


def create_ssh_key() -> bool:
    """
    Create a new ssh key (ECDSA-P521 format).

        Returns:
            retval (bool): True on success, False on failure
    """
    if not os.path.exists(f"{HOMEDIR}/.ssh/id_ecdsa"):
        args = [SSH_KEYGEN_BIN_PATH] + SSH_KEYGEN_ARGS + ["-f", f"{SSH_DIR}/id_ecdsa"]
        retval = execute_command(args)
    else:
        retval = True
    return retval


def add_ssh_keys_to_authorized_keys() -> bool:
    """
    Add all public keys for this host to authorized keys on this host;
    in other words, make the host able to SSH to itself.

        Returns:
            retval (bool): True on success, False on failure
    """
    retval: bool = False

    # Since we run create_ssh_key() before this task we can safely assume that .ssh/ exists
    for path in Path(SSH_DIR).iterdir():
        if not path.name.endswith(".pub"):
            continue

        tmp = secure_read_string(FilePath(path))
        if tmp is not None:
            tmplines = tmp.splitlines()
            pubkey = tmplines[0]
            if len(pubkey) == 0:
                continue

        exists = False

        authorized_keys_path = SSH_DIR.joinpath("authorized_keys")

        try:
            tmp = secure_read_string(authorized_keys_path)
            if tmp is not None and tmp:
                for line in tmp:
                    if line == pubkey[0]:
                        exists = True
                        break
        except FilePathAuditError as e:
            if "SecurityStatus.DOES_NOT_EXIST" in str(e):
                pass

        if not exists:
            secure_write_string(authorized_keys_path, f"{pubkey}\n", write_mode="a")

        # We've added at least one public key
        retval = True

    return retval


# pylint: disable-next=unused-argument
def __task_check_and_create_ssh_key(installation_info: dict, verbose: bool = False) -> None:
    """
    An installer task that checks whether the system has an ssh-key,
    and creates one if not.

        Parameters:
            installation_info (dict): A dict with installation information (Unused)
    """
    # FIXME
    confirm: bool = True

    if not check_for_ssh_key():
        if confirm:
            input_retval = \
                ansithemeinput([ANSIThemeStr("Warning", "warning"),
                                ANSIThemeStr(": No ssh key found in ", "default"),
                                ANSIThemeStr(f"{HOMEDIR}/.ssh", "path"),
                                ANSIThemeStr("; create one now? (No will abort the installation) "
                                             "[y/", "default"),
                                ANSIThemeStr("N", "emphasis"),
                                ANSIThemeStr("]: ", "default")])
            if input_retval.lower() not in ("y", "yes"):
                ansithemeprint([ANSIThemeStr("\nAborting:", "error"),
                                ANSIThemeStr(" No ssh key available.", "default")], stderr=True)
                sys.exit(errno.ENOENT)
        check_and_print_status(create_ssh_key())


# pylint: disable-next=unused-argument
def __task_scan_and_add_ssh_keys(installation_info: dict, verbose: bool = False) -> None:
    """
    An installer task that scans all specified control planes for public ssh keys
    and adds them to known hosts.

        Parameters:
            installation_info (dict): A dict with installation information (Unused)
    """
    cluster_name = installation_info["installation_target"]
    nodes = deep_get(installation_info, DictPath(f"{cluster_name}#nodes"), [])

    controlplanes = get_control_planes()
    hosts = [controlplane[0] for controlplane in controlplanes]
    hosts += [f"{controlplane[0]}.local" for controlplane in controlplanes]
    hosts += [
        "localhost",
    ]
    hosts += nodes

    scan_and_add_ssh_keys(hosts)


# pylint: disable-next=unused-argument
def __task_add_ssh_keys_to_inventory(installation_info: dict, verbose: bool = False) -> None:
    """
    An installer task that adds SSH keys to the Ansible inventory.

        Parameters:
            installation_info (dict): A dict with installation information (Unused)
    """
    pubkey = None
    d = ansible_get_inventory_dict()
    __vars = deep_get(d, DictPath("all#vars"), {})
    __authorized_keys = deep_get(__vars, DictPath("authorized_keys"), [])
    found_key = False

    for path in Path(SSH_DIR).iterdir():
        if not str(path).endswith(".pub"):
            continue

        tmp = secure_read_string(FilePath(path))
        if tmp is not None:
            tmplines = tmp.splitlines()
            pubkey = tmplines[0]

        if tmp is None or len(pubkey) == 0:
            ansithemeprint([ANSIThemeStr("Warning", "warning"),
                            ANSIThemeStr(": Failed to read a key from ", "default"),
                            ANSIThemeStr(str(path), "path"),
                            ANSIThemeStr("; skipping.", "default")], stderr=True)
            continue

        if pubkey not in __authorized_keys:
            __authorized_keys.append(pubkey)

        found_key = True

    if not found_key:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": Could not find a valid public key in ", "default"),
                        ANSIThemeStr(SSH_DIR, "path"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    __vars["authorized_keys"] = __authorized_keys
    ansible_set_vars(ANSIBLE_INVENTORY, "all", __vars)


# pylint: disable-next=unused-argument
def __task_check_and_add_ssh_keys_to_authorized_keys(**kwargs: Any) -> None:
    """
    An installer task that adds SSH keys to authorized keys.

        Parameters:
            **kwargs (dict[str, Any]): Keyword arguments
    """
    add_ssh_keys_to_authorized_keys()


def install_ansible_posix() -> bool:
    """
    Install ansible-posix using ansible-galaxy; this is necessary on systems
    where the version of Ansible is too old to support certain actions.

        Returns:
            (bool): True on success, False on failure
    """
    # Old versions of ansible-galaxy does not have the list command;
    # if it does not work we just assume that ansible.posix is missing
    args = ["/usr/bin/ansible-galaxy", "collection", "list"]
    result = execute_command_with_response(args)

    if "COLLECTION_ACTION: invalid choice" in result or "ansible.posix" not in result:
        http_proxy = deep_get(cmtlib.cmtconfig, DictPath("Network#http_proxy"), "")
        https_proxy = deep_get(cmtlib.cmtconfig, DictPath("Network#https_proxy"), "")
        no_proxy = deep_get(cmtlib.cmtconfig, DictPath("Network#no_proxy"), "")
        env = {
            "http_proxy": http_proxy,
            "https_proxy": https_proxy,
            "no_proxy": no_proxy,
        }
        args = ["/usr/bin/ansible-galaxy", "collection", "install", "ansible.posix"]
        return execute_command(args, env=env)

    return True


# pylint: disable-next=unused-argument
def __task_check_and_install_ansible_posix(installation_info: dict, verbose: bool = False) -> None:
    """
    An installer task that installs Ansible-posix.

        Parameters:
            installation_info (dict): A dict with installation information (Unused)
    """
    check_and_print_status(install_ansible_posix())


def update_apt_cache() -> bool:
    """
    Update the APT cache.

        Returns:
            (bool): True on success, False on failure
    """
    sudo_path = cmtio.secure_which(FilePath("sudo"), fallback_allowlist=["/bin", "/usr/bin"],
                                   security_policy=SecurityPolicy.ALLOWLIST_STRICT)
    apt_get_path = cmtio.secure_which(FilePath("apt-get"), fallback_allowlist=["/bin", "/usr/bin"],
                                      security_policy=SecurityPolicy.ALLOWLIST_STRICT)
    args = [sudo_path, apt_get_path, "update"]
    return execute_command(args)


def update_zypper_cache() -> bool:
    """
    Update the Zypper cache.

        Returns:
            (bool): True on success, False on failure
    """
    sudo_path = cmtio.secure_which(FilePath("sudo"), fallback_allowlist=["/bin", "/usr/bin"],
                                   security_policy=SecurityPolicy.ALLOWLIST_STRICT)
    apt_get_path = cmtio.secure_which(FilePath("zypper"), fallback_allowlist=["/bin", "/usr/bin"],
                                      security_policy=SecurityPolicy.ALLOWLIST_STRICT)
    args = [sudo_path, apt_get_path, "refresh"]
    return execute_command(args)


def deb_compare_versions(current_version: str, candidate_version: str) -> bool:
    """
    Compare two package versions.

        Returns:
            (bool): True if current version < candidate version, else False
    """
    args = ["/usr/bin/dpkg", "--compare-versions", current_version, "lt", candidate_version]
    return execute_command(args, comparison=1)


def __get_theme(string: str, default: str) -> str:
    """
    Return the suitable format reference for a particular string.

        Parameters:
            string (str): A string to return the format reference for
            default (str): The default format reference to use if there's no matching translation
        Returns:
            (str): A format reference
    """
    translation: dict[str, str] = {
        "<none>": "none",
        "<unknown>": "unknown",
    }
    return translation.get(string, default)


# pylint: disable-next=too-many-locals,too-many-branches,too-many-statements
def check_versions(pkg_packages: list[str], version_checks: list[tuple[str, list[str], str]],
                   verbose: bool = False) -> tuple[list[tuple[str, str, str, list[str]]],
                                                   list[tuple[str, str, str, list[str]]]]:
    """
    Check versions for all relevant software.

        Parameters:
            deb_packages ([str]): A list of debian packages
            version_checks ([(str, [str], str)]): A list of component, the command needed
                                                  to check its version, and a regex to
                                                  extract the version number
            verbose (bool): If the results are printed, should skipped tasks be printed too?
        Returns:
            ([(str, str, str, [str]), [(str, str, str, [str])]): A tuple of output from
                                                                 check_versions_apt(),
                                                                 (software, installed_version, "")
    """
    global kh  # pylint: disable=global-statement

    del verbose

    os_distro = identify_distro()

    other_versions: list[tuple[str, str, str, list[str]]] = []

    if os_distro in ("debian",):
        pkg_versions = check_versions_apt(pkg_packages)
    elif os_distro in ("suse",):
        pkg_versions = check_versions_zypper(pkg_packages)
    elif os_distro in ("fedora", "rhel",):
        pkg_versions = check_versions_yum(pkg_packages)

    kubernetes_installed_version = "<unknown>"
    if not (kubernetes_upstream_version := get_latest_upstream_version("kubernetes")):
        kubernetes_upstream_version = "<unknown>"

    # Get the API-server version (if available)
    # This will only work for running clusters
    if kh is None:
        kh = kubernetes_helper.KubernetesHelper(about.PROGRAM_SUITE_NAME,
                                                about.PROGRAM_SUITE_VERSION, None)

    _server_major_version, _server_minor_version, kubernetes_installed_version = \
        kh.get_api_server_version()

    if kubernetes_installed_version == kubernetes_upstream_version:
        kubernetes_upstream_version = ""

    if not (crc_upstream_version := get_latest_upstream_version("crc")):
        crc_upstream_version = "<unknown>"
    crc_installed_version = fieldgetters.fieldgetter_crc_version(fields=[0])[0]

    # Now check versions that need special checks
    for version_check in version_checks:
        tmp_software: str = version_check[0]
        args = version_check[1]
        regex = version_check[2]

        try:
            response = execute_command_with_response(args)
        except FileNotFoundError:
            other_versions.append((tmp_software, "<none>", "", []))
            continue
        except CalledProcessError:
            other_versions.append((tmp_software, "<unknown>", "", []))
            continue

        if regex is None:
            tmp_installed_version = response
        else:
            tmp_match = re.match(regex, response)
            if tmp_match is not None:
                tmp_installed_version = tmp_match[1]
            else:
                tmp_installed_version = "<none>"
        other_versions.append((tmp_software, tmp_installed_version, "", []))

    # Finally, display the gathered version information;
    # find the longest string of each type
    # Create lists of header + all values belonging to that header,
    # and get the length of the longest element
    slen = len(max(["Software:"] + [tmp[0] for tmp in pkg_versions + other_versions], key=len))
    slen = max(slen, len("Kubernetes (upstream)"))
    ilen = len(max(["Installed Version:"] + [tmp[1] for tmp in pkg_versions + other_versions],
                   key=len))

    # Print a header
    ansithemeprint([ANSIThemeStr("Software:", "header"),
                    ANSIThemeStr(f"{''.ljust(slen - len('Software:') + 2)}", "default"),
                    ANSIThemeStr("Installed Version:", "header"),
                    ANSIThemeStr(f"{''.ljust(ilen - len('Installed Version:') + 2)}", "default"),
                    ANSIThemeStr("Candidate Version:", "header")])

    ansithemeprint([ANSIThemeStr("Kubernetes ", "default"),
                    ANSIThemeStr("(upstream)".ljust(slen - len("Kubernetes ") + 2), "note"),
                    ANSIThemeStr(f"{kubernetes_installed_version}".ljust(ilen + 2), "version"),
                    ANSIThemeStr(f"{kubernetes_upstream_version}", "version")])

    if crc_installed_version:
        ansithemeprint([ANSIThemeStr("CRC ", "default"),
                        ANSIThemeStr("(upstream)".ljust(slen - len("CRC ") + 2), "note"),
                        ANSIThemeStr(f"{crc_installed_version}".ljust(ilen + 2), "version"),
                        ANSIThemeStr(f"{crc_upstream_version}", "version")])

    for software, installed_version, candidate_version, _ in \
            cast(list[tuple[str, str, str, list[str]]], natsorted(pkg_versions)):
        iformat = __get_theme(installed_version, "version")
        cformat = __get_theme(candidate_version, "version")
        ansithemeprint([ANSIThemeStr(f"{software.ljust(slen + 2)}", "default"),
                        ANSIThemeStr(f"{installed_version.ljust(ilen + 2)}", iformat),
                        ANSIThemeStr(f"{candidate_version}", cformat)])
    print()
    for software, installed_version, candidate_version, _ in \
            cast(list[tuple[str, str, str, list[str]]], natsorted(other_versions)):
        iformat = __get_theme(installed_version, "version")
        cformat = __get_theme(candidate_version, "version")
        ansithemeprint([ANSIThemeStr(f"{software.ljust(slen + 2)}", "default"),
                        ANSIThemeStr(f"{installed_version.ljust(ilen + 2)}", iformat),
                        ANSIThemeStr(f"{candidate_version}", cformat)])

    return pkg_versions, other_versions


# pylint: disable-next=too-many-locals
def run_playbook(playbookpath: FilePath, hosts: list[str], **kwargs: Any) -> tuple[int, dict]:
    """
    Run a playbook.

        Parameters:
            playbookpath (FilePath): A path to the playbook to run
            hosts (list[str]): A list of hosts to run the playbook on
            **kwargs (dict[str, Any]): Keyword arguments
                extra_values (dict): A dict of values to set before running the playbook
                quiet (bool): Should the results of the run be printed?
                verbose (bool): If the results are printed, should skipped tasks be printed too?
        Returns:
            ((int, dict)):
                (int): The return value from ansible_run_playbook_on_selection()
                (dict): A dict with the results from the run
    """
    extra_values: Optional[dict] = deep_get(kwargs, DictPath("extra_values"))
    quiet: bool = deep_get(kwargs, DictPath("quiet"), False)
    verbose: bool = deep_get(kwargs, DictPath("verbose"), False)

    # Set necessary Ansible keys before running playbooks
    http_proxy = deep_get(cmtlib.cmtconfig, DictPath("Network#http_proxy"), "")
    https_proxy = deep_get(cmtlib.cmtconfig, DictPath("Network#https_proxy"), "")
    no_proxy = deep_get(cmtlib.cmtconfig, DictPath("Network#no_proxy"), "")
    insecure_registries = deep_get(cmtlib.cmtconfig, DictPath("Docker#insecure_registries"), [])
    registry_mirrors = deep_get(cmtlib.cmtconfig, DictPath("Containerd#registry_mirrors"), [])

    use_proxy = "no"
    if http_proxy or https_proxy:
        use_proxy = "yes"

    if extra_values is None:
        extra_values = {}

    values = {
        "http_proxy": http_proxy,
        "https_proxy": https_proxy,
        "no_proxy": no_proxy,
        "insecure_registries": insecure_registries,
        "registry_mirrors": registry_mirrors,
        "use_proxy": use_proxy,
    }
    merged_values = {**values, **extra_values}

    retval, ansible_results = \
        ansible_run_playbook_on_selection(playbookpath, selection=hosts, values=merged_values,
                                          verbose=verbose, quiet=False)

    if not quiet:
        ansible_print_play_results(retval, ansible_results, verbose=verbose)

    return retval, ansible_results


def run_playbooks(playbooks: list[tuple[list[ANSIThemeStr], FilePath]],
                  hosts: Optional[list[str]] = None,
                  extra_values: Optional[dict] = None, verbose: bool = False) -> bool:
    """
    Run a set of playbooks.

        Parameters:
            playbooks ([([ANSIThemeStr], FilePath)]): A list of playbooks
            hosts ([str]): The hosts to run the playbooks on
            extra_values (dict): Variables to set before running the playbooks
            verbose (bool): If the results are printed, should skipped tasks be printed too?
        Returns:
            (bool): True on success, False on failure
    """
    if not playbooks or hosts is None:
        return True

    for string, playbookpath in playbooks:
        ansithemeprint(string)
        retval, _ansible_results = \
            run_playbook(playbookpath, hosts=hosts, extra_values=extra_values, verbose=verbose)

        # We do not want to continue executing playbooks if the first one failed
        if retval != 0:
            break

    return retval == 0


def __playbook_paths_from_path(playbook_path: FilePath) -> list[FilePath]:
    """
    Scan a directory and return a list of playbook paths.

        Parameters:
            playbook_path (str): A path to a directory
        Returns:
            ([FilePath]): A list of paths
    """
    if playbook_path is None:
        raise ValueError("No path passed to __playbook_paths_from_path(); "
                         "this is a programming error.")

    playbook_paths = []

    # Populate list of playbooks
    for path in Path(playbook_path).iterdir():
        # Do not process backups, etc.
        if path.name.startswith(("~", ".")):
            continue
        if not path.name.endswith((".yml", ".yaml")):
            continue
        playbook_paths.append(FilePath(str(path)))

    return playbook_paths


# Add all playbooks in the directory
def populate_playbooks_from_dir(path: FilePath) -> list[tuple[list[ANSIThemeStr], FilePath]]:
    """
    Populate a playbook list from path.

        Parameters:
            paths (FilePath): A directory to populate playbooks from
        Returns:
            ([([ANSIThemeStr], FilePath)]): A playbook list for use with run_playbooks()
    """
    playbook_paths = __playbook_paths_from_path(path)

    return populate_playbooks_from_filenames(playbook_paths)


# pylint: disable-next=unused-argument
def __task_request_ansible_password(installation_info: dict, verbose: bool = False) -> None:
    """
    An installer task that requests the ansible password.

        Parameters:
            installation_info (dict): A dict with installation information (Unused)
    """
    # Check whether ansible_password is defined or not
    if deep_get(ansible_configuration, DictPath("ansible_password")) is None and not no_password:
        ansithemeprint([ANSIThemeStr("Attention", "warning"),
                        ANSIThemeStr(": To be able to run playbooks you need to provide the "
                                     "ansible/ssh", "default")])
        ansithemeprint([ANSIThemeStr("password. Since the systems will be reconfigured to use "
                                     "passwordless sudo", "default")])
        ansithemeprint([ANSIThemeStr("and ssh keys this is a one-time thing.\n", "default")])
        ansithemeprint([ANSIThemeStr("Note", "note"),
                        ANSIThemeStr(": If the remote host is already configured for passwordless "
                                     "sudo", "default")])
        ansithemeprint([ANSIThemeStr("and allows for login using a pre-generated SSH-key "
                                     "you can use", "default")])
        ansithemeprint([ANSIThemeStr("the “", "default"),
                        ANSIThemeStr("--no-password", "option"),
                        ANSIThemeStr("“ option to bypass this check.", "default")])
        ansible_password = ansithemeinput_password([ANSIThemeStr("\nPassword: ", "default")])
        if not ansible_password:
            ansithemeprint([ANSIThemeStr("\nError", "error"),
                            ANSIThemeStr(": Empty password; aborting.", "default")], stderr=True)
            sys.exit(errno.EINVAL)
        else:
            ansible_configuration["ansible_password"] = ansible_password


def __run_playbooks_on_selection(playbooks: list[tuple[list[ANSIThemeStr], FilePath]],
                                 selection: list[str], extra_values: Optional[dict] = None,
                                 verbose: bool = False) -> None:
    """
    A helper that runs a playbook list on a selection.

        Parameters:
            playbooks ([(description, playbookpath)]): A playbook list
            selection ([str]): A list of hosts
            extra_values (dict): A dict of values to set before running the playbook
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    check_and_print_status(run_playbooks(playbooks, hosts=selection,
                                         extra_values=extra_values, verbose=verbose))


def __selection_control_planes() -> list[str]:
    """
    Return a selection with all control planes.

        Returns:
            ([str]): A list of control planes
    """
    controlplanes_ = get_control_planes(fail_on_empty=False)
    return [controlplane[0] for controlplane in controlplanes_]


def __selection_localhost() -> list[str]:
    """
    Returns a list with the hostname of localhost.

        Returns:
            ([str]): A list with the hostname of localhost
    """
    return [socket.gethostname()]


# pylint: disable-next=unused-argument
def __generic_task_run_playbooks_on_selection(selection: list[str], installation_info: dict,
                                              playbook_dir: FilePath,
                                              extra_values: Optional[dict] = None,
                                              verbose: bool = False) -> None:
    """
    A generic helper to run playbooks on selection.

        Parameters:
            selection ([str]): A list of hosts
            installation_info (dict): A dict with installation information
            playbook_dir (FilePath): The directory of playbooks
            extra_values (dict): Additional values to pass along when running the playbooks
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    paths = __playbook_paths_from_path(playbook_dir)
    playbooks = populate_playbooks_from_filenames(paths)
    if extra_values is None:
        extra_values = {}
    __run_playbooks_on_selection(playbooks=playbooks, selection=selection,
                                 extra_values=extra_values, verbose=verbose)


def __task_run_prepreparation_playbooks(installation_info: dict, verbose: bool = False) -> None:
    """
    A task that runs pre-preparation playbooks.

        Parameters:
            installation_info (dict): A dict with installation information
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    # The default selection is control planes;
    # the playbooks need to explicitly specify if the play should be run on other hosts
    selection = __selection_control_planes()
    try:
        __generic_task_run_playbooks_on_selection(selection=selection,
                                                  installation_info=installation_info,
                                                  playbook_dir=CMT_PRE_PREPARE_DIR,
                                                  extra_values={}, verbose=verbose)
    except FileNotFoundError as e:
        if CMT_PRE_PREPARE_DIR not in str(e):
            raise


# pylint: disable-next=unused-argument
def __task_run_preparation_playbooks_on_localhost(installation_info: dict,
                                                  verbose: bool = False) -> None:
    """
    An installer task that runs preparation playbooks on localhost.

        Parameters:
            installation_info (dict): A dict with installation information (Unused)
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    os_distro = identify_distro()
    selection = __selection_localhost()
    playbooks = populate_playbooks_from_filenames(prepare_targets["localhost"]["playbooks"])
    packages = []
    held_packages = []

    # The first minor Kubernetes version available from the new repository is 1.24.0;
    # let's add all minor versions from 24 and up; the old repository is no longer usable.
    if not (kubernetes_upstream_version := get_latest_upstream_version("kubernetes")):
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": Could not get the latest upstream Kubernetes version; ",
                                     "default"),
                        ANSIThemeStr("this is either a network error or a bug; aborting.",
                                     "default")], stderr=True)
        sys.exit(errno.ENOENT)

    # Split the version tuple
    _upstream_major, upstream_minor, _rest = kubernetes_upstream_version.split(".")
    minor_versions = []
    for minor_version in range(24, int(upstream_minor) + 1):
        minor_versions.append(f"{minor_version}")

    if os_distro in ("debian",):
        packages = deep_get(prepare_targets, DictPath("localhost#deb_packages"), [])
        held_packages = deep_get(prepare_targets, DictPath("localhost#deb_packages_held"), [])
    elif os_distro in ("suse",):
        packages = deep_get(prepare_targets, DictPath("localhost#suse_packages"), [])
        held_packages = deep_get(prepare_targets, DictPath("localhost#suse_packages_held"), [])
    elif os_distro in ("fedora", "rhel",):
        packages = deep_get(prepare_targets, DictPath("localhost#fedora_packages"), [])
        held_packages = deep_get(prepare_targets, DictPath("localhost#fedora_packages_held"), [])

    extra_values = {
        "packages": packages,
        "held_packages": held_packages,
        "ansible_become_pass": deep_get(ansible_configuration, DictPath("ansible_password")),
        "ansible_ssh_pass": deep_get(ansible_configuration, DictPath("ansible_password")),
        "minor_versions": minor_versions,
    }
    __run_playbooks_on_selection(playbooks=playbooks, selection=selection,
                                 extra_values=extra_values, verbose=verbose)


# pylint: disable-next=too-many-locals
def __task_run_preparation_playbooks_on_hosts(installation_info: dict,
                                              verbose: bool = False) -> None:
    """
    An installer task that runs preparation playbooks on hosts.

        Parameters:
            installation_info (dict): A dict with installation information
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    cluster_name = installation_info["installation_target"]
    k8s_distro = installation_info[cluster_name]["distro"]
    os_distro = identify_distro()
    control_planes = deep_get(installation_info, DictPath(f"{cluster_name}#control_planes"), [])
    nodes = deep_get(installation_info, DictPath(f"{cluster_name}#nodes"), [])
    packages = []
    held_packages = []

    if not control_planes:
        selection = __selection_control_planes()
    else:
        selection = control_planes + nodes
    playbooks = populate_playbooks_from_filenames(prepare_targets[k8s_distro]["playbooks"])

    # The first minor Kubernetes version available from the new repository is 1.24.0;
    # let's add all minor versions from 24 and up; the old repository is no longer usable.
    if not (kubernetes_upstream_version := get_latest_upstream_version("kubernetes")):
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": Could not get the latest upstream Kubernetes version; ",
                                     "default"),
                        ANSIThemeStr("this is either a network error or a bug; aborting.",
                                     "default")], stderr=True)
        sys.exit(errno.ENOENT)

    # Split the version tuple
    _upstream_major, upstream_minor, _rest = kubernetes_upstream_version.split(".")
    minor_versions = []
    for minor_version in range(24, int(upstream_minor) + 1):
        minor_versions.append(f"{minor_version}")

    if os_distro in ("debian",):
        packages = deep_get(prepare_targets, DictPath(f"{k8s_distro}#deb_packages"), [])
        held_packages = deep_get(prepare_targets, DictPath(f"{k8s_distro}#deb_packages_held"), [])
    elif os_distro in ("suse",):
        packages = deep_get(prepare_targets, DictPath(f"{k8s_distro}#suse_packages"), [])
        held_packages = deep_get(prepare_targets, DictPath(f"{k8s_distro}#suse_packages_held"), [])
    elif os_distro in ("fedora", "rhel",):
        packages = deep_get(prepare_targets, DictPath(f"{k8s_distro}#fedora_packages"), [])
        held_packages = deep_get(prepare_targets,
                                 DictPath(f"{k8s_distro}#fedora_packages_held"), [])

    extra_values = {
        "packages": packages,
        "held_packages": held_packages,
        "ansible_become_pass": deep_get(ansible_configuration, DictPath("ansible_password")),
        "ansible_ssh_pass": deep_get(ansible_configuration, DictPath("ansible_password")),
        "minor_versions": minor_versions,
    }
    __run_playbooks_on_selection(playbooks=playbooks, selection=selection,
                                 extra_values=extra_values, verbose=verbose)


def __task_run_postpreparation_playbooks(installation_info: dict, verbose: bool = False) -> None:
    """
    A task that runs post-preparation playbooks.

        Parameters:
            installation_info (dict): A dict with installation information
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    # The default selection is control planes;
    # the playbooks need to explicitly specify if the play should be run on other hosts
    selection = __selection_control_planes()
    try:
        __generic_task_run_playbooks_on_selection(selection=selection,
                                                  installation_info=installation_info,
                                                  playbook_dir=CMT_POST_PREPARE_DIR,
                                                  extra_values={}, verbose=verbose)
    except FileNotFoundError as e:
        if CMT_POST_PREPARE_DIR not in str(e):
            raise


# pylint: disable-next=unused-argument
def __task_setup_bash_completion(installation_info: dict, verbose: bool = False) -> None:
    """
    An installer task that sets up bash completion.

        Parameters:
            installation_info (dict): A dict with installation information (Unused)
    """
    secure_mkdir(FilePath(HOMEDIR).joinpath(".local"), verbose=True)
    secure_mkdir(FilePath(HOMEDIR).joinpath(".local", "share"), verbose=True)
    secure_mkdir(BASH_COMPLETION_BASE_DIR, verbose=True)
    secure_mkdir(BASH_COMPLETION_DIR, verbose=True)

    args = ["/usr/bin/kubectl", "completion", "bash"]
    result = execute_command_with_response(args)
    secure_write_string(BASH_COMPLETION_DIR.joinpath("kubectl"), result)


def __task_run_presetup_playbooks(installation_info: dict, verbose: bool = False) -> None:
    """
    A task that runs pre-setup playbooks.

        Parameters:
            installation_info (dict): A dict with installation information
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    # The default selection is control planes;
    # the playbooks need to explicitly specify if the play should be run on other hosts
    selection = __selection_control_planes()
    try:
        __generic_task_run_playbooks_on_selection(selection=selection,
                                                  installation_info=installation_info,
                                                  playbook_dir=CMT_PRE_SETUP_DIR,
                                                  extra_values={}, verbose=verbose)
    except FileNotFoundError as e:
        if CMT_PRE_SETUP_DIR not in str(e):
            raise


def __task_run_setup_playbooks_on_localhost(installation_info: dict, verbose: bool = False) -> None:
    """
    An installer task that runs playbooks on localhost.

        Parameters:
            installation_info (dict): A dict with installation information
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    cluster_name = installation_info["installation_target"]
    pod_network_cidr = installation_info[cluster_name]["pod_network_cidr"]
    os_distro = identify_distro()
    selection = __selection_localhost()
    packages = []
    held_packages = []

    playbooks = \
        populate_playbooks_from_filenames(setup_control_plane_targets["localhost"]["playbooks"])

    if os_distro in ("debian",):
        packages = deep_get(setup_control_plane_targets, DictPath("localhost#deb_packages"), [])
        held_packages = \
            deep_get(setup_control_plane_targets, DictPath("localhost#deb_packages_held"), [])
    elif os_distro in ("suse",):
        packages = deep_get(setup_control_plane_targets, DictPath("localhost#suse_packages"), [])
        held_packages = \
            deep_get(setup_control_plane_targets, DictPath("localhost#suse_packages_held"), [])
    elif os_distro in ("fedora", "rhel",):
        packages = deep_get(setup_control_plane_targets, DictPath("localhost#fedora_packages"), [])
        held_packages = \
            deep_get(setup_control_plane_targets, DictPath("localhost#fedora_packages_held"), [])

    extra_values = {
        "cluster_name": cluster_name,
        "pod_network_cidr": pod_network_cidr,
        "packages": packages,
        "held_packages": held_packages,
    }
    __run_playbooks_on_selection(playbooks=playbooks, selection=selection,
                                 extra_values=extra_values, verbose=verbose)


def __task_run_setup_playbooks_on_control_planes(installation_info: dict,
                                                 verbose: bool = False) -> None:
    """
    An installer task that runs playbooks on control planes.

        Parameters:
            installation_info (dict): A dict with installation information
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    cluster_name = installation_info["installation_target"]
    k8s_distro = installation_info[cluster_name]["distro"]
    os_distro = identify_distro()
    pod_network_cidr = installation_info[cluster_name]["pod_network_cidr"]
    selection = __selection_control_planes()
    packages = []
    held_packages = []

    playbooks = \
        populate_playbooks_from_filenames(setup_control_plane_targets[k8s_distro]["playbooks"])
    cri = installation_info[cluster_name]["cri"]
    cri_socket = deep_get(cri_data[cri], DictPath("socket"))
    extra_values_setup = \
        deep_get(setup_control_plane_targets[k8s_distro], DictPath("extra_values"), {})

    if os_distro in ("debian",):
        packages = deep_get(setup_control_plane_targets, DictPath(f"{k8s_distro}#deb_packages"), [])
        held_packages = \
            deep_get(setup_control_plane_targets, DictPath(f"{k8s_distro}#deb_packages_held"), [])
    elif os_distro in ("suse",):
        packages = \
            deep_get(setup_control_plane_targets, DictPath(f"{k8s_distro}#suse_packages"), [])
        held_packages = \
            deep_get(setup_control_plane_targets, DictPath(f"{k8s_distro}#suse_packages_held"), [])
    elif os_distro in ("fedora", "rhel",):
        packages = \
            deep_get(setup_control_plane_targets, DictPath(f"{k8s_distro}#fedora_packages"), [])
        held_packages = \
            deep_get(setup_control_plane_targets,
                     DictPath(f"{k8s_distro}#fedora_packages_held"), [])

    if k8s_distro == "rke2":
        cri_socket = cri_socket.removeprefix("unix://")

    extra_values = {
        "cluster_name": cluster_name,
        "pod_network_cidr": pod_network_cidr,
        "cri_socket": cri_socket,
        "packages": packages,
        "held_packages": held_packages,
        **extra_values_setup,
    }
    __run_playbooks_on_selection(playbooks=playbooks, selection=selection,
                                 extra_values=extra_values, verbose=verbose)


def __task_run_postsetup_playbooks(installation_info: dict, verbose: bool = False) -> None:
    """
    A task that runs post-setup playbooks.

        Parameters:
            installation_info (dict): A dict with installation information
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    # The default selection is control planes;
    # the playbooks need to explicitly specify if the play should be run on other hosts
    selection = __selection_control_planes()
    try:
        __generic_task_run_playbooks_on_selection(selection=selection,
                                                  installation_info=installation_info,
                                                  playbook_dir=CMT_POST_SETUP_DIR,
                                                  extra_values={}, verbose=verbose)
    except FileNotFoundError as e:
        if CMT_POST_SETUP_DIR not in str(e):
            raise


# pylint: disable-next=unused-argument,too-many-locals
def __task_import_kubeconfig(installation_info: dict, verbose: bool = False) -> None:
    """
    An installer task that merges a Kube config into ~/.kube/config.

        Parameters:
            installation_info (dict): A dict with installation information
    """
    cluster_name = installation_info["installation_target"]
    config_file_name = FilePath(f"{HOMEDIR}/.kube/config.{cluster_name}")
    admin_name = f"kubernetes-admin+{cluster_name}"
    context_name = f"{admin_name}@{cluster_name}"

    # This read must not fail, so no exceptions
    d2 = secure_read_yaml(config_file_name)

    # This file will not exist if this is the first cluster
    security_checks = [
        SecurityChecks.PARENT_RESOLVES_TO_SELF,
        SecurityChecks.OWNER_IN_ALLOWLIST,
        SecurityChecks.PARENT_OWNER_IN_ALLOWLIST,
        SecurityChecks.PERMISSIONS,
        SecurityChecks.PARENT_PERMISSIONS,
        SecurityChecks.IS_FILE,
    ]

    try:
        d1 = secure_read_yaml(KUBE_CONFIG_FILE, checks=security_checks)
    except FileNotFoundError:
        d1 = dict(d2)
        # We will be renaming things anyway,
        # so instead of using a lot of special casing below we just empty these fields
        d1["clusters"] = []
        d1["contexts"] = []
        d1["current-context"] = context_name
        d1["users"] = []

    for cluster in d1["clusters"]:
        if cluster_name == cluster["name"]:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": A cluster named ", "default"),
                            ANSIThemeStr(cluster_name, "hostname"),
                            ANSIThemeStr(" already exists in ", "default"),
                            ANSIThemeStr(f"{HOMEDIR}/.kube/config", "path"),
                            ANSIThemeStr("; manual merge is necessary.", "default")], stderr=True)
            return

    for user in d1["users"]:
        if admin_name == user["name"]:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": A user named ", "default"),
                            ANSIThemeStr(admin_name, "hostname"),
                            ANSIThemeStr(" already exists in ", "default"),
                            ANSIThemeStr(f"{HOMEDIR}/.kube/config", "path"),
                            ANSIThemeStr("; manual merge is necessary.", "default")], stderr=True)
            return

    for context in d1["contexts"]:
        if context_name == context["name"]:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": A context named ", "default"),
                            ANSIThemeStr(context_name, "hostname"),
                            ANSIThemeStr(" already exists in ", "default"),
                            ANSIThemeStr(f"{HOMEDIR}/.kube/config", "path"),
                            ANSIThemeStr("; manual merge is necessary.", "default")], stderr=True)
            return

    cad = d2["clusters"][0]["cluster"].get("certificate-authority-data")
    server = d2["clusters"][0]["cluster"]["server"]

    insecure_skip_tls_verify = d2["clusters"][0]["cluster"].get("insecure-skip-tls-verify")
    cluster = {
        "cluster": {
            "server": server,
        },
        "name": cluster_name,
    }
    if cad is not None:
        cluster["cluster"]["certificate-authority-data"] = cad
    if insecure_skip_tls_verify is not None:
        cluster["cluster"]["insecure-skip-tls-verify"] = insecure_skip_tls_verify

    d1["clusters"].append(cluster)

    context = {
        "context": {
            "cluster": cluster_name,
            "user": admin_name,
        },
        "name": context_name,
    }
    d1["contexts"].append(context)

    ccd = d2["users"][0]["user"].get("client-certificate-data")
    ckd = d2["users"][0]["user"].get("client-key-data")
    token = d2["users"][0]["user"].get("token")
    user = {
        "user": {
        },
        "name": admin_name,
    }
    if ccd is not None:
        user["user"]["client-certificate-data"] = ccd
    if ckd is not None:
        user["user"]["client-key-data"] = ckd
    if token is not None:
        user["user"]["token"] = token

    d1["users"].append(user)

    secure_write_yaml(KUBE_CONFIG_FILE, d1, permissions=0o600, sort_keys=False)

    check_and_print_status(True)


# pylint: disable-next=unused-argument,too-many-locals
def __task_adjust_kubeconfig_server_address(installation_info: dict, verbose: bool = False) -> None:
    """
    An installer task that replaces localhost for the real IP-address of the API-server.

        Parameters:
            installation_info (dict): A dict with installation information
    """
    global kh  # pylint: disable=global-statement

    control_plane_name = __selection_control_planes()[0]
    cluster_name = installation_info["installation_target"]

    try:
        d = secure_read_yaml(KUBE_CONFIG_FILE)
    except FileNotFoundError:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": Could not open ", "default"),
                        ANSIThemeStr(f"{KUBE_CONFIG_FILE}", "path"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    for i, cluster in enumerate(deep_get(d, DictPath("clusters"), [])):
        if deep_get(cluster, DictPath("name"), "") == cluster_name:
            # Is this cluster a localhost?
            server = deep_get(cluster, DictPath("cluster#server"), "")
            if server.startswith("https://127") or server == "::1":
                tmp = re.match(r"^(https://)([^:]+)(:.*)", server)
                if tmp is None:
                    ansithemeprint([ANSIThemeStr("Error", "error"),
                                    ANSIThemeStr(": Failed to extract server address.",
                                                 "default")])
                    ansithemeprint([ANSIThemeStr("This is probably a programming error",
                                                 "default"),
                                    ANSIThemeStr("; aborting.", "default")], stderr=True)
                    sys.exit(errno.EINVAL)

                prefix = tmp[1]
                server_address = tmp[2]
                server_port_and_path = tmp[3]

                # Try to figure out the internal IP of the control plane;
                # this only works if we're running this on the control plane;
                # we should probably do this while fetching the config-file instead
                # This will only work for running clusters
                if kh is None:
                    kh = kubernetes_helper.KubernetesHelper(about.PROGRAM_SUITE_NAME,
                                                            about.PROGRAM_SUITE_VERSION, None)

                obj = kh.get_ref_by_kind_name_namespace(("Node", ""), control_plane_name, "")
                if obj is None:
                    ansithemeprint([ANSIThemeStr("Error", "error"),
                                    ANSIThemeStr(": Could not find control plane ", "default"),
                                    ANSIThemeStr(f"{control_plane_name}", "hostname"),
                                    ANSIThemeStr("; aborting.", "default")], stderr=True)
                    sys.exit(errno.ENOENT)
                addresses = deep_get(obj, DictPath("status#addresses"), [])
                internal_ip = None
                for address in addresses:
                    address_type = deep_get(address, DictPath("type"), "")
                    if address_type == "InternalIP":
                        internal_ip = deep_get(address, DictPath("address"), "")
                        break
                if internal_ip is not None and internal_ip and internal_ip != server_address:
                    d["clusters"][i]["cluster"]["server"] = \
                        f"{prefix}{internal_ip}{server_port_and_path}"

    secure_write_yaml(KUBE_CONFIG_FILE, d, permissions=0o600, sort_keys=False)


# pylint: disable-next=unused-argument
def __task_setup_cni(installation_info: dict, verbose: bool = False) -> None:
    """
    An installer task that sets up a CNI.

        Parameters:
            installation_info (dict): A dict with installation information
    """
    cluster_name = installation_info["installation_target"]
    cni = installation_info[cluster_name]["cni"]
    admin_name = f"kubernetes-admin+{cluster_name}"
    context_name = f"{admin_name}@{cluster_name}"

    # If everything is successful so far we deploy the pod network
    __setup_cni(cni, context_name, cluster_name)


# pylint: disable-next=unused-argument
def __task_drain_control_planes(installation_info: dict, verbose: bool = False) -> None:
    """
    An installer task that drains control plains.

        Parameters:
            installation_info (dict): A dict with installation information (Unused)
    """
    controlplanes = __selection_control_planes()

    # Check if there is an API-server running that will listen to our request;
    # drain is issued during teardown--so if we are resuming a teardown the cluster
    # might already be partially deconfigured
    # XXX: This should be done on the control plane(s), not on localhost
    args = ["/usr/bin/sudo", "/usr/bin/lsof", "-i", "-P", "-n"]
    response = execute_command_with_response(args)
    running = False
    for line in response.splitlines():
        if "6443 (LISTEN)" in line:
            running = True
            break
    if not running:
        check_and_print_status(True)
        return

    kubectl_major_version, kubectl_minor_version, *_ignore = kubernetes_helper.kubectl_get_version()
    if kubectl_major_version is None or kubectl_minor_version is None:
        ansithemeprint([ANSIThemeStr("Critical", "critical"),
                        ANSIThemeStr(": Could not extract ", "default"),
                        ANSIThemeStr("kubectl", "programname"),
                        ANSIThemeStr(" version; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    # Build the drain command line based on what version of kubectl is installed
    args = ["/usr/bin/kubectl", "drain", "--ignore-daemonsets"]
    if kubectl_major_version >= 1 and kubectl_minor_version >= 20:
        args.append("--delete-emptydir-data")
    else:
        args.append("--delete-local-data")
    if kubectl_major_version >= 1 and kubectl_minor_version >= 18:
        args.append("--disable-eviction")

    for controlplane in controlplanes:
        check_and_print_status(execute_command(args + [controlplane]))


# pylint: disable-next=unused-argument
def __task_uncordon_control_planes(installation_info: dict, verbose: bool = False) -> None:
    """
    An installer task that uncordons control planes.

        Parameters:
            installation_info (dict): A dict with installation information (Unused)
    """
    global kh  # pylint: disable=global-statement

    controlplanes = __selection_control_planes()

    # This will only work for running clusters
    if kh is None:
        kh = kubernetes_helper.KubernetesHelper(about.PROGRAM_SUITE_NAME,
                                                about.PROGRAM_SUITE_VERSION, None)

    for controlplane in controlplanes:
        ansithemeprint([ANSIThemeStr(f"  {controlplane}:", "hostname")])
        message, status = kh.uncordon_node(controlplane)
        if status in (200, 204):
            ansithemeprint([ANSIThemeStr("    Uncordoned", "success")])
            print(message)
        elif status == 42503:
            ansithemeprint([ANSIThemeStr("\n  ", "default"),
                            ANSIThemeStr("Critical", "critical"),
                            ANSIThemeStr(": Cluster not available; aborting", "default")],
                           stderr=True)
            sys.exit(errno.ENOENT)
        else:
            ansithemeprint([ANSIThemeStr("\n  ", "default"),
                            ANSIThemeStr("Critical", "critical"),
                            ANSIThemeStr(": API call returned error:", "default")], stderr=True)
            ansithemeprint([ANSIThemeStr(f"    {message}", "error")], stderr=True)
            sys.exit(errno.EINVAL)


def __task_run_preupgrade_playbooks(installation_info: dict, verbose: bool = False) -> None:
    """
    A task that runs pre-upgrade playbooks.

        Parameters:
            installation_info (dict): A dict with installation information
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    # The default selection is control planes;
    # the playbooks need to explicitly specify if the play should be run on other hosts
    selection = __selection_control_planes()
    try:
        __generic_task_run_playbooks_on_selection(selection=selection,
                                                  installation_info=installation_info,
                                                  playbook_dir=CMT_PRE_UPGRADE_DIR,
                                                  extra_values={}, verbose=verbose)
    except FileNotFoundError as e:
        if CMT_PRE_UPGRADE_DIR not in str(e):
            raise


# Run upgrade playbooks on control planes
def __task_run_upgrade_playbooks_on_control_planes(installation_info: dict,
                                                   verbose: bool = False) -> None:
    """
    A task that runs upgrade playbooks on control planes.

        Parameters:
            installation_info (dict): A dict with installation information
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    cluster_name = installation_info["installation_target"]
    k8s_distro = installation_info[cluster_name]["distro"]
    os_distro = identify_distro()
    pod_network_cidr = installation_info[cluster_name]["pod_network_cidr"]
    requested_version = installation_info[cluster_name]["requested_version"]
    packages = []
    held_packages = []
    selection = __selection_control_planes()
    playbooks = \
        populate_playbooks_from_filenames(upgrade_control_plane_targets[k8s_distro]["playbooks"])
    cri = installation_info[cluster_name]["cri"]
    cri_socket = deep_get(cri_data[cri], DictPath("socket"))
    extra_values_upgrade = \
        deep_get(upgrade_control_plane_targets[k8s_distro], DictPath("extra_values"), {})

    if os_distro in ("debian",):
        packages = \
            deep_get(upgrade_control_plane_targets,
                     DictPath(f"{k8s_distro}#deb_packages"), [])
        held_packages = \
            deep_get(upgrade_control_plane_targets,
                     DictPath(f"{k8s_distro}#deb_packages_held"), [])
    elif os_distro in ("suse",):
        packages = \
            deep_get(upgrade_control_plane_targets,
                     DictPath(f"{k8s_distro}#suse_packages"), [])
        held_packages = \
            deep_get(upgrade_control_plane_targets,
                     DictPath(f"{k8s_distro}#suse_packages_held"), [])
    elif os_distro in ("fedora", "rhel",):
        packages = \
            deep_get(upgrade_control_plane_targets,
                     DictPath(f"{k8s_distro}#fedora_packages"), [])
        held_packages = \
            deep_get(upgrade_control_plane_targets,
                     DictPath(f"{k8s_distro}#fedora_packages_held"), [])

    extra_values = {
        "cluster_name": cluster_name,
        "pod_network_cidr": pod_network_cidr,
        "cri_socket": cri_socket,
        "packages": packages,
        "held_packages": held_packages,
        "requested_control_plane_k8s_version": requested_version,
        **extra_values_upgrade,
    }
    __run_playbooks_on_selection(playbooks=playbooks, selection=selection,
                                 extra_values=extra_values, verbose=verbose)


def __task_run_postupgrade_playbooks(installation_info: dict, verbose: bool = False) -> None:
    """
    A task that runs post-upgrade playbooks.

        Parameters:
            installation_info (dict): A dict with installation information (Unused)
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    # The default selection is control planes;
    # the playbooks need to explicitly specify if the play should be run on other hosts
    selection = __selection_control_planes()
    try:
        __generic_task_run_playbooks_on_selection(selection=selection,
                                                  installation_info=installation_info,
                                                  playbook_dir=CMT_POST_UPGRADE_DIR,
                                                  extra_values={}, verbose=verbose)
    except FileNotFoundError as e:
        if CMT_POST_UPGRADE_DIR not in str(e):
            raise


# pylint: disable-next=unused-argument
def __task_verify_that_cluster_has_no_nodes(installation_info: dict, verbose: bool = False) -> None:
    """
    A task that verifies that no non-control plane nodes remain in the cluster.

        Parameters:
            installation_info (dict): A dict with installation information (Unused)
    """
    node_statuses = __get_node_info()

    if node_statuses is not None:
        for node in node_statuses:
            if "control-plane" not in node["roles"]:
                ansithemeprint([ANSIThemeStr("\nPlease delete all nodes (except the ", "warning"),
                                ANSIThemeStr("control plane", "emphasis"),
                                ANSIThemeStr(") from the cluster before attempting a teardown.",
                                             "warning")])
                sys.exit(errno.EAGAIN)

    ansithemeprint([ANSIThemeStr("OK", "ok")])


def __task_run_preteardown_playbooks(installation_info: dict, verbose: bool = False) -> None:
    """
    A task that runs pre-teardown playbooks.

        Parameters:
            installation_info (dict): A dict with installation information
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    # The default selection is control planes;
    # the playbooks need to explicitly specify if the play should be run on other hosts
    selection = __selection_control_planes()
    try:
        __generic_task_run_playbooks_on_selection(selection=selection,
                                                  installation_info=installation_info,
                                                  playbook_dir=CMT_PRE_TEARDOWN_DIR,
                                                  extra_values={}, verbose=verbose)
    except FileNotFoundError as e:
        if CMT_PRE_TEARDOWN_DIR not in str(e):
            raise


# Run teardown playbooks on control planes
def __task_run_teardown_playbooks_on_control_planes(installation_info: dict,
                                                    verbose: bool = False) -> None:
    """
    A task that runs teardown playbooks on control planes.

        Parameters:
            installation_info (dict): A dict with installation information
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    cluster_name = installation_info["installation_target"]
    k8s_distro = installation_info[cluster_name]["distro"]
    selection = __selection_control_planes()
    playbooks = \
        populate_playbooks_from_filenames(teardown_control_plane_targets[k8s_distro]["playbooks"])

    __run_playbooks_on_selection(playbooks=playbooks, selection=selection,
                                 extra_values={}, verbose=verbose)


def __task_run_postteardown_playbooks(installation_info: dict, verbose: bool = False) -> None:
    """
    A task that runs post-teardown playbooks.

        Parameters:
            installation_info (dict): A dict with installation information (Unused)
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    # The default selection is control planes;
    # the playbooks need to explicitly specify if the play should be run on other hosts
    selection = __selection_control_planes()
    try:
        __generic_task_run_playbooks_on_selection(selection=selection,
                                                  installation_info=installation_info,
                                                  playbook_dir=CMT_POST_TEARDOWN_DIR,
                                                  extra_values={}, verbose=verbose)
    except FileNotFoundError as e:
        if CMT_POST_TEARDOWN_DIR not in str(e):
            raise


# pylint: disable-next=unused-argument,too-many-branches
def __task_remove_kubeconfig(installation_info: dict, verbose: bool = False) -> None:
    """
    An installer task that removes a cluster configuration from ~/.kube/config.

        Parameters:
            installation_info (dict): A dict with installation information
    """
    if not Path(KUBE_CONFIG_DIR).is_dir():
        check_and_print_status(True)
        return

    cluster_name = installation_info["installation_target"]
    admin_name = f"kubernetes-admin+{cluster_name}"
    context_name = f"{admin_name}@{cluster_name}"
    d1 = None

    if os.path.exists(f"{HOMEDIR}/.kube/config.{cluster_name}"):
        os.remove(f"{HOMEDIR}/.kube/config.{cluster_name}")
    if os.path.exists(f"{HOMEDIR}/.kube/token.{cluster_name}"):
        os.remove(f"{HOMEDIR}/.kube/token.{cluster_name}")
    if os.path.exists(f"{HOMEDIR}/.kube/rke2.config.{cluster_name}.yaml"):
        os.remove(f"{HOMEDIR}/.kube/rke2.config.{cluster_name}.yaml")

    try:
        d1 = secure_read_yaml(KUBE_CONFIG_FILE)

        # This is not perfect--there might be leftover users and contexts
        # belonging to this cluster; but we better not go on a killing spree--better leave
        # cruft behind than remove everything.
        for i in range(0, len(d1["clusters"])):
            if d1["clusters"][i].get("name") == cluster_name:
                d1["clusters"].pop(i)
                break
        for i in range(0, len(d1["users"])):
            if d1["users"][i].get("name") == admin_name:
                d1["users"].pop(i)
                break
        for i in range(0, len(d1["contexts"])):
            if d1["contexts"][i].get("name") == context_name:
                d1["contexts"].pop(i)
                break
    except FileNotFoundError:
        pass

    # If there's no cluster left, remove the config completely
    if d1 is None or len(d1["clusters"]) == 0:
        secure_rm(KUBE_CONFIG_FILE, ignore_non_existing=True)

        _dir = os.listdir(f"{HOMEDIR}/.kube")
        try:
            _dir.remove("cache")
        except ValueError:
            pass
        try:
            _dir.remove("http-cache")
        except ValueError:
            pass

        # If there's nothing else than the cache subdirectories in .kube, remove .kube too
        if not _dir:
            shutil.rmtree(f"{HOMEDIR}/.kube")
    else:
        # Set the context to the first remaining context, if any
        if d1["contexts"]:
            d1["current-context"] = d1["contexts"][0]["name"]

        secure_write_yaml(KUBE_CONFIG_FILE, d1, permissions=0o600, sort_keys=False)

    check_and_print_status(True)


def __task_run_prepurge_playbooks(installation_info: dict, verbose: bool = False) -> None:
    """
    A task that runs pre-purge playbooks.

        Parameters:
            installation_info (dict): A dict with installation information
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    # The default selection is control planes;
    # the playbooks need to explicitly specify if the play should be run on other hosts
    selection = __selection_control_planes()
    try:
        __generic_task_run_playbooks_on_selection(selection=selection,
                                                  installation_info=installation_info,
                                                  playbook_dir=CMT_PRE_PURGE_DIR,
                                                  extra_values={}, verbose=verbose)
    except FileNotFoundError as e:
        if CMT_PRE_PURGE_DIR not in str(e):
            raise


# Run purge playbooks on control planes
def __task_run_purge_playbooks_on_control_planes(installation_info: dict,
                                                 verbose: bool = False) -> None:
    """
    An installer task that runs purge playbooks on control planes.

        Parameters:
            installation_info (dict): A dict with installation information
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    cluster_name = installation_info["installation_target"]
    k8s_distro = installation_info[cluster_name]["distro"]
    os_distro = identify_distro()
    selection = __selection_control_planes()
    packages = []
    held_packages = []
    playbooks = \
        populate_playbooks_from_filenames(purge_control_plane_targets[k8s_distro]["playbooks"])

    if os_distro in ("debian",):
        packages = \
            deep_get(purge_control_plane_targets, DictPath(f"{k8s_distro}#deb_packages"), [])
        held_packages = \
            deep_get(purge_control_plane_targets, DictPath(f"{k8s_distro}#deb_packages_held"), [])
    elif os_distro in ("suse",):
        packages = \
            deep_get(purge_control_plane_targets, DictPath(f"{k8s_distro}#suse_packages"), [])
        held_packages = \
            deep_get(purge_control_plane_targets, DictPath(f"{k8s_distro}#suse_packages_held"), [])
    elif os_distro in ("fedora", "rhel",):
        packages = \
            deep_get(purge_control_plane_targets, DictPath(f"{k8s_distro}#fedora_packages"), [])
        held_packages = \
            deep_get(purge_control_plane_targets,
                     DictPath(f"{k8s_distro}#fedora_packages_held"), [])

    extra_values = {
        "packages": packages,
        "held_packages": held_packages,
    }
    __run_playbooks_on_selection(playbooks=playbooks, selection=selection,
                                 extra_values=extra_values, verbose=verbose)


def __task_run_postpurge_playbooks(installation_info: dict, verbose: bool = False) -> None:
    """
    A task that runs post-purge playbooks.

        Parameters:
            installation_info (dict): A dict with installation information
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    # The default selection is control planes;
    # the playbooks need to explicitly specify if the play should be run on other hosts
    selection = __selection_control_planes()
    try:
        __generic_task_run_playbooks_on_selection(selection=selection,
                                                  installation_info=installation_info,
                                                  playbook_dir=CMT_POST_PURGE_DIR,
                                                  extra_values={}, verbose=verbose)
    except FileNotFoundError as e:
        if CMT_POST_PURGE_DIR not in str(e):
            raise


def __validate_task_index(tasks: list[tuple[list[ANSIThemeStr], Callable[..., None]]],
                          phase: Union[str, int]) -> int:
    """
    Helper that validates the installer task index.

        Parameters:
            tasks ([tasks]): A list of tasks
            phase (int): An integer index
        Returns:
            (int): The integer index on success; exits if the task is out of range
    """
    try:
        phase = int(phase)
    except ValueError as e:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": ", "default"),
                        ANSIThemeStr("TASK", "argument"),
                        ANSIThemeStr(" needs to be an integer index in the range "
                                     f"[0, {len(tasks) - 1}]. Aborting.", "default")], stderr=True)
        sys.exit(f"Exception: {e}")

    if phase < 0 or phase >= len(tasks):
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": ", "default"),
                        ANSIThemeStr("TASK", "argument"),
                        ANSIThemeStr(" needs to be in the range "
                                     f"[0, {len(tasks) - 1}]. Aborting.", "default")], stderr=True)
        sys.exit(errno.ERANGE)
    return phase


def __list_phases(phases: list[tuple[list[ANSIThemeStr], Callable[..., None]]],
                  start_phase: Optional[int] = 0, phase_skiplist: Optional[list] = None) -> None:
    """
    Helper that lists installation phases.
    """
    if start_phase is None or start_phase == "<none>":
        start_phase = 0
    if phase_skiplist is None:
        phase_skiplist = []

    for i, phase in enumerate(phases):
        if i < start_phase or i in phase_skiplist:
            tmp = [ANSIThemeStr(f"{str(i).rjust(2)}: ", "skip")]
            override_formatting = "skip"
        else:
            tmp = [ANSIThemeStr(f"{str(i).rjust(2)}: ", "emphasis")]
            override_formatting = None
        tmp += themearray_override_formatting(phase[0], override_formatting)
        ansithemeprint(tmp)


def __expand_index_list(index_list: str) -> set[int]:
    indexes = set()

    for index in index_list.split(","):
        try:
            if "-" in index:
                tmp_first, tmp_last = index.split("-")
                first = int(tmp_first)
                last = int(tmp_last)
            else:
                first = int(index)
                last = first
        except ValueError as e:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": ", "default"),
                            ANSIThemeStr(f"{index}", "argument"),
                            ANSIThemeStr(" is not an integer or range of integers. Aborting.",
                                         "default")], stderr=True)
            sys.exit(f"Exception: {e}")
        for i in range(first, last + 1):
            indexes.add(i)
    return indexes


def __format_none(string: Optional[str], fmt: str) -> ANSIThemeStr:
    if string is None or string == "<none>":
        __string = ANSIThemeStr("<none>", "none")
    else:
        __string = ANSIThemeStr(string, fmt)
    return __string


prepare_tasks: list[tuple[list[ANSIThemeStr], Callable[..., None]]] = [
    ([ANSIThemeStr("Check for ssh host key and create if needed", "action")],
     __task_check_and_create_ssh_key),
    ([ANSIThemeStr("Add ssh keys for localhost, control plane(s) and nodes to ", "action"),
      ANSIThemeStr(f"{HOMEDIR}/.ssh/known_hosts", "path")],
     __task_scan_and_add_ssh_keys),
    ([ANSIThemeStr("Add public ssh keys to ", "action"),
      ANSIThemeStr(f"{HOMEDIR}/.ssh/authorized_keys", "path")],
     __task_check_and_add_ssh_keys_to_authorized_keys),
    ([ANSIThemeStr("Add public ssh keys to inventory", "action")],
     __task_add_ssh_keys_to_inventory),
    ([ANSIThemeStr("Request the ansible password if necessary", "action")],
     __task_request_ansible_password),
    ([ANSIThemeStr("Install ", "action"),
      ANSIThemeStr("ansible.posix", "programname"),
      ANSIThemeStr(" if necessary", "action")],
     __task_check_and_install_ansible_posix),
    ([ANSIThemeStr("Run pre-prepare playbooks", "action")],
     __task_run_prepreparation_playbooks),
    ([ANSIThemeStr("Run playbooks on ", "action"),
      ANSIThemeStr("localhost", "hostname")],
     __task_run_preparation_playbooks_on_localhost),
    ([ANSIThemeStr("Run playbooks on ", "action"),
      ANSIThemeStr("hosts", "hostname")],
     __task_run_preparation_playbooks_on_hosts),
    ([ANSIThemeStr("Run post-prepare playbooks", "action")],
     __task_run_postpreparation_playbooks),
]


__setup_kubeadm_control_plane_tasks: list[tuple[list[ANSIThemeStr], Callable[..., None]]] = [
    ([ANSIThemeStr("Run pre-setup playbooks", "action")],
     __task_run_presetup_playbooks),
    ([ANSIThemeStr("Run playbooks on ", "action"),
      ANSIThemeStr("localhost", "hostname")],
     __task_run_setup_playbooks_on_localhost),
    ([ANSIThemeStr("Setup ", "action"),
      ANSIThemeStr("bash completion", "emphasis"),
      ANSIThemeStr(" for ", "action"),
      ANSIThemeStr("kubectl", "programname")],
     __task_setup_bash_completion),
    ([ANSIThemeStr("Run playbooks on ", "action"),
      ANSIThemeStr("control planes", "hostname")],
     __task_run_setup_playbooks_on_control_planes),
    ([ANSIThemeStr("Import kubeconfig", "action")],
     __task_import_kubeconfig),
    ([ANSIThemeStr("Setup ", "action"),
      ANSIThemeStr("Container Network Interface (CNI)", "action")],
     __task_setup_cni),
    ([ANSIThemeStr("Run post-setup playbooks", "action")],
     __task_run_postsetup_playbooks),
]


__setup_rke2_control_plane_tasks: list[tuple[list[ANSIThemeStr], Callable[..., None]]] = [
    ([ANSIThemeStr("Run pre-setup playbooks", "action")],
     __task_run_presetup_playbooks),
    ([ANSIThemeStr("Run playbooks on ", "action"),
      ANSIThemeStr("localhost", "hostname")],
     __task_run_setup_playbooks_on_localhost),
    ([ANSIThemeStr("Setup ", "action"),
      ANSIThemeStr("bash completion", "emphasis"),
      ANSIThemeStr(" for ", "action"),
      ANSIThemeStr("kubectl", "programname")],
     __task_setup_bash_completion),
    ([ANSIThemeStr("Run playbooks on ", "action"),
      ANSIThemeStr("control planes", "hostname")],
     __task_run_setup_playbooks_on_control_planes),
    ([ANSIThemeStr("Import kubeconfig", "action")],
     __task_import_kubeconfig),
    ([ANSIThemeStr("Adjust kubeconfig server-address (if necessary)", "action")],
     __task_adjust_kubeconfig_server_address),
    ([ANSIThemeStr("Setup ", "action"),
      ANSIThemeStr("Container Network Interface (CNI)", "action")],
     __task_setup_cni),
    ([ANSIThemeStr("Run post-setup playbooks", "action")],
     __task_run_postsetup_playbooks),
]


setup_control_plane_tasks = __setup_kubeadm_control_plane_tasks

upgrade_control_plane_tasks: list[tuple[list[ANSIThemeStr], Callable[..., None]]] = [
    ([ANSIThemeStr("Drain the ", "action"),
      ANSIThemeStr("control planes", "hostname")], __task_drain_control_planes),
    ([ANSIThemeStr("Run pre-upgrade playbooks", "action")], __task_run_preupgrade_playbooks),
    ([ANSIThemeStr("Run upgrade playbooks on ", "action"),
      ANSIThemeStr("control planes", "hostname")], __task_run_upgrade_playbooks_on_control_planes),
    ([ANSIThemeStr("Run post-upgrade playbooks", "action")], __task_run_postupgrade_playbooks),
    ([ANSIThemeStr("Uncordon the ", "action"),
      ANSIThemeStr("control planes", "hostname")], __task_uncordon_control_planes),
]

teardown_control_plane_tasks: list[tuple[list[ANSIThemeStr], Callable[..., None]]] = [
    ([ANSIThemeStr("Verify that only ", "action"),
      ANSIThemeStr("control planes", "hostname"),
      ANSIThemeStr(" remain in the cluster", "action")], __task_verify_that_cluster_has_no_nodes),
    ([ANSIThemeStr("Drain the ", "action"),
      ANSIThemeStr("control planes", "hostname")], __task_drain_control_planes),
    ([ANSIThemeStr("Run pre-teardown playbooks", "action")], __task_run_preteardown_playbooks),
    ([ANSIThemeStr("Run teardown playbooks on ", "action"),
      ANSIThemeStr("control planes", "hostname")], __task_run_teardown_playbooks_on_control_planes),
    ([ANSIThemeStr("Remove kube config", "action")], __task_remove_kubeconfig),
    ([ANSIThemeStr("Run post-teardown playbooks", "action")], __task_run_postteardown_playbooks),
]

purge_control_plane_tasks: list[tuple[list[ANSIThemeStr], Callable[..., None]]] = [
    ([ANSIThemeStr("Run pre-purge playbooks", "action")], __task_run_prepurge_playbooks),
    ([ANSIThemeStr("Run purge playbooks on ", "action"),
      ANSIThemeStr("control planes", "hostname")], __task_run_purge_playbooks_on_control_planes),
    ([ANSIThemeStr("Run post-purge playbooks", "action")], __task_run_postpurge_playbooks),
]


def run_tasks(tasks: list[tuple[list[ANSIThemeStr], Callable[..., None]]],
              phase: int, phase_skiplist: set[int],
              final_state: str, verbose: bool = False) -> None:
    """
    Run tasks.

        Parameters:
            tasks [([themestring], Callable)]:
                ([themestring]): A description of the task in the form of list[ANSIThemeStr]
                (Callable): A function areference
            phase (int): The installation phase
            phase_skiplist (set(int)): A set of phases to skip
            final_state (str): The state to set if the task completes successfully
            verbose (bool): If the results are printed, should skipped tasks be printed too?
    """
    installation_info = get_installation_info()

    if not isinstance(phase, int):
        phase = 0

    for i in range(phase, len(tasks)):
        ansithemeprint([ANSIThemeStr("\n• ", "separator")] + tasks[i][0])
        if i not in set(phase_skiplist):
            tasks[i][1](installation_info=installation_info, verbose=verbose)
        if i < len(tasks) - 1:
            installation_info = update_installation_info(phase=i)
        else:
            update_installation_info(state=final_state, phase="<done>")


# pylint: disable-next=too-many-branches
def __find_requested_version(k8s_distro: str, version: Optional[str] = None) -> str:
    """
    Based on a provided version, try to find a matching package;
    Passing None will give the latest version, passing major, minor
    will give the latest patch revision of that version.
    Passing an exact version will give that version.

        Parameters:
            k8s_distro (str): Currently only kubeadm and rke2 are supported
            version (str): The requested version string or version substring
        Returns:
            requested_version (str): The best matching package version
    """
    os_distro = identify_distro()
    versions: list[tuple[str, str, str, list[str]]] = []

    requested_version: str = ""

    if k8s_distro == "kubeadm":
        # If version is an exact match for a package version, use it.
        # If version is a match for a package version with the package revision ("-nn") removed,
        # use the latest matching package revision.
        # If no version is specified, use the latest package revision.
        if os_distro in ("debian",):
            versions = check_versions_apt(["kubeadm"])
        elif os_distro in ("suse",):
            versions = check_versions_zypper([f"kubernetes{version}-kubeadm"])
        elif os_distro in ("fedora", "rhel",):
            versions = check_versions_yum(["kubeadm"])

        if not versions:
            ansithemeprint([ANSIThemeStr("Critical", "critical"),
                            ANSIThemeStr(": No candidate version for ", "default"),
                            ANSIThemeStr("kubeadm", "programname"),
                            ANSIThemeStr(" available; aborting.", "default")], stderr=True)
            sys.exit(errno.ENOENT)
        elif version is not None and version != "<none>":
            # The list is sorted in falling order, so the first match is the one we want,
            # since that's the newest package revision.
            # We're only checking versions for kubeadm,
            # so that's always gonna be entry 0 in the list.
            # field 3 is the list of available versions.
            for package_version in versions[0][3]:
                if package_version.split("-")[0] == version:
                    requested_version = package_version
                    break
                if package_version.split(".")[0] + "." + package_version.split(".")[1] == version:
                    requested_version = package_version
                    break
                if version.split(".")[0] == version and package_version.split(".")[0] == version:
                    requested_version = package_version
                    break
        else:
            # If no version was requested, we pick the candidate version
            requested_version = versions[0][2]
            if not requested_version:
                requested_version = versions[0][1]

        if not requested_version:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": Could not find a matching kubeadm package "
                                         "for version ", "default"),
                            ANSIThemeStr(f"{version}", "version"),
                            ANSIThemeStr("; aborting.", "default")], stderr=True)
            sys.exit(errno.ENOENT)
    elif k8s_distro == "rke2":
        if version is None or version == "<none>":
            return "latest"

        # FIXME: we could fetch the channel file from upstream to get the version list,
        # but for now we just accept any version
        tmp = re.match(r"^v?(\d+\.\d+).*", version)
        if tmp is not None:
            requested_version = tmp[1]
    else:
        raise ValueError(f"No support for distro {k8s_distro} implemented")

    if k8s_distro == "kubeadm":
        # First remove the distro revision
        major_minor_patchrev = requested_version.split("-")
        # Now split the version tuple
        _major, minor, _patchrev = major_minor_patchrev[0].split(".")
    else:
        # Split the version tuple
        _major, minor = requested_version.split(".")
        requested_version = f"v{requested_version}"

    if int(minor) < 24:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": kubeadm versions older than ", "default"),
                        ANSIThemeStr("1.24", "version"),
                        ANSIThemeStr(" are not supported by ", "default"),
                        ANSIThemeStr(f"{about.PROGRAM_SUITE_NAME}", "programname"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)
    elif int(minor) < 21 and k8s_distro == "rke2":
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": RKE2 versions older than ", "default"),
                        ANSIThemeStr("1.21", "version"),
                        ANSIThemeStr(" are not supported by ", "default"),
                        ANSIThemeStr(f"{about.PROGRAM_SUITE_NAME}", "programname"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)

    return requested_version


# pylint: disable-next=too-many-locals,too-many-branches,too-many-statements
def prepare_installation(options: list[tuple[str, Any]], args: list[str]) -> None:
    """
    Install and configure pre-requisites for a control plane.

        Parameters:
            options ([(str, str)]): Options to use when executing this action
            args ([str]): Arguments to use when executing this action
    """
    global no_password  # pylint: disable=global-statement
    confirm = True

    # We do not need a cluster name for --list-tasks, so check this first of all
    if "--list-tasks" in (tmp[0] for tmp in options):
        __list_phases(prepare_tasks)
        sys.exit(0)
    elif len(args) == 0:
        ansithemeprint([ANSIThemeStr(f"{about.ADMIN_PROGRAM_NAME}", "programname"),
                        ANSIThemeStr(": “", "default"),
                        ANSIThemeStr("prepare", "command"),
                        ANSIThemeStr("“ requires at least 1 arguments.", "default")], stderr=True)
        ansithemeprint([ANSIThemeStr("Try “", "default"),
                        ANSIThemeStr(f"{about.ADMIN_PROGRAM_NAME} ", "programname"),
                        ANSIThemeStr("help", "command"),
                        ANSIThemeStr("“ for more information.", "default")], stderr=True)
        sys.exit(errno.EINVAL)

    # Always require the user to specify the name of the cluster to operate against,
    # even when resuming prepare; this avoids a lot of hairy corner-cases and attempts
    # to figure out what cluster the user intended to operate against
    cluster_name = args[0]

    installation_info = get_installation_info(cluster_name=cluster_name)
    k8s_distro = "kubeadm"
    os_distro = identify_distro()
    requested_version = installation_info[cluster_name]["requested_version"]
    state = installation_info[cluster_name]["state"]
    phase: Union[str, int] = installation_info[cluster_name]["phase"]
    phase_skiplist = set(installation_info[cluster_name]["phase_skiplist"])
    verbose = False
    hide_next_step = False
    nodes: list[str] = []

    if state is not None and state not in ("<none>", "preparing", "prepared", "torn_down"):
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": Invalid installation state; the system cannot be in a "
                                     "configured or semi-configured state when running prepare; "
                                     "aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)

    # The user has restarted preparation
    if phase == "<done>" and state == "preparing":
        phase = 0

    if phase == "<none>" or state in ("prepared", "torn_down") and phase == "<done>":
        phase = 0
        requested_version = None

    hostname = socket.gethostname()

    # default options
    installation_info = \
        update_installation_info(cluster_name=cluster_name, distro=k8s_distro,
                                 state="preparing", phase=phase)

    controlplanes_ = get_control_planes(fail_on_empty=False)
    controlplanes = [controlplane[0] for controlplane in controlplanes_]
    if not controlplanes and "--control-plane" not in (tmp[0] for tmp in options):
        input_retval = \
            ansithemeinput([ANSIThemeStr("\nWarning", "warning"),
                            ANSIThemeStr(": No control plane defined in the inventory; "
                                         "do you want to add ", "default"),
                            ANSIThemeStr(f"{hostname}", "hostname"),
                            ANSIThemeStr(" as control plane? (No will abort the "
                                         "installation) [y/", "default"),
                            ANSIThemeStr("N", "emphasis"),
                            ANSIThemeStr("]: ", "default")])
        if input_retval.lower() not in ("y", "yes"):
            ansithemeprint([ANSIThemeStr("\nAborting:", "error"),
                            ANSIThemeStr(" No available control plane in inventory.", "default")],
                           stderr=True)
            sys.exit(errno.ENOENT)
        ansible_add_hosts(inventory=ANSIBLE_INVENTORY, hosts=[hostname],
                          group="controlplane", skip_all=False)
        ansible_add_hosts(inventory=ANSIBLE_INVENTORY, hosts=[hostname],
                          group=cluster_name, skip_all=True)

    for opt, optarg in options:
        if opt == "--control-plane":
            ansible_add_hosts(inventory=ANSIBLE_INVENTORY, hosts=[optarg],
                              group="controlplane", skip_all=False)
            ansible_add_hosts(inventory=ANSIBLE_INVENTORY, hosts=[optarg],
                              group=cluster_name, skip_all=True)
        elif opt == "--no-password":
            no_password = True
        elif opt == "--start-at-task":
            phase = __validate_task_index(prepare_tasks, optarg)
            if "--skip-tasks" not in options:
                phase_skiplist = set()
            if phase == 0:
                requested_version = None
        elif opt == "--skip-tasks":
            skip_phases = __expand_index_list(optarg)
            for __phase in skip_phases:
                __validate_task_index(prepare_tasks, __phase)
            phase_skiplist = set(skip_phases)
        elif opt == "--resume":
            if phase is None or phase == "<none>":
                phase = 0
            elif phase == "<done>":
                ansithemeprint([ANSIThemeStr("Warning", "warning"),
                                ANSIThemeStr(": Ignoring request to resume a completed "
                                             "preparation. Exiting.", "default")], stderr=True)
                sys.exit(0)
            phase = __validate_task_index(prepare_tasks, phase)
        elif opt == "--hide-next-step":
            hide_next_step = True
        elif opt == "--nodes":
            nodes = optarg
        elif opt == "--verbose":
            verbose = True
        elif opt == "--save-ansible-logs":
            ansible_configuration["save_logs"] = True
        elif opt == "-Y":
            confirm = False

    installation_info = \
        update_installation_info(cluster_name=cluster_name,
                                 control_planes=controlplanes, nodes=nodes)

    if len(args) > 1:
        requested_version = args[1]

        if requested_version in ("kubeadm", "rke2"):
            k8s_distro = requested_version
            requested_version = None
        elif "=" in requested_version:
            k8s_distro, requested_version = requested_version.split("=")
            if k8s_distro not in ("kubeadm", "rke2"):
                ansithemeprint([ANSIThemeStr(f"{about.ADMIN_PROGRAM_NAME}", "programname"),
                                ANSIThemeStr(": “", "default"),
                                ANSIThemeStr(f"{k8s_distro}", "option"),
                                ANSIThemeStr("“ is not a supported Kubernetes distribution.",
                                             "default")], stderr=True)
                sys.exit(errno.EINVAL)

    if requested_version is not None and requested_version in ("<none>", ""):
        requested_version = None

    if requested_version is None and os_distro == "suse":
        ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                        ANSIThemeStr("Updating version cache", "action")])

        # The first minor Kubernetes version available from the new repository is 1.24.0;
        # thus we don't support any version older than that.
        if not (kubernetes_upstream_version := get_latest_upstream_version("kubernetes")):
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": Could not get the latest upstream Kubernetes version; ",
                                         "default"),
                            ANSIThemeStr("this is either a network error or a bug; aborting.",
                                         "default")], stderr=True)
            sys.exit(errno.ENOENT)

        # Split the version tuple
        upstream_major, upstream_minor, _rest = kubernetes_upstream_version.split(".")
        requested_version = f"{upstream_major}.{upstream_minor}"

    installation_info = \
        update_installation_info(cluster_name=cluster_name, distro=k8s_distro, phase=phase,
                                 phase_skiplist=list(phase_skiplist),
                                 requested_version=requested_version)

    show_configuration(action="Preparing cluster:", tasks=prepare_tasks)

    if confirm:
        input_retval = \
            ansithemeinput([ANSIThemeStr("\nStart control plane preparation? [y/", "default"),
                            ANSIThemeStr("N", "emphasis"),
                            ANSIThemeStr("]: ", "default")])
        if input_retval.lower() not in ("y", "yes"):
            ansithemeprint([ANSIThemeStr("\nAborting:", "error"),
                            ANSIThemeStr(" User stopped control plane preparation.", "default")],
                           stderr=True)
            os.remove(CMT_INSTALLATION_INFO_FILE)
            sys.exit(errno.EINTR)

    ansithemeprint([ANSIThemeStr("\n[Preparing localhost and control plane(s)]", "phase")])

    update_version_cache(verbose=verbose)

    run_tasks(tasks=prepare_tasks, phase=cast(int, phase), phase_skiplist=phase_skiplist,
              final_state="prepared", verbose=verbose)

    # Adjust the kube* packages for the control plane(s) and localhost to match
    # the requested cluster version if necessary; we could not do this before
    # since the apt repository was not available then
    if requested_version is not None:
        requested_version = __find_requested_version(k8s_distro, requested_version)
    else:
        requested_version = __find_requested_version(k8s_distro)

    ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                    ANSIThemeStr("Updating installation information", "action")])

    update_installation_info(requested_version=requested_version)

    ansithemeprint([ANSIThemeStr("   Requested version: ", "action"),
                    ANSIThemeStr(requested_version, "version")])

    ansithemeprint([ANSIThemeStr("\nControl plane preparation successful", "success")])

    if not hide_next_step:
        print("\nNext step:")
        ansithemeprint([ANSIThemeStr("• ", "separator"),
                        ANSIThemeStr(f"{about.ADMIN_PROGRAM_NAME}", "programname"),
                        ANSIThemeStr(" setup-control-plane ", "command"),
                        ANSIThemeStr("[", "separator"),
                        ANSIThemeStr("CNI", "argument"),
                        ANSIThemeStr("]", "separator"),
                        ANSIThemeStr(" [", "separator"),
                        ANSIThemeStr("POD_NETWORK_CIDR", "argument"),
                        ANSIThemeStr("]", "separator")])
        ansithemeprint([ANSIThemeStr("\nSee “", "separator"),
                        ANSIThemeStr(f"{about.ADMIN_PROGRAM_NAME}", "programname"),
                        ANSIThemeStr(" help", "command"),
                        ANSIThemeStr("“ for more information about valid ", "default"),
                        ANSIThemeStr("CNI", "argument"),
                        ANSIThemeStr(" options.\n", "default")])


# pylint: disable-next=too-many-locals,too-many-branches,too-many-statements
def import_cluster(options: list[tuple[str, str]], args: list[str]) -> None:
    """
    Import an existing cluster to the CMT Ansible inventory.

        Parameters:
            options ([(str, str)]): Options to use when executing this action
            args ([str]): Arguments to use when executing this action
    """
    global kh  # pylint: disable=global-statement
    global no_password  # pylint: disable=global-statement

    # default options
    confirm: bool = True
    verbose: bool = False

    for opt, _optarg in options:
        if opt == "-Y":
            confirm = False
        elif opt == "--no-password":
            no_password = True
        elif opt == "--save-ansible-logs":
            ansible_configuration["save_logs"] = True
        elif opt == "--verbose":
            verbose = True

    __task_check_and_create_ssh_key(installation_info={})
    __task_add_ssh_keys_to_inventory(installation_info={})
    print()

    if kh is None:
        kh = kubernetes_helper.KubernetesHelper(about.PROGRAM_SUITE_NAME,
                                                about.PROGRAM_SUITE_VERSION, None)
    clusters = kh.list_clusters()
    available_clusters = [tmp[0] for tmp in clusters]

    pad = len("Cluster:")
    match = False
    for cluster_name, context in clusters:
        if args and cluster_name not in args[0].split(","):
            continue
        pad = max(len(cluster_name), pad)
        match = True

    if args:
        for cluster in args[0].split(","):
            if cluster not in available_clusters and match:
                ansithemeprint([ANSIThemeStr("Warning", "warning"),
                                ANSIThemeStr(": Ignoring non-existing cluster ", "default"),
                                ANSIThemeStr(cluster, "hostname"),
                                ANSIThemeStr("\n", "default")], stderr=True)

    if not match:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": No matching clusters found; available clusters are:",
                                     "default")], stderr=True)
        for cluster in available_clusters:
            ansithemeprint([ANSIThemeStr("• ", "separator"),
                            ANSIThemeStr(cluster, "hostname")])
        ansithemeprint([ANSIThemeStr("\nAborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    ansithemeprint([ANSIThemeStr("Cluster:", "header"),
                    ANSIThemeStr("".ljust(pad + 2 - len("Cluster:")), "default"),
                    ANSIThemeStr("Context:", "header")])
    for cluster_name, context in clusters:
        if args and cluster_name not in args[0]:
            continue
        ansithemeprint([ANSIThemeStr(cluster_name, "hostname"),
                        ANSIThemeStr("".ljust(pad + 2 - len(cluster_name)), "default"),
                        ANSIThemeStr(context, "default")])

    if confirm:
        input_retval = \
            ansithemeinput([ANSIThemeStr("\nImport the following clusters? [y/", "default"),
                            ANSIThemeStr("N", "emphasis"),
                            ANSIThemeStr("]: ", "default")])

        if input_retval.lower() not in ("y", "yes"):
            ansithemeprint([ANSIThemeStr("\nAborting:", "error"),
                            ANSIThemeStr(" User stopped cluster import.", "default")], stderr=True)
            sys.exit(errno.EINTR)

    ansithemeprint([ANSIThemeStr("\n[Importing cluster(s)]", "phase")])

    # Since the clusters might use different versions of Kubernetes
    # we always install the latest kubectl on localhost, so no need to adjust version

    contexts = kh.list_contexts()
    current_context = None
    for ctx in contexts:
        if ctx[0]:
            current_context = ctx[1]

    failures: list[str] = []

    for cluster_name, context in clusters:
        if args and cluster_name not in args[0]:
            continue

        if not kh.set_context(name=context, unchanged_is_success=True):
            ansithemeprint([ANSIThemeStr("Warning", "warning"),
                            ANSIThemeStr(": Failed to change to context ", "default"),
                            ANSIThemeStr(f"{context}", "hostname"),
                            ANSIThemeStr(" for cluster ", "default"),
                            ANSIThemeStr(f"{cluster_name}", "hostname"),
                            ANSIThemeStr("; skipping.", "default")])
            failures.append(cluster_name)
            continue

        vlist, status = kh.get_list_by_kind_namespace(("Node", ""), "")
        if status != 200:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": The API-server for cluster ", "default"),
                            ANSIThemeStr(f"{cluster_name}", "hostname"),
                            ANSIThemeStr(" returned ", "default"),
                            ANSIThemeStr(f"{status}", "errorvalue"),
                            ANSIThemeStr("; skipping.", "default")], stderr=True)
            failures.append(cluster_name)
            continue

        if vlist is None:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": The API-server for cluster ", "default"),
                            ANSIThemeStr(f"{cluster_name}", "hostname"),
                            ANSIThemeStr(" did not return any data", "default"),
                            ANSIThemeStr("; skipping.", "default")], stderr=True)
            failures.append(cluster_name)
            continue

        controlplanes = []
        nodes = []
        controlplane = None

        for node in vlist:
            node_name = deep_get(node, DictPath("metadata#name"))
            node_roles = get_node_roles(cast(dict, node))
            if "control-plane" in node_roles:
                controlplanes.append(node_name)
                if controlplane is None:
                    controlplane = node_name
            else:
                nodes.append(node_name)

        if controlplane is None:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": Could not find a control plane", "default"),
                            ANSIThemeStr(" for cluster ", "default"),
                            ANSIThemeStr(f"{cluster_name}", "hostname"),
                            ANSIThemeStr("; skipping.", "default")], stderr=True)
            failures.append(cluster_name)
            continue

        __task_request_ansible_password(installation_info={})

        extra_values = {
            "ansible_become_pass": deep_get(ansible_configuration, DictPath("ansible_password")),
            "ansible_ssh_pass": deep_get(ansible_configuration, DictPath("ansible_password")),
        }

        install_ansible_posix()

        retval, ansible_results = \
            run_playbook(get_playbook_path(FilePath("prepare_passwordless_ansible.yaml")),
                         hosts=[controlplane], extra_values=extra_values, verbose=verbose)

        if not ansible_results:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": The Ansible playbook “", "default"),
                            ANSIThemeStr("prepare_passwordless_ansible.yaml", "path"),
                            ANSIThemeStr("“ failed on ", "default"),
                            ANSIThemeStr(controlplane, "hostname"),
                            ANSIThemeStr(" (retval: ", "default"),
                            ANSIThemeStr(f"{retval}", "errorvalue"),
                            ANSIThemeStr("); skipping.", "default")], stderr=True)
            failures.append(cluster_name)
            continue

        tmp_cni = kh.identify_cni()
        if len(tmp_cni) != 1:
            cni = "<unknown>"
        else:
            cni = tmp_cni[0][0]

        pod_network_cidr = kh.get_pod_network_cidr()
        if pod_network_cidr is None:
            pod_network_cidr = "<unknown>"

        get_versions_path = get_playbook_path(FilePath("get_versions.yaml"))
        retval, ansible_results = \
            ansible_run_playbook_on_selection(get_versions_path,
                                              selection=[controlplane], quiet=False)

        if not ansible_results:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": The Ansible playbook “", "default"),
                            ANSIThemeStr("get_versions.yaml", "path"),
                            ANSIThemeStr("“ failed on ", "default"),
                            ANSIThemeStr(controlplane, "hostname"),
                            ANSIThemeStr(" (retval: ", "default"),
                            ANSIThemeStr(f"{retval}", "errorvalue"),
                            ANSIThemeStr("); skipping.", "default")], stderr=True)
            failures.append(cluster_name)
            continue

        k8s_distro = "<unknown>"
        version = "<unknown>"

        for result in deep_get(ansible_results, DictPath(controlplane), []):
            if deep_get(result, DictPath("task"), "") == "Package versions":
                tmp = deep_get(result, DictPath("msg_lines"), [])
                break

        if not tmp:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": Received empty version data from ", "default"),
                            ANSIThemeStr(controlplane, "hostname"),
                            ANSIThemeStr("; skipping.", "default")], stderr=True)
            failures.append(cluster_name)
            continue

        package_version_regex: re.Pattern[str] = re.compile(r"^(.*?): (.*)")

        for line in tmp:
            tmp2 = package_version_regex.match(line)
            if tmp2 is None:
                continue
            package = tmp2[1]
            version = tmp2[2]
            if package == "kubeadm":
                k8s_distro = "kubeadm"
                update_installation_info(cluster_name=cluster_name, distro=k8s_distro,
                                         version=version, requested_version="<none>", cni=cni,
                                         state="installed", phase="<none>",
                                         pod_network_cidr=pod_network_cidr,
                                         control_planes=controlplanes, nodes=nodes)
                break
        if k8s_distro == "<unknown>":
            ansithemeprint([ANSIThemeStr("Warning", "warning"),
                            ANSIThemeStr(": Failed to import cluster ", "default"),
                            ANSIThemeStr(f"{cluster_name}", "hostname"),
                            ANSIThemeStr("; could not identify distro; skipping.", "default")])
            failures.append(cluster_name)
            continue

    kh.set_context(name=current_context)

    if not failures:
        ansithemeprint([ANSIThemeStr("\nCluster import successful", "success")])
    else:
        ansithemeprint([ANSIThemeStr("\nFailed to import the following clusters:", "warning")],
                       stderr=True)

        for cluster_name in failures:
            ansithemeprint([ANSIThemeStr("• ", "separator"),
                            ANSIThemeStr(f"{cluster_name}", "hostname")], stderr=True)


# pylint: disable-next=unused-argument,too-many-locals
def check_for_updates(options: list[tuple[str, str]], args: list[str]) -> None:
    """
    Check whether there are newer versions of packages that are related to the cluster.

        Parameters:
            options ([(str, str)]): Options to use when executing this action
            args (list[str]): Arguments to use when executing this action
    """
    os_distro = identify_distro()

    # default options
    update_cache = True
    verbose = False
    force = False

    for opt, _optarg in options:
        if opt == "--force":
            force = True
        if opt == "--no-cache-update":
            update_cache = False
        elif opt == "--verbose":
            verbose = True

    ansithemeprint([ANSIThemeStr("\n[Checking for software updates]", "phase")])
    ansithemeprint([ANSIThemeStr("Note", "note"),
                    ANSIThemeStr(": Currently only local versions are checked.", "default")])

    if update_cache:
        if os_distro in ("debian",):
            ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                            ANSIThemeStr("Updating APT cache", "action")])
            check_and_print_status(update_apt_cache())
        elif os_distro in ("suse",):
            ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                            ANSIThemeStr("Updating Zypper cache", "action")])
            check_and_print_status(update_zypper_cache())
        ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                        ANSIThemeStr("Updating version cache", "action")])
        update_version_cache(verbose=verbose, force=force)

    pkg_packages: list[str] = [
        "ansible",
        "ansible-core",
        "containerd",
        "containerd.io",
        "cri-o",
        "cri-tools",
        "docker-ce",
        "docker-engine",
        "docker.io",
        "kubeadm",
        "kubectl",
        "kubelet",
        "kubernetes-cni",
        "python3-ansible-runner",
        "python3-natsort",
        "python3-paramiko",
        "python3-pip",
        "python3-ujson",
        "python3-urllib3",
        "runc",
    ]

    if os_distro == "suse":
        if (kubernetes_upstream_version := get_latest_upstream_version("kubernetes")):
            _upstream_major, upstream_minor, _rest = kubernetes_upstream_version.split(".")

            for package in ("client", "kubeadm", "kubelet"):
                for i in range(17, int(upstream_minor) + 1):
                    pkg_packages.append(f"kubernetes1.{i}-{package}")

    pkg_packages = cast(list[str], natsorted(pkg_packages))
    version_checks: list = []

    print()
    check_versions(pkg_packages, version_checks, verbose=verbose)
    print()


# pylint: disable-next=too-many-locals,too-many-branches,too-many-statements
def show_configuration(action: str,
                       tasks: list[tuple[list[ANSIThemeStr], Callable[..., None]]]) -> None:
    """
    Show cluster configuration.

        Parameters:
            action (str): The action to take
            tasks ([([ANSIThemeStr], callable, args)])): The tasks list
    """
    installation_info = get_installation_info()
    cluster_name = deep_get(installation_info, DictPath("installation_target"))
    k8s_distro = deep_get(installation_info, DictPath(f"{cluster_name}#distro"))
    version = deep_get(installation_info, DictPath(f"{cluster_name}#version"))
    requested_version = deep_get(installation_info, DictPath(f"{cluster_name}#requested_version"))
    cni = deep_get(installation_info, DictPath(f"{cluster_name}#cni"))
    pod_network_cidr = deep_get(installation_info, DictPath(f"{cluster_name}#pod_network_cidr"))
    cri = deep_get(installation_info, DictPath(f"{cluster_name}#cri"))
    controlplanes = deep_get(installation_info, DictPath(f"{cluster_name}#control_planes"), [])
    nodes = deep_get(installation_info, DictPath(f"{cluster_name}#nodes"), [])

    failures: dict = {}

    control_plane_info = []
    for controlplane in controlplanes:
        ip = None
        try:
            ip = socket.gethostbyname(controlplane)
        except socket.gaierror as e:
            if str(e) in ("[Errno -2] Name or service not known",
                          "[Errno -3] Temporary failure in name resolution",
                          "[Errno -5] No address associated with hostname"):
                tmp = re.match(r"^\[Errno (-\d+)\] (.+)", str(e))
                if tmp is None:
                    # If we cannot extract the error we re-raise it since we don't know
                    # how to handle it
                    raise
                if not deep_get(failures, DictPath(tmp[2]), []):
                    failures[tmp[2]] = []
                failures[tmp[2]].append(controlplane)
        control_plane_info.append((controlplane, ip))

    node_info = []
    for node in nodes:
        ip = None
        try:
            ip = socket.gethostbyname(node)
        except socket.gaierror as e:
            if str(e) in ("[Errno -2] Name or service not known",
                          "[Errno -3] Temporary failure in name resolution",
                          "[Errno -5] No address associated with hostname"):
                tmp = re.match(r"^\[Errno (-\d+)\] (.+)", str(e))
                if tmp is None:
                    # If we cannot extract the error we re-raise it since we don't know
                    # how to handle it
                    raise
                if not deep_get(failures, DictPath(tmp[2]), []):
                    failures[tmp[2]] = []
                failures[tmp[2]].append(node)
        node_info.append((node, ip))

    # If we fail to get the IP-address of any of the hostnames
    # we output them here and abort.
    if failures:
        ansithemeprint([ANSIThemeStr("Errors", "error"),
                        ANSIThemeStr(":", "default")], stderr=True)
    for error, hosts in failures.items():
        hosts = ansithemestr_join_list(hosts, formatting="hostname",
                                       separator=ANSIThemeStr(", ", "separator"))
        ansithemeprint([ANSIThemeStr(f"  {error} (hosts: ", "default")]
                       + hosts
                       + [ANSIThemeStr(")", "default")], stderr=True)
    if failures:
        ansithemeprint([ANSIThemeStr("Aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    http_proxy = deep_get(cmtlib.cmtconfig, DictPath("Network#http_proxy"), "")
    if http_proxy is not None and http_proxy == "":
        http_proxy = None
    http_proxy_env = os.getenv("http_proxy")
    if http_proxy_env is not None and http_proxy_env == "":
        http_proxy_env = None
    https_proxy = deep_get(cmtlib.cmtconfig, DictPath("Network#https_proxy"), "")
    if https_proxy is not None and https_proxy == "":
        https_proxy = None
    https_proxy_env = os.getenv("https_proxy")
    if https_proxy_env is not None and https_proxy_env == "":
        https_proxy_env = None
    no_proxy = deep_get(cmtlib.cmtconfig, DictPath("Network#no_proxy"), "")
    if no_proxy is not None and no_proxy == "":
        no_proxy = None
    no_proxy_env = os.getenv("no_proxy")
    if no_proxy_env is not None and no_proxy_env == "":
        no_proxy_env = None

    if action:
        ansithemeprint([ANSIThemeStr(action, "emphasis")])

    ansithemeprint([ANSIThemeStr("\n[Summary]", "phase")])
    ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                    ANSIThemeStr("Configuration:", "action")])
    ansithemeprint([ANSIThemeStr("        Cluster Name: ", "action"),
                    ANSIThemeStr(f"{cluster_name}", "hostname")])
    ansithemeprint([ANSIThemeStr("        Distribution: ", "action"),
                    ANSIThemeStr(f"{k8s_distro}", "programname")])
    ansithemeprint([ANSIThemeStr("   Installed Version: ", "action"),
                    __format_none(version, "version")])
    if requested_version is not None and requested_version != "<none>":
        ansithemeprint([ANSIThemeStr("   Requested Version: ", "action"),
                        __format_none(requested_version, "version")])
    if cni is not None and cni != "<none>":
        ansithemeprint([ANSIThemeStr("                 CNI: ", "action"),
                        ANSIThemeStr(f"{cni}", "programname")])
    if pod_network_cidr is not None and pod_network_cidr != "<none>":
        ansithemeprint([ANSIThemeStr("    Pod Network CIDR: ", "action"),
                        ANSIThemeStr(f"{pod_network_cidr}", "emphasis")])
    if cri is not None and cri != "<none>":
        ansithemeprint([ANSIThemeStr("                 CRI: ", "action"),
                        ANSIThemeStr(f"{cri}", "programname")])
    ansithemeprint([ANSIThemeStr("          HTTP Proxy: ", "action"),
                    __format_none(http_proxy, "url"),
                    ANSIThemeStr(" (", "default"),
                    ANSIThemeStr(f"{CMT_CONFIG_FILE}", "path"),
                    ANSIThemeStr(")", "default")])
    ansithemeprint([ANSIThemeStr("          HTTP Proxy: ", "action"),
                    __format_none(http_proxy_env, "url"),
                    ANSIThemeStr(" (Environment)", "default")])
    ansithemeprint([ANSIThemeStr("         HTTPS Proxy: ", "action"),
                    __format_none(https_proxy, "url"),
                    ANSIThemeStr(" (", "default"),
                    ANSIThemeStr(f"{CMT_CONFIG_FILE}", "path"),
                    ANSIThemeStr(")", "default")])
    ansithemeprint([ANSIThemeStr("         HTTPS Proxy: ", "action"),
                    __format_none(https_proxy_env, "url"),
                    ANSIThemeStr(" (Environment)", "default")])
    ansithemeprint([ANSIThemeStr("            No Proxy: ", "action"),
                    __format_none(no_proxy, "url"),
                    ANSIThemeStr(" (", "default"),
                    ANSIThemeStr(f"{CMT_CONFIG_FILE}", "path"),
                    ANSIThemeStr(")", "default")])
    ansithemeprint([ANSIThemeStr("            No Proxy: ", "action"),
                    __format_none(no_proxy_env, "url"),
                    ANSIThemeStr(" (Environment)", "default")])
    if control_plane_info:
        ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                        ANSIThemeStr("Control Plane(s):", "action")])
        for hostname, ip in control_plane_info:
            ansithemeprint([ANSIThemeStr("            Hostname: ", "action"),
                            ANSIThemeStr(f"{hostname} ", "emphasis"),
                            ANSIThemeStr("(", "default"),
                            ANSIThemeStr(f"{ip}", "emphasis"),
                            ANSIThemeStr(")", "default")])
    if node_info:
        ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                        ANSIThemeStr("Node(s):", "action")])
        for hostname, ip in node_info:
            ansithemeprint([ANSIThemeStr("            Hostname: ", "action"),
                            ANSIThemeStr(f"{hostname} ", "emphasis"),
                            ANSIThemeStr("(", "default"),
                            ANSIThemeStr(f"{ip}", "emphasis"),
                            ANSIThemeStr(")", "default")])

    if tasks is not None:
        ansithemeprint([ANSIThemeStr("", "default")])
        ansithemeprint([ANSIThemeStr("Pending tasks (", "action"),
                        ANSIThemeStr("Dimmed", "skip"),
                        ANSIThemeStr(" tasks are either skipped or completed):\n", "action")])

        phase = installation_info[cluster_name]["phase"]
        phase_skiplist = installation_info[cluster_name]["phase_skiplist"]
        __list_phases(tasks, start_phase=phase, phase_skiplist=phase_skiplist)


def __get_configuration_data(configuration_file: FilePath, configuration_type: str,
                             configuration: dict, default: FilePath) -> FilePath:
    """
    Given a configuration dict, make sure that it doesn't contains both data and a path.
    If given data write it to a temporary file. Returns the path on success, exists on failure.

        Parameters:
            configuration_file (str): The name of the configuration file; used for error messages
            configuration_type (str): The type of the configuration we're expecting
            configuration (dict): The configuration to verify and return a path to
            default (FilePath): If the configuration is empty use path as fallback
        Returns:
            (FilePath): The path to a file with the configuration on success
    """
    data = deep_get(configuration, DictPath(f"{configuration_type}#data"))
    path = deep_get(configuration, DictPath(f"{configuration_type}#path"))

    if data is None and path is None:
        return default

    if data is not None and path is not None:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": ", "default"),
                        ANSIThemeStr("The configuration in ", "default"),
                        ANSIThemeStr(f"{configuration_file}", "path"),
                        ANSIThemeStr(" specifies both data and a path for ", "default"),
                        ANSIThemeStr(f"{configuration_type}", "command"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)

    if data is not None:
        # pylint: disable-next=consider-using-with
        tf = tempfile.NamedTemporaryFile(suffix=".yaml.j2", delete=False)
        path = FilePath(tf.name)
        secure_write_string(path, data, temporary=True)

    return FilePath(path)


# pylint: disable-next=too-many-locals,too-many-branches,too-many-statements
def add_nodes(options: list[tuple[str, Any]], args: list[Any]) -> None:
    """
    Add nodes to the cluster.

        Parameters:
            options ([(str, str)]): Options to use when executing this action
            args ([str]): Arguments to use when executing this action
    """
    global kh  # pylint: disable=global-statement

    if kh is None:
        kh = kubernetes_helper.KubernetesHelper(about.PROGRAM_SUITE_NAME,
                                                about.PROGRAM_SUITE_VERSION, None)

    cordon_on_join = False
    cri = "containerd"
    cri_version = None
    verbose: bool = False

    installation_info = get_installation_info()
    cluster_name = deep_get(installation_info, DictPath("installation_target"))
    k8s_distro = deep_get(installation_info, DictPath(f"{cluster_name}#distro"))
    control_planes = deep_get(installation_info, DictPath(f"{cluster_name}#control_planes"))

    configuration_paths: list[FilePath] = []

    if k8s_distro == "kubeadm":
        configuration_paths = [
            FilePath("templates").joinpath("config", "join_configuration.yaml.j2"),
        ]
    elif k8s_distro == "rke2":
        configuration_paths = [
            FilePath("templates").joinpath("etc", "rancher", "rke2", "config.agent.yaml.j2"),
        ]

    for opt, optarg in options:
        if opt == "--verbose":
            verbose = True
        elif opt == "--save-ansible-logs":
            ansible_configuration["save_logs"] = True
        elif opt == "--configuration-paths":
            configuration_paths = optarg
        elif opt == "--cordon-on-join":
            cordon_on_join = optarg
        elif opt == "--cri":
            cri = optarg
        elif opt == "--cri-version":
            cri_version = optarg
        elif opt == "--forks":
            ansible_configuration["ansible_forks"] = optarg

    configuration_path = merge_configurations(configuration_paths)

    nodes: list[str] = args[0]
    groups = args[1]

    control_plane_version = get_control_plane_version(control_planes[0], k8s_distro)

    # First remove the distro revision
    major_minor_patchrev = control_plane_version.split("-")
    # Now split the version tuple
    requested_major, requested_minor, _rest = major_minor_patchrev[0].split(".")

    control_plane_ip, control_plane_port, control_plane_path = kh.get_control_plane_address()
    if control_plane_ip is None:
        ansithemeprint([ANSIThemeStr("\nAborting", "error"),
                        ANSIThemeStr(": Could not get the IP-address for the control plane.",
                                     "default")], stderr=True)
        sys.exit(errno.ENOENT)

    # Add all nodes to the inventory
    ansible_add_hosts(inventory=ANSIBLE_INVENTORY, hosts=nodes, skip_all=False)

    crio_major_version = None
    crio_minor_version = None

    add_playbooks: list[FilePath] = []

    # Add the CRI to the setup playbooks for the control plane
    if cri == "containerd":
        add_playbooks = [FilePath("setup_containerd.yaml")]
    elif cri == "cri-o":
        add_playbooks = [FilePath("setup_cri-o.yaml")]
        if cri_version is None:
            tmp_crio_version = get_crio_version((int(requested_major), int(requested_minor)))
            if tmp_crio_version is None:
                check_and_print_status(False)
            # The type ignore below is necessary because mypy doesn't recognise that
            # check_and_print_status always aborts with sys.exit() when called with False
            crio_major_version, crio_minor_version = tmp_crio_version  # type: ignore
        else:
            # This case shouldn't be necessary; mypy should recognise that we're doing
            # a None-check first.
            tmp_crio_version = cast(tuple, cri_version.split("."))
            if len(tmp_crio_version) < 2:
                check_and_print_status(False)
            crio_major_version = tmp_crio_version[0]
            crio_minor_version = tmp_crio_version[1]

    cri_socket = deep_get(cri_data[cri], DictPath("socket"))

    add_playbooks += [
        # cmtadm does not use run_before/run_after/add_to_groups/remove_from_groups,
        # so we have to do all of that explicitly
        FilePath("add_kubernetes_repo.yaml"),
    ]
    if k8s_distro == "kubeadm":
        add_playbooks += [
            FilePath("kubeadm_setup_node.yaml"),
            FilePath("kubeadm_join_node.yaml"),
        ]
    elif k8s_distro == "rke2":
        add_playbooks += [
            FilePath("rke2_setup_node.yaml"),
        ]

    if int(requested_minor) < 31:
        join_configuration_api_version = "kubeadm.k8s.io/v1beta3"
    else:
        join_configuration_api_version = "kubeadm.k8s.io/v1beta4"

    minor_versions = []
    for minor_version in range(24, int(requested_minor) + 1):
        minor_versions.append(f"{minor_version}")

    extra_values = {
        "ansible_become_pass": deep_get(ansible_configuration, DictPath("ansible_password")),
        "ansible_ssh_pass": deep_get(ansible_configuration, DictPath("ansible_password")),
        "cluster_name": cluster_name,
        "control_plane_ip": control_plane_ip,
        "control_plane_port": control_plane_port,
        "control_plane_path": control_plane_path,
        "cordon_nodes_on_join": cordon_on_join,
        "control_plane_k8s_version": control_plane_version,
        "crio_major_version": crio_major_version,
        "crio_minor_version": crio_minor_version,
        "cri_socket": cri_socket,
        "configuration_path": configuration_path,
        "join_configuration_api_version": join_configuration_api_version,
        "kubernetes_major_minor_version": f"{requested_major}.{requested_minor}",
        "minor_versions": minor_versions,
    }

    if k8s_distro == "kubeadm":
        join_token = kh.get_join_token()
        if not join_token:
            ansithemeprint([ANSIThemeStr("\nAborting", "error"),
                            ANSIThemeStr(": No join token available.", "default")], stderr=True)
            sys.exit(errno.ENOENT)
        ca_cert_hash = kh.get_ca_cert_hash()
        if not ca_cert_hash:
            ansithemeprint([ANSIThemeStr("\nAborting", "error"),
                            ANSIThemeStr(": No CA certificate available.", "default")], stderr=True)
            sys.exit(errno.ENOENT)

        extra_values["join_token"] = join_token
        extra_values["ca_cert_hash"] = ca_cert_hash
    elif k8s_distro == "rke2":
        extra_values["requested_version"] = control_plane_version

    playbooks = populate_playbooks_from_filenames(add_playbooks)

    retval = run_playbooks(playbooks=playbooks, hosts=nodes,
                           extra_values=extra_values, verbose=verbose)
    if retval:
        ansithemeprint([ANSIThemeStr("OK", "ok")])
    else:
        ansithemeprint([ANSIThemeStr("NOT OK", "notok"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)

    for group in ["nodes"] + groups + [cluster_name]:
        ansible_add_hosts(inventory=ANSIBLE_INVENTORY, hosts=nodes, group=group, skip_all=True)


def set_context(cluster_name: str, context_name: str) -> None:
    """
    Set context.

        Parameters:
            cluster_name (str): The name of the cluster (used for error messages)
            context_name (str): The context to switch to
    """
    global kh  # pylint: disable=global-statement

    # This will only work for running clusters
    if kh is None:
        kh = kubernetes_helper.KubernetesHelper(about.PROGRAM_SUITE_NAME,
                                                about.PROGRAM_SUITE_VERSION, None)

    if not kh.set_context(name=context_name, unchanged_is_success=True):
        ansithemeprint([ANSIThemeStr("Warning", "warning"),
                        ANSIThemeStr(": Failed to change to context ", "default"),
                        ANSIThemeStr(f"{context_name}", "hostname"),
                        ANSIThemeStr(" for cluster ", "default"),
                        ANSIThemeStr(f"{cluster_name}", "hostname"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)


# pylint: disable-next=too-many-locals,too-many-branches,too-many-statements
def create_cluster(options: list[tuple[str, str]], args: list[str]) -> None:
    """
    Create a cluster based on a ClusterDeployment template.

        Parameters:
            options (list[(opt, optarg)]): Options to use when executing this action
            args (list[str]): Arguments to use when executing this action
    """
    global kh  # pylint: disable=global-statement
    global no_password  # pylint: disable=global-statement

    join_configuration_path: str = ""

    http_proxy = deep_get(cmtlib.cmtconfig, DictPath("Network#http_proxy"), "")
    if http_proxy is None:
        http_proxy = ""
    https_proxy = deep_get(cmtlib.cmtconfig, DictPath("Network#https_proxy"), "")
    if https_proxy is None:
        https_proxy = ""
    no_proxy = deep_get(cmtlib.cmtconfig, DictPath("Network#no_proxy"), "")
    if no_proxy is None:
        no_proxy = ""

    env = {
        "KUBECONFIG": HOMEDIR.joinpath(".kube", "config"),
        "PATH": deep_get(dict(os.environ), DictPath("PATH"), ""),
        "HOME": HOMEDIR,
        "http_proxy": http_proxy,
        "https_proxy": https_proxy,
        "no_proxy": no_proxy,
    }

    installation_info = get_installation_info()
    confirm = True
    redeploy = False

    for opt, _optarg in options:
        if opt == "-Y":
            confirm = False
        elif opt == "--redeploy":
            redeploy = True

    cluster_deployment_file = FilePath(args[0])

    if not os.path.exists(cluster_deployment_file):
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": ", "default"),
                        ANSIThemeStr(f"{cluster_deployment_file}", "path"),
                        ANSIThemeStr(" does not exist; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)
    elif not os.path.isfile(cluster_deployment_file):
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": ", "default"),
                        ANSIThemeStr(f"{cluster_deployment_file}", "path"),
                        ANSIThemeStr(" is not a file; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    security_checks = [
        SecurityChecks.OWNER_IN_ALLOWLIST,
        SecurityChecks.PARENT_OWNER_IN_ALLOWLIST,
        SecurityChecks.PERMISSIONS,
        SecurityChecks.PARENT_PERMISSIONS,
        SecurityChecks.IS_FILE,
    ]
    cd = secure_read_yaml(cluster_deployment_file, checks=security_checks)
    cd_kind = deep_get(cd, DictPath("kind"), "")
    cd_api_version = deep_get(cd, DictPath("apiVersion"), "")

    # Is this a cluster deployment and do we support this version?
    if cd_kind != "ClusterDeployment":
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": ", "default"),
                        ANSIThemeStr(f"{cluster_deployment_file}", "path"),
                        ANSIThemeStr(" is not a valid ", "default"),
                        ANSIThemeStr("ClusterDeployment", "command"),
                        ANSIThemeStr(" configuration; aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)

    if cd_api_version not in ("v1alpha1", "v1alpha2", "v1alpha3", "v1alpha4"):
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": ", "default"),
                        ANSIThemeStr("The apiVersion ", "default"),
                        ANSIThemeStr(f"{cd_api_version}", "version"),
                        ANSIThemeStr(" used in ", "default"),
                        ANSIThemeStr(f"{cluster_deployment_file}", "path"),
                        ANSIThemeStr(" is not supported; aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)

    # Fetch all configuration options
    k8s_distro = deep_get(cd, DictPath("options#distro"), "kubeadm")
    requested_version = deep_get(cd, DictPath("options#version"), "latest")
    verbose = deep_get(cd, DictPath("options#verbose"), False)
    pod_network_cidr = deep_get(cd, DictPath("options#podNetworkCIDR"), "10.244.0.0/16")
    cni = deep_get(cd, DictPath("options#cni"), "cilium")
    cni_version = deep_get(cd, DictPath("options#cni_version"))
    if cni_version == "latest":
        cni_version = None
    cni_executable_version = deep_get(cd, DictPath("options#cni_executable_version"))
    if cni_executable_version == "latest":
        cni_executable_version = None
    no_password = not deep_get(cd, DictPath("options#promptForPassword"), True)
    save_ansible_logs = deep_get(cd, DictPath("options#saveAnsibleLogs"), False)

    control_planes = deep_get(cd, DictPath("cluster#controlPlanes"))
    if control_planes is None or not control_planes:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": ", "default"),
                        ANSIThemeStr("The configuration in ", "default"),
                        ANSIThemeStr(f"{cluster_deployment_file}", "path"),
                        ANSIThemeStr(" does not specify a control plane; aborting.", "default")],
                       stderr=True)
        sys.exit(errno.EINVAL)

    cri = deep_get(cd, DictPath("cluster#cri"), "containerd")
    cri_version = deep_get(cd, DictPath("cluster#cri_version"))
    if cri_version == "auto":
        cri_version = None

    cluster_name = deep_get(cd, DictPath("cluster#clusterName"))
    if cluster_name is None or not cluster_name:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": ", "default"),
                        ANSIThemeStr("The configuration in ", "default"),
                        ANSIThemeStr(f"{cluster_deployment_file}", "path"),
                        ANSIThemeStr(" does not specify a cluster name; aborting.", "default")],
                       stderr=True)
        sys.exit(errno.EINVAL)

    if deep_get(installation_info, DictPath("installation_target")) != cluster_name:
        installation_info = \
            update_installation_info(installation_target=cluster_name,
                                     cluster_name=cluster_name, distro=k8s_distro)

    rke2_configuration_path = \
        __get_configuration_data(cluster_deployment_file, "rke2Configuration",
                                 deep_get(cd, DictPath("cluster#configurationTemplates"), {}),
                                 HOMEDIR.joinpath("templates", "config",
                                                  "rke2_configuration.yaml.j2"))
    cluster_configuration_path = \
        __get_configuration_data(cluster_deployment_file, "clusterConfiguration",
                                 deep_get(cd, DictPath("cluster#configurationTemplates"), {}),
                                 HOMEDIR.joinpath("templates", "config",
                                                  "cluster_configuration.yaml.j2"))
    init_configuration_path = \
        __get_configuration_data(cluster_deployment_file, "initConfiguration",
                                 deep_get(cd, DictPath("cluster#configurationTemplates"), {}),
                                 HOMEDIR.joinpath("templates", "config",
                                                  "init_configuration.yaml.j2"))
    kubelet_configuration_path = \
        __get_configuration_data(cluster_deployment_file, "kubeletConfiguration",
                                 deep_get(cd, DictPath("cluster#configurationTemplates"), {}),
                                 HOMEDIR.joinpath("templates", "config",
                                                  "kubelet_configuration.yaml.j2"))
    kube_proxy_configuration_path = \
        __get_configuration_data(cluster_deployment_file, "kubeProxyConfiguration",
                                 deep_get(cd, DictPath("cluster#configurationTemplates"), {}),
                                 HOMEDIR.joinpath("templates", "config",
                                                  "kube_proxy_configuration.yaml.j2"))

    nodes = []

    # Get a list of all nodes in the cluster
    for _node_group, data in deep_get(cd, DictPath("nodes#groups"), {}).items():
        nodes += deep_get(data, DictPath("nodes"), [])

    prepare_options = [
        ("--control-plane", control_planes[0]),
        ("--hide-next-step", None),
        ("--nodes", nodes),
    ]
    if not confirm:
        prepare_options.append(("-Y", None))
    if verbose:
        prepare_options.append(("--verbose", None))
    if save_ansible_logs:
        prepare_options.append(("--save-ansible-logs", None))
    if not no_password:
        prepare_options.append(("--no-password", None))

    prepare_args = [
        cluster_name,
    ]

    if requested_version == "latest":
        prepare_args.append(f"{k8s_distro}=")
    else:
        prepare_args.append(f"{k8s_distro}={requested_version}")

    setup_control_plane_options = [
        ("-Y", None),
        ("--cri", cri),
        ("--cri-version", cri_version),
    ]

    if k8s_distro == "kubeadm":
        setup_control_plane_options += [
            ("--configuration-paths", [
                cluster_configuration_path,
                init_configuration_path,
                kubelet_configuration_path,
                kube_proxy_configuration_path,
            ])
        ]
    elif k8s_distro == "rke2":
        setup_control_plane_options += [
            ("--configuration-paths", [rke2_configuration_path])
        ]

    taint_control_planes = not deep_get(cd, DictPath("cluster#untaintControlPlanes"), False)
    setup_control_plane_options.append(("--extra-values",
                                        {"taint_control_plane": taint_control_planes}))

    if verbose:
        setup_control_plane_options.append(("--verbose", None))
    if save_ansible_logs:
        setup_control_plane_options.append(("--save-ansible-logs", None))

    setup_control_plane_args = [
        # Pass "none" as CNI; since we'll potentially be adding nodes
        # we want to postpone deploying the CNI until after those
        # have been added, to avoid extra traffic
        "none",
        pod_network_cidr,
    ]

    cluster_name = installation_info["installation_target"]
    state = deep_get(installation_info, DictPath(f"{cluster_name}#state"))
    phase = deep_get(installation_info, DictPath(f"{cluster_name}#phase"))

    if state is None or state in ("<none>", "preparing", "torn_down"):
        # First prepare all hosts
        prepare_installation(prepare_options, prepare_args)
    elif state not in ("adding_nodes", "installing", "installed", "deploying",
                       "<done>", "setup_cni", "deploy_workloads", "deployed", "prepared",
                       "upgraded"):
        ansithemeprint([ANSIThemeStr("\nError", "error"),
                        ANSIThemeStr(": Invalid installation state ", "default"),
                        ANSIThemeStr(f"{state}", "phase"),
                        ANSIThemeStr("; aborting.", "default")],
                       stderr=True)
        sys.exit(errno.EINVAL)

    installation_info = get_installation_info()
    state = deep_get(installation_info, DictPath(f"{cluster_name}#state"))
    phase = deep_get(installation_info, DictPath(f"{cluster_name}#phase"))

    # Now it's time to setup the control plane(s)
    if state in ("prepared", "installing"):
        setup_control_planes(setup_control_plane_options, setup_control_plane_args)
        phase = 0
        state = "adding_nodes"
        installation_info = \
            update_installation_info(cluster_name=cluster_name, state=state, phase=phase)

    state = deep_get(installation_info, DictPath(f"{cluster_name}#state"))
    phase = deep_get(installation_info, DictPath(f"{cluster_name}#phase"))

    admin_name = f"kubernetes-admin+{cluster_name}"
    context_name = f"{admin_name}@{cluster_name}"

    if kh is None:
        kh = kubernetes_helper.KubernetesHelper(about.PROGRAM_SUITE_NAME,
                                                about.PROGRAM_SUITE_VERSION, None)

    # We (hopefully) have a cluster now; change the context to that cluster
    if not kh.set_context(name=context_name, unchanged_is_success=True):
        ansithemeprint([ANSIThemeStr("Warning", "warning"),
                        ANSIThemeStr(": Failed to change to context ", "default"),
                        ANSIThemeStr(f"{context_name}", "hostname"),
                        ANSIThemeStr(" for cluster ", "default"),
                        ANSIThemeStr(f"{cluster_name}", "hostname"),
                        ANSIThemeStr("; aborting.", "default")])
        sys.exit(errno.ENOENT)

    if state == "installed" and phase == "<done>":
        phase = 0
        if nodes:
            state = "adding_nodes"
            control_plane_version = \
                get_control_plane_version(controlplane=control_planes[0], k8s_distro=k8s_distro)
            installation_info = \
                update_installation_info(cluster_name=cluster_name, state=state, phase=phase,
                                         version=control_plane_version)
        else:
            state = "setup_cni"
            installation_info = \
                update_installation_info(cluster_name=cluster_name, state=state, phase=phase)

    # Set necessary Ansible keys before running playbooks
    insecure_registries = deep_get(cmtlib.cmtconfig, DictPath("Docker#insecure_registries"), [])
    registry_mirrors = deep_get(cmtlib.cmtconfig, DictPath("Containerd#registry_mirrors"), [])
    retval = 0

    use_proxy = "no"
    if http_proxy or https_proxy:
        use_proxy = "yes"

    extra_values = {
        "http_proxy": http_proxy,
        "https_proxy": https_proxy,
        "no_proxy": no_proxy,
        "insecure_registries": insecure_registries,
        "registry_mirrors": registry_mirrors,
        "use_proxy": use_proxy,
    }

    # pylint: disable-next=too-many-nested-blocks
    if state == "adding_nodes":
        i = 0
        # We (hopefully) don't need to check whether the nodes are part of the cluster already,
        # since we've just setup the control plane...
        ansithemeprint([ANSIThemeStr("\n[Adding nodes]", "phase")])

        # Check whether ansible_password is defined or not
        if deep_get(ansible_configuration,
                    DictPath("ansible_password")) is None and not no_password:
            ansithemeprint([ANSIThemeStr("Attention", "warning"),
                            ANSIThemeStr(": To be able to run playbooks you need to provide the "
                                         "ansible/ssh password.", "default")])
            ansithemeprint([ANSIThemeStr("Since the systems will be reconfigured to use "
                                         "passwordless sudo and ssh keys "
                                         "this is a one-time thing.\n", "default")])
            ansithemeprint([ANSIThemeStr("Note", "note"),
                            ANSIThemeStr(": If the remote host is already configured for "
                                         "passwordless sudo and allows for login using a "
                                         "pre-generated SSH-key you can use “", "default"),
                            ANSIThemeStr("--no-password", "option"),
                            ANSIThemeStr("“ to bypass this check.", "default")])
            ansible_password = ansithemeinput_password([ANSIThemeStr("\nPassword: ", "default")])
            if not ansible_password:
                ansithemeprint([ANSIThemeStr("\nError", "error"),
                                ANSIThemeStr(": Empty password; aborting.", "default")],
                               stderr=True)
                sys.exit(errno.EINVAL)
            else:
                ansible_configuration["ansible_password"] = ansible_password

        node_global_cri = deep_get(cd, DictPath("nodes#options#cri"), cri)
        node_global_cri_version = deep_get(cd, DictPath("nodes#options#cri_version"), cri_version)
        if node_global_cri_version == "auto":
            node_global_cri_version = None
        node_global_forks = deep_get(cd, DictPath("nodes#options#forks"), 5)
        node_global_cordon_on_join = deep_get(cd, DictPath("nodes#options#cordonOnJoin"), False)
        node_global_configuration = \
            deep_get(cd, DictPath("nodes#options#configurationTemplates"), {})

        # We've potentially got multiple groups of nodes,
        # so calling add_nodes() once isn't gonna cut it
        for group, data in deep_get(cd, DictPath("nodes#groups"), {}).items():
            # Don't add node groups again
            if i < phase:
                continue
            tmp_nodes: list[str] = deep_get(data, DictPath("nodes"), [])
            hosts = []
            hostname_template: str = deep_get(data, DictPath("vmNameTemplate"),
                                              "vmnode-<<<item.0>>>-<<<item.1>>>")
            ranges: list = deep_get(data, DictPath("ranges"), [])
            virtualise: bool = deep_get(data, DictPath("virtualise"), False)
            vmhost: str = deep_get(data, DictPath("vmHost"), "")
            if virtualise:
                ipaddress_template: str = \
                    deep_get(data, DictPath("ipTemplate"),
                             "192.168.<<<item.0>>>.<<<item.1>>>")
                macaddress_template: \
                    str = deep_get(data, DictPath("macTemplate"),
                                   "52:54:00:42.<<<item.0>>>.<<<item.1>>>")
            else:
                ipaddress_template = ""
                macaddress_template = ""
            os_image: FilePath = FilePath(deep_get(data, DictPath("osImage"), ""))
            os_image = FilePath(substitute_string(os_image, {"{HOME}": HOMEDIR}))
            os_variant: str = deep_get(data, DictPath("osVariant"), "")
            vcpus: int = deep_get(data, DictPath("vCPUs"), 4)
            extra_values["vcpus"] = vcpus
            vmram: int = deep_get(data, DictPath("vmRAM"), 4096)
            extra_values["vmram"] = vmram
            template_balloon_size: str = \
                FilePath(deep_get(data, DictPath("templateBalloonSize"), "5G"))
            template_name: str = deep_get(data, DictPath("vmTemplateImageName"), "template")
            template_mac: str = ""
            template_ip: str = ""
            nodes_tuple: list[tuple[str, str, str]] = []
            if tmp_nodes:
                for node in tmp_nodes:
                    nodes_tuple.append((node, "", ""))
                    hosts.append(node)
            else:
                if not hostname_template:
                    sys.exit("Either nodes or vmNameTemplate is mandatory")
                for address_range in ranges:
                    if len(address_range) != 2:
                        sys.exit("An address-range must be [[high-digit range], [low-digit range]]")
                    tmp_highrange, tmp_lowrange = address_range
                    if isinstance(tmp_highrange, int):
                        highrange = [tmp_highrange]
                    else:
                        if len(tmp_highrange) != 2:
                            sys.exit("An address-range must be [start, end]")
                        highrange = list(range(tmp_highrange[0], tmp_highrange[1] + 1))
                    if isinstance(tmp_lowrange, int):
                        lowrange = [tmp_lowrange]
                    else:
                        if len(tmp_lowrange) != 2:
                            sys.exit("An address-range must be [start, end]")
                        lowrange = list(range(tmp_lowrange[0], tmp_lowrange[1] + 1))
                    for highdigit in highrange:
                        for lowdigit in lowrange:
                            ipaddress = substitute_string(ipaddress_template, {
                                                          "<<<item.1>>>": str(lowdigit),
                                                          "<<<item.0>>>": str(highdigit)})
                            macaddress = substitute_string(macaddress_template, {
                                                           "<<<item.1>>>": f"{lowdigit:0>2x}",
                                                           "<<<item.0>>>": f"{highdigit:0>2x}"})
                            # The first IP is reserved for the template
                            if not template_ip:
                                template_ip = ipaddress
                                template_mac = macaddress
                                continue
                            nodename = substitute_string(hostname_template, {
                                                         "<<<item.1>>>": str(lowdigit),
                                                         "<<<item.0>>>": str(highdigit)})
                            nodes_tuple.append((nodename, ipaddress, macaddress))
                            hosts.append(nodename)

            node_cri = deep_get(cd, DictPath("options#cri"), node_global_cri)
            node_cri_version = deep_get(cd, DictPath("options#cri_version"),
                                        node_global_cri_version)
            if node_cri_version == "auto":
                node_cri_version = None
            node_forks = deep_get(cd, DictPath("options#forks"), node_global_forks)
            node_cordon_on_join = \
                deep_get(cd, DictPath("options#cordonOnJoin"), node_global_cordon_on_join)

            node_configuration = \
                deep_get(group,
                         DictPath("options#configurationTemplates"), node_global_configuration)
            if k8s_distro == "kubeadm":
                join_configuration_path = \
                    __get_configuration_data(cluster_deployment_file, "joinConfiguration",
                                             node_configuration,
                                             HOMEDIR.joinpath("templates", "config",
                                                              "join_configuration.yaml.j2"))
            elif k8s_distro == "rke2":
                join_configuration_path = \
                    __get_configuration_data(cluster_deployment_file, "rke2Configuration",
                                             node_configuration,
                                             HOMEDIR.joinpath("templates", "config",
                                                              "rke2_configuration.yaml.j2"))

            add_nodes_options = [
                ("--cri", node_cri),
                ("--cri-version", node_cri_version),
                ("--configuration-paths", [
                    join_configuration_path,
                ]),
                ("--cordon-on-join", node_cordon_on_join),
                ("--forks", node_forks),
            ]
            if verbose:
                add_nodes_options.append(("--verbose", None))
            if save_ansible_logs:
                setup_control_plane_options.append(("--save-ansible-logs", None))

            add_nodes_args = [
                # List of nodes
                hosts,
                # Group(s) to add the nodes to
                [group],
            ]

            if not nodes_tuple:
                continue

            # Prepare nodes
            if virtualise:
                retval = prepare_vm_template(vmhost,
                                             [(template_name, template_ip, template_mac)],
                                             os_image=os_image,
                                             os_variant=os_variant,
                                             template_name=template_name,
                                             template_balloon_size=template_balloon_size,
                                             extra_values=extra_values,
                                             kubernetes_helper=kh, cri=cri, verbose=verbose)
                if retval:
                    ansithemeprint([ANSIThemeStr("Error", "error"),
                                    ANSIThemeStr(": failed to prepare a VM image template"
                                                 "; aborting.", "default")], stderr=True)
                    sys.exit(retval)
                retval = create_vm_hosts(vmhost, nodes_tuple,
                                         os_image=os_image,
                                         os_variant=os_variant,
                                         extra_values=extra_values, verbose=verbose)
                if retval:
                    ansithemeprint([ANSIThemeStr("Error", "error"),
                                    ANSIThemeStr(": failed to instantiate VM hosts"
                                                 "; aborting.", "default")], stderr=True)
                    sys.exit(retval)
                retval = join_nodes(hosts,
                                    kubernetes_helper=kh,
                                    k8s_distro=k8s_distro,
                                    cri=cri,
                                    extra_values=extra_values, verbose=verbose)
                if retval:
                    ansithemeprint([ANSIThemeStr("Error", "error"),
                                    ANSIThemeStr(": failed to join VM nodes to the cluster"
                                                 "; aborting.", "default")], stderr=True)
                    sys.exit(retval)
            else:
                retval = prepare_nodes(hosts=hosts, extra_values=extra_values, verbose=verbose)
                if retval:
                    ansithemeprint([ANSIThemeStr("Error", "error"),
                                    ANSIThemeStr(": failed to prepare hosts for use as "
                                                 "worker nodes; aborting.", "default")],
                                   stderr=True)
                    sys.exit(retval)
                add_nodes(add_nodes_options, add_nodes_args)
            i += 1
            installation_info = \
                update_installation_info(cluster_name=cluster_name, state=state, phase=i)

        ansithemeprint([ANSIThemeStr("\nNodes successfully added", "success")])
        phase = 0
        state = "setup_cni"
        installation_info = \
            update_installation_info(cluster_name=cluster_name, state=state, phase=phase)

    # Now we can setup the CNI
    admin_name = f"kubernetes-admin+{cluster_name}"
    context_name = f"{admin_name}@{cluster_name}"

    if state == "setup_cni":
        __setup_cni(cni, context_name, cluster_name,
                    cni_version=cni_version, cni_executable_version=cni_executable_version)
        phase = 0
        state = "deploy_workloads"
        installation_info = \
            update_installation_info(cluster_name=cluster_name, state=state, phase=phase, cni=cni)

    if redeploy:
        phase = 0
        state = "deploy_workloads"
        installation_info = \
            update_installation_info(cluster_name=cluster_name, state=state, phase=phase)

    # Finally we attempt to deploy workloads
    workloads = deep_get(cd, DictPath("workloads"), [])

    # pylint: disable-next=too-many-nested-blocks
    if state in ("deploy_workloads", "deploying") and workloads:
        ansithemeprint([ANSIThemeStr("\n[Deploying workloads]", "phase")])

        for i, workload in enumerate(workloads):
            if i < phase:
                continue

            name = deep_get(workload, DictPath("name"))
            skip = deep_get(workload, DictPath("skip"), False)
            if skip:
                installation_info = \
                    update_installation_info(cluster_name=cluster_name, state="deploying", phase=i)
                continue
            description = deep_get(workload, DictPath("description"), "")
            split_description = description.splitlines()
            # For now this is unused
            _error_policy = deep_get(workload, DictPath("errorPolicy"), "ignore")

            if name is None:
                ansithemeprint([ANSIThemeStr("Error", "error"),
                                ANSIThemeStr(": ", "default"),
                                ANSIThemeStr("name", "emphasis"),
                                ANSIThemeStr(" is a mandatory field for workloads; aborting.",
                                             "default")], stderr=True)
                sys.exit(errno.EINVAL)

            args = []

            ansithemeprint([ANSIThemeStr("\n  • ", "separator"),
                            ANSIThemeStr("Deploying ", "subaction"),
                            ANSIThemeStr(f"{name}", "argument")])

            if verbose:
                for line in split_description:
                    ansithemeprint([ANSIThemeStr("    ", "separator"),
                                    ANSIThemeStr(f"{line}", "default")])
                print()

            deployments = deep_get(workload, DictPath("deployments"), [])
            kubectl_path = \
                secure_which(FilePath("/usr/bin/kubectl"),
                             fallback_allowlist=["/etc/alternatives"],
                             security_policy=SecurityPolicy.ALLOWLIST_RELAXED)
            sudo_path = \
                cmtio.secure_which(FilePath("sudo"),
                                   fallback_allowlist=["/bin", "/usr/bin"],
                                   security_policy=SecurityPolicy.ALLOWLIST_STRICT)
            try:
                helm_path = \
                    secure_which(FilePath("/usr/bin/helm"),
                                 fallback_allowlist=["/etc/alternatives",
                                                     "/usr/local/bin", f"{HOMEDIR}/bin"],
                                 security_policy=SecurityPolicy.ALLOWLIST_RELAXED)
            except FileNotFoundError:
                helm_path = None

            try:
                git_path = \
                    secure_which(FilePath("/usr/bin/git"),
                                 fallback_allowlist=["/etc/alternatives", "/usr/lib/git",
                                                     "/usr/local/bin", f"{HOMEDIR}/bin"],
                                 security_policy=SecurityPolicy.ALLOWLIST_RELAXED)
            except FileNotFoundError:
                git_path = None

            for data in deployments:
                intersection = {"kubectl", "helm", "git"} & data.keys()
                if len(intersection) != 1:
                    ansithemeprint([ANSIThemeStr("Error", "error"),
                                    ANSIThemeStr(": The workload ", "default"),
                                    ANSIThemeStr(f"{name}", "argument"),
                                    ANSIThemeStr(" is invalid; ", "default"),
                                    ANSIThemeStr("each deployment must contain exactly one of ",
                                                 "default"),
                                    ANSIThemeStr("git", "emphasis"),
                                    ANSIThemeStr(", ", "default"),
                                    ANSIThemeStr("kubectl", "emphasis"),
                                    ANSIThemeStr(", or ", "default"),
                                    ANSIThemeStr("helm", "emphasis"),
                                    ANSIThemeStr("; aborting.", "default")], stderr=True)
                    sys.exit(errno.EINVAL)

                if "git" in data:
                    if git_path is None:
                        ansithemeprint([ANSIThemeStr("Error", "error"),
                                        ANSIThemeStr(": The workload ", "default"),
                                        ANSIThemeStr(f"{name}", "argument"),
                                        ANSIThemeStr(" uses ", "default"),
                                        ANSIThemeStr("git", "emphasis"),
                                        ANSIThemeStr(", but it's not installed; aborting.",
                                                     "default")], stderr=True)
                        sys.exit(errno.ENOENT)

                    deployment = deep_get(data, DictPath("git"), {})
                    repo = deep_get(deployment, DictPath("repo"))
                    destination = deep_get(deployment, DictPath("destination"))
                    if destination.startswith("{HOME}"):
                        destination = destination.replace("{HOME}", HOMEDIR, 1)
                    branch = deep_get(deployment, DictPath("branch"))
                    fetch = deep_get(deployment, DictPath("fetch"), False)

                    if destination is None:
                        ansithemeprint([ANSIThemeStr("Error", "error"),
                                        ANSIThemeStr(": The workload ", "default"),
                                        ANSIThemeStr(f"{name}", "argument"),
                                        ANSIThemeStr(" is invalid; ", "default"),
                                        ANSIThemeStr("destination", "emphasis"),
                                        ANSIThemeStr(" is a mandatory field for ", "default"),
                                        ANSIThemeStr("git", "emphasis"),
                                        ANSIThemeStr(" deployments; aborting.",
                                                     "default")], stderr=True)
                        sys.exit(errno.EINVAL)

                    security_checks = [
                        SecurityChecks.PARENT_RESOLVES_TO_SELF,
                        SecurityChecks.OWNER_IN_ALLOWLIST,
                        SecurityChecks.PARENT_OWNER_IN_ALLOWLIST,
                        SecurityChecks.PERMISSIONS,
                        SecurityChecks.PARENT_PERMISSIONS,
                        SecurityChecks.EXISTS,
                        SecurityChecks.IS_DIR,
                    ]

                    if check_path(destination, checks=security_checks) == [SecurityStatus.OK]:
                        if fetch:
                            args = [git_path, "-C", destination, "fetch"]
                            check_and_print_status(execute_command(args, env=env))

                        if branch is not None:
                            args = [git_path, "-C", destination, "checkout", f"tags/{branch}"]
                            check_and_print_status(execute_command(args, env=env))
                    elif retval == [SecurityStatus.DOES_NOT_EXIST]:
                        security_checks = [
                            SecurityChecks.PARENT_RESOLVES_TO_SELF,
                            SecurityChecks.PARENT_OWNER_IN_ALLOWLIST,
                            SecurityChecks.PARENT_PERMISSIONS,
                        ]

                        if check_path(destination, checks=security_checks) != [SecurityStatus.OK]:
                            ansithemeprint([ANSIThemeStr("Error", "error"),
                                            ANSIThemeStr(": The workload ", "default"),
                                            ANSIThemeStr(f"{name}", "argument"),
                                            ANSIThemeStr(" has an invalid destionation path; "
                                                         "aborting.", "default")], stderr=True)
                            sys.exit(errno.EINVAL)

                        if repo is None or not repo:
                            ansithemeprint([ANSIThemeStr("Error", "error"),
                                            ANSIThemeStr(": No local checkout of the workload ",
                                                         "default"),
                                            ANSIThemeStr(f"{name}", "argument"),
                                            ANSIThemeStr(" exists, and ", "default"),
                                            ANSIThemeStr("repo", "emphasis"),
                                            ANSIThemeStr(" has not been specified; aborting.",
                                                         "default")], stderr=True)
                            sys.exit(errno.ENOENT)

                        if branch is not None:
                            args = [git_path, "clone", "--depth", "1", "--branch",
                                    branch, repo, destination]
                        else:
                            args = [git_path, "clone", "--depth", "1", repo, destination]

                        check_and_print_status(execute_command(args, env=env))
                elif "kubectl" in data:
                    deployment = deep_get(data, DictPath("kubectl"), {})
                    method = deep_get(deployment, DictPath("method"), "apply")
                    wtype = deep_get(deployment, DictPath("type"), "file")
                    uris = deep_get(deployment, DictPath("uri"))
                    data = deep_get(deployment, DictPath("data"))
                    cert = deep_get(deployment, DictPath("cert"))
                    key = deep_get(deployment, DictPath("key"))
                    pname = deep_get(deployment, DictPath("name"))
                    pnamespace = deep_get(deployment, DictPath("namespace"))
                    resource = deep_get(deployment, DictPath("resource"))
                    version = deep_get(deployment, DictPath("version"))

                    if wtype is None:
                        ansithemeprint([ANSIThemeStr("Error", "error"),
                                        ANSIThemeStr(": The workload ", "default"),
                                        ANSIThemeStr(f"{name}", "argument"),
                                        ANSIThemeStr(" is invalid; ", "default"),
                                        ANSIThemeStr("type", "emphasis"),
                                        ANSIThemeStr(" is a mandatory field for ", "default"),
                                        ANSIThemeStr("kubectl", "emphasis"),
                                        ANSIThemeStr(" deployments; aborting.", "default")],
                                       stderr=True)
                        sys.exit(errno.EINVAL)

                    if wtype not in ("embedded", "file", "kustomization",
                                     "namespace", "secret+tls"):
                        ansithemeprint([ANSIThemeStr("Error", "error"),
                                        ANSIThemeStr(": The type ", "default"),
                                        ANSIThemeStr(f"{wtype}", "argument"),
                                        ANSIThemeStr(" for the workload ", "default"),
                                        ANSIThemeStr(f"{name}", "argument"),
                                        ANSIThemeStr(" is invalid; only ", "default"),
                                        ANSIThemeStr("embedded", "emphasis"),
                                        ANSIThemeStr(", ", "default"),
                                        ANSIThemeStr("namespace", "emphasis"),
                                        ANSIThemeStr(", ", "default"),
                                        ANSIThemeStr("secret+tls", "emphasis"),
                                        ANSIThemeStr(", and ", "default"),
                                        ANSIThemeStr("file", "emphasis"),
                                        ANSIThemeStr(", and ", "default"),
                                        ANSIThemeStr("kustomization", "emphasis"),
                                        ANSIThemeStr(" are supported for .", "default"),
                                        ANSIThemeStr("kubectl", "emphasis"),
                                        ANSIThemeStr(" deployments; aborting.", "default")],
                                       stderr=True)
                        sys.exit(errno.EINVAL)

                    if method not in ("apply", "create", "patch+strategic",
                                      "patch+merge", "patch+json"):
                        ansithemeprint([ANSIThemeStr("Error", "error"),
                                        ANSIThemeStr(": The method ", "default"),
                                        ANSIThemeStr(f"{method}", "argument"),
                                        ANSIThemeStr(" for the workload ", "default"),
                                        ANSIThemeStr(f"{name}", "argument"),
                                        ANSIThemeStr(" is invalid; only ", "default"),
                                        ANSIThemeStr("apply", "emphasis"),
                                        ANSIThemeStr(", ", "default"),
                                        ANSIThemeStr("create", "emphasis"),
                                        ANSIThemeStr(", ", "default"),
                                        ANSIThemeStr("patch+strategic", "emphasis"),
                                        ANSIThemeStr(", ", "default"),
                                        ANSIThemeStr("patch+merge", "emphasis"),
                                        ANSIThemeStr(", and ", "default"),
                                        ANSIThemeStr("patch+json", "emphasis"),
                                        ANSIThemeStr(" are supported for .", "default"),
                                        ANSIThemeStr("kubectl", "emphasis"),
                                        ANSIThemeStr(" deployments; aborting.", "default")],
                                       stderr=True)
                        sys.exit(errno.EINVAL)

                    if wtype not in ("namespace", "secret+tls") \
                            and (uris is None and data is None
                                 or uris is not None and data is not None):
                        ansithemeprint([ANSIThemeStr("Error", "error"),
                                        ANSIThemeStr(": The workload ", "default"),
                                        ANSIThemeStr(f"{name}", "argument"),
                                        ANSIThemeStr(" is invalid; exactly one of ", "default"),
                                        ANSIThemeStr("uri", "emphasis"),
                                        ANSIThemeStr(" or ", "default"),
                                        ANSIThemeStr("data", "emphasis"),
                                        ANSIThemeStr(" is a mandatory field for ", "default"),
                                        ANSIThemeStr("kubectl", "emphasis"),
                                        ANSIThemeStr(" deployments; aborting.", "default")],
                                       stderr=True)
                        sys.exit(errno.EINVAL)

                    if wtype == "secret+tls" and (cert is None or key is None):
                        ansithemeprint([ANSIThemeStr("Error", "error"),
                                        ANSIThemeStr(": The workload ", "default"),
                                        ANSIThemeStr(f"{name}", "argument"),
                                        ANSIThemeStr(" is invalid; when type is ", "default"),
                                        ANSIThemeStr("secret+tls", "argument"),
                                        ANSIThemeStr(" both ", "default"),
                                        ANSIThemeStr("key", "emphasis"),
                                        ANSIThemeStr(" and ", "default"),
                                        ANSIThemeStr("cert", "emphasis"),
                                        ANSIThemeStr(" are mandatory fields.", "default")],
                                       stderr=True)
                        sys.exit(errno.EINVAL)

                    if uris is None:
                        uris = []
                    if isinstance(uris, str):
                        uris = [uris]

                    args = [kubectl_path]
                    if method == "create":
                        args += ["create"]
                    elif method == "apply":
                        args += ["apply", "--server-side"]
                    elif method == "patch+strategic":
                        args += ["patch", resource, pname, "--type", "strategic"]
                    elif method == "patch+merge":
                        args += ["patch", resource, pname, "--type", "merge"]
                    elif method == "patch+json":
                        args += ["patch", resource, pname, "--type", "json"]

                    if method.startswith("patch"):
                        if pnamespace is not None:
                            args += ["--namespace", pnamespace]
                        args += ["--patch-file"]
                    elif wtype in ("embedded", "file"):
                        args += ["-f"]
                    elif wtype == "kustomization":
                        args += ["-k"]
                    elif wtype == "namespace":
                        args = [kubectl_path, "create", "namespace", pname]
                    elif wtype == "secret+tls":
                        # secret + tls needs to run as sudo to be able to access /etc/kubernetes/pki
                        args = [sudo_path, kubectl_path, "create", "--kubeconfig", KUBE_CONFIG_FILE,
                                "secret", "tls", pname, "--cert", cert, "--key", key]
                        if pnamespace is not None:
                            args += ["--namespace", pnamespace]

                    for uri in uris:
                        if not uri.startswith(("file://", "http://", "https://")):
                            ansithemeprint([ANSIThemeStr("Error", "error"),
                                            ANSIThemeStr(": The URI ", "default"),
                                            ANSIThemeStr(f"{uri}", "argument"),
                                            ANSIThemeStr(" for the workload ", "default"),
                                            ANSIThemeStr(f"{name}", "argument"),
                                            ANSIThemeStr(" is invalid; only ", "default"),
                                            ANSIThemeStr("http://", "emphasis"),
                                            ANSIThemeStr(", ", "default"),
                                            ANSIThemeStr("https://", "emphasis"),
                                            ANSIThemeStr(", and ", "default"),
                                            ANSIThemeStr("file://", "emphasis"),
                                            ANSIThemeStr(" are supported URI-prefixes; aborting.",
                                                         "default")], stderr=True)
                            sys.exit(errno.EINVAL)

                        if uri.startswith("file://"):
                            uri = uri.removeprefix("file://")
                            if uri.startswith("{HOME}"):
                                uri = uri.replace("{HOME}", HOMEDIR, 1)
                            _uri_type = "file"
                        elif uri.startswith("http://"):
                            _uri_type = "http"
                        elif uri.startswith("https://"):
                            _uri_type = "https"
                        if version is not None:
                            uri = uri.replace("<<<version>>>", version)

                        if verbose:
                            check_and_print_status(execute_command(args + [uri], env=env))
                        else:
                            execute_command_with_response(args + [uri], env=env)
                    if wtype in ("namespace", "secret+tls"):
                        dry_run_result = \
                            execute_command_with_response(args + ["--dry-run=server"], env=env)
                        if "already exists" not in dry_run_result:
                            if verbose:
                                check_and_print_status(execute_command(args, env=env))
                            else:
                                execute_command_with_response(args, env=env)
                    if data is not None:
                        # pylint: disable-next=consider-using-with
                        tf = tempfile.NamedTemporaryFile(delete=True)
                        path = FilePath(tf.name)
                        secure_write_string(path, data, temporary=True)
                        if verbose:
                            check_and_print_status(execute_command(args + [path], env=env))
                        else:
                            execute_command_with_response(args + [path], env=env)
                elif "helm" in data:
                    if helm_path is None:
                        ansithemeprint([ANSIThemeStr("Warning", "warning"),
                                        ANSIThemeStr(": The workload ", "default"),
                                        ANSIThemeStr(f"{name}", "argument"),
                                        ANSIThemeStr(" uses ", "default"),
                                        ANSIThemeStr("Helm", "emphasis"),
                                        ANSIThemeStr(", but it's not installed; skipping.",
                                                     "default")], stderr=True)
                        continue

                    deployment = deep_get(data, DictPath("helm"), {})
                    hname = deep_get(deployment, DictPath("name"))
                    chart = deep_get(deployment, DictPath("chart"))
                    namespace = deep_get(deployment, DictPath("namespace"))
                    create_namespace = deep_get(deployment, DictPath("createNamespace"), False)
                    repo_url = deep_get(deployment, DictPath("repoURL"))
                    repo_name = deep_get(deployment, DictPath("repoName"))
                    version = deep_get(deployment, DictPath("version"))

                    if None in (hname, chart, namespace, repo_url, repo_name, version):
                        ansithemeprint([ANSIThemeStr("Error", "error"),
                                        ANSIThemeStr(": The workload ", "default"),
                                        ANSIThemeStr(f"{name}", "argument"),
                                        ANSIThemeStr(" is invalid; the following fields are "
                                                     "mandatory for Helm deployments: ", "default"),
                                        ANSIThemeStr("name", "emphasis"),
                                        ANSIThemeStr(", ", "default"),
                                        ANSIThemeStr("chart", "emphasis"),
                                        ANSIThemeStr(", ", "default"),
                                        ANSIThemeStr("namespace", "emphasis"),
                                        ANSIThemeStr(", ", "default"),
                                        ANSIThemeStr("repoURL", "emphasis"),
                                        ANSIThemeStr(", ", "default"),
                                        ANSIThemeStr("repoName", "emphasis"),
                                        ANSIThemeStr(", and ", "default"),
                                        ANSIThemeStr("version", "emphasis"),
                                        ANSIThemeStr("; aborting.", "default")], stderr=True)
                        sys.exit(errno.EINVAL)

                    # Add helm repo
                    args = [helm_path, "repo", "add", repo_name, repo_url]
                    if verbose:
                        check_and_print_status(execute_command(args))
                    else:
                        execute_command_with_response(args)

                    # Check whether the component is already installed
                    args = [helm_path, "list", "--namespace", namespace,
                            "--filter", f"^{hname}$", "--no-headers"]
                    response = execute_command_with_response(args, env=env)
                    tmp = re.match(fr"^{hname}\s.+{version}$", response)
                    # We've got an exact match for version and name; for now we just ignore.
                    # We might add a remove + install policy later.
                    if tmp is not None:
                        ansithemeprint([ANSIThemeStr("Warning", "warning"),
                                        ANSIThemeStr(": The component ", "default"),
                                        ANSIThemeStr(f"{hname}", "argument"),
                                        ANSIThemeStr(" is already installed using ", "default"),
                                        ANSIThemeStr("Helm", "emphasis"),
                                        ANSIThemeStr(" with the specified version; skipping.",
                                                     "default")], stderr=True)
                        continue

                    # If redeploying workloads we might need to update the repo
                    args = [helm_path, "repo", "update"]

                    if verbose:
                        check_and_print_status(execute_command(args, env=env))
                    else:
                        execute_command_with_response(args, env=env)

                    args = [helm_path, "upgrade", hname, chart, "--namespace", namespace,
                            "--version", version, "--install"]
                    if create_namespace:
                        args.append("--create-namespace")

                    if verbose:
                        check_and_print_status(execute_command(args, env=env))
                    else:
                        execute_command_with_response(args, env=env)

            installation_info = \
                update_installation_info(cluster_name=cluster_name, state="deploying", phase=i)

        installation_info = \
            update_installation_info(cluster_name=cluster_name, state="deployed", phase="<done>")


def merge_configurations(configuration_paths: list[FilePath]) -> FilePath:
    """
    Given a list of configurations in YAML-format,
    load them all and write them out to a single, temporary, file.

        Parameters:
            configuration_paths ([str]): A list of paths to configuration files
    """
    security_checks = [
        SecurityChecks.PARENT_OWNER_IN_ALLOWLIST,
        SecurityChecks.OWNER_IN_ALLOWLIST,
        SecurityChecks.PERMISSIONS,
        SecurityChecks.IS_FILE,
    ]

    configurations = ""
    # pylint: disable-next=consider-using-with
    tf = tempfile.NamedTemporaryFile(suffix=".yaml.j2", delete=False)

    for path in configuration_paths:
        configuration = secure_read_string(path, checks=security_checks)
        if configurations:
            configurations += "---\n"
        configurations += configuration

    secure_write_string(FilePath(tf.name), configurations, temporary=True)

    return FilePath(tf.name)


# pylint: disable-next=too-many-locals,too-many-branches,too-many-statements
def setup_control_planes(options: list[tuple[str, Any]], args: list[str]) -> None:
    """
    Setup a control plane.

        Parameters:
            options ([(str, str)]): Options to use when executing this action
            args (list[str]): Arguments to use when executing this action
    """
    global setup_control_plane_tasks  # pylint: disable=global-statement

    confirm = True
    override_cni = False

    installation_info = get_installation_info()
    cluster_name = deep_get(installation_info, DictPath("installation_target"))
    k8s_distro = deep_get(installation_info, DictPath(f"{cluster_name}#distro"))
    configuration_paths: list[FilePath] = []

    if k8s_distro == "kubeadm":
        setup_control_plane_tasks = __setup_kubeadm_control_plane_tasks
        configuration_paths = [
            FilePath("templates").joinpath("config", "cluster_configuration.yaml.j2"),
            FilePath("templates").joinpath("config", "init_configuration.yaml.j2"),
            FilePath("templates").joinpath("config", "kubelet_configuration.yaml.j2"),
            FilePath("templates").joinpath("config", "kube_proxy_configuration.yaml.j2"),
        ]
    elif k8s_distro == "rke2":
        setup_control_plane_tasks = __setup_rke2_control_plane_tasks
        configuration_paths = [
            FilePath("templates").joinpath("etc", "rancher", "rke2", "config.yaml.j2"),
        ]

    if "--list-tasks" in (tmp[0] for tmp in options):
        __list_phases(setup_control_plane_tasks)
        sys.exit(0)

    requested_version = installation_info[cluster_name]["requested_version"]
    state = installation_info[cluster_name]["state"]
    phase: Union[str, int] = installation_info[cluster_name]["phase"]
    phase_skiplist = set(installation_info[cluster_name]["phase_skiplist"])
    cni = installation_info[cluster_name]["cni"]
    pod_network_cidr = installation_info[cluster_name]["pod_network_cidr"]
    verbose = False

    if state is None or state in ("<none>", "preparing"):
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": The system is not prepared for installation yet; run ",
                                     "default"),
                        ANSIThemeStr(f"{about.ADMIN_PROGRAM_NAME}", "programname"),
                        ANSIThemeStr(" prepare", "command"),
                        ANSIThemeStr(" before continuing. Aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)
    elif state is not None and state not in ("installing", "prepared"):
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": Invalid installation state; there is a partial "
                                     "or full installation already; aborting.", "default")],
                       stderr=True)
        sys.exit(errno.EINVAL)

    if state == "prepared":
        phase = 0

    cri = None
    cri_version = None
    enable_dra = False
    extra_values: dict = {}

    for opt, optarg in options:
        if opt == "--start-at-task":
            phase = __validate_task_index(setup_control_plane_tasks, optarg)
            if "--skip-tasks" not in options:
                phase_skiplist = set()
        elif opt == "--skip-tasks":
            skip_phases = __expand_index_list(optarg)
            for __phase in skip_phases:
                __validate_task_index(setup_control_plane_tasks, __phase)
            phase_skiplist = set(skip_phases)
        elif opt == "--resume":
            if phase is None or phase == "<none>":
                phase = 0
            elif phase == "<done>":
                ansithemeprint([ANSIThemeStr("Warning", "warning"),
                                ANSIThemeStr(": Ignoring request to resume a completed "
                                             "installation. Exiting.", "default")], stderr=True)
                sys.exit(0)
            phase = __validate_task_index(setup_control_plane_tasks, phase)
        elif opt == "--cri":
            if optarg in ("containerd", "cri-o"):
                cri = optarg
            else:
                ansithemeprint([ANSIThemeStr("Error", "error"),
                                ANSIThemeStr(": Unknown CRI “", "default"),
                                ANSIThemeStr(f"{optarg}", "argument"),
                                ANSIThemeStr("“ specified; aborting.", "default")], stderr=True)
                sys.exit(errno.EINVAL)
        elif opt == "--cri-version":
            cri_version = optarg
        elif opt == "--override-cni":
            override_cni = True
        elif opt == "--save-ansible-logs":
            ansible_configuration["save_logs"] = True
        elif opt == "--enable-dra":
            enable_dra = True
        elif opt == "--verbose":
            verbose = True
        elif opt == "-Y":
            confirm = False
        elif opt == "--extra-values":
            extra_values = optarg
        elif opt == "--configuration-paths":
            configuration_paths = optarg

    configuration_path = merge_configurations(configuration_paths)

    setup_extras = setup_control_plane_targets[k8s_distro]["extra_values"]
    setup_extras = {
        **extra_values,
        "configuration_path": configuration_path,
    }

    if enable_dra:
        dra_extra_values = {
            "api_server_feature_gates": "DynamicResourceAllocation=true",
            "api_server_runtime_config_api_alpha_enabled": "true",
            "controller_manager_feature_gates": "DynamicResourceAllocation=true",
            "scheduler_feature_gates": "DynamicResourceAllocation=true",
            "kubelet_feature_gates": "{ DynamicResourceAllocation: true }",
            "kubeproxy_feature_gates": "{ DynamicResourceAllocation: true }",
        }
        setup_extras = {
            **setup_extras,
            **dra_extra_values,
        }

        # Override the feature-gate format if the parameters are passed via command-line;
        # e.g. when using RKE2
        if k8s_distro == "rke2":
            setup_extras["kubelet_feature_gates"] = \
                "DynamicResourceAllocation=true"
            setup_extras["kubeproxy_feature_gates"] = \
                "DynamicResourceAllocation=true"
    else:
        no_dra_extra_values = {
            "api_server_feature_gates": "",
            "api_server_runtime_config_api_alpha_enabled": "false",
            "controller_manager_feature_gates": "",
            "kubelet_feature_gates": "{}",
            "kubeproxy_feature_gates": "{}",
            "scheduler_feature_gates": "",
        }
        setup_extras = {
            **setup_extras,
            **no_dra_extra_values,
        }

        # Override the feature-gate format if the parameters are passed via command-line;
        # e.g. when using RKE2
        if k8s_distro == "rke2":
            setup_extras["kubelet_feature_gates"] = ""
            setup_extras["kubeproxy_feature_gates"] = ""

    if k8s_distro == "kubeadm":
        # First remove the distro revision
        major_minor_patchrev = requested_version.split("-")
        # Now split the version tuple
        split_tuple = major_minor_patchrev[0].split(".")
        if len(split_tuple) < 2:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": Could not parse version “", "default"),
                            ANSIThemeStr(f"{major_minor_patchrev[0]}", "version"),
                            ANSIThemeStr("“; aborting.",
                                         "default")], stderr=True)
            sys.exit(errno.EINVAL)
        requested_major = split_tuple[0]
        requested_minor = split_tuple[1]

        if int(requested_minor) < 31:
            cluster_configuration_api_version = "kubeadm.k8s.io/v1beta3"
            init_configuration_api_version = "kubeadm.k8s.io/v1beta3"
        else:
            cluster_configuration_api_version = "kubeadm.k8s.io/v1beta4"
            init_configuration_api_version = "kubeadm.k8s.io/v1beta4"

        setup_extras["cluster_configuration_api_version"] = cluster_configuration_api_version
        setup_extras["init_configuration_api_version"] = init_configuration_api_version
        requested_major_minor = f"{requested_major}.{requested_minor}"

        # Add major/minor to extra_values
        setup_extras["kubernetes_major_minor_version"] = requested_major_minor
    elif k8s_distro == "rke2":
        if requested_version == "latest":
            if not (kubernetes_upstream_version := get_latest_upstream_version("kubernetes")):
                ansithemeprint([ANSIThemeStr("Error", "error"),
                                ANSIThemeStr(": “", "default"),
                                ANSIThemeStr("latest", "version"),
                                ANSIThemeStr("“ was specified as version, but latest upstream "
                                             "version could not be found; aborting.",
                                             "default")], stderr=True)
                sys.exit(errno.ENOENT)
            version_split = kubernetes_upstream_version.split(".")
            requested_major = version_split[0]
            requested_minor = version_split[1]
            requested_major_minor = f"{requested_major}.{requested_minor}"
            # Add major/minor to extra_values
            setup_extras["kubernetes_major_minor_version"] = requested_major_minor
        else:
            # First remove the distro revision, if any
            major_minor_patchrev = requested_version.lstrip("v").split("-")
            # Now split the version tuple
            version_split = major_minor_patchrev[0].split(".")
            requested_major = version_split[0]
            requested_minor = version_split[1]
            requested_major_minor = f"{requested_major}.{requested_minor}"
            # Add major/minor to extra_values
            setup_extras["kubernetes_major_minor_version"] = requested_major_minor

        # Add requested_version to extra_values;
        # for rke2 this is either vX.YY or latest
        setup_extras["requested_version"] = requested_version

        # For now at least we disable the cloud controller
        rke2_disable_cloud_controller = True
        setup_extras["disable_cloud_controller"] = rke2_disable_cloud_controller
        # Disable canal since we install the CNI separately
        rke2_disabled_options_list = ["rke2-canal"]
        setup_extras["disabled_options_list"] = rke2_disabled_options_list

    if cri is None:
        if k8s_distro == "rke2":
            cri = "containerd"
        elif enable_dra:
            cri = "cri-o"
        else:
            cri = "containerd"

    installation_info = update_installation_info(cri=cri)

    # Add the CRI to the setup playbooks for the control plane;
    # the list is short enough that doing prepend isn't a performance issue
    if cri == "containerd":
        cri_playbook_path = FilePath("setup_containerd.yaml")
        setup_control_plane_targets[k8s_distro]["playbooks"].insert(0, cri_playbook_path)
    elif cri == "cri-o":
        cri_playbook_path = FilePath("setup_cri-o.yaml")
        setup_control_plane_targets[k8s_distro]["playbooks"].insert(0, cri_playbook_path)
        if cri_version is None:
            tmp_crio_version = get_crio_version((int(requested_major), int(requested_minor)))
            if tmp_crio_version is None:
                check_and_print_status(False)
            # The type ignore below is necessary because mypy doesn't recognise that
            # check_and_print_status always aborts with sys.exit() when called with False
            crio_major_version, crio_minor_version = tmp_crio_version  # type: ignore
        else:
            # This case shouldn't be necessary; mypy should recognise that we're doing
            # a None-check first.
            tmp_crio_version = cast(tuple, cri_version.split("."))
            if len(tmp_crio_version) < 2:
                check_and_print_status(False)
            crio_major_version = tmp_crio_version[0]
            crio_minor_version = tmp_crio_version[1]
        setup_extras["crio_major_version"] = crio_major_version
        setup_extras["crio_minor_version"] = crio_minor_version

    if cni is None or cni == "<none>":
        if k8s_distro == "rke2":
            default_cni = "canal"
        else:
            default_cni = DEFAULT_CNI
        if len(args) == 0:
            ansithemeprint([ANSIThemeStr("Warning", "warning"),
                            ANSIThemeStr(": No CNI specified; defaulting to ", "default"),
                            ANSIThemeStr(f"{default_cni}", "programname"),
                            ANSIThemeStr(".", "default")], stderr=True)
            cni = default_cni
        elif cni != args[0]:
            cni = args[0]
    else:
        new_cni = cni
        if args:
            new_cni = args[0]

        if new_cni != cni:
            if not override_cni:
                ansithemeprint([ANSIThemeStr("Error", "error"),
                                ANSIThemeStr(": Installation was previous initiated with ",
                                             "default"),
                                ANSIThemeStr(f"{cni}", "programname"),
                                ANSIThemeStr(" as CNI; ", "default"),
                                ANSIThemeStr("you can attempt to override this by using ",
                                             "default"),
                                ANSIThemeStr("--override-cni", "option"),
                                ANSIThemeStr(", but success is not guaranteed. Aborting.",
                                             "default")], stderr=True)
                sys.exit(errno.EINVAL)
            else:
                ansithemeprint([ANSIThemeStr("Warning", "warning"),
                                ANSIThemeStr(": Installation was previous initiated with ",
                                             "default"),
                                ANSIThemeStr(f"{cni}", "programname"),
                                ANSIThemeStr(" as CNI, ", "default"),
                                ANSIThemeStr("but ", "default"),
                                ANSIThemeStr("--override-cni", "option"),
                                ANSIThemeStr(" was specified; continuing installation.",
                                             "default")], stderr=True)
        cni = new_cni

    if cni != "none" and cni not in cni_data:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": ", "default"),
                        ANSIThemeStr(f"{cni}", "argument"),
                        ANSIThemeStr(" is not a valid/supported CNI; aborting.", "default")],
                       stderr=True)
        sys.exit(errno.EINVAL)

    if len(args) > 1:
        new_pod_network_cidr = args[1]
    else:
        new_pod_network_cidr = DEFAULT_POD_NETWORK_CIDR
    if pod_network_cidr == "<none>" or pod_network_cidr == new_pod_network_cidr or phase == 0:
        pod_network_cidr = new_pod_network_cidr
    else:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": Installation was previously initiated with ", "default"),
                        ANSIThemeStr(f"{pod_network_cidr}", "emphasis"),
                        ANSIThemeStr(" as Pod Network CIDR;\n", "default"),
                        ANSIThemeStr("changing this value in the middle of installation is not "
                                     "supported. Aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)

    update_installation_info(cni=cni, pod_network_cidr=pod_network_cidr, state="installing",
                             phase=phase, phase_skiplist=list(phase_skiplist))

    show_configuration(action="Setting up control plane:", tasks=setup_control_plane_tasks)

    if confirm:
        input_retval = \
            ansithemeinput([ANSIThemeStr("\nStart installation? [y/", "default"),
                            ANSIThemeStr("N", "emphasis"),
                            ANSIThemeStr("]: ", "default")])
        if input_retval.lower() not in ("y", "yes"):
            ansithemeprint([ANSIThemeStr("\nAborting:", "error"),
                            ANSIThemeStr(" User stopped installation.", "default")], stderr=True)
            sys.exit(errno.EINTR)

    ansithemeprint([ANSIThemeStr("\n[Setting up control plane]", "phase")])

    setup_localhost = setup_control_plane_targets["localhost"]
    setup_k8s_distro = setup_control_plane_targets[k8s_distro]

    # Add the package versions to the package list
    if k8s_distro == "kubeadm":
        setup_localhost["deb_packages"].remove("kubectl")
        setup_localhost["deb_packages"].append(f"kubectl={requested_version}")
        setup_k8s_distro["deb_packages"].remove("kubectl")
        setup_k8s_distro["deb_packages"].append(f"kubectl={requested_version}")
        setup_k8s_distro["deb_packages"].remove("kubeadm")
        setup_k8s_distro["deb_packages"].append(f"kubeadm={requested_version}")
        setup_k8s_distro["deb_packages"].remove("kubelet")
        setup_k8s_distro["deb_packages"].append(f"kubelet={requested_version}")

        setup_localhost["fedora_packages"].remove("kubectl")
        setup_localhost["fedora_packages"].append(f"kubectl-{requested_version}")
        setup_localhost["fedora_packages_held"].remove("kubectl")
        setup_localhost["fedora_packages_held"].append(f"kubectl-{requested_version}")
        setup_k8s_distro["fedora_packages"].remove("kubectl")
        setup_k8s_distro["fedora_packages"].append(f"kubectl-{requested_version}")
        setup_k8s_distro["fedora_packages"].remove("kubeadm")
        setup_k8s_distro["fedora_packages"].append(f"kubeadm-{requested_version}")
        setup_k8s_distro["fedora_packages"].remove("kubelet")
        setup_k8s_distro["fedora_packages"].append(f"kubelet-{requested_version}")
        setup_k8s_distro["fedora_packages_held"].remove("kubectl")
        setup_k8s_distro["fedora_packages_held"].append(f"kubectl-{requested_version}")
        setup_k8s_distro["fedora_packages_held"].remove("kubeadm")
        setup_k8s_distro["fedora_packages_held"].append(f"kubeadm-{requested_version}")
        setup_k8s_distro["fedora_packages_held"].remove("kubelet")
        setup_k8s_distro["fedora_packages_held"].append(f"kubelet-{requested_version}")

        setup_localhost["suse_packages"].remove("kubectl")
        setup_localhost["suse_packages"].append(f"kubernetes{requested_major_minor}-client")
        setup_localhost["suse_packages_held"].remove("kubectl")
        setup_localhost["suse_packages_held"].append(f"kubernetes{requested_major_minor}-client")
        setup_k8s_distro["suse_packages"].remove("kubectl")
        setup_k8s_distro["suse_packages"].append(f"kubernetes{requested_major_minor}-client")
        setup_k8s_distro["suse_packages"].remove("kubeadm")
        setup_k8s_distro["suse_packages"].append(f"kubernetes{requested_major_minor}-kubeadm")
        setup_k8s_distro["suse_packages"].remove("kubelet")
        setup_k8s_distro["suse_packages"].append(f"kubernetes{requested_major_minor}-kubelet")
        setup_k8s_distro["suse_packages_held"].remove("kubectl")
        setup_k8s_distro["suse_packages_held"].append(f"kubernetes{requested_major_minor}-client")
        setup_k8s_distro["suse_packages_held"].remove("kubeadm")
        setup_k8s_distro["suse_packages_held"].append(f"kubernetes{requested_major_minor}-kubeadm")
        setup_k8s_distro["suse_packages_held"].remove("kubelet")
        setup_k8s_distro["suse_packages_held"].append(f"kubernetes{requested_major_minor}-kubelet")

    setup_k8s_distro["extra_values"] = setup_extras
    run_tasks(tasks=setup_control_plane_tasks, phase=cast(int, phase),
              phase_skiplist=phase_skiplist, final_state="installed", verbose=verbose)
    update_installation_info(version=requested_version)

    ansithemeprint([ANSIThemeStr("\nControl plane setup successful", "success")])

    if k8s_distro == "rke2":
        ansithemeprint([ANSIThemeStr("\nNote", "note"),
                        ANSIThemeStr(": If this is an RKE2 installation you may need "
                                     "to substitute ", "default"),
                        ANSIThemeStr("127.0.0.1", "hostname")])
        ansithemeprint([ANSIThemeStr("in ", "default"),
                        ANSIThemeStr(f"{KUBE_CONFIG_FILE}", "path"),
                        ANSIThemeStr(" with the real IP-address of this host.\n", "default")])


def __get_node_info() -> list[dict]:
    global kh  # pylint: disable=global-statement

    if not Path(KUBE_CONFIG_FILE).is_file():
        ansithemeprint([ANSIThemeStr("Critical", "critical"),
                        ANSIThemeStr(": Could not find ", "default"),
                        ANSIThemeStr(f"{KUBE_CONFIG_FILE}", "path"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    if kh is None:
        kh = kubernetes_helper.KubernetesHelper(about.PROGRAM_SUITE_NAME,
                                                about.PROGRAM_SUITE_VERSION, None)

    # If kh is *still* None here something has gone wrong. It is time to exit.
    if kh is None:
        ansithemeprint([ANSIThemeStr("Critical", "critical"),
                        ANSIThemeStr(": Failed to initialise connection to API-server; "
                                     "do you have a running cluster? Aborting.",
                                     "default")], stderr=True)
        sys.exit(errno.ENXIO)

    node_statuses = []

    vlist, status = kh.get_list_by_kind_namespace(("Node", ""), "")
    if status != 200:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": API-server returned ", "default"),
                        ANSIThemeStr(f"{status}", "errorvalue"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)
    if vlist is None:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": API-server did not return any data",
                                     "default")], stderr=True)
        sys.exit(errno.EINVAL)

    for node in vlist:
        node_name = deep_get(node, DictPath("metadata#name"))
        node_schedulable = not deep_get(node, DictPath("spec#unschedulable"), False)
        node_roles = get_node_roles(cast(dict, node))
        node_taints = deep_get(node, DictPath("spec#taints"), [])
        node_statuses.append({
            "name": node_name,
            "schedulable": node_schedulable,
            "roles": node_roles,
            "taints": node_taints,
        })

    return node_statuses


def get_control_planes(fail_on_empty: bool = True) -> list[tuple[str, str]]:
    """
    Get a list of control planes defined in the inventory.

        Parameters:
            fail_on_empty (bool): If True the action will fail
                                  if no control planes are defined in the inventory
        Returns:
            ([(str, str)]): A list of control planes
                (str): The hostname
                (str): The IP-address
    """
    installation_info = get_installation_info()
    cluster_name = installation_info["installation_target"]
    if cluster_name is None:
        cluster_name = get_cluster_name()
    inventory = ansible_get_inventory_dict()
    controlplanes_ = deep_get(inventory, DictPath("controlplane#hosts"), {})
    clusterhosts_ = deep_get(inventory, DictPath(f"{cluster_name}#hosts"), {})
    controlplanes = []
    if not controlplanes_ and fail_on_empty:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": No control plane(s) defined in the inventory; "
                                     "rebuilding the inventory using “", "default"),
                        ANSIThemeStr(f"{about.INVENTORY_PROGRAM_NAME} ", "programname"),
                        ANSIThemeStr("rebuild-inventory", "command"),
                        ANSIThemeStr("“ might help. Aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)
    if not clusterhosts_ and fail_on_empty:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": The cluster ", "default"),
                        ANSIThemeStr(f"{cluster_name}", "hostname"),
                        ANSIThemeStr(" has no hosts in the inventory; "
                                     "rebuilding the inventory using “", "default"),
                        ANSIThemeStr(f"{about.INVENTORY_PROGRAM_NAME} ", "programname"),
                        ANSIThemeStr("rebuild-inventory", "command"),
                        ANSIThemeStr("“ might help. Aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    for controlplane in controlplanes_:
        # Only include control planes belonging to this cluster
        if controlplane not in clusterhosts_:
            continue
        try:
            ip = socket.gethostbyname(controlplane)
            controlplanes.append((controlplane, ip))
        except socket.gaierror as e:
            if str(e) == "[Errno -5] No address associated with hostname":
                ansithemeprint([ANSIThemeStr("Error", "error"),
                                ANSIThemeStr(": No address associated with hostname “", "default"),
                                ANSIThemeStr(f"{controlplane}", "hostname"),
                                ANSIThemeStr("“. Aborting.", "default")], stderr=True)
                sys.exit(errno.ENOENT)
    return controlplanes


# pylint: disable-next=unused-argument,too-many-branches,too-many-statements
def teardown_control_plane(options: list[tuple[str, str]], args: list[str]) -> None:
    """
    Teardown an existing control plane.

        Parameters:
            options ([(str, str)]): Options to use when executing this action
            args ([str]): Arguments to use when executing this action
    """
    confirm: bool = True

    # We do not need a cluster name for --list-tasks, so check this first of all
    if "--list-tasks" in (tmp[0] for tmp in options):
        __list_phases(teardown_control_plane_tasks)
        sys.exit(0)

    if not Path(CMT_INSTALLATION_INFO_FILE).is_file():
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": Could not find reliable installation information. "
                                     "Aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)

    installation_info = get_installation_info()
    cluster_name = deep_get(installation_info, DictPath("installation_target"))
    k8s_distro = deep_get(installation_info, DictPath(f"{cluster_name}#distro"))
    state = deep_get(installation_info, DictPath(f"{cluster_name}#state"))
    phase: Union[str, int] = deep_get(installation_info, DictPath(f"{cluster_name}#phase"))
    phase_skiplist = set(deep_get(installation_info, DictPath(f"{cluster_name}#phase_skiplist")))
    verbose = False

    if k8s_distro is None or k8s_distro == "<none>":
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": Could not find reliable installation information. "
                                     "Aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)

    if state is None:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": Cannot determine installation state; aborting.",
                                     "default")], stderr=True)
        sys.exit(errno.EINVAL)
    elif state in ("preparing", "prepared", "torn_down"):
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": The system has no installed cluster; try ", "default"),
                        ANSIThemeStr(f"{about.ADMIN_PROGRAM_NAME}", "programname"),
                        ANSIThemeStr(" purge-control-plane", "command"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    # If phase is "<done>" it's a holdover from a previous state
    if phase == "<done>":
        phase = 0

    for opt, optarg in options:
        if opt == "--start-at-task":
            phase = __validate_task_index(teardown_control_plane_tasks, optarg)
            if "--skip-tasks" not in options:
                phase_skiplist = set()
        elif opt == "--skip-tasks":
            skip_phases = __expand_index_list(optarg)
            for __phase in skip_phases:
                __validate_task_index(teardown_control_plane_tasks, __phase)
            phase_skiplist = set(skip_phases)
        elif opt == "--resume":
            if phase is None or phase == "<none>":
                phase = 0
            elif phase == "<done>":
                ansithemeprint([ANSIThemeStr("Warning", "warning"),
                                ANSIThemeStr(": Ignoring request to resume a completed teardown. "
                                             "Exiting.", "default")], stderr=True)
                sys.exit(0)
            phase = __validate_task_index(teardown_control_plane_tasks, phase)
        elif opt == "--save-ansible-logs":
            ansible_configuration["save_logs"] = True
        elif opt == "--verbose":
            verbose = True
        elif opt == "-Y":
            confirm = False

    installation_info = \
        update_installation_info(state="tearing_down", phase=phase,
                                 phase_skiplist=list(phase_skiplist))
    show_configuration(action="Tearing down control plane", tasks=teardown_control_plane_tasks)

    if confirm:
        input_retval = \
            ansithemeinput([ANSIThemeStr("\nStart teardown? [y/", "default"),
                            ANSIThemeStr("N", "emphasis"),
                            ANSIThemeStr("]: ", "default")])
        if input_retval.lower() not in ("y", "yes"):
            ansithemeprint([ANSIThemeStr("\nAborting:", "error"),
                            ANSIThemeStr(" User stopped teardown.", "default")], stderr=True)
            sys.exit(errno.EINTR)

    ansithemeprint([ANSIThemeStr("\n[Tearing down cluster]", "phase")])

    installation_info = update_installation_info(state="tearing_down", phase=0)

    run_tasks(tasks=teardown_control_plane_tasks, phase=cast(int, phase),
              phase_skiplist=phase_skiplist, final_state="torn_down", verbose=verbose)
    ansithemeprint([ANSIThemeStr("\nCluster teardown successful", "success")])


# pylint: disable-next=unused-argument,too-many-branches,too-many-statements
def purge_control_plane(options: list[tuple[str, str]], args: list[str]) -> None:
    """
    Purge an existing control plane.

        Parameters:
            options ([(str, str)]): Options to use when executing this action
            args (list[str]): Arguments to use when executing this action
    """
    confirm = True

    # We do not need a cluster name for --list-tasks, so check this first of all
    if "--list-tasks" in (tmp[0] for tmp in options):
        __list_phases(purge_control_plane_tasks)
        sys.exit(0)

    if not Path(CMT_INSTALLATION_INFO_FILE).is_file():
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": Could not find reliable installation information. "
                                     "Aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    installation_info = get_installation_info()
    cluster_name = installation_info["installation_target"]
    state = installation_info[cluster_name]["state"]
    phase: Union[str, int] = installation_info[cluster_name]["phase"]
    phase_skiplist = set(installation_info[cluster_name]["phase_skiplist"])
    verbose = False

    if state is None or state not in ("prepared", "torn_down", "purging"):
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": Invalid installation state; there's no prepared or torn "
                                     "down cluster; aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)

    # If phase is "<done>" it's a holdover from a previous state;
    # if this is the case we need to start from phase 0 and ignore the skiplist
    if not options or phase == "<done>":
        phase = 0
        phase_skiplist = set()

    for opt, optarg in options:
        if opt == "--start-at-task":
            phase = __validate_task_index(purge_control_plane_tasks, optarg)
            if "--skip-tasks" not in options:
                phase_skiplist = set()
        elif opt == "--skip-tasks":
            skip_phases = __expand_index_list(optarg)
            for __phase in skip_phases:
                __validate_task_index(purge_control_plane_tasks, __phase)
            phase_skiplist = set(skip_phases)
        elif opt == "--resume":
            if phase is None or phase == "<none>":
                phase = 0
            elif phase == "<done>":
                ansithemeprint([ANSIThemeStr("Warning", "warning"),
                                ANSIThemeStr(": Ignoring request to resume a completed purge. "
                                             "Exiting.", "default")], stderr=True)
                sys.exit(0)
            phase = __validate_task_index(purge_control_plane_tasks, phase)
        elif opt == "--verbose":
            verbose = True
        elif opt == "--save-ansible-logs":
            ansible_configuration["save_logs"] = True
        elif opt == "-Y":
            confirm = False

    installation_info = \
        update_installation_info(state="purging", phase=phase, phase_skiplist=list(phase_skiplist))
    show_configuration(action="Purging control plane:", tasks=purge_control_plane_tasks)

    if confirm:
        input_retval = \
            ansithemeinput([ANSIThemeStr("\nStart purge? [y/", "default"),
                            ANSIThemeStr("N", "emphasis"),
                            ANSIThemeStr("]: ", "default")])
        if input_retval.lower() not in ("y", "yes"):
            ansithemeprint([ANSIThemeStr("\nAborting:", "error"),
                            ANSIThemeStr(" User stopped purge.", "default")], stderr=True)
            sys.exit(errno.EINTR)

    ansithemeprint([ANSIThemeStr("\n\n[Purging cluster configuration and software]", "phase")])

    # If we are purging the control plane we want to leave kubectl behind;
    # if we have a setup with multiple control planes and we run this from one of them,
    # then kubectl will remain on the other two.
    if __selection_localhost() == __selection_control_planes():
        purge_control_plane_targets["kubeadm"]["deb_packages"].remove("kubectl")
        purge_control_plane_targets["kubeadm"]["deb_packages_held"].remove("kubectl")

    installation_info = update_installation_info(state="purging", phase=0)

    run_tasks(tasks=purge_control_plane_tasks, phase=cast(int, phase),
              phase_skiplist=phase_skiplist, final_state="purged", verbose=verbose)
    secure_rm(CMT_INSTALLATION_INFO_FILE, ignore_non_existing=True)

    ansithemeprint([ANSIThemeStr("\nCluster purge successful", "success")])

    ansithemeprint([ANSIThemeStr("\nNote", "note"),
                    ANSIThemeStr(": It is recommended to reboot the control plane(s) after "
                                 "purging the cluster\n", "default")])


# pylint: disable-next=too-many-locals,too-many-branches,too-many-statements
def upgrade_control_plane(options: list[tuple[str, str]], args: list[str]) -> None:
    """
    Upgrade a control plane.

        Parameters:
            options ([(str, str)]): Options to use when executing this action
            args ([str]): Arguments to use when executing this action
    """
    global kh  # pylint: disable=global-statement

    confirm = True

    # default options
    update_cache = True
    allow_reinstall = False
    requested_version = None

    # We do not need a cluster name for --list-tasks or --override, so check this first of all
    if "--list-tasks" in (tmp[0] for tmp in options):
        __list_phases(upgrade_control_plane_tasks)
        sys.exit(0)
    elif "--override" in (tmp[0] for tmp in options):
        if Path(CMT_INSTALLATION_INFO_FILE).is_file():
            ansithemeprint([ANSIThemeStr("Warning", "warning"),
                            ANSIThemeStr(": Overriding ", "default"),
                            ANSIThemeStr(f"{CMT_INSTALLATION_INFO_FILE}", "path"),
                            ANSIThemeStr("; this may cause issues.", "default")], stderr=True)
        else:
            ansithemeprint([ANSIThemeStr("Note", "note"),
                            ANSIThemeStr(": ", "default"),
                            ANSIThemeStr(f"{CMT_INSTALLATION_INFO_FILE}", "path"),
                            ANSIThemeStr(" does not exist; rebuilding.", "default")])
        rebuild_installation_info(state="upgrading")

    # This is the cluster name according to .kube/config; this is what we care about when upgrading,
    # not whatever is currently set as the target in installation info
    cluster_name = get_cluster_name()
    installation_info = get_installation_info(ignore_non_existing=False)
    if installation_info is None:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": ", "default"),
                        ANSIThemeStr(f"{CMT_INSTALLATION_INFO_FILE}", "path"),
                        ANSIThemeStr(" does not exist; ", "default"),
                        ANSIThemeStr("most likely ", "default")], stderr=True)
        ansithemeprint([ANSIThemeStr("the cluster was not ", "default"),
                        ANSIThemeStr("installed using ", "default"),
                        ANSIThemeStr(f"{about.PROGRAM_SUITE_NAME}", "programname"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)
    installation_info = update_installation_info(installation_target=cluster_name)

    cluster_name = deep_get(installation_info, DictPath("installation_target"))
    os_distro = identify_distro()
    k8s_distro = deep_get(installation_info, DictPath(f"{cluster_name}#distro"))
    version = deep_get(installation_info, DictPath(f"{cluster_name}#version"))
    requested_version = deep_get(installation_info, DictPath(f"{cluster_name}#requested_version"))
    state = deep_get(installation_info, DictPath(f"{cluster_name}#state"))
    phase: Union[str, int] = deep_get(installation_info, DictPath(f"{cluster_name}#phase"))
    phase_skiplist = set(deep_get(installation_info, DictPath(f"{cluster_name}#phase_skiplist")))
    verbose = False
    ignore_feature_gates = False

    if state is None or state == "<none>":
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": Unknown installation state; if you believe this is OK "
                                     "(such as when upgrading a cluster not installed using ",
                                     "default"),
                        ANSIThemeStr(f"{about.ADMIN_PROGRAM_NAME}", "programname"),
                        ANSIThemeStr(" you can try using the “", "default"),
                        ANSIThemeStr("--override", "option"),
                        ANSIThemeStr("“ option; aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)

    if state in ("preparing", "prepared", "installing", "tearing_down", "torn_down", "purging"):
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": Invalid installation state; Kubernetes does not seems "
                                     "to be fully installed; aborting.", "default")], stderr=True)
        sys.exit(errno.EINVAL)

    if state in ("installed", "upgraded"):
        phase = 0

    for opt, optarg in options:
        if opt == "--start-at-task":
            phase = __validate_task_index(upgrade_control_plane_tasks, optarg)
            if "--skip-tasks" not in options:
                phase_skiplist = set()
        elif opt == "--skip-tasks":
            skip_phases = __expand_index_list(optarg)
            for __phase in skip_phases:
                __validate_task_index(upgrade_control_plane_tasks, __phase)
            phase_skiplist = set(skip_phases)
        elif opt == "--resume":
            if phase is None or phase == "<none>":
                phase = 0
            elif phase == "<done>":
                ansithemeprint([ANSIThemeStr("Warning", "warning"),
                                ANSIThemeStr(": Ignoring request to resume a completed upgrade. "
                                             "Exiting.", "default")], stderr=True)
                sys.exit(0)
            phase = __validate_task_index(upgrade_control_plane_tasks, phase)
        elif opt == "--no-cache-update":
            update_cache = False
        elif opt == "--reinstall":
            allow_reinstall = True
        elif opt in ("--override", "--list-tasks"):
            continue
        elif opt == "--verbose":
            verbose = True
        elif opt == "--save-ansible-logs":
            ansible_configuration["save_logs"] = True
        elif opt == "--ignore-feature-gates":
            ignore_feature_gates = True
        elif opt == "-Y":
            confirm = False

    if kh is None:
        kh = kubernetes_helper.KubernetesHelper(about.PROGRAM_SUITE_NAME,
                                                about.PROGRAM_SUITE_VERSION, None)
    if kh is not None:
        fg = kh.check_for_feature_gates()

        has_fg = []

        for config, feature_gates in fg.items():
            if feature_gates is not None and feature_gates:
                has_fg.append(config)
        if has_fg:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": the following configurations have ", "default"),
                            ANSIThemeStr("changes to feature gates:", "default")], stderr=True)
            for key in has_fg:
                ansithemeprint([ANSIThemeStr("• ", "separator"),
                                ANSIThemeStr(f"{key}", "default")], stderr=True)
            ansithemeprint([ANSIThemeStr("\nUpgrading is not supported when", "default"),
                            ANSIThemeStr(" feature gates have been altered.", "default")],
                           stderr=True)

            if not ignore_feature_gates:
                ansithemeprint([ANSIThemeStr("To override, pass “", "default"),
                                ANSIThemeStr("--ignore-feature-gates", "option"),
                                ANSIThemeStr("“. Aborting.", "default")], stderr=True)
                sys.exit(errno.ENOTSUP)
            ansithemeprint([ANSIThemeStr("Warning", "warning"),
                            ANSIThemeStr(": “", "default"),
                            ANSIThemeStr("--ignore-feature-gates", "option"),
                            ANSIThemeStr("“ passed; continuing.", "default")], stderr=True)

    # XXX: How do we check that nodes are drained? We can check whether they are cordoned
    node_status = __get_node_info()

    if node_status is None:
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": No ", "default"),
                        ANSIThemeStr(f"{HOMEDIR}/.kube/config", "path"),
                        ANSIThemeStr(" available; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    ansithemeprint([ANSIThemeStr("\n[Preparing upgrade]", "phase")])

    # The first minor Kubernetes version available from the new repository is 1.24.0;
    # let's add all minor versions from 24 and up; the old repository is no longer usable.
    if not (kubernetes_upstream_version := get_latest_upstream_version("kubernetes")):
        ansithemeprint([ANSIThemeStr("Error", "error"),
                        ANSIThemeStr(": Could not get the latest upstream Kubernetes version; ",
                                     "default"),
                        ANSIThemeStr("this is either a network error or a bug; aborting.",
                                     "default")], stderr=True)
        sys.exit(errno.ENOENT)

    # Split the version tuple
    upstream_major, upstream_minor, _rest = kubernetes_upstream_version.split(".")
    minor_versions = []
    for minor_version in range(24, int(upstream_minor) + 1):
        minor_versions.append(f"{minor_version}")

    if update_cache:
        ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                        ANSIThemeStr("Updating version cache", "action")])

        extra_values = {
            "minor_versions": minor_versions,
        }

        ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                        ANSIThemeStr("Updating Kubernetes repository files on localhost "
                                     "and controlplane(s)", "action")])
        selection = __selection_control_planes() + ["localhost"]
        _retval, _ansible_results = \
            run_playbook(get_playbook_path(FilePath("add_kubernetes_repo.yaml")),
                         hosts=selection, extra_values=extra_values, quiet=True)

        if os_distro == "debian" and update_cache:
            ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                            ANSIThemeStr("Updating APT cache", "action")])
            check_and_print_status(update_apt_cache())
        ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                        ANSIThemeStr("Updating version cache", "action")])
        update_version_cache()

    if args:
        requested_version = __find_requested_version(k8s_distro, args[0])
    else:
        requested_version = \
            __find_requested_version(k8s_distro, f"{upstream_major}.{upstream_minor}")

    ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                    ANSIThemeStr("Running sanity checks", "action")])

    pkg_versions: list[tuple[str, str, str, list[str]]] = []

    if k8s_distro == "kubeadm":
        if os_distro in ("debian",):
            pkg_versions = check_versions_apt(["kubeadm"])
        elif os_distro in ("suse",):
            pkg_versions = check_versions_zypper(["kubeadm"])
        elif os_distro in ("fedora", "rhel",):
            pkg_versions = check_versions_yum(["kubeadm"])

        if not pkg_versions:
            ansithemeprint([ANSIThemeStr("Critical", "critical"),
                            ANSIThemeStr(": No candidate version for ", "default"),
                            ANSIThemeStr("kubeadm", "programname"),
                            ANSIThemeStr(" available; aborting.", "default")], stderr=True)
            sys.exit(errno.ENOENT)

        if requested_version is not None:
            if requested_version not in pkg_versions[0][3]:
                ansithemeprint([ANSIThemeStr("Error", "error"),
                                ANSIThemeStr(": The requested version ", "default"),
                                ANSIThemeStr(f"{requested_version}", "version"),
                                ANSIThemeStr(" is not available; aborting.", "default")],
                               stderr=True)
                sys.exit(errno.ENOENT)
            elif requested_version == version:
                if not allow_reinstall:
                    ansithemeprint([ANSIThemeStr("Warning", "warning"),
                                    ANSIThemeStr(": The requested version ", "default"),
                                    ANSIThemeStr(f"{requested_version}", "version"),
                                    ANSIThemeStr(" is already installed; "
                                                 "to reinstall use the option “", "default"),
                                    ANSIThemeStr("--reinstall", "option"),
                                    ANSIThemeStr("“.", "default")], stderr=True)
                    sys.exit(errno.EINVAL)
            elif os_distro == "debian" and deb_compare_versions(version, requested_version):
                ansithemeprint([ANSIThemeStr("Error", "error"),
                                ANSIThemeStr(": The requested version ", "default"),
                                ANSIThemeStr(f"{requested_version}", "version"),
                                ANSIThemeStr(" is older than than the installed version ",
                                             "default"),
                                ANSIThemeStr(f"{pkg_versions[0][1]}", "version"),
                                ANSIThemeStr(". Downgrades are not supported; aborting.",
                                             "default")], stderr=True)
                sys.exit(errno.EINVAL)
            # FIXME: Implement version checking for Fedora/RHEL packages
        else:
            requested_version = pkg_versions[0][2]
            if not requested_version or requested_version == "<none>":
                ansithemeprint([ANSIThemeStr("Note", "note"),
                                ANSIThemeStr(": The latest version ", "default"),
                                ANSIThemeStr(f"{pkg_versions[0][1]}", "version"),
                                ANSIThemeStr(" is already installed.", "default")])
                sys.exit(0)

        # Is this an major, minor, or patchrev upgrade?
        installed_version_tuple = version.split(".")
        requested_version_tuple = requested_version.split(".")

        upgrade_type = None

        if installed_version_tuple[0] != requested_version_tuple[0]:
            ansithemeprint([ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": Upgrades between ", "default"),
                            ANSIThemeStr("MAJOR", "emphasis"),
                            ANSIThemeStr(" versions is currently not supported "
                                         "(installed version: ", "default"),
                            ANSIThemeStr(f"{pkg_versions[0][1]}", "version"),
                            ANSIThemeStr(", requested version: ", "default"),
                            ANSIThemeStr(f"{requested_version}", "version"),
                            ANSIThemeStr("; aborting.", "default")], stderr=True)
            sys.exit(errno.ENOTSUP)
        elif installed_version_tuple[1] != requested_version_tuple[1]:
            if int(requested_version_tuple[1]) - int(installed_version_tuple[1]) > 1:
                ansithemeprint([ANSIThemeStr("Error", "error"),
                                ANSIThemeStr(": Skipping ", "default"),
                                ANSIThemeStr("MINOR", "emphasis"),
                                ANSIThemeStr(" versions is not supported, please perform the "
                                             "following upgrades sequentially:",
                                             "default")], stderr=True)
                minor_versions_dict: dict = {}
                for version in reversed(pkg_versions[0][3]):
                    version_tuple = version.split(".")
                    major_minor = f"{version_tuple[0]}.{version_tuple[1]}"
                    if (int(version_tuple[1]) > int(installed_version_tuple[1])
                            and int(version_tuple[1]) <= int(requested_version_tuple[1])
                            and (major_minor not in minor_versions_dict
                                 or minor_versions_dict[major_minor] < version_tuple[2])):
                        minor_versions_dict[major_minor] = version_tuple[2].split("-")[0]
                for version, minor_version in minor_versions_dict.items():
                    ansithemeprint([ANSIThemeStr("• ", "separator"),
                                    ANSIThemeStr(f"{about.ADMIN_PROGRAM_NAME}", "programname"),
                                    ANSIThemeStr(" upgrade-control-plane", "command"),
                                    ANSIThemeStr(f" {version}.{minor_version}", "version")])
                sys.exit(errno.EINVAL)
            else:
                upgrade_type = "minor"
        else:
            upgrade_type = "patchrev"

        # If this is an upgrade to a new patchrev the cluster does not have to be drained
        if upgrade_type == "minor":
            schedulable_count = 0
            nodes_to_drain = []
            for node in node_status:
                if not node["schedulable"]:
                    continue

                if "control-plane" in node["roles"]:
                    continue

                if schedulable_count == 0:
                    ansithemeprint([ANSIThemeStr("\nError", "error"),
                                    ANSIThemeStr(": The following nodes need to be drained; "
                                                 "aborting.", "default")])
                schedulable_count += 1
                ansithemeprint([ANSIThemeStr("• ", "separator"),
                                ANSIThemeStr(f"{node['name']}", "emphasis")])
                nodes_to_drain.append(node['name'])

            if schedulable_count:
                ansithemeprint([ANSIThemeStr("\nThis can be achieved with: “", "default"),
                                ANSIThemeStr("kubectl ", "programname"),
                                ANSIThemeStr("drain ", "command"),
                                ANSIThemeStr("--ignore-daemonsets --delete-emptydir-data ",
                                             "option"),
                                ANSIThemeStr(f"{' '.join(nodes_to_drain)}", "url"),
                                ANSIThemeStr("“", "default")])
                sys.exit(errno.EBUSY)

        # First remove the distro revision
        major_minor_patchrev = requested_version.split("-")
        # Now split the version tuple
        requested_major, requested_minor, _rest = major_minor_patchrev[0].split(".")

        if int(requested_minor) < 31:
            cluster_configuration_api_version = "kubeadm.k8s.io/v1beta3"
            init_configuration_api_version = "kubeadm.k8s.io/v1beta3"
            upgrade_configuration_api_version = "<unused>"
        else:
            cluster_configuration_api_version = "kubeadm.k8s.io/v1beta4"
            init_configuration_api_version = "kubeadm.k8s.io/v1beta4"
            upgrade_configuration_api_version = "kubeadm.k8s.io/v1beta4"

        upgrade_control_plane_targets[k8s_distro]["extra_values"] = {
            "cluster_configuration_api_version": cluster_configuration_api_version,
            "init_configuration_api_version": init_configuration_api_version,
            "upgrade_configuration_api_version": upgrade_configuration_api_version,
            # Add major/minor to extra_values
            "kubernetes_major_minor_version": f"{requested_major}.{requested_minor}",
        }

        # Adjust package version for localhost
        localhost_targets = upgrade_control_plane_targets["localhost"]
        localhost_targets["deb_packages"].remove("kubectl")
        localhost_targets["deb_packages"].append(f"kubectl={requested_version}")

        localhost_targets["fedora_packages"].remove("kubectl")
        localhost_targets["fedora_packages"].append(f"kubectl-{requested_version}")
    elif k8s_distro == "rke2":
        # Add requested_version to extra_values;
        # for rke2 this is either vX.YY or latest
        upgrade_control_plane_targets["rke2"]["extra_values"]["requested_version"] = \
            requested_version

    installation_info = \
        update_installation_info(state="upgrading", requested_version=requested_version, phase=0)

    # If we got here all the sanity checks were successful
    ansithemeprint([ANSIThemeStr("OK", "ok")])

    show_configuration(action="Upgrading control plane:", tasks=upgrade_control_plane_tasks)

    if confirm:
        input_retval = \
            ansithemeinput([ANSIThemeStr("\nStart upgrade? [y/", "default"),
                            ANSIThemeStr("N", "emphasis"),
                            ANSIThemeStr("]: ", "default")])
        if input_retval.lower() not in ("y", "yes"):
            ansithemeprint([ANSIThemeStr("\nAborting:", "error"),
                            ANSIThemeStr(" User stopped upgrade.", "default")], stderr=True)
            sys.exit(errno.EINTR)

    ansithemeprint([ANSIThemeStr("\n[Upgrading control plane]", "phase")])

    run_tasks(tasks=upgrade_control_plane_tasks, phase=cast(int, phase),
              phase_skiplist=phase_skiplist, final_state="upgraded", verbose=verbose)
    values = {
        "control_plane_k8s_version": requested_version,
    }
    ansible_set_vars(ANSIBLE_INVENTORY, "all", values)
    update_installation_info(version=requested_version)

    ansithemeprint([ANSIThemeStr("\nControl plane upgrade successful", "success")])


# pylint: disable-next=unused-argument,too-many-branches
def taint_control_plane(options: list[tuple[str, str]], args: list[str]) -> None:
    """
    Taint a control plane.

        Parameters:
            options ([(str, str)]): Options to use when executing this action
            args ([str]): Arguments to use when executing this action
    """
    node_statuses = __get_node_info()

    if node_statuses is None:
        ansithemeprint([ANSIThemeStr("Critical", "critical"),
                        ANSIThemeStr(": Failed to get node information; do you have a running "
                                     "cluster? Aborting.", "default")], stderr=True)
        sys.exit(errno.ENXIO)

    first = True

    for node in node_statuses:
        if "control-plane" in node["roles"] and (not args or node["name"] in args):
            if first:
                ansithemeprint([ANSIThemeStr("Tainting control plane(s):", "header")])
                first = False
            ansithemeprint([ANSIThemeStr(node["name"], "hostname")])
            _message, status = \
                kh.taint_node(node=node["name"],
                              taints=node["taints"],
                              new_taint=("node-role.kubernetes.io/control-plane",
                                         None, None, "NoSchedule"))
            if status == 304:
                ansithemeprint([ANSIThemeStr("  Not modified", "none")])
            elif status == 200:
                ansithemeprint([ANSIThemeStr("  Tainted", "success")])
            else:
                ansithemeprint([ANSIThemeStr("  Failed to modify taint", "error"),
                                ANSIThemeStr(f"; HTTP error {status}; aborting.",
                                             "default")], stderr=True)
                sys.exit(errno.EINVAL)

        if "master" in node["roles"] and (not args or node["name"] in args):
            if first:
                ansithemeprint([ANSIThemeStr("Tainting control plane(s):", "header")])
                first = False
            ansithemeprint([ANSIThemeStr(node["name"], "hostname")])
            _message, status = \
                kh.taint_node(node=node["name"],
                              taints=node["taints"],
                              new_taint=("node-role.kubernetes.io/master",
                                         None, None, "NoSchedule"))
            if status == 304:
                ansithemeprint([ANSIThemeStr("  Not modified", "none")])
            elif status == 200:
                ansithemeprint([ANSIThemeStr("  Tainted", "success")])
            else:
                ansithemeprint([ANSIThemeStr("  Failed to modify taint", "error"),
                                ANSIThemeStr(f"; HTTP error {status}; aborting.",
                                             "default")], stderr=True)
                sys.exit(errno.EINVAL)

    if first:
        ansithemeprint([ANSIThemeStr("Warning", "warning"),
                        ANSIThemeStr(": No matching control planes found. Aborting.",
                                     "default")], stderr=True)
        sys.exit(errno.ENOENT)


# pylint: disable-next=unused-argument,too-many-branches
def untaint_control_plane(options: list[tuple[str, str]], args: list[str]) -> None:
    """
    Untaint a control plane.

        Parameters:
            options ([(str, str)]): Options to use when executing this action
            args ([str]): Arguments to use when executing this action
    """
    node_statuses = __get_node_info()

    if node_statuses is None:
        ansithemeprint([ANSIThemeStr("Critical", "critical"),
                        ANSIThemeStr(": Failed to get node information; do you have a running "
                                     "cluster? Aborting.", "default")], stderr=True)
        sys.exit(errno.ENXIO)

    first = True

    for node in node_statuses:
        if "control-plane" in node["roles"] and (not args or node["name"] in args):
            if first:
                ansithemeprint([ANSIThemeStr("Untainting control plane(s):", "header")])
                first = False
            ansithemeprint([ANSIThemeStr(node["name"], "hostname")])
            _message, status = \
                kh.taint_node(node=node["name"],
                              taints=node["taints"],
                              new_taint=("node-role.kubernetes.io/control-plane",
                                         None, None, None))
            if status == 304:
                ansithemeprint([ANSIThemeStr("  Not modified", "none")])
            elif status == 200:
                ansithemeprint([ANSIThemeStr("  Untainted", "success")])
            else:
                ansithemeprint([ANSIThemeStr("  Failed to modify taint", "error"),
                                ANSIThemeStr(f"; HTTP error {status}; aborting.",
                                             "default")], stderr=True)
                sys.exit(errno.EINVAL)

        if "master" in node["roles"] and (not args or node["name"] in args):
            if first:
                ansithemeprint([ANSIThemeStr("Untainting control plane(s):", "header")])
                first = False
            ansithemeprint([ANSIThemeStr(node["name"], "hostname")])
            _message, status = \
                kh.taint_node(node=node["name"],
                              taints=node["taints"],
                              new_taint=("node-role.kubernetes.io/master",
                                         None, None, None))
            if status == 304:
                ansithemeprint([ANSIThemeStr("  Not modified", "none")])
            elif status == 200:
                ansithemeprint([ANSIThemeStr("  Untainted", "success")])
            else:
                ansithemeprint([ANSIThemeStr("  Failed to modify taint", "error"),
                                ANSIThemeStr(f"; HTTP error {status}; aborting.",
                                             "default")], stderr=True)
                sys.exit(errno.EINVAL)

    if first:
        ansithemeprint([ANSIThemeStr("Warning", "warning"),
                        ANSIThemeStr(": No matching control planes found. Aborting.",
                                     "default")], stderr=True)
        sys.exit(errno.ENOENT)


# pylint: disable=redefined-outer-name
def run_checks(checks: list[dict], **kwargs: Any) -> bool:
    """
    Run a batch of checks, and output the result.
    The checks can return 4 different severities; critical, error, warning, and note.

        Parameters:
            checks ([dict]): A list with all checks to run
        Return:
            (bool): True if nothing higher than "note" was returned, False otherwise
    """
    abort: bool = False
    critical: int = 0
    error: int = 0
    warning: int = 0
    note: int = 0
    skip: int = 0

    user = ""
    cluster_name = ""
    kubeconfig: dict = {}

    for check in checks:
        preconditions = deep_get(check, DictPath("preconditions"), [])
        if "user" in preconditions and len(user) == 0:
            user = getuser()
        if "clusterinfo" in preconditions and len(kubeconfig) == 0:
            if not Path(KUBE_CONFIG_DIR).is_dir():
                ansithemeprint([ANSIThemeStr(deep_get(check,
                                                      DictPath("description"), ""), "phase")])
                ansithemeprint([ANSIThemeStr("  Warning:", "warning"),
                                ANSIThemeStr(f" {KUBE_CONFIG_DIR} ", "path"),
                                ANSIThemeStr("does not exist; most likely there's no cluster; "
                                             "skipping.\n", "default")], stderr=True)
                skip += 1
                continue
            if not Path(KUBE_CONFIG_FILE).is_file():
                ansithemeprint([ANSIThemeStr(deep_get(check,
                                             DictPath("description"), ""), "phase")])
                ansithemeprint([ANSIThemeStr("  Warning:", "warning"),
                                ANSIThemeStr(f" {KUBE_CONFIG_FILE} ", "path"),
                                ANSIThemeStr("does not exist; most likely there's no cluster; "
                                             "skipping.\n", "default")], stderr=True)
                skip += 1
                continue
            kubeconfig = secure_read_yaml(KUBE_CONFIG_FILE)
            cluster_name = cast(str, get_cluster_name())

        call = deep_get(check, DictPath("call"))
        if call is not None:
            if (kh := deep_get(kwargs, DictPath("kubernetes_helper"))) is None:
                # This will only work for running clusters
                kh = kubernetes_helper.KubernetesHelper(about.PROGRAM_SUITE_NAME,
                                                        about.PROGRAM_SUITE_VERSION, None)
                kwargs["kubernetes_helper"] = kh
            abort, critical, error, warning, note = \
                call(cluster_name=cluster_name,
                     kubeconfig=kubeconfig,
                     cmtconfig_dict=cmtlib.cmtconfig,
                     user=user,
                     critical=critical,
                     error=error,
                     warning=warning,
                     note=note,
                     **kwargs)
            if abort:
                break

    ansithemeprint([ANSIThemeStr("Summary:", "header")])
    ansithemeprint([ANSIThemeStr("Critical", "critical"),
                    ANSIThemeStr(f": {critical}", "error")])
    ansithemeprint([ANSIThemeStr(f"  Errors: {error}", "error")])
    ansithemeprint([ANSIThemeStr(f"Warnings: {warning}", "warning")])
    ansithemeprint([ANSIThemeStr(f" Skipped: {skip}", "emphasis")])
    ansithemeprint([ANSIThemeStr(f"   Notes: {note}", "note")])
    if abort:
        ansithemeprint([ANSIThemeStr("Testing aborted due to critical error.", "error")])

    return critical + error + warning == 0


security_audit_checks: list[dict[str, Any]] = [
    {
        "description": "Check whether strict host key checking has been disabled",
        "call": checks.check_disable_strict_host_key_checking,
    },
    {
        "description": "Check for insecure kubeconfig options",
        "call": checks.check_insecure_kube_config_options,
        "preconditions": [
            "clusterinfo",
        ],
    }, {
        "description": "Check for insecure file permissions",
        "call": checks.check_file_permissions,
        "preconditions": [
            "user",
        ],
    },
]


# pylint: disable-next=unused-argument,too-many-locals
def audit(options: list[tuple[str, str]], args: list[str]) -> None:
    """
    Run security audit checks.

        Parameters:
            options (list[(opt, optarg)]): Options to use when executing this action
            args (list[str]): Arguments to use when executing this action
    """
    usergroup: str = ""
    usergroup_autodetect: bool = True

    for opt, optarg in options:
        if opt == "--usergroup":
            usergroup = optarg
        elif opt == "--disable-usergroup-autodetect":
            usergroup_autodetect = False

    # Get the username and group for the user;
    # if they match it's highly likely that the system
    # is configured to use usergroups
    uid = os.getuid()
    username = pwd.getpwuid(uid).pw_name
    gid = pwd.getpwuid(uid).pw_gid
    groupname = grp.getgrgid(gid).gr_name

    if len(usergroup) == 0 and usergroup_autodetect:
        logindefs = secure_read_string(FilePath("/etc/login.defs"))
        usergroups_regex: re.Pattern[str] = re.compile(r"^USERGROUPS_ENAB\s+yes")
        for line in logindefs.splitlines():
            tmp = usergroups_regex.match(line)
            if tmp is not None:
                usergroup = groupname

    kwargs: dict[str, str] = {
        "usergroup": usergroup
    }

    clean_run = run_checks(security_audit_checks, **kwargs)

    if clean_run:
        ansithemeprint([ANSIThemeStr("\nImportant", "emphasis"),
                        ANSIThemeStr(":", "default")])
        ansithemeprint([ANSIThemeStr(f"  {about.ADMIN_PROGRAM_NAME}", "programname"),
                        ANSIThemeStr(" audit", "command"),
                        ANSIThemeStr(" currently only checks for a very limited set of issues;",
                                     "default")])
        ansithemeprint([ANSIThemeStr("  a perfect score is not a guarantee that your "
                                     "installation is secure.", "default")])
    elif len(usergroup) == 0:
        if username == groupname:
            ansithemeprint([ANSIThemeStr("\nNote", "emphasis"),
                            ANSIThemeStr(":", "default"),
                            ANSIThemeStr(" the username for the current user “", "default"),
                            ANSIThemeStr(f"{username}", "argument"),
                            ANSIThemeStr("“ is the same", "default")])
            ansithemeprint([ANSIThemeStr("as its primary group name. This may be an indication ",
                                         "default"),
                            ANSIThemeStr("that the system", "default")])
            ansithemeprint([ANSIThemeStr("is configured to use usergroups; if you're seeing "
                                         "warnings related", "default")])
            ansithemeprint([ANSIThemeStr("to file permissions this ", "default"),
                            ANSIThemeStr("might", "emphasis"),
                            ANSIThemeStr(" be the cause; if you know (or think) that", "default")])
            ansithemeprint([ANSIThemeStr("the system is configured with usergroups you "
                                         "can try:\n", "default")])
            ansithemeprint([ANSIThemeStr(f"  {about.ADMIN_PROGRAM_NAME} ", "programname"),
                            ANSIThemeStr("audit ", "command"),
                            ANSIThemeStr("--usergroup ", "option"),
                            ANSIThemeStr(f"{username}", "argument")])
            ansithemeprint([ANSIThemeStr("\nto see if that resolves the warnings.", "default")])


preflight_checks: list[dict[str, Any]] = [
    {
        "description": "check whether ownership for .ansible is correct",
        "call": checks.check_ansible_dir_permissions,
        "preconditions": [
            "user",
        ],
    }, {
        "description": "check whether permissions for .netrc are strict enough",
        "call": checks.check_netrc_permissions,
        "preconditions": [
            "user",
        ],
    }, {
        "description": "Check whether the user can sudo without a password on localhost",
        "call": checks.check_sudo_configuration,
        "preconditions": [
            "user",
        ],
    }, {
        "description": "Check whether SSH known_hosts hashing is enabled on localhost",
        "call": checks.check_known_hosts_hashing,
    }, {
        "description": "Check control plane for suitability",
        "call": checks.check_control_plane,
    }
]


def preflight_check(options: list[tuple[str, str]], args: list[str]) -> None:
    """
    Run preflight checks before creating a new cluster.

        Parameters:
            options (list[(opt, optarg)]): Options to use when executing this action
            args (list[str]): Arguments to use when executing this action
    """
    global no_password  # pylint: disable=global-statement

    for opt, _optarg in options:
        if opt == "--no-password":
            no_password = True

    control_planes = [args[0]]

    ansible_add_hosts(inventory=ANSIBLE_INVENTORY, hosts=control_planes, skip_all=False)

    __task_request_ansible_password(installation_info={})

    clean_run = run_checks(preflight_checks, hosts=control_planes)

    if clean_run:
        ansithemeprint([ANSIThemeStr("\nImportant", "emphasis"),
                        ANSIThemeStr(":", "default")])
        ansithemeprint([ANSIThemeStr(f"  {about.ADMIN_PROGRAM_NAME}", "programname"),
                        ANSIThemeStr(" preflight-check", "command"),
                        ANSIThemeStr(" currently only checks for a very limited set of issues;",
                                     "default")])
        ansithemeprint([ANSIThemeStr("  a perfect score is not a guarantee that installation "
                                     "will succeed.", "default")])


troubleshoot_checks: list[dict[str, Any]] = [
    # First start with the really basic: does the cluster respond?
    # Check no_proxy, perhaps?
    {
        "description": "Check whether client / server versions match",
        "call": checks.check_client_server_version_match,
        "preconditions": [
            "clusterinfo",
        ],
    }, {
        "description": "Check kubelet and kube-proxy versions",
        "call": checks.check_kubelet_and_kube_proxy_versions,
        "preconditions": [
            "clusterinfo",
        ],
    }, {
        "description": "Check required pods and their statuses",
        "call": checks.check_running_pods,
    },
    # Check kube-controller-manager, kube-scheduler, and cloud-controller-manager;
    # they should match, but can be up to one version older,
    # but must not be newer

    # check whether role bindings and cluster role bindings
    # refer to non-existing roles/cluster roles
]


# pylint: disable-next=unused-argument
def troubleshoot(options: list[tuple[str, str]], args: list[str]) -> None:
    """
    Troubleshoot issues in the cluster.

        Parameters:
            options (list[(opt, optarg)]): Options to use when executing this action
            args (list[str]): Arguments to use when executing this action
    """
    clean_run = run_checks(troubleshoot_checks)

    if clean_run:
        ansithemeprint([ANSIThemeStr("\nImportant", "emphasis"),
                        ANSIThemeStr(":", "default")])
        ansithemeprint([ANSIThemeStr(f"  {about.ADMIN_PROGRAM_NAME}", "programname"),
                        ANSIThemeStr(" troubleshoot", "command"),
                        ANSIThemeStr(" currently only checks for a very limited set of issues;",
                                     "default")])
        ansithemeprint([ANSIThemeStr("  a perfect score is not a guarantee that your cluster is "
                                     "problem free.", "default")])


COMMANDLINE: dict[str, Any] = {
    "Check Versions": {
        "command": ["check-versions", "cv"],
        "description": [
            ANSIThemeStr("Update the package cache and show software versions", "description")],
        "extended_description": [
            [ANSIThemeStr("Note", "note"),
             ANSIThemeStr(": some of the listed software may not be", "description")],
            [ANSIThemeStr("relevant to the configuration in use", "description")],
        ],
        "options": {
            "--force": {
                "description": [
                    ANSIThemeStr("Force update of the version cache", "description")],
                "extended_description": [
                    [ANSIThemeStr("Typically version cache updates are", "description")],
                    [ANSIThemeStr("rate-limited to once per hour; this option", "description")],
                    [ANSIThemeStr("forces an update during the cooldown period.", "description")],
                ],
            },
            "--no-cache-update": {
                "description": [
                    ANSIThemeStr("Do not update the APT cache", "description")],
            },
            "--verbose": {
                "description": [
                    ANSIThemeStr("Be more verbose", "description")],
            },
        },
        "callback": check_for_updates,
    },
    "Import Cluster": {
        "command": ["import-cluster"],
        "values": [
            ANSIThemeStr("[", "separator"),
            ANSIThemeStr("CLUSTER_NAME", "argument"),
            ANSIThemeStr(",", "separator"),
            ANSIThemeStr("...", "argument"),
            ANSIThemeStr("]", "separator")],
        "description": [
            ANSIThemeStr(f"Import existing cluster(s) for use with {about.PROGRAM_SUITE_NAME}",
                         "description")],
        "extended_description": [
            [ANSIThemeStr("If ", "description"),
             ANSIThemeStr("CLUSTER_NAME", "argument"),
             ANSIThemeStr(",", "separator"),
             ANSIThemeStr("...", "argument"),
             ANSIThemeStr(" is not specified", "description")],
            [ANSIThemeStr("all clusters in ", "description"),
             ANSIThemeStr("~/.kube/config", "path")],
            [ANSIThemeStr("will be imported", "description")],
        ],
        "options": {
            "-Y": {
                "description": [
                    ANSIThemeStr("Do not ask for confirmation", "description")],
            },
            "--no-password": {
                "description": [
                    ANSIThemeStr("Do not prompt for a password", "description")],
                "extended_description": [
                    [ANSIThemeStr("Use this if the hosts you are importing", "description")],
                    [ANSIThemeStr("are already configured for login using an SSH key",
                                  "description")],
                ],
            },
            "--save-ansible-logs": {
                "description": [
                    ANSIThemeStr("Save logs from Ansible runs", "description")],
                "extended_description": [
                    [ANSIThemeStr("The logs can be viewed using “", "description"),
                     ANSIThemeStr("cmu", "programname"),
                     ANSIThemeStr(" logs", "command"),
                     ANSIThemeStr("“", "description")]
                ],
            },
            "--verbose": {
                "description": [
                    ANSIThemeStr("Be more verbose", "description")],
            },
        },
        "optional_args": [
            {
                "name": "clustername",
                "string": [
                    ANSIThemeStr("CLUSTER_NAME", "argument")],
                "validation": {
                    "validator": "hostname",
                    "list_separator": ",",
                },
            },
        ],
        "callback": import_cluster,
    },
    "Prepare Installation": {
        "command": ["prepare"],
        "values": [
            ANSIThemeStr("CLUSTER_NAME", "argument"),
            ANSIThemeStr(" [[", "separator"),
            ANSIThemeStr("KUBERNETES_DISTRO", "argument"),
            ANSIThemeStr("=]", "separator"),
            ANSIThemeStr("KUBERNETES_VERSION", "argument"),
            ANSIThemeStr("]", "separator")],
        "description": [
            ANSIThemeStr("Install and configure pre-requisites", "description")],
        "extended_description": [
            [ANSIThemeStr("Run this before ", "description"),
             ANSIThemeStr("setup-control-plane", "command"),
             ANSIThemeStr(".", "description")],
            [ANSIThemeStr("If ", "description"),
             ANSIThemeStr("KUBERNETES_VERSION", "argument"),
             ANSIThemeStr(" is not specified", "description")],
            [ANSIThemeStr("the newest available version will be used.", "description")],
            [ANSIThemeStr("Supported versions for ", "description"),
             ANSIThemeStr("KUBERNETES_DISTRO", "argument"),
             ANSIThemeStr(" are:", "description")],
            [ANSIThemeStr("kubeadm", "argument"),
             ANSIThemeStr(" (default)", "description")],
            [ANSIThemeStr("rke2", "argument")],
        ],
        "options": {
            "--control-plane": {
                "values": [
                    ANSIThemeStr("HOST", "argument")],
                "extended_description": [
                    [ANSIThemeStr("Note", "note"),
                     ANSIThemeStr(": if possible ", "description"),
                     ANSIThemeStr("HOST", "argument"),
                     ANSIThemeStr(" should be a resolvable", "description")],
                    [ANSIThemeStr("hostname; using an IP-address may cause issues", "description")],
                ],
                "description": [
                    ANSIThemeStr("Use ", "description"),
                    ANSIThemeStr("HOST", "argument"),
                    ANSIThemeStr(" as control plane ", "description")],
                "requires_arg": True,
                "validation": {
                    "validator": "hostname_or_ip",
                },
            },
            "--resume": {
                "description": [
                    ANSIThemeStr("Resume preparation", "description")],
                "extended_description": [
                    [ANSIThemeStr("This can be used to resume operations", "description")],
                    [ANSIThemeStr("if preparation was aborted", "description")],
                ],
            },
            "--start-at-task": {
                "values": [
                    ANSIThemeStr("TASK", "argument")],
                "description": [
                    ANSIThemeStr("Start at ", "description"),
                    ANSIThemeStr("TASK", "argument"),
                    ANSIThemeStr(" instead of running all tasks", "description")],
                "requires_arg": True,
                "validation": {
                    "validator": "int",
                    "valid_range": (0, None),
                },
            },
            "--skip-tasks": {
                "values": [
                    ANSIThemeStr("TASK", "argument"),
                    ANSIThemeStr(",", "separator"),
                    ANSIThemeStr("...", "argument")],
                "description": [
                    ANSIThemeStr("Skip ", "description"),
                    ANSIThemeStr("TASK", "argument"),
                    ANSIThemeStr(",", "separator"),
                    ANSIThemeStr("...", "argument")],
                "requires_arg": True,
                "validation": {
                    "validator": "int",
                    "valid_range": (0, None),
                    "list_separator": ",",
                },
            },
            "--list-tasks": {
                "description": [
                    ANSIThemeStr("List valid values for ", "description"),
                    ANSIThemeStr("TASK", "argument")],
                "extended_description": [
                    [ANSIThemeStr("List valid values to use with ", "description"),
                     ANSIThemeStr("--start-at-task", "option")],
                    [ANSIThemeStr("and ", "description"),
                     ANSIThemeStr("--skip-tasks", "option")],
                ],
            },
            "--no-password": {
                "description": [
                    ANSIThemeStr("Do not prompt for a password", "description")],
                "extended_description": [
                    [ANSIThemeStr("Use this if the hosts you are preparing", "description")],
                    [ANSIThemeStr("are already configured for login using an SSH key",
                                  "description")],
                ],
            },
            "--save-ansible-logs": {
                "description": [
                    ANSIThemeStr("Save logs from Ansible runs", "description")],
                "extended_description": [
                    [ANSIThemeStr("The logs can be viewed using “", "description"),
                     ANSIThemeStr("cmu", "programname"),
                     ANSIThemeStr(" logs", "command"),
                     ANSIThemeStr("“", "description")]
                ],
            },
            "--verbose": {
                "description": [
                    ANSIThemeStr("Be more verbose", "description")],
            },
            "-Y": {
                "description": [
                    ANSIThemeStr("Do not ask for confirmation", "description")],
            },
        },
        "required_args": [
            {
                "name": "clustername",
                "string": [
                    ANSIThemeStr("CLUSTER_NAME", "argument")],
                "validation": {
                    "validator": "regex",
                    "regex": r"^[a-z_][a-z0-9_]*$",
                },
            },
        ],
        "optional_args": [
            {
                "name": "version",
                "string": [
                    ANSIThemeStr("KUBERNETES_VERSION", "argument")],
                "validation": {
                    "validator": "regex",
                    "regex": r"^(kubeadm=|rke2=v?|)\d+\.\d+(\.\d+-\d+|\.\d+|)$"
                },
            },
        ],
        "callback": prepare_installation,
    },
    "Create cluster": {
        "command": ["create-cluster"],
        "values": [
            ANSIThemeStr("PATH", "argument"),
        ],
        "description": [
            ANSIThemeStr("Create a cluster based on template in ", "description"),
            ANSIThemeStr("PATH", "path")],
        "extended_description": [
            [ANSIThemeStr("Create a cluster based on a ", "description"),
             ANSIThemeStr("ClusterDeployment", "command")],
            [ANSIThemeStr("template.", "description")],
            [ANSIThemeStr("This combines all the necessary steps to prepare", "description")],
            [ANSIThemeStr("and setup control planes and worker nodes,", "description")],
            [ANSIThemeStr("as well as CNI for a cluster.", "description")],
        ],
        "options": {
            "--redeploy": {
                "description": [
                    ANSIThemeStr("Redeploy workloads", "description")],
            },
            "-Y": {
                "description": [
                    ANSIThemeStr("Do not ask for confirmation", "description")],
            },
        },
        "required_args": [
            {
                "name": "path",
                "string": [
                    ANSIThemeStr("PATH", "argument")],
                "validation": {
                    "validator": "path",
                },
            },
        ],
        "callback": create_cluster,
    },
    "Setup Control Plane": {
        "command": ["setup-control-plane"],
        "values": [
            ANSIThemeStr("[", "separator"),
            ANSIThemeStr("CNI", "argument"),
            ANSIThemeStr("]", "separator"),
            ANSIThemeStr(" [", "separator"),
            ANSIThemeStr("POD_NETWORK_CIDR", "argument"),
            ANSIThemeStr("]", "separator"),
        ],
        "description": [
            ANSIThemeStr("Setup and launch the control plane", "description")],
        "extended_description": [
            [ANSIThemeStr("Valid options for CNI", "description")],
            [ANSIThemeStr("(Container Network Interface, aka Pod Network): ", "description")],
            [ANSIThemeStr("antrea", "argument"),
             ANSIThemeStr(", ", "separator"),
             ANSIThemeStr("calico", "argument"),
             ANSIThemeStr(", ", "separator"),
             ANSIThemeStr("canal", "argument"),
             ANSIThemeStr(", ", "separator"),
             ANSIThemeStr("cilium", "argument"),
             ANSIThemeStr(",", "separator")],
            [ANSIThemeStr("flannel", "argument"),
             ANSIThemeStr(", ", "separator"),
             ANSIThemeStr("kube-router", "argument"),
             ANSIThemeStr(", ", "separator"),
             ANSIThemeStr("weave", "argument"),
             ANSIThemeStr(", ", "separator"),
             ANSIThemeStr("none", "argument")],
            [ANSIThemeStr("By default ", "description"),
             ANSIThemeStr("cilium", "argument"),
             ANSIThemeStr(" will be used as CNI", "description")],
            [ANSIThemeStr("and ", "description"),
             ANSIThemeStr("10.244.0.0/16", "argument"),
             ANSIThemeStr(" will be used as pod network CIDR.", "description")],
            [ANSIThemeStr("If you wish to postpone the choice of CNI", "description")],
            [ANSIThemeStr("you can specify ", "description"),
             ANSIThemeStr("none", "argument")],
        ],
        "options": {
            "--resume": {
                "description": [
                    ANSIThemeStr("Resume setup", "description")],
                "extended_description": [
                    [ANSIThemeStr("This can be used to resume operations", "description")],
                    [ANSIThemeStr("if control plane setup was aborted", "description")],
                ],
            },
            "--start-at-task": {
                "values": [
                    ANSIThemeStr("TASK", "argument")],
                "description": [
                    ANSIThemeStr("Start at ", "description"),
                    ANSIThemeStr("TASK", "argument"),
                    ANSIThemeStr(" instead of running all tasks", "description")],
                "requires_arg": True,
                "validation": {
                    "validator": "int",
                    "valid_range": (0, None),
                },
            },
            "--skip-tasks": {
                "values": [
                    ANSIThemeStr("TASK", "argument"),
                    ANSIThemeStr(",", "separator"),
                    ANSIThemeStr("...", "argument")],
                "description": [
                    ANSIThemeStr("Skip ", "description"),
                    ANSIThemeStr("TASK", "argument"),
                    ANSIThemeStr(",", "separator"),
                    ANSIThemeStr("...", "argument")],
                "requires_arg": True,
                "validation": {
                    "validator": "int",
                    "valid_range": (0, None),
                    "list_separator": ",",
                },
            },
            "--list-tasks": {
                "description": [
                    ANSIThemeStr("List valid values for ", "description"),
                    ANSIThemeStr("TASK", "argument")],
                "extended_description": [
                    [ANSIThemeStr("List valid values to use with ", "description"),
                     ANSIThemeStr("--start-at-task", "option")],
                    [ANSIThemeStr("and ", "description"),
                     ANSIThemeStr("--skip-tasks", "option")],
                ],
            },
            "--save-ansible-logs": {
                "description": [
                    ANSIThemeStr("Save logs from Ansible runs", "description")],
                "extended_description": [
                    [ANSIThemeStr("The logs can be viewed using “", "description"),
                     ANSIThemeStr("cmu", "programname"),
                     ANSIThemeStr(" logs", "command"),
                     ANSIThemeStr("“", "description")]
                ],
            },
            "--cri": {
                "values": [
                    ANSIThemeStr("CRI", "argument")],
                "description": [
                    ANSIThemeStr("Use ", "description"),
                    ANSIThemeStr("CRI", "argument"),
                    ANSIThemeStr(" instead of the default CRI", "description")],
                "extended_description": [
                    [ANSIThemeStr("Valid options for CRI", "description")],
                    [ANSIThemeStr("(Container Runtime Interface) are:", "description")],
                    [ANSIThemeStr("containerd", "argument"),
                     ANSIThemeStr(", ", "separator"),
                     ANSIThemeStr("cri-o", "argument")],
                    [ANSIThemeStr("Default CRI:", "description")],
                    [ANSIThemeStr("containerd", "argument"),
                     ANSIThemeStr(" (", "description"),
                     ANSIThemeStr("rke2", "programname"),
                     ANSIThemeStr("),", "description")],
                    [ANSIThemeStr("containerd", "argument"),
                     ANSIThemeStr(" (", "description"),
                     ANSIThemeStr("kubeadm", "programname"),
                     ANSIThemeStr(")", "description"),
                     ANSIThemeStr(",", "separator")],
                    [ANSIThemeStr("cri-o", "argument"),
                     ANSIThemeStr(" (", "description"),
                     ANSIThemeStr("kubeadm", "programname"),
                     ANSIThemeStr(" with ", "description"),
                     ANSIThemeStr("--enable-dra", "option"),
                     ANSIThemeStr(")", "description")],
                    [ANSIThemeStr("Note", "note"),
                     ANSIThemeStr(": ", "description"),
                     ANSIThemeStr("Kubernetes", "programname"),
                     ANSIThemeStr(" >= ", "description"),
                     ANSIThemeStr("1.26", "version"),
                     ANSIThemeStr(" requires", "description")],
                    [ANSIThemeStr("containerd", "argument"),
                     ANSIThemeStr(" >= ", "description"),
                     ANSIThemeStr("1.6", "version"),
                     ANSIThemeStr(" or ", "description"),
                     ANSIThemeStr("cri-o", "argument")],
                ],
                "requires_arg": True,
                "validation": {
                    "validator": "allowlist",
                    "allowlist": ["containerd", "cri-o"],
                },
            },
            "--enable-dra": {
                "description": [
                    ANSIThemeStr("Enable DRA", "description")],
                "extended_description": [
                    [ANSIThemeStr("Enables the feature gates necessary to use", "description")],
                    [ANSIThemeStr("Dynamic Resource Allocation (DRA).", "description")],
                    [ANSIThemeStr("Currently only ", "description"),
                     ANSIThemeStr("cri-o", "argument"),
                     ANSIThemeStr(" supports DRA", "description")],
                ],
            },
            "--override-cni": {
                "description": [
                    ANSIThemeStr("Override CNI", "description")],
                "extended_description": [
                    [ANSIThemeStr("Allow a change of CNI even if installation", "description")],
                    [ANSIThemeStr("started with a different CNI", "description")],
                ],
            },
            "--verbose": {
                "description": [
                    ANSIThemeStr("Be more verbose", "description")],
            },
            "-Y": {
                "description": [
                    ANSIThemeStr("Do not ask for confirmation", "description")],
            },
        },
        "optional_args": [
            {
                "name": "cni",
                "string": [
                    ANSIThemeStr("CNI", "argument")],
                "validation": {
                    "validator": "allowlist",
                    "allowlist": ["antrea", "calico", "canal", "cilium",
                                  "flannel", "kube-router", "weave", "none"],
                },
            },
            {
                "name": "pod_network_cidr",
                "string": [
                    ANSIThemeStr("POD_NETWORK_CIDR", "argument")],
                "validation": {
                    "validator": "cidr",
                },
            },
        ],
        "callback": setup_control_planes,
    },
    "Setup CNI": {
        "command": ["setup-cni"],
        "values": [
            ANSIThemeStr("[", "separator"),
            ANSIThemeStr("CNI", "argument"),
            ANSIThemeStr("]", "separator")],
        "description": [
            ANSIThemeStr("Install and configure CNI", "description")],
        "extended_description": [
            [ANSIThemeStr("Valid options for CNI", "description")],
            [ANSIThemeStr("(Container Network Interface, aka Pod Network): ", "description")],
            [ANSIThemeStr("antrea", "argument"),
             ANSIThemeStr(", ", "separator"),
             ANSIThemeStr("calico", "argument"),
             ANSIThemeStr(", ", "separator"),
             ANSIThemeStr("canal", "argument"),
             ANSIThemeStr(", ", "separator"),
             ANSIThemeStr("cilium", "argument"),
             ANSIThemeStr(", ", "separator")],
            [ANSIThemeStr("flannel", "argument"),
             ANSIThemeStr(", ", "separator"),
             ANSIThemeStr("kube-router", "argument"),
             ANSIThemeStr(", ", "separator"),
             ANSIThemeStr("weave", "argument")],
            [ANSIThemeStr("By default ", "description"),
             ANSIThemeStr("cilium", "argument"),
             ANSIThemeStr(" will be used as CNI", "description")],
        ],
        "options": {
            "--reinstall": {
                "description": [
                    ANSIThemeStr("Try to reinstall the already installed CNI", "description")],
            },
            "-Y": {
                "description": [
                    ANSIThemeStr("Do not ask for confirmation", "description")],
            },
            "--verbose": {
                "description": [
                    ANSIThemeStr("Be more verbose", "description")],
            },
        },
        "optional_args": [
            {
                "name": "cni",
                "string": [
                    ANSIThemeStr("CNI", "argument")],
                "validation": {
                    "validator": "allowlist",
                    "allowlist": ["antrea", "calico", "canal", "cilium",
                                  "flannel", "kube-router", "weave"],
                },
                "default": DEFAULT_CNI,
            },
        ],
        "callback": setup_cni,
    },
    "Uninstall CNI": {
        "command": ["uninstall-cni"],
        "description": [
            ANSIThemeStr("Uninstall the CNI", "description")],
        "extended_description": [
            [ANSIThemeStr("This should be used in case", "description")],
            [ANSIThemeStr("you want to switch to another CNI", "description")],
        ],
        "options": {
            "--cni-version": {
                "values": [
                    ANSIThemeStr("CNI_VERSION", "argument")],
                "description": [
                    ANSIThemeStr("The installed CNI version", "description")],
                "extended_description": [
                    [ANSIThemeStr("Use this option to specify", "description")],
                    [ANSIThemeStr("the version in case it cannot", "description")],
                    [ANSIThemeStr("be autodetected", "description")],
                ],
                "requires_arg": True,
                "validation": {
                    "validator": "regex",
                    "regex": "[a-zA-Z0-9+-]+",
                },
            },
            "-Y": {
                "description": [
                    ANSIThemeStr("Do not ask for confirmation", "description")],
            },
            "--verbose": {
                "description": [
                    ANSIThemeStr("Be more verbose", "description")],
            },
        },
        "callback": teardown_cni,
    },
    "Upgrade CNI": {
        "command": ["upgrade-cni"],
        "values": [
            ANSIThemeStr("[", "separator"),
            ANSIThemeStr("CNI", "argument"),
            ANSIThemeStr("]", "separator")],
        "description": [
            ANSIThemeStr("Upgrade the CNI", "description")],
        "options": {
            "--verbose": {
                "description": [
                    ANSIThemeStr("Be more verbose", "description")],
            },
        },
        "optional_args": [
            {
                "name": "cni",
                "string": [
                    ANSIThemeStr("CNI", "argument")],
                "validation": {
                    "validator": "allowlist",
                    "allowlist": ["antrea", "calico", "canal", "cilium",
                                  "flannel", "kube-router", "weave"],
                },
            },
        ],
        "callback": upgrade_cni,
    },
    "Upgrade Control Plane": {
        "command": ["upgrade-control-plane"],
        "values": [
            ANSIThemeStr("[", "separator"),
            ANSIThemeStr("KUBERNETES_VERSION", "argument"),
            ANSIThemeStr("]", "separator")],
        "description": [
            ANSIThemeStr("Upgrade the control plane", "description")],
        "extended_description": [
            [ANSIThemeStr("If ", "description"),
             ANSIThemeStr("KUBERNETES_VERSION", "argument"),
             ANSIThemeStr(" is not specified", "description")],
            [ANSIThemeStr("the newest available version will be used.", "description")],
            [ANSIThemeStr("Upgrading requires all nodes to be drained first.", "description")],
            [ANSIThemeStr("Once the control plane has been uppgraded you ", "description"),
             ANSIThemeStr("must", "emphasis")],
            [ANSIThemeStr("upgrade all nodes to the same version.", "description")],
            [ANSIThemeStr("Important", "emphasis"),
             ANSIThemeStr(": skipping PATCH REVISIONS is acceptable,", "description")],
            [ANSIThemeStr("but when upgrading to a newer MINOR version", "description")],
            [ANSIThemeStr("all intermediate MINOR versions must be", "description")],
            [ANSIThemeStr("installed first; this applies to nodes too.", "description")],
        ],
        "options": {
            "--ignore-feature-gates": {
                "description": [
                    ANSIThemeStr("Ignore the result of the feature gates check", "description")],
                "extended_description": [
                    [ANSIThemeStr("Upgrading a cluster may fail if the default", "description")],
                    [ANSIThemeStr("set of feature gates has been altered; ", "description"),
                     ANSIThemeStr("by default", "description")],
                    [ANSIThemeStr("upgrades are aborted if such changes", "description"),
                     ANSIThemeStr("are detected.", "description")],
                    [ANSIThemeStr("This option makes the check non-aborting.", "description")],
                    [ANSIThemeStr("Note", "note"),
                     ANSIThemeStr(": this may yield a failed installation", "description")],
                    [ANSIThemeStr("or non-operational cluster", "description")],
                ],
            },
            "--no-cache-update": {
                "description": [
                    ANSIThemeStr("Do not update the APT cache", "description")],
            },
            "--resume": {
                "description": [
                    ANSIThemeStr("Resume upgrade", "description")],
                "extended_description": [
                    [ANSIThemeStr("This can be used to resume operations", "description")],
                    [ANSIThemeStr("if upgrade was aborted", "description")],
                ],
            },
            "--start-at-task": {
                "values": [
                    ANSIThemeStr("TASK", "argument")],
                "description": [
                    ANSIThemeStr("Start at ", "description"),
                    ANSIThemeStr("TASK", "argument"),
                    ANSIThemeStr(" instead of running all tasks", "description")],
                "requires_arg": True,
                "validation": {
                    "validator": "int",
                    "valid_range": (0, None),
                },
            },
            "--skip-tasks": {
                "values": [
                    ANSIThemeStr("TASK", "argument"),
                    ANSIThemeStr(",", "separator"),
                    ANSIThemeStr("...", "argument")],
                "description": [
                    ANSIThemeStr("Skip ", "description"),
                    ANSIThemeStr("TASK", "argument"),
                    ANSIThemeStr(",", "separator"),
                    ANSIThemeStr("...", "argument")],
                "requires_arg": True,
                "validation": {
                    "validator": "int",
                    "valid_range": (0, None),
                    "list_separator": ",",
                },
            },
            "--list-tasks": {
                "description": [
                    ANSIThemeStr("List valid values for ", "description"),
                    ANSIThemeStr("TASK", "argument")],
                "extended_description": [
                    [ANSIThemeStr("List valid values to use with ", "description"),
                     ANSIThemeStr("--start-at-task", "option")],
                    [ANSIThemeStr("and ", "description"),
                     ANSIThemeStr("--skip-tasks", "option")],
                ],
            },
            "--reinstall": {
                "description": [
                    ANSIThemeStr("Allow installing the same version", "description")],
                "extended_description": [
                    [ANSIThemeStr("This option allows you to install the same", "description")],
                    [ANSIThemeStr("version that's already running in the cluster", "description")],
                ],
            },
            "--override": {
                "description": [
                    ANSIThemeStr("Override/rebuild installation info", "description")],
            },
            "--save-ansible-logs": {
                "description": [
                    ANSIThemeStr("Save logs from Ansible runs", "description")],
                "extended_description": [
                    [ANSIThemeStr("The logs can be viewed using “", "description"),
                     ANSIThemeStr("cmu", "programname"),
                     ANSIThemeStr(" logs", "command"),
                     ANSIThemeStr("“", "description")]
                ],
            },
            "--verbose": {
                "description": [
                    ANSIThemeStr("Be more verbose", "description")],
            },
            "-Y": {
                "description": [
                    ANSIThemeStr("Do not ask for confirmation", "description")],
            },
        },
        "optional_args": [
            {
                "name": "version",
                "string": [
                    ANSIThemeStr("KUBERNETES_VERSION", "argument")],
                "validation": {
                    "validator": "regex",
                    "regex": r"^\d+\.\d+(\.\d+-\d+|\.\d+|)$"
                },
            },
        ],
        "callback": upgrade_control_plane,
    },
    "Teardown Control Plane": {
        "command": ["teardown-control-plane"],
        "description": [
            ANSIThemeStr("Tear down the control plane", "description")],
        "extended_description": [
            [ANSIThemeStr("Note", "emphasis"),
             ANSIThemeStr(": Before running this command all nodes", "description")],
            [ANSIThemeStr("must have been removed first.", "description")],
            [ANSIThemeStr("The configuration for the control plane", "description")],
            [ANSIThemeStr("and any software installed during setup", "description")],
            [ANSIThemeStr("will NOT be removed", "description")],

        ],
        "options": {
            "--resume": {
                "description": [
                    ANSIThemeStr("Resume teardown", "description")],
                "extended_description": [
                    [ANSIThemeStr("This can be used to resume operations", "description")],
                    [ANSIThemeStr("if teardown was aborted", "description")],
                ],
            },
            "--start-at-task": {
                "values": [
                    ANSIThemeStr("TASK", "argument")],
                "description": [
                    ANSIThemeStr("Start at ", "description"),
                    ANSIThemeStr("TASK", "argument"),
                    ANSIThemeStr(" instead of running all tasks", "description")],
                "requires_arg": True,
                "validation": {
                    "validator": "int",
                    "valid_range": (0, None),
                },
            },
            "--skip-tasks": {
                "values": [
                    ANSIThemeStr("TASK", "argument"),
                    ANSIThemeStr(",", "separator"),
                    ANSIThemeStr("...", "argument")],
                "description": [
                    ANSIThemeStr("Skip ", "description"),
                    ANSIThemeStr("TASK", "argument"),
                    ANSIThemeStr(",", "separator"),
                    ANSIThemeStr("...", "argument")],
                "requires_arg": True,
                "validation": {
                    "validator": "int",
                    "valid_range": (0, None),
                    "list_separator": ",",
                },
            },
            "--list-tasks": {
                "description": [
                    ANSIThemeStr("List valid values for ", "description"),
                    ANSIThemeStr("TASK", "argument")],
                "extended_description": [
                    [ANSIThemeStr("List valid values to use with ", "description"),
                     ANSIThemeStr("--start-at-task", "option")],
                    [ANSIThemeStr("and ", "description"),
                     ANSIThemeStr("--skip-tasks", "option")],
                ],
            },
            "--save-ansible-logs": {
                "description": [
                    ANSIThemeStr("Save logs from Ansible runs", "description")],
                "extended_description": [
                    [ANSIThemeStr("The logs can be viewed using “", "description"),
                     ANSIThemeStr("cmu", "programname"),
                     ANSIThemeStr(" logs", "command"),
                     ANSIThemeStr("“", "description")]
                ],
            },
            "--verbose": {
                "description": [
                    ANSIThemeStr("Be more verbose", "description")],
            },
            "-Y": {
                "description": [
                    ANSIThemeStr("Do not ask for confirmation", "description")],
            },
        },
        "callback": teardown_control_plane,
    },
    "Purge Control Plane": {
        "command": ["purge-control-plane"],
        "description": [
            ANSIThemeStr("Purge configuration and packages", "description")],
        "extended_description": [
            [ANSIThemeStr("Software and configuration needed", "description")],
            [ANSIThemeStr("for ", "description"),
             ANSIThemeStr(f"{about.PROGRAM_SUITE_NAME}", "programname"),
             ANSIThemeStr(" itself will not be purged", "description")],
        ],
        "options": {
            "--resume": {
                "description": [
                    ANSIThemeStr("Resume purge; can be used if purge was aborted", "description")],
            },
            "--start-at-task": {
                "values": [
                    ANSIThemeStr("TASK", "argument")],
                "description": [
                    ANSIThemeStr("Start at ", "description"),
                    ANSIThemeStr("TASK", "argument"),
                    ANSIThemeStr(" instead of running all tasks", "description")],
                "requires_arg": True,
                "validation": {
                    "validator": "int",
                    "valid_range": (0, None),
                },
            },
            "--skip-tasks": {
                "values": [
                    ANSIThemeStr("TASK", "argument"),
                    ANSIThemeStr(",", "separator"),
                    ANSIThemeStr("...", "argument")],
                "description": [
                    ANSIThemeStr("Skip ", "description"),
                    ANSIThemeStr("TASK", "argument"),
                    ANSIThemeStr(",", "separator"),
                    ANSIThemeStr("...", "argument")],
                "requires_arg": True,
                "validation": {
                    "validator": "int",
                    "valid_range": (0, None),
                    "list_separator": ",",
                },
            },
            "--list-tasks": {
                "description": [
                    ANSIThemeStr("List valid values for ", "description"),
                    ANSIThemeStr("TASK", "argument")],
                "extended_description": [
                    [ANSIThemeStr("List valid values to use with ", "description"),
                     ANSIThemeStr("--start-at-task", "option")],
                    [ANSIThemeStr("and ", "description"),
                     ANSIThemeStr("--skip-tasks", "option")],
                ],
            },
            "--save-ansible-logs": {
                "description": [
                    ANSIThemeStr("Save logs from Ansible runs", "description")],
                "extended_description": [
                    [ANSIThemeStr("The logs can be viewed using “", "description"),
                     ANSIThemeStr("cmu", "programname"),
                     ANSIThemeStr(" logs", "command"),
                     ANSIThemeStr("“", "description")]
                ],
            },
            "--verbose": {
                "description": [
                    ANSIThemeStr("Be more verbose", "description")],
            },
            "-Y": {
                "description": [
                    ANSIThemeStr("Do not ask for confirmation", "description")],
            },
        },
        "callback": purge_control_plane,
    },
    "Taint Control Plane": {
        "command": ["taint-control-plane"],
        "values": [
            ANSIThemeStr("[", "separator"),
            ANSIThemeStr("CONTROLPLANE", "argument"),
            ANSIThemeStr(",", "separator"),
            ANSIThemeStr("...", "argument"),
            ANSIThemeStr("]", "separator")],
        "description": [
            ANSIThemeStr("Mark control plane(s) as tainted", "description")],
        "extended_description": [
            [ANSIThemeStr("If you have previously marked your control plane(s)", "description")],
            [ANSIThemeStr("as untainted you can mark them as tainted", "description")],
            [ANSIThemeStr("using this command.", "description")],
            [ANSIThemeStr("If ", "description"),
             ANSIThemeStr("CONTROLPLANE", "argument"),
             ANSIThemeStr(",", "separator"),
             ANSIThemeStr("...", "argument"),
             ANSIThemeStr(" is not specified", "description")],
            [ANSIThemeStr("all control planes will be tainted", "description")],
        ],
        "optional_args": [
            {
                "name": "controlplane",
                "string": [
                    ANSIThemeStr("CONTROLPLANE", "argument")],
                "validation": {
                    "validator": "hostname_or_ip",
                    "list_separator": ",",
                },
            },
        ],
        "callback": taint_control_plane,
    },
    "Untaint Control Plane": {
        "command": ["untaint-control-plane"],
        "values": [
            ANSIThemeStr("[", "separator"),
            ANSIThemeStr("CONTROLPLANE", "argument"),
            ANSIThemeStr(",", "separator"),
            ANSIThemeStr("...", "argument"),
            ANSIThemeStr("]", "separator")],
        "description": [
            ANSIThemeStr("Mark control plane(s) as untainted", "description")],
        "extended_description": [
            [ANSIThemeStr("By default control planes are marked as tainted;", "description")],
            [ANSIThemeStr("workloads that lack tolerations will not be", "description")],
            [ANSIThemeStr("scheduled to control planes. If you are running", "description")],
            [ANSIThemeStr("a single-node cluster, or if the control plane", "description")],
            [ANSIThemeStr("is very powerful it might be useful to permit", "description")],
            [ANSIThemeStr("workloads on control plane(s) too.", "description")],
            [ANSIThemeStr("If ", "description"),
             ANSIThemeStr("CONTROLPLANE", "argument"),
             ANSIThemeStr(",", "separator"),
             ANSIThemeStr("...", "argument"),
             ANSIThemeStr(" is not specified", "description")],
            [ANSIThemeStr("all control planes will be untainted", "description")],

        ],
        "optional_args": [
            {
                "name": "controlplane",
                "string": [
                    ANSIThemeStr("CONTROLPLANE", "argument")],
                "validation": {
                    "validator": "hostname_or_ip",
                    "list_separator": ",",
                },
            },
        ],
        "callback": untaint_control_plane,
    },
    "Audit": {
        "command": ["audit"],
        "description": [
            ANSIThemeStr("Search for potential security issues in the cluster", "description")],
        "extended_description": [
            [ANSIThemeStr("Note", "emphasis"),
             ANSIThemeStr(": If the system is configured", "description")],
            [ANSIThemeStr("to use ", "description"),
             ANSIThemeStr("usergroups", "emphasis"),
             ANSIThemeStr(" (every user have their own group", "description")],
            [ANSIThemeStr("that only they belong to); if the name of that", "description")],
            [ANSIThemeStr("group differs from the username be sure to specify", "description")],
            [ANSIThemeStr("that group using the ", "description"),
             ANSIThemeStr("--usergroup ", "option"),
             ANSIThemeStr("USERGROUP", "argument"),
             ANSIThemeStr(" option,", "description")],
            [ANSIThemeStr("to prevent the permission checker", "description")],
            [ANSIThemeStr("from complaining about insecure permissions", "description")],
        ],
        "options": {
            "--disable-usergroup-autodetect": {
                "description": [
                    ANSIThemeStr("Disable usergroup autodetect", "description")],
                "extended_description": [
                    [ANSIThemeStr("Note", "emphasis"),
                     ANSIThemeStr(": the audit command attempts", "description")],
                    [ANSIThemeStr("to autodetect whether ", "description"),
                     ANSIThemeStr("usergroups", "emphasis"),
                     ANSIThemeStr(" are in use;", "description")],
                    [ANSIThemeStr("use this option to disable autodetect", "description")],
                ],
            },
            "--usergroup": {
                "values": [
                    ANSIThemeStr("USERGROUP", "argument")],
                "description": [
                    ANSIThemeStr("The name of the usergroup", "description")],
                "requires_arg": True,
                "validation": {
                    "validator": "regex",
                    # regex taken from debian;
                    # other distros might be stricter or possible more permissive
                    "regex": r"^[a-z_][a-z0-9_-]*[$]?$",
                },
            },
        },
        "callback": audit,
    },
    "Preflight Check": {
        "command": ["preflight-check"],
        "values": [
            ANSIThemeStr("CONTROLPLANE", "argument")],
        "description": [
            ANSIThemeStr("Preflight check", "description")],
        "extended_description": [
            [ANSIThemeStr("Check for potential pitfalls that may prevent", "description")],
            [ANSIThemeStr("preparation or setup from succeeding", "description")],
        ],
        "options": {
            "--no-password": {
                "description": [
                    ANSIThemeStr("Do not prompt for a password", "description")],
                "extended_description": [
                    [ANSIThemeStr("Use this if the hosts you are preparing", "description")],
                    [ANSIThemeStr("are already configured for login using an SSH key",
                                  "description")],
                ],
            },
        },
        "required_args": [
            {
                "name": "controlplane",
                "string": [
                    ANSIThemeStr("CONTROLPLANE", "argument")],
                "validation": {
                    "validator": "hostname_or_ip",
                },
            },
        ],
        "callback": preflight_check,
    },
    "Troubleshoot": {
        "command": ["troubleshoot"],
        "description": [
            ANSIThemeStr("Search for potential problems in the cluster", "description")],
        "callback": troubleshoot,
    },
    "spacer1": {
        "command": [""],
        "description": [
            ANSIThemeStr("", "description")],
    },
}


def main() -> None:
    """
    Main function for the program.
    """
    # Before doing anything else, make sure that the user is not running as root
    if os.geteuid() == 0:
        sys.exit("CRITICAL: This program should not be run as the root user; aborting.")

    # Now check if all necessary paths exist
    if (violations := cmtlib.setup_paths()) != [SecurityStatus.OK]:
        violations_joined = cmtio.join_securitystatus_set(",", set(violations))
        sys.exit(f"Failed to create necessary directories; violated rules: {violations_joined}")

    # Then initialise the configuration file
    read_cmtconfig()

    command, options, args = \
        parse_commandline(about.ADMIN_PROGRAM_NAME, about.ADMIN_PROGRAM_VERSION,
                          PROGRAMDESCRIPTION, PROGRAMAUTHORS, sys.argv, COMMANDLINE,
                          theme=DEFAULT_THEME_FILE)

    for key, value in options:
        if key == "__commandname" and value in ("help", "version", "preflight-check"):
            break
    else:
        checks.check_ansible_dir_permissions(verbose=False, exit_on_error=True,
                                             quiet_on_ok=True, user=getuser())
        checks.check_netrc_permissions(verbose=False, exit_on_error=True, quiet_on_ok=True)

    # Used by the ansible module
    ansible_configuration["ansible_forks"] = \
        deep_get(cmtlib.cmtconfig, DictPath("Ansible#forks"), 10)
    ansible_user = deep_get(cmtlib.cmtconfig, DictPath("Ansible#ansible_user"))
    if ansible_user is None or len(ansible_user) == 0:
        ansible_user = getuser()
    ansible_configuration["ansible_user"] = ansible_user
    ansible_password = deep_get(cmtlib.cmtconfig, DictPath("Ansible#ansible_password"))
    if ansible_password is not None and ansible_password:
        ansible_configuration["ansible_password"] = ansible_password
    ansible_configuration["disable_strict_host_key_checking"] = \
        deep_get(cmtlib.cmtconfig, DictPath("Nodes#disablestricthostkeychecking"), False)
    ansible_configuration["save_logs"] = \
        deep_get(cmtlib.cmtconfig, DictPath("Ansible#save_logs"), False)

    return command(options, args)


if __name__ == "__main__":
    main()