SWAMPy

✎ CARVE FLESH

EDITING: base.py

# coding=utf-8
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENCE.TXT
#
import os
import re
from abc import abstractmethod
from typing import Dict, List, Optional  # NOQA

from clcommon import cpapi
from clcommon.group_info_reader import GroupInfoReader
from clcommon.utils import (
    ExternalProgramFailed,
    get_cl_version,
    get_package_db_errors,
    get_passenger_package_name,
    is_ea4,
    is_package_installed,
    is_ubuntu,
    run_command,
)

from clwizard.exceptions import InstallationFailedException
from clwizard.utils import convert_package_version, setup_logger

class WizardInstaller:
    """
    General interface for modules installers
    """

LOG_FILE = ""

_REQUIRED_CL_COMPONENT_SUPPORT = None

# Keyring shipped by cloudlinux-release. CloudLinux signs with more than one key - the EA4
    # release RPM uses the Packager key up to el9 and the Extension Signer key on el10 - and the
    # keyring is per-major, carrying the key(s) its own major's artifact needs rather than every
    # key (el7 7.9-1 Packager only, el8 8.10-15 and el9 9.8-1 both, el10 10-5 Extension Signer
    # only). Updating cloudlinux-release is what widens it, so the recovery for a signature
    # rejected because the host's keyring predates the key it was signed with is that update plus
    # an import of this file. On el7 the build carrying both keys is currently only in the
    # cloudlinux-updates-testing repo, which ships disabled, so that update needs --enablerepo.
    _CL_RPM_GPG_KEY = "/etc/pki/rpm-gpg/RPM-GPG-KEY-CloudLinux"

# Lowercased substrings yum/dnf/rpm print when a signature cannot be verified against the
    # rpmdb keyring: "Public key for <file> is not installed" and "Package <file> is not signed"
    # on yum 3 and dnf 4, dnf 4's aggregate "Error: GPG check FAILED", and rpm's own "key ID
    # <id>: NOKEY". No shipped CloudLinux major runs dnf5 - CL7 is yum 3 and CL8/CL9/CL10 are all
    # dnf 4 - but these same substrings also cover dnf5's "The package is not signed."/"the key is
    # not trusted.", so they stay as forward compatibility for free. That last one keeps its "key
    # is" qualifier: a bare "not trusted" also matches curl's TLS trust errors, which name a
    # certificate issuer rather than a signing key and which no keyring import can repair.
    _GPG_FAILURE_MARKERS = ("public key", "not signed", "key is not trusted", "nokey", "gpg check failed")

def __init__(self):
        self.app_logger = setup_logger("wizard." + self.__class__.__name__, self.LOG_FILE)

def _run_command(self, cmd, cmd_env=None):
        # type: (List[str], Dict[str, str]) -> str
        """Run external tool and log results"""
        self.app_logger.info("~" * 60)
        self.app_logger.info("Executing command %s...", " ".join(cmd))
        try:
            output = run_command(cmd, env_data=cmd_env)
        except ExternalProgramFailed as err:
            self.app_logger.info("...external command failed, see the following lines for tracebacks, errors, etc.")
            self.app_logger.error(str(err))
            raise
        self.app_logger.info("...external command successfully ended with output:")
        self.app_logger.info(output)
        self.app_logger.info("-" * 60)
        return output

def _is_package_installed(self, package_name):
        # type: (str) -> bool
        """Check if package exists on server"""
        error_message = get_package_db_errors()
        if error_message:
            # DB corrupted
            log_message = f"Can't check package {package_name} presence. Errors:\n{error_message}"
            self.app_logger.error(log_message)
            raise InstallationFailedException()
        return is_package_installed(package_name)

def _install_package(self, *packages):
        # type: (str) -> Optional[str]
        """Install a package and log results"""
        if is_ubuntu():
            return self._install_apt_package(*packages)
        return self._install_yum_package(*packages)

def _install_apt_package(self, *packages):
        # type: (str) -> Optional[str]
        """Install an apt package and log results"""

apt_env = {}
        # For debconf to avoid making any interactive queries.
        apt_env["DEBIAN_FRONTEND"] = "noninteractive"
        apt_env["DEBCONF_NONINTERACTIVE_SEEN"] = "true"

if packages:
            return self._run_command(["apt-get", "install", "-y"] + list(packages), cmd_env=apt_env)
        return None

def _install_yum_package(self, *packages):
        # type: (str) -> Optional[str]
        """Install a yum package and log results"""
        if packages:
            # A package given as a URL or a local file is gated on localpkg_gpgcheck, not on the
            # repo-level gpgcheck, and it defaults to off - force it so an RPM whose signature
            # does not verify is rejected before rpm runs its scriptlets as root.
            try:
                return self._run_command(["yum", "install", "-y", "--setopt=localpkg_gpgcheck=1"] + list(packages))
            except ExternalProgramFailed as err:
                # Stay fail-closed, but do not leave a rejected signature undiagnosable: a
                # command-line package has no repo gpgkey= for yum to auto-import from, so name
                # the keyring the admin has to import instead of only logging yum's output.
                if any(marker in str(err).lower() for marker in self._GPG_FAILURE_MARKERS):
                    self.app_logger.error(
                        "Signature verification failed for %s, so nothing was installed. If this host's "
                        "keyring predates the key CloudLinux signed the package with, update the "
                        "cloudlinux-release package and import the keys it ships with "
                        "'rpm --import %s', then retry.",
                        ", ".join(packages),
                        self._CL_RPM_GPG_KEY,
                    )
                raise
        return None

def _install_groups(self, *groups):
        # type: (str) -> Optional[str]
        """
        Install a package group.
        On Ubuntu, this will install the meta packages.
        Currently, we only support PHP Selector on Ubuntu.
        If other selectors are added, they will need to have a meta packages.
        """
        if is_ubuntu():
            groups = [group + "-meta" for group in groups]
            return self._install_apt_package(*groups)
        return self._install_yum_groups(*groups)

def _install_yum_groups(self, *groups):
        # type: (str) -> Optional[str]
        """Install package group with yum and log results"""
        if groups:
            return self._run_command(["yum", "groupinstall", "-y"] + list(groups))
        return None

def _ensure_cl_ea4_repo_exists(self):
        """
        Check whether cloudlinux-ea4.repo is present and install it if not.
        It's required only on EA4 to install proper Passenger package
        """
        # CLOS-2147: should also be made compatible with Ubuntu, once Python/Node/Ruby selector
        # support comes into play.
        # As of 02.11.2023, only PHP Selector is supposed to be supported.
        cl_ea4_repofile = "/etc/yum.repos.d/cloudlinux-ea4.repo"
        if os.path.exists(cl_ea4_repofile):
            return

dist = get_cl_version() or "Unknown"
        if "ubuntu" in dist:
            raise InstallationFailedException("Functionality currently not supported for Ubuntu")

# From "cl7h" to "7", "cl9" to "9", etc.
        maj_ver_match = re.search(r"\d+", dist)
        if not maj_ver_match:
            raise InstallationFailedException("Failed to get the distribution's major version")

dist_maj_ver = int(maj_ver_match.group())

# NOTE(rprilipskii): We already have a Debian repo for EA4:
        # https://1.mirror.g.cdn.mycache.org/cloudlinux-ubuntu/cloudlinux-ea4/stable/20.04/
        # but there's no cloudlinux-ea4-release in them.
        # In CLOS-2147 see how we're supposed to ensure that the cloudlinux-ubuntu/cloudlinux-ea4
        # repo is enabled on the system.
        package_url = (
            "https://1.mirror.g.cdn.mycache.org/cloudlinux/EA4/" f"cloudlinux-ea4-release-latest-{dist_maj_ver}.noarch.rpm"
        )
        self.app_logger.info("Unable to find cloudlinux-ea4 repo. Trying to install it using url: %s", package_url)
        self._install_yum_package(package_url)

def _install_passenger(self):
        """
        Install proper passenger package for Selectors if it's not yet
        installed
        """
        # CLOS-2147: should also be made compatible with Ubuntu, once Python/Node/Ruby selector
        # support comes into play.
        # As of 02.11.2023, only PHP Selector is supposed to be supported.
        # Per Alexander Demeshko, Passenger is not supported on Ubuntu at the moment, todo later.
        if is_ubuntu():
            raise InstallationFailedException("Passenger currently not supported on Ubuntu")
        if is_ea4():
            self._ensure_cl_ea4_repo_exists()

passenger = get_passenger_package_name()
        self.app_logger.info("Trying to install Passenger package: %s", passenger)
        try:
            self._install_yum_package(passenger)
        except ExternalProgramFailed as extern_failed:
            raise InstallationFailedException() from extern_failed

@staticmethod
    def _get_available_versions(group):
        # type: (str) -> List
        """
        Get a list of available Python, NodeJS, PHP or Ruby module versions.
        :param group: - group name, e.g: python
        :return: list of available versions
        """
        available_groups_info = GroupInfoReader.get_group_info(group)
        versions = []
        # we need to format version:
        # python: 2.3.3 -> 2.3
        # ruby: 2.1.1 -> 2.1
        # php: 7.2.0 -> 7.2
        # nodejs: 8.1.0 -> 8
        ver_size = 1 if group == "nodejs" else 2
        for group_data in available_groups_info.values():
            versions.append(convert_package_version(group_data["version"], version_size=ver_size))
        return versions

@abstractmethod
    def run_installation(self, options):
        # type: (Dict) -> None
        raise NotImplementedError()

@classmethod
    def supported_options(cls):
        return set()

@abstractmethod
    def initial_status(self):
        # type: () -> Dict
        """
        Method that returns dictionary with two required keys:
        {installed: True|False and options: {...}]
        This will be used by lvemanager to properly display wizard.
        """
        raise NotImplementedError()

@classmethod
    def is_supported_by_control_panel(cls):
        """
        Checks whether we must
        :return:
        """
        if cls._REQUIRED_CL_COMPONENT_SUPPORT is None:
            return True
        return cpapi.is_panel_feature_supported(cls._REQUIRED_CL_COMPONENT_SUPPORT)

CANCEL

Name	Type	Size	Modified	Actions
↩ ..	DIR	—	—
📁 __pycache__/	DIR	—	2026-11-02 22:59
📄 __init__.py	PY	5.2 KB	2026-10-16 11:34	EDIT
📄 base.py	PY	10.9 KB	2026-10-16 11:34	EDIT
📄 cagefs.py	PY	3.9 KB	2026-10-16 11:34	EDIT
📄 governor.py	PY	6.2 KB	2026-10-16 11:34	EDIT
📄 lsapi.py	PY	8.3 KB	2026-10-16 11:34	EDIT
📄 nodejs.py	PY	5.3 KB	2026-10-16 11:34	EDIT
📄 php.py	PY	7.7 KB	2026-10-16 11:34	EDIT
📄 python.py	PY	5.6 KB	2026-10-16 11:34	EDIT
📄 ruby.py	PY	4 KB	2026-10-16 11:34	EDIT