cmt-install

#! /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)
#
# Copyright the Cluster Management Toolkit for Kubernetes contributors.
# SPDX-License-Identifier: MIT

# pylint: disable=invalid-name,too-many-lines

import errno
import os
from pathlib import Path
import re
import subprocess  # nosec
from subprocess import PIPE, STDOUT  # nosec
import sys
from typing import Any, Union

from clustermanagementtoolkit.cmttypes import deep_get, DictPath, FilePath, SecurityPolicy

from clustermanagementtoolkit.cmtpaths import BASH_COMPLETION_BASE_DIR, BASH_COMPLETION_DIR
from clustermanagementtoolkit.cmtpaths import BASH_COMPLETION_DIRNAME
from clustermanagementtoolkit.cmtpaths import BINDIR, HOMEDIR, CMTDIR, CMT_LOGS_DIR
from clustermanagementtoolkit.cmtpaths import VERSION_CACHE_DIR
from clustermanagementtoolkit.cmtpaths import CHANGELOG_DIR, CHANGELOG_DIRNAME
from clustermanagementtoolkit.cmtpaths import DEPLOYMENT_DIR, ANSIBLE_DIR, ANSIBLE_LOG_DIR
from clustermanagementtoolkit.cmtpaths import CMT_CONFIG_FILE_DIR, CMT_HOOKS_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_PLAYBOOK_DIR, CMT_CONFIG_FILE
from clustermanagementtoolkit.cmtpaths import PARSER_DIR, THEME_DIR, VIEW_DIR
from clustermanagementtoolkit.cmtpaths import ANSIBLE_PLAYBOOK_DIRNAME, CMT_CONFIG_FILENAME
from clustermanagementtoolkit.cmtpaths import PARSER_DIRNAME, THEME_DIRNAME, VIEW_DIRNAME
from clustermanagementtoolkit.cmtpaths import SOFTWARE_SOURCES_DIRNAME, SOFTWARE_SOURCES_DIR

from clustermanagementtoolkit.cmtio import execute_command_with_response
from clustermanagementtoolkit.cmtio import secure_copy, secure_mkdir, secure_symlink, secure_which

from clustermanagementtoolkit.commandparser import parse_commandline

from clustermanagementtoolkit.ansithemeprint import ANSIThemeStr, ansithemeprint, ansithemeinput

from clustermanagementtoolkit import about

PROGRAMDESCRIPTION = f"Setup {about.PROGRAM_SUITE_FULL_NAME}"
PROGRAMAUTHORS = "Written by David Weinehall."

DIRECTORIES = [
    (BINDIR, 0o755),
    (CMTDIR, 0o755),
    (CMT_LOGS_DIR, 0o700),
    (DEPLOYMENT_DIR, 0o755),
    (VERSION_CACHE_DIR, 0o700),
    (ANSIBLE_DIR, 0o755),
    (ANSIBLE_LOG_DIR, 0o700),
    (CMT_CONFIG_FILE_DIR, 0o755),
    (CMT_HOOKS_DIR, 0o755),
    (CMT_PRE_PREPARE_DIR, 0o755),
    (CMT_POST_PREPARE_DIR, 0o755),
    (CMT_PRE_SETUP_DIR, 0o755),
    (CMT_POST_SETUP_DIR, 0o755),
    (CMT_PRE_UPGRADE_DIR, 0o755),
    (CMT_POST_UPGRADE_DIR, 0o755),
    (CMT_PRE_TEARDOWN_DIR, 0o755),
    (CMT_POST_TEARDOWN_DIR, 0o755),
    (CMT_PRE_PURGE_DIR, 0o755),
    (CMT_POST_PURGE_DIR, 0o755),
    (BASH_COMPLETION_BASE_DIR, 0o755),
    (BASH_COMPLETION_DIR, 0o755),
]

EXECUTABLE_SYMLINKS = [
    # These are Python scripts
    (FilePath(os.path.join(os.getcwd(), "cmtadm")), FilePath(os.path.join(BINDIR, "cmtadm"))),
    (FilePath(os.path.join(os.getcwd(), "cmtinv")), FilePath(os.path.join(BINDIR, "cmtinv"))),
    (FilePath(os.path.join(os.getcwd(), "cmt")), FilePath(os.path.join(BINDIR, "cmt"))),
    (FilePath(os.path.join(os.getcwd(), "cmu")), FilePath(os.path.join(BINDIR, "cmu"))),
]

MISC_SYMLINKS = [
    (FilePath(os.path.join(os.getcwd(), SOFTWARE_SOURCES_DIRNAME)), SOFTWARE_SOURCES_DIR),
    (FilePath(os.path.join(os.getcwd(), ANSIBLE_PLAYBOOK_DIRNAME)), ANSIBLE_PLAYBOOK_DIR),
    (FilePath(os.path.join(os.getcwd(), PARSER_DIRNAME)), PARSER_DIR),
    (FilePath(os.path.join(os.getcwd(), THEME_DIRNAME)), THEME_DIR),
    (FilePath(os.path.join(os.getcwd(), VIEW_DIRNAME)), VIEW_DIR),
    (FilePath(os.path.join(os.getcwd(), CHANGELOG_DIRNAME)), CHANGELOG_DIR),
    (FilePath(os.path.join(os.getcwd(), BASH_COMPLETION_DIRNAME, "cmt")),
     FilePath(os.path.join(BASH_COMPLETION_DIR, "cmt"))),
    (FilePath(os.path.join(os.getcwd(), BASH_COMPLETION_DIRNAME, "cmtadm")),
     FilePath(os.path.join(BASH_COMPLETION_DIR, "cmtadm"))),
    (FilePath(os.path.join(os.getcwd(), BASH_COMPLETION_DIRNAME, "cmtinv")),
     FilePath(os.path.join(BASH_COMPLETION_DIR, "cmtinv"))),
    (FilePath(os.path.join(os.getcwd(), BASH_COMPLETION_DIRNAME, "cmu")),
     FilePath(os.path.join(BASH_COMPLETION_DIR, "cmu"))),
]

# Distros in always-fallback-distros will always use pip
# Distros in never-fallback-distros will never use pip
PACKAGES_WITH_PIP_FALLBACK = {
    "python3-ansible-runner": {
        "fallback": "ansible-runner",
        "always-fallback-distros": [
            "suse",
            "rhel",
        ],
    },
    "python3-cryptography": {
        "fallback": "cryptography",
        "always-fallback-distros": [
            "suse",
            # Until RHEL 9
            "rhel",
        ],
        "never-fallback-distros": [
            "debian",
        ],
    },
    "python3-jinja2": {
        "fallback": "jinja2",
        "always-fallback-distros": [
            "suse",
            "rhel",
        ],
        "never-fallback-distros": [
            "debian",
        ],
    },
    "python3-natsort": {
        "fallback": "natsort",
        "always-fallback-distros": [
            "suse",
            "rhel",
        ],
        "never-fallback-distros": [
            "debian",
        ],
    },
    "python3-paramiko": {
        "fallback": "paramiko",
        "always-fallback-distros": [
            "suse",
            "rhel",
        ],
        "never-fallback-distros": [
            "debian",
        ],
    },
    "python3-pyyaml": {
        "fallback": "PyYAML",
        "distros": [
            # Until RHEL 9
            "fedora",
            "rhel",
        ],
        "always-fallback-distros": [
            # Until RHEL 9
            "rhel",
        ],
    },
    "python3-ujson": {
        "fallback": "ujson",
        "always-fallback-distros": [
            "suse",
            "rhel",
        ],
        "never-fallback-distros": [
            "debian",
        ],
    },
    "python3-urllib3": {
        "recommended minimum": "1.26.0",
        "reason": "TLSv1.1 is still enabled in older versions",
        "fallback": "urllib3",
        "always-fallback-distros": [
            "suse",
            # Until RHEL 9
            "rhel",
        ],
        "never-fallback-distros": [
            "debian",
        ],
    },
    "python3-validators": {
        "fallback": "validators",
        "always-fallback-distros": [
            "rhel",
        ],
        "never-fallback-distros": [
            "debian",
        ],
    },
    "python3-yaml": {
        "fallback": "PyYAML",
        "distros": [
            "debian",
            "suse",
        ],
        "always-fallback-distros": [
            "suse",
        ],
        "never-fallback-distros": [
            "debian",
        ],
    },
}

PACKAGES_WITHOUT_FALLBACK = {
    "ansible": {
        "distros": [
            "debian",
            "suse",
        ],
    },
    "ansible-core": {
        "distros": [
            "fedora",
            "rhel",
        ],
    },
    "python3-pip": {
        "distros": [
            "debian",
        ],
    },
    "sshpass": {},
}

CONFIGURATION_FILES = [
    (FilePath(os.path.join(os.getcwd(), CMT_CONFIG_FILENAME)), CMT_CONFIG_FILE, 0o644)
]


def create_directories(directories: list[tuple[FilePath, int]], verbose: bool = False) -> None:
    """
    Batch create directories
    If a directory already exists it is silently ignored

        Parameters:
            directories ([(FilePath, int)]): A list of directories (path, permissions) to create
            verbose (bool): Be more verbose
    """
    for directory, permissions in directories:
        secure_mkdir(directory, permissions=permissions, verbose=verbose)


def create_symlinks(symlinks: list[tuple[FilePath, FilePath]], verbose: bool = False) -> None:
    """
    Batch create symlinks
    If a symlink already exists it is *replaced*

        Parameters:
            symlinks ([(FilePath, FilePath)]): A list of symlinks (full path) to create;
                                               (src, dst)
            verbose (bool): Be more verbose
    """
    for src, dst in symlinks:
        try:
            _violations = secure_symlink(src, dst, verbose=verbose, replace_existing=True)
        except PermissionError:
            print()
            ansithemeprint([ANSIThemeStr("Critical", "critical"),
                            ANSIThemeStr(": Could not write to ", "default"),
                            ANSIThemeStr(f"{dst}", "path"),
                            ANSIThemeStr("; permissions or ownership are incorrect; aborting.",
                                         "default")], stderr=True)
            sys.exit(errno.EPERM)


def copy_files(files: list[tuple[FilePath, FilePath, int]], verbose: bool = False) -> None:
    """
    Batch copy files
    If a file already exists it is silently ignored

        Parameters:
            files ([(FilePath, FilePath, int)]): A list of files (full path) to copy;
                                                 (src, dst, permissions)
            verbose (bool): Be more verbose
    """
    for src, dst, permissions in files:
        _violations = secure_copy(src, dst, verbose=verbose, permissions=permissions)


# pylint: disable-next=unused-argument
def install_software_suse(packages: Union[dict, set], verbose: bool = False) -> None:
    """
    Batch install SUSE packages

        Parameters:
            packages (union(dict, set)): A dict or set of packages to install
            verbose (bool): Unused
    """
    try:
        sudo_path = secure_which(FilePath("/usr/bin/sudo"),
                                 fallback_allowlist=[],
                                 security_policy=SecurityPolicy.STRICT)
    except FileNotFoundError:
        ansithemeprint([ANSIThemeStr("Critical", "critical"),
                        ANSIThemeStr(": Could not find ", "default"),
                        ANSIThemeStr("sudo", "path"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    try:
        zypper_path = secure_which(FilePath("/usr/bin/zypper"),
                                   fallback_allowlist=[],
                                   security_policy=SecurityPolicy.STRICT)
    except FileNotFoundError:
        ansithemeprint([ANSIThemeStr("Critical", "critical"),
                        ANSIThemeStr(": Could not find ", "default"),
                        ANSIThemeStr("zypper", "path"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    if isinstance(packages, set):
        packages = dict.fromkeys(packages, {})
    args = [sudo_path, zypper_path, "-n", "install", "-y"] + list(packages.keys())

    _retval = subprocess.run(args, check=False).returncode

    # Check if the installed versions are new enough
    for pkg, data in packages.items():
        recommended_minimum = deep_get(data, DictPath("recommended minimum"), "")
        reason = deep_get(data, DictPath("reason"), "")
        if not recommended_minimum:
            continue

        args = [zypper_path, "info"] + [pkg]
        result = subprocess.run(args, stdout=PIPE, stderr=STDOUT, check=False)
        pkg_version = None
        compiled_regex: re.Pattern[str] = re.compile(r"^.*Version: (.+?)")
        for line in result.stdout.decode("utf-8").splitlines():
            tmp = compiled_regex.match(line)
            if tmp is None:
                continue

            pkg_version = tmp[1]
            break
        if pkg_version is not None:
            args = [zypper_path, "versioncmp", pkg_version, recommended_minimum]
            result = subprocess.run(args, stdout=PIPE, stderr=STDOUT, check=False)
            if "older" in result.stdout.decode("utf-8").splitlines():
                ansithemeprint([ANSIThemeStr("\nWarning:", "warning"),
                                ANSIThemeStr(" The installed version of package ", "default"),
                                ANSIThemeStr(f"{pkg}", "path"),
                                ANSIThemeStr(" (", "default"),
                                ANSIThemeStr(f"{pkg_version}", "version"),
                                ANSIThemeStr(") is older than the recommended "
                                             "minimum version: ", "default"),
                                ANSIThemeStr(f"{recommended_minimum}", "version")], stderr=True)
                ansithemeprint([ANSIThemeStr("   Risk:", "emphasis"),
                                ANSIThemeStr(f" {reason}\n", "default")], stderr=True)


# pylint: disable-next=unused-argument,too-many-locals
def install_software_fedora(packages: Union[dict, set], verbose: bool = False) -> None:
    """
    Batch install Red Hat (Fedora) packages

        Parameters:
            packages (union(dict, set)): A dict or set of packages to install
            verbose (bool): Unused
    """
    try:
        sudo_path = secure_which(FilePath("/usr/bin/sudo"),
                                 fallback_allowlist=[],
                                 security_policy=SecurityPolicy.STRICT)
    except FileNotFoundError:
        ansithemeprint([ANSIThemeStr("Critical", "critical"),
                        ANSIThemeStr(": Could not find ", "default"),
                        ANSIThemeStr("sudo", "path"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    try:
        yum_path = secure_which(FilePath("/usr/bin/yum"),
                                fallback_allowlist=["/usr/bin"],
                                security_policy=SecurityPolicy.ALLOWLIST_RELAXED)
    except FileNotFoundError:
        ansithemeprint([ANSIThemeStr("Critical", "critical"),
                        ANSIThemeStr(": Could not find ", "default"),
                        ANSIThemeStr("yum", "path"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    rpmdev_vercmp_path = None

    try:
        rpmdev_vercmp_path = secure_which(FilePath("/usr/bin/rpmdev-vercmp"),
                                          fallback_allowlist=[],
                                          security_policy=SecurityPolicy.STRICT)
    except FileNotFoundError:
        ansithemeprint([ANSIThemeStr("Warning", "warning"),
                        ANSIThemeStr(": Could not find ", "default"),
                        ANSIThemeStr("rpmdev-vercmp", "path"),
                        ANSIThemeStr("; will not be able to compare package versions.",
                                     "default")], stderr=True)

    if isinstance(packages, set):
        packages = dict.fromkeys(packages, {})
    args = [sudo_path, yum_path, "-y", "install"] + list(packages.keys())

    _retval = subprocess.run(args, check=False).returncode

    # Check if the installed versions are new enough
    for pkg, data in packages.items():
        recommended_minimum = deep_get(data, DictPath("recommended minimum"), "")
        reason = deep_get(data, DictPath("reason"), "")
        if not recommended_minimum:
            continue

        args = [yum_path, "info"] + [pkg]
        result = subprocess.run(args, stdout=PIPE, stderr=STDOUT, check=False)
        pkg_version = None
        compiled_regex: re.Pattern[str] = re.compile(r"^.*Version: (.+?)")
        for line in result.stdout.decode("utf-8").splitlines():
            tmp = compiled_regex.match(line)
            if tmp is None:
                continue

            pkg_version = tmp[1]
            break

        if pkg_version is not None and rpmdev_vercmp_path is not None:
            args = [yum_path, "versioncmp", pkg_version, recommended_minimum]
            result = subprocess.run(args, stdout=PIPE, stderr=STDOUT, check=False)
            if "older" in result.stdout.decode("utf-8").splitlines():
                ansithemeprint([ANSIThemeStr("\nWarning:", "warning"),
                                ANSIThemeStr(" The installed version of package ", "default"),
                                ANSIThemeStr(f"{pkg}", "path"),
                                ANSIThemeStr(" (", "default"),
                                ANSIThemeStr(f"{pkg_version}", "version"),
                                ANSIThemeStr(") is older than the recommended "
                                             "minimum version: ", "default"),
                                ANSIThemeStr(f"{recommended_minimum}", "version")], stderr=True)
                ansithemeprint([ANSIThemeStr("   Risk:", "emphasis"),
                                ANSIThemeStr(f" {reason}\n", "default")], stderr=True)


# pylint: disable-next=unused-argument,too-many-locals
def install_software_deb(packages: dict, verbose: bool = False) -> None:
    """
    Batch install Debian packages

        Parameters:
            packages (union(dict, set)): A dict or set of packages to install
            verbose (bool): Unused
    """
    try:
        sudo_path = secure_which(FilePath("/usr/bin/sudo"),
                                 fallback_allowlist=[],
                                 security_policy=SecurityPolicy.STRICT)
    except FileNotFoundError:
        ansithemeprint([ANSIThemeStr("Critical", "critical"),
                        ANSIThemeStr(": Could not find ", "default"),
                        ANSIThemeStr("sudo", "path"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    try:
        apt_get_path = secure_which(FilePath("/usr/bin/apt-get"),
                                    fallback_allowlist=[],
                                    security_policy=SecurityPolicy.STRICT)
    except FileNotFoundError:
        ansithemeprint([ANSIThemeStr("Critical", "critical"),
                        ANSIThemeStr(": Could not find ", "default"),
                        ANSIThemeStr("apt-get", "path"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    args = [sudo_path, apt_get_path, "install"] + list(packages.keys())

    _retval = subprocess.run(args, check=False).returncode

    try:
        apt_cache_path = secure_which(FilePath("/usr/bin/apt-cache"),
                                      fallback_allowlist=[],
                                      security_policy=SecurityPolicy.STRICT)
    except FileNotFoundError:
        ansithemeprint([ANSIThemeStr("Critical", "critical"),
                        ANSIThemeStr(": Could not find ", "default"),
                        ANSIThemeStr("apt-cache", "path"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    try:
        dpkg_path = secure_which(FilePath("/usr/bin/dpkg"),
                                 fallback_allowlist=[],
                                 security_policy=SecurityPolicy.STRICT)
    except FileNotFoundError:
        ansithemeprint([ANSIThemeStr("Critical", "critical"),
                        ANSIThemeStr(": Could not find ", "default"),
                        ANSIThemeStr("dpkg", "path"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)

    # Check if the installed versions are new enough
    for pkg, data in packages.items():
        recommended_minimum = deep_get(data, DictPath("recommended minimum"), "")
        reason = deep_get(data, DictPath("reason"), "")
        if not recommended_minimum:
            continue
        args = [apt_cache_path, "policy"] + [pkg]
        result = subprocess.run(args, stdout=PIPE, stderr=STDOUT, check=False)
        pkg_version = None
        compiled_regex: re.Pattern[str] = re.compile(r"^  Installed: (.+)")
        for line in result.stdout.decode("utf-8").splitlines():
            tmp = compiled_regex.match(line)
            if tmp is None:
                continue

            pkg_version = tmp[1]
            break
        if pkg_version is not None:
            args = [dpkg_path, "--compare-versions", recommended_minimum, "lt", pkg_version]
            retval = subprocess.run(args, check=False).returncode
            if retval == 1:
                ansithemeprint([ANSIThemeStr("\nWarning:", "warning"),
                                ANSIThemeStr(" The installed version of package ", "default"),
                                ANSIThemeStr(f"{pkg}", "path"),
                                ANSIThemeStr(" (", "default"),
                                ANSIThemeStr(f"{pkg_version}", "version"),
                                ANSIThemeStr(") is older than the recommended "
                                             "minimum version: ", "default"),
                                ANSIThemeStr(f"{recommended_minimum}", "version")], stderr=True)
                ansithemeprint([ANSIThemeStr("   Risk:", "emphasis"),
                                ANSIThemeStr(f" {reason}\n", "default")], stderr=True)


def trim_package_list(packages: dict, distro: str = "", fallback: bool = False) -> dict:
    """
    Given a dict of packages, remove any that shouldn't be installed

        Parameters:
            packages (union(dict, set)): A dict of packages to install
            distro (str): Supported options "debian", "fedora" "rhel", "suse"
            fallback (bool): Is a package list with fallbacks being processed
        Returns:
            trimmed_packages (dict): The trimmed dict of packages
    """
    trimmed_packages: dict = {}

    for pkg, data in packages.items():
        distros = deep_get(data, DictPath("distros"), [])
        never_fallback_distros = deep_get(data, DictPath("never-fallback-distros"), [])

        if distros and distro not in distros:
            continue
        trimmed_packages[pkg] = data
        if never_fallback_distros and distro in never_fallback_distros and fallback:
            trimmed_packages[pkg].pop("fallback")

    return trimmed_packages


def install_software(packages: dict, verbose: bool = False, distro: str = "") -> None:
    """
    Batch install packages

        Parameters:
            packages (union(dict, set)): A dict of packages to install
            verbose (bool): Unused
            distro (str): Supported options "debian", "suse"
    """
    if distro == "debian":
        install_software_deb(packages=packages, verbose=verbose)
    elif distro == "suse":
        install_software_suse(packages=packages, verbose=verbose)
    elif distro in ("fedora", "rhel"):
        install_software_fedora(packages=packages, verbose=verbose)


def install_software_with_pip(packages: list[str],
                              pip_proxy: str = "",
                              verbose: bool = False) -> None:  # pylint: disable=unused-argument
    """
    Installs Python modules using pip

        Parameters:
            packages ([str]): A list of packages to install
            pip_proxy (str): A proxy to use when installing packages
            verbose (bool): Unused
    """
    try:
        pip3_path = secure_which(FilePath("pip3"),
                                 fallback_allowlist=["/bin", "/usr/bin"],
                                 security_policy=SecurityPolicy.ALLOWLIST_RELAXED)
    except FileNotFoundError:
        ansithemeprint([ANSIThemeStr("Critical", "critical"),
                        ANSIThemeStr(": Could not find ", "default"),
                        ANSIThemeStr("pip3", "path"),
                        ANSIThemeStr("; aborting.", "default")], stderr=True)
        sys.exit(errno.ENOENT)
    args = [pip3_path, "install"]
    if pip_proxy:
        args += ["--proxy", pip_proxy]
    args += list(packages)
    _retval = subprocess.run(args, check=False).returncode


# pylint: disable-next=too-many-locals,too-many-statements,too-many-branches
def install_software_with_pip_fallback(packages: dict, pip_proxy: str = "",
                                       verbose: bool = False,
                                       dryrun: bool = False,
                                       distro: str = "") -> tuple[list[str], list[str]]:
    """
    Batch install packages with pip fallback
    This function tries to install a Python module from a package;
    if that package does not exist it installs the pip version of the module instead

        Parameters:
            packages (union(dict, set)): A dict of packages to install
            pip_proxy (str): A proxy to use when installing packages
            verbose (bool): Unused
            dryrun (bool): Simulate a run without installing packages
            distro (str): Supported options "debian", "suse"
        Returns:
            ([str], [str]):
                ([str]): Packages that were, or would be, installed as packages
                ([str]): Packages that were, or would be, installed using pip
    """
    pkgs: dict = {}
    fallbacks: set[str] = set()

    if distro == "debian":
        try:
            apt_cache_path = secure_which(FilePath("/usr/bin/apt-cache"),
                                          fallback_allowlist=[],
                                          security_policy=SecurityPolicy.STRICT)
        except FileNotFoundError:
            ansithemeprint([ANSIThemeStr("Critical", "critical"),
                            ANSIThemeStr(": Could not find ", "default"),
                            ANSIThemeStr("apt-cache", "path"),
                            ANSIThemeStr("; aborting.", "default")], stderr=True)
            sys.exit(errno.ENOENT)

        args = [apt_cache_path, "madison"] + list(packages.keys())
        response = execute_command_with_response(args)
        split_response = response.splitlines()

        for line in split_response:
            _pkg = line.split(' ', 1)
            # If we explicitly specify distros there are different names
            # for the same package in different distros, so skip if irrelevant
            if "distros" in deep_get(packages, DictPath(f"{_pkg[0]}"), {}) and \
                    distro not in deep_get(packages, DictPath(f"{_pkg[0]}#distros"), []):
                continue
            if distro not in deep_get(packages, DictPath(f"{_pkg[0]}#always-fallback-distros"), []):
                if _pkg[0] in packages.keys() and _pkg[0] not in pkgs:
                    pkgs[_pkg[0]] = packages[_pkg[0]]

        for pkg in packages:
            if pkg not in pkgs:
                fallbacks.add(deep_get(packages, DictPath(f"{pkg}#fallback")))
    elif distro == "suse":
        try:
            zypper_path = secure_which(FilePath("/usr/bin/zypper"),
                                       fallback_allowlist=[],
                                       security_policy=SecurityPolicy.STRICT)
        except FileNotFoundError:
            ansithemeprint([ANSIThemeStr("Critical", "critical"),
                            ANSIThemeStr(": Could not find ", "default"),
                            ANSIThemeStr("zypper", "path"),
                            ANSIThemeStr("; aborting.", "default")], stderr=True)
            sys.exit(errno.ENOENT)

        args = [zypper_path, "info"] + list(packages.keys())
        response = execute_command_with_response(args)
        split_response = response.splitlines()

        information_for_regex: re.Pattern[str] = re.compile(r"^Information for package (.+):$")
        not_found_regex: re.Pattern[str] = re.compile(r"^package '([^']+)' not found\.$")

        for line in split_response:
            tmp = information_for_regex.match(line)
            if tmp is not None:
                if distro not in deep_get(packages,
                                          DictPath(f"{tmp[1]}#always-fallback-distros"), []):
                    if tmp[1] in packages.keys() and tmp[1] not in pkgs:
                        pkgs[tmp[1]] = packages[tmp[1]]
                    continue
                fallbacks.add(deep_get(packages, DictPath(f"{tmp[1]}#fallback")))
                continue
            tmp = not_found_regex.match(line)
            if tmp is not None:
                fallbacks.add(deep_get(packages, DictPath(f"{tmp[1]}#fallback")))
    elif distro in ("fedora", "rhel"):
        try:
            yum_path = secure_which(FilePath("/usr/bin/yum"),
                                    fallback_allowlist=["/usr/bin"],
                                    security_policy=SecurityPolicy.ALLOWLIST_RELAXED)
        except FileNotFoundError:
            ansithemeprint([ANSIThemeStr("Critical", "critical"),
                            ANSIThemeStr(": Could not find ", "default"),
                            ANSIThemeStr("yum", "path"),
                            ANSIThemeStr("; aborting.", "default")], stderr=True)
            sys.exit(errno.ENOENT)

        args = [yum_path, "list", "-q"] + list(packages.keys())
        response = execute_command_with_response(args)
        split_response = response.splitlines()

        information_for_regex = re.compile(r"^([^.]+).+$")

        all_packages = set(packages.keys())
        found_packages = set()

        if split_response[0] != "Error: No matching Packages to list":
            for line in split_response:
                if line.startswith(("Available Packages", "Installed Packages")):
                    continue
                tmp = information_for_regex.match(line)

                if tmp is not None:
                    found_packages.add(tmp[1])
                    if distro in deep_get(packages,
                                          DictPath(f"{tmp[1]}#always-fallback-distros"), []):
                        fallbacks.add(deep_get(packages, DictPath(f"{tmp[1]}#fallback")))
                        continue
                    if tmp[1] in packages.keys() and tmp[1] not in pkgs:
                        pkgs[tmp[1]] = packages[tmp[1]]
                    continue
        for missing in all_packages.difference(found_packages):
            fallbacks.add(deep_get(packages, DictPath(f"{missing}#fallback")))

    if pkgs and not dryrun:
        install_software(pkgs, verbose=verbose, distro=distro)

    if fallbacks and not dryrun:
        install_software_with_pip(list(fallbacks), pip_proxy=pip_proxy, verbose=verbose)

    return list(pkgs), list(fallbacks)


# pylint: disable-next=unused-argument,too-many-locals,too-many-statements,too-many-branches
def install(options: list[tuple[str, str]], args: list[str]) -> int:
    """
    Create directories, install configuration files, create symlinks,
    and optionally install packages necessary to run CMT

        Parameters:
            options ([(opt, optarg)]): Options to use when executing this action
            args ([str]): Options to use when executing this action
        Returns:
            (int): 0 on success, errno on failure
    """
    verbose: bool = False
    install_dependencies: bool = True
    confirm: bool = True
    allow_fallback: bool = True
    pip_proxy = ""

    bindir_exists = Path(BINDIR).is_dir()

    for opt, optarg in options:
        if opt == "--verbose":
            verbose = True
        elif opt == "--no-dependencies":
            install_dependencies = False
        elif opt == "--no-fallback":
            allow_fallback = False
        elif opt == "--pip-proxy":
            pip_proxy = optarg
        elif opt == "-Y":
            confirm = False

    # Find out what distro this is run on
    try:
        distro_path = secure_which(FilePath("/etc/os-release"),
                                   fallback_allowlist=["/usr/lib", "/lib"],
                                   security_policy=SecurityPolicy.ALLOWLIST_STRICT,
                                   executable=False)
    except FileNotFoundError:
        ansithemeprint([ANSIThemeStr("Error:", "error"),
                        ANSIThemeStr(" Cannot find an “", "default"),
                        ANSIThemeStr("os-release", "path"),
                        ANSIThemeStr("“ file to determine OS distribution; aborting.",
                                     "default")])
        sys.exit(errno.ENOENT)

    distro = None

    distro_id_like = ""
    distro_id = ""

    with open(distro_path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            key, value = line.split("=")
            value = value.strip("\"'")
            if key == "ID_LIKE":
                distro_id_like = value
                # If there's an ID_LIKE in the file we're done
                break
            if key == "ID":
                distro_id = value
                # But if we've only found an ID we cannot be sure
                # that there won't be an ID_LIKE later on

    if distro_id_like:
        distro = distro_id_like
    else:
        distro = distro_id

    if distro is None or not distro:
        ansithemeprint([ANSIThemeStr("Error:", "error"),
                        ANSIThemeStr(" Cannot read ID_LIKE from “", "default"),
                        ANSIThemeStr("os-release", "path"),
                        ANSIThemeStr("“ file to determine OS distribution; aborting.",
                                     "default")])
        sys.exit(errno.ENOENT)

    if distro == "suse opensuse":
        distro = "suse"

    if distro not in ("debian", "suse", "rhel", "fedora"):
        ansithemeprint([ANSIThemeStr("Error:", "error"),
                        ANSIThemeStr(" distro “", "default"),
                        ANSIThemeStr(f"{distro}", "path"),
                        ANSIThemeStr("“ is not supported.", "default")])
        ansithemeprint([ANSIThemeStr("Currently only distros based on ", "default"),
                        ANSIThemeStr("Debian", "programname"),
                        ANSIThemeStr(", ", "default"),
                        ANSIThemeStr("Red Hat", "programname"),
                        ANSIThemeStr(", or ", "default"),
                        ANSIThemeStr("SUSE", "programname"),
                        ANSIThemeStr(" are supported.", "default")])
        sys.exit(errno.ENOTSUP)

    if distro == "suse":
        ansithemeprint([ANSIThemeStr("Note:", "note"),
                        ANSIThemeStr(" on SUSE distros older than OpenSUSE 16 "
                                     "you need to install “", "default"),
                        ANSIThemeStr("pip", "programname"),
                        ANSIThemeStr("“ manually;", "default")])
        ansithemeprint([ANSIThemeStr("depending on your SP-level this may be, for instance, “",
                                     "default"),
                        ANSIThemeStr("python310-pip", "programname"),
                        ANSIThemeStr("“.\n", "default")])
    elif distro in ("fedora", "rhel"):
        ansithemeprint([ANSIThemeStr("Note:", "note"),
                        ANSIThemeStr(" on Red Hat distros older than RHEL 9 "
                                     "you need to install a newer version", "default")])
        ansithemeprint([ANSIThemeStr("of python3 and set it as the default version "
                                     "(using ", "default"),
                        ANSIThemeStr("update-alternatives", "programname"),
                        ANSIThemeStr(").\n", "default")])

    pkgs_without_fallback: dict = trim_package_list(packages=PACKAGES_WITHOUT_FALLBACK,
                                                    distro=distro)
    pkgs_with_pip_fallback: dict = trim_package_list(packages=PACKAGES_WITH_PIP_FALLBACK,
                                                     distro=distro, fallback=True)

    if confirm:
        ansithemeprint([ANSIThemeStr("Important:", "warning"),
                        ANSIThemeStr(" The following actions will be taken "
                                     "that will modify the system:", "default")])
        pkgs: list[str] = []
        pkgs_fallback: list[str] = []

        if install_dependencies:
            pkgs, pkgs_fallback = install_software_with_pip_fallback(pkgs_with_pip_fallback,
                                                                     dryrun=True, distro=distro)
            ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                            ANSIThemeStr("Install the following system software "
                                         "(+ any dependencies):", "action")])
            ansithemeprint([ANSIThemeStr("  Note", "warning"),
                            ANSIThemeStr(": This requires sudo permissions", "emphasis")])
            for pkg in pkgs_without_fallback:
                ansithemeprint([ANSIThemeStr(f"    {pkg}", "path")])
            for pkg in pkgs:
                ansithemeprint([ANSIThemeStr(f"    {pkg}", "path")])
            for pkg in pkgs_fallback:
                ansithemeprint([ANSIThemeStr(f"    {pkg}", "path"),
                                ANSIThemeStr(" (Using package from PIP)", "emphasis")])

        ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                        ANSIThemeStr("Create the following new directories:", "action")])
        for name, _permissions in DIRECTORIES:
            ansithemeprint([ANSIThemeStr(f"    {name}", "path")])
        ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                        ANSIThemeStr("Install the following configuration files:", "action")])
        for src, dst, _permissions in CONFIGURATION_FILES:
            ansithemeprint([ANSIThemeStr(f"     {dst}", "path")])
        ansithemeprint([ANSIThemeStr("\n• ", "separator"),
                        ANSIThemeStr("Create the following symlinks:", "action")])
        for src, dst in MISC_SYMLINKS:
            ansithemeprint([ANSIThemeStr(f"    {src}", "path"),
                            ANSIThemeStr(" ⇨ ", "emphasis"),
                            ANSIThemeStr(f"{dst}", "path")])
        for src, dst in EXECUTABLE_SYMLINKS:
            ansithemeprint([ANSIThemeStr(f"    {src}", "path"),
                            ANSIThemeStr(" ⇨ ", "emphasis"),
                            ANSIThemeStr(f"{dst}", "path")])

        if install_dependencies and not allow_fallback and pkgs_fallback:
            ansithemeprint([ANSIThemeStr("\n", "default"),
                            ANSIThemeStr("Error", "error"),
                            ANSIThemeStr(": Installation would require fallback to PIP, but “",
                                         "default"),
                            ANSIThemeStr("--no-fallback", "option"),
                            ANSIThemeStr("“ was specified; aborting.", "default")], stderr=True)
            sys.exit(errno.EINVAL)

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

    if install_dependencies:
        install_software(pkgs_without_fallback, verbose=verbose, distro=distro)
        install_software_with_pip_fallback(pkgs_with_pip_fallback,
                                           verbose=verbose, pip_proxy=pip_proxy, distro=distro)

    create_directories(directories=DIRECTORIES, verbose=verbose)
    copy_files(files=CONFIGURATION_FILES, verbose=verbose)
    create_symlinks(symlinks=MISC_SYMLINKS + EXECUTABLE_SYMLINKS, verbose=verbose)

    if not bindir_exists:
        ansithemeprint([ANSIThemeStr("\nWarning", "warning"),
                        ANSIThemeStr(": ", "default"),
                        ANSIThemeStr(f"{BINDIR}", "path"),
                        ANSIThemeStr(" did not exist before installation; "
                                     "you will most likely need to ", "default"),
                        ANSIThemeStr("logout and login again (or spawn a new "
                                     "shell/terminal) to have it added to ", "default"),
                        ANSIThemeStr("$PATH", "argument"),
                        ANSIThemeStr(".", "default")])

    return 0


COMMANDLINE: dict[str, Any] = {
    # Default command
    "__global_options": {
        "command": ["__global_options"],
        "description": [
            ANSIThemeStr("", "")],
        "options": {
            "--no-dependencies": {
                "description": [ANSIThemeStr("Do not install dependencies", "description")],
            },
            "--no-fallback": {
                "description": [
                    ANSIThemeStr("Do not fallback to Python packages from PIP", "description")],
                "extended_description": [
                    [ANSIThemeStr("If a distribution package cannot be found", "description")],
                    [ANSIThemeStr("cmt-install", "programname"),
                     ANSIThemeStr(" will, by default, install packages", "description")],
                    [ANSIThemeStr("using PIP.", "description")],
                    [ANSIThemeStr("This option can be used to disable that behaviour",
                                  "description")],
                ],
            },
            "--pip-proxy": {
                "values": [
                    ANSIThemeStr("PROXY", "argument")],
                "description": [
                    ANSIThemeStr("Proxy to use for PIP", "description")],
                "extended_description": [
                    [ANSIThemeStr("HTTPS proxy to use if fallback packages", "description")],
                    [ANSIThemeStr("are installed from PIP. Format:", "description")],
                    [ANSIThemeStr("[user:passwd@]proxy:port", "argument")],
                ],
                "requires_arg": True,
                "validation": {
                    "validator": "url",
                },
            },
            "--verbose": {
                "description": [
                    ANSIThemeStr("Be more verbose", "description")],
            },
            "-Y": {
                "description": [
                    ANSIThemeStr("Do not ask for confirmation", "description")],
            },
        },
    },
    "__default": {
        "command": ["__default"],
        "description": [
            ANSIThemeStr("", "")],
        "callback": install,
    },
    "spacer1": {
        "command": [""],
        "description": [
            ANSIThemeStr("", "")],
    },
}


def main() -> int:
    """
    Main function for the program

        Returns:
           (int): 0 on success, non-zero on failure
    """
    # 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.")

    homepath = Path(HOMEDIR)
    path_stat = homepath.stat()
    path_permissions = path_stat.st_mode & 0o777

    if path_permissions & 0o002:
        sys.exit(f"CRITICAL: The destination directory {HOMEDIR} is world-writable; aborting.")

    command, options, args = \
        parse_commandline(about.INSTALL_PROGRAM_NAME, about.INSTALL_PROGRAM_VERSION,
                          PROGRAMDESCRIPTION, PROGRAMAUTHORS, sys.argv, COMMANDLINE,
                          default_command="__default",
                          theme=FilePath(os.path.join(os.getcwd(), "themes", "default.yaml")))

    return command(options, args)


if __name__ == "__main__":
    main()