ꦫꦣꦺꦤ꧀ꦄꦤ꧀ꦠꦱꦺꦤ

PATH: opt / imunify360 / venv / lib / python3.11 / site-packages / im360 / subsys / features

import base64
import configparser
import logging
import os
import re
import shlex
import distro
from abc import abstractmethod
from typing import List, Optional

from defence360agent.contracts.config import Core, Packaging
from defence360agent.contracts.license import LicenseCLN
from defence360agent.subsys.features.abstract_feature import (
    AbstractFeature,
    FeatureError,
    FeatureStatus,
    ea4_only,
)
from defence360agent.subsys.panels.cpanel import cPanel
from defence360agent.utils import (
    OsReleaseInfo,
    check_run,
    run,
    run_cmd_and_log,
    os_version,
)

logger = logging.getLogger(__name__)

_ELS_SETUP_TMPL = (
    "export PATH=/opt/imunify360/venv/bin:$PATH;"
    " _els_tmp=$(mktemp)"
    ' && curl -sf -o "$_els_tmp" {url}'
    ' && sh "$_els_tmp" -i;'
    ' rm -f "$_els_tmp";'
)
_ELS_RPM_SETUP = _ELS_SETUP_TMPL.format(
    url=(
        "https://repo.alt.tuxcare.com/alt-php-els/"
        "install-els-alt-php-rpm-repo.sh"
    )
)
_ELS_DEB_SETUP = _ELS_SETUP_TMPL.format(
    url=(
        "https://repo.alt.tuxcare.com/alt-php-els/"
        "install-els-alt-php-deb-repo.sh"
    )
)

class SimpleInstallerMixIn:
    """This is a mixin class implementing common case installation scenario.

Installation is supposed to be through a single command cls.INSTALL_CMD.

Removal is done through interpolating a space separated list of package
    names to remove into cls.REMOVE_CMD_TMPL. List of packages to remove is
    obtained by collecting all installed alt-php* packages except those we want
    to keep (as returned by required_packages()).
    """

INSTALL_CMD = "/bin/false"
    REMOVE_CMD_TMPL = "/bin/false"

@abstractmethod
    def generate_repo(self, enabled: Optional[bool] = None):
        return

@abstractmethod
    async def pre_install_cmd(self, enabled: bool):
        return

@abstractmethod
    def remove_repo(self):
        return

@staticmethod
    @abstractmethod
    async def _list_alt_php_packages() -> set:
        """Set of installed package names matching alt-php*"""
        return set()

@classmethod
    def _keep_installed(cls, pkg):
        # this packages should not be managed by this class, as required by
        # Imunify360 to work. Should be updated every time major php version
        # used by ai-bolit is updated
        return (
            pkg.startswith("alt-php-internal")
            or pkg == "alt-php-config"
            or pkg == "alt-php-hyperscan"
        )

@classmethod
    async def _feature_packages(cls) -> set:
        """Set of installed alt-php packages except those we keep installed"""
        all_alt_php = await cls._list_alt_php_packages()
        return set(pkg for pkg in all_alt_php if not cls._keep_installed(pkg))

@AbstractFeature.raise_if_shouldnt_install_now
    async def install(self):
        self.generate_repo(enabled=True)
        await self.pre_install_cmd(enabled=True)
        return await run_cmd_and_log(
            self.INSTALL_CMD, self.INSTALL_LOG_FILE_MASK
        )

@AbstractFeature.raise_if_shouldnt_remove_now
    async def remove(self):
        self.remove_repo()
        cmd = self.REMOVE_CMD_TMPL.format(
            " ".join(map(shlex.quote, await self._feature_packages()))
        )
        await self.pre_install_cmd(enabled=False)
        return await run_cmd_and_log(cmd, self.REMOVE_LOG_FILE_MASK)

async def _check_installed_impl(self) -> bool:
        return bool(await self._feature_packages())

class HardenedPHPCentos(SimpleInstallerMixIn, AbstractFeature):
    LEGACY_REPO_FILE = "/etc/yum.repos.d/imunify360-alt-php.repo"
    NAME = "Hardened-PHP"
    LOG_DIR = "/var/log/%s" % Core.PRODUCT
    INSTALL_LOG_FILE_MASK = "%s/install-hardenedphp.log.*" % LOG_DIR
    REMOVE_LOG_FILE_MASK = "%s/remove-hardenedphp.log.*" % LOG_DIR
    INSTALL_CMD = (  # noqa: E501
        _ELS_RPM_SETUP
        + " yum group mark remove alt-php; yum -y groupinstall alt-php"
    )
    REMOVE_CMD_TMPL = "yum group mark install alt-php; yum -y remove {}"
    ENABLE_CRB_CMD = "dnf config-manager --enable crb"
    DISABLE_CRB_CMD = "dnf config-manager --disable crb"
    _CMD_LIST = [INSTALL_CMD, REMOVE_CMD_TMPL.format("")]

def _remove_legacy_repo(self):
        try:
            os.remove(self.LEGACY_REPO_FILE)
        except FileNotFoundError:
            pass
        except OSError:
            logger.error("Can't delete %s", self.LEGACY_REPO_FILE)

def generate_repo(self, enabled: Optional[bool] = None):
        if enabled is None:
            # License update path — ELS repos are token-free, nothing to
            # regenerate.  Leave the legacy repo in place until the user
            # explicitly installs/removes the feature (which runs INSTALL_CMD
            # and sets up the ELS replacement).
            return
        self._remove_legacy_repo()

async def pre_install_cmd(self, enabled: bool):
        if not os_version().startswith("9"):
            return
        elif enabled:
            await check_run(self.ENABLE_CRB_CMD.split())
        else:
            await check_run(self.DISABLE_CRB_CMD.split())

def remove_repo(self):
        self._remove_legacy_repo()

@staticmethod
    async def _list_alt_php_packages() -> set:
        raw_output = await check_run(
            ["rpm", "-qa", "--queryformat", "%{NAME}\n", "alt-php*"]
        )
        return set(raw_output.decode().split())

class HardenedPHPUbuntu(SimpleInstallerMixIn, AbstractFeature):
    NAME = "Hardened-PHP"
    LOG_DIR = "/var/log/%s" % Core.PRODUCT
    INSTALL_LOG_FILE_MASK = "%s/install-hardenedphp.log.*" % LOG_DIR
    REMOVE_LOG_FILE_MASK = "%s/remove-hardenedphp.log.*" % LOG_DIR
    INSTALL_CMD = _ELS_DEB_SETUP + " apt-get install -y alt-php"
    REMOVE_CMD_TMPL = "apt-get purge -y {}"
    _CMD_LIST = [INSTALL_CMD, REMOVE_CMD_TMPL.format("")]

def generate_repo(self, enabled: Optional[bool] = None):
        return

async def pre_install_cmd(self, enabled: bool):
        return

def remove_repo(self):
        return

@staticmethod
    async def _list_alt_php_packages() -> set:
        pkgs_in_dpkg_db = (
            (
                await check_run(
                    [
                        "dpkg-query",
                        "-W",
                        "-f",
                        "${Package} ${db:Status-Status}\n",
                        "alt-php*",
                    ]
                )
            )
            .decode()
            .strip()
            .split("\n")
        )
        return set(
            pkg
            for line in pkgs_in_dpkg_db
            for pkg, status in [line.split()]
            if status == "installed"
        )

class HardenedPHPCloudLinux(AbstractFeature):
    MSG = "HardenedPHP is managed by lvemanager in CloudLinuxOS"
    INSTALL_LOG_FILE_MASK = "empty"
    REMOVE_LOG_FILE_MASK = "empty"

async def init(self):
        return self

async def status(self):
        rc, _, _ = await run(["rpm", "-q", "lvemanager"])
        return {
            "items": {
                "status": FeatureStatus.MANAGED_BY_LVE,
                "lve_installed": rc == 0,
                "message": self.MSG,
            }
        }

async def install(self):
        raise FeatureError(self.MSG)

async def remove(self):
        raise FeatureError(self.MSG)

async def _check_installed_impl(self) -> bool:
        # does not matter
        return True

class HardenedPHPCloudLinuxSolo(HardenedPHPCloudLinux):
    MSG = "HardenedPHP is not supported in CloudLinuxOS Solo"

async def status(self):
        return {
            "items": {
                "status": FeatureStatus.NOT_SUPPORTED_BY_CL_SOLO,
                "message": self.MSG,
            }
        }

class EaPHPCentos(HardenedPHPCentos):
    REPO_FILE = "/etc/yum.repos.d/imunify360-ea-php-hardened.repo"
    LOG_DIR = "/var/log/%s" % Core.PRODUCT
    INSTALL_LOG_FILE_MASK = "%s/install-ea_php.log.*" % LOG_DIR
    REMOVE_LOG_FILE_MASK = "%s/remove-ea_php.log.*" % LOG_DIR
    INSTALL_CMD = "yum -y groupremove ea-php; yum -y groupinstall ea-php"

REMOVE_SCRIPT = "/opt/imunify360/venv/share/imunify360/scripts/remove_hardened_php.py"  # noqa: E501
    REPO_NAME = "imunify360-ea-php-hardened"
    _CMD_LIST = [INSTALL_CMD, REMOVE_SCRIPT]

@classmethod
    def _repo_tmpl_filepath(cls):
        return os.path.join(Packaging.DATADIR, os.path.basename(cls.REPO_FILE))

@classmethod
    def _prepare_token(cls, token):
        try:
            sep = ":"
            fields = "".join(
                str(token[k]) + sep for k in LicenseCLN.VERIFY_FIELDS_V1
            )
        except KeyError as e:
            raise FeatureError(
                f"License token can not be created by error {e}"
            )
        sign_bytes = base64.b64decode(token["sign"])
        data = fields.encode() + sign_bytes
        return base64.urlsafe_b64encode(data).decode()

@classmethod
    def _prepare_repo_conf(cls, token, enabled: bool):
        enabled_flag = "1" if enabled else "0"
        try:
            token = cls._prepare_token(token)
        except FeatureError as e:
            if not enabled:
                token_placeholder = "unregister-token-placeholder"
                token = base64.urlsafe_b64encode(
                    token_placeholder.encode()
                ).decode()
            else:
                raise e

with open(cls._repo_tmpl_filepath(), "r") as repo_template:
            template = repo_template.read()
            return template.format(token=token, enabled=enabled_flag)

def generate_repo(self, enabled: Optional[bool] = None):
        if enabled is None:
            # called on CLN license update
            repo = configparser.ConfigParser()
            try:
                repo.read(self.REPO_FILE)
                enabled = repo[self.REPO_NAME]["enabled"] == "1"
            except Exception:
                enabled = True
        token = LicenseCLN.get_token()
        server_id = LicenseCLN.get_server_id()
        if not server_id:
            if enabled:
                raise FeatureError(
                    "tried to enable repo but server_id is empty (not"
                    " registered?)"
                )
            logger.warning(
                "server_id is empty (not registered?) ignoring due to removal"
                " of repo"
            )
        with open(self.REPO_FILE, "w") as repo_file:
            repo_file.write(self._prepare_repo_conf(token, enabled))
        os.chmod(self.REPO_FILE, os.stat(self._repo_tmpl_filepath()).st_mode)

def remove_repo(self):
        self.generate_repo(enabled=False)

@staticmethod
    async def _query_eaphp_versions() -> List[dict]:
        raw_output = await check_run(
            'rpm -qa --queryformat "%{NAME} %{RELEASE}\n" "ea-php*"',
            shell=True,
        )
        words = raw_output.decode().split()
        return [
            {"name": words[i], "release": words[i + 1]}
            for i in range(0, len(words), 2)
        ]

async def _check_installed_impl(self) -> bool:
        versioned_re = re.compile(r"ea-php\d+")
        for pkg in await self._query_eaphp_versions():
            if (
                versioned_re.search(pkg["name"]) is not None
                and "cloudlinux" in pkg["release"]
            ):
                return True
        return False

@ea4_only
    async def status(self):
        return await super().status()

@ea4_only
    @AbstractFeature.raise_if_shouldnt_install_now
    async def install(self):
        self.generate_repo(enabled=True)
        return await run_cmd_and_log(
            self.INSTALL_CMD, self.INSTALL_LOG_FILE_MASK
        )

@ea4_only
    @AbstractFeature.raise_if_shouldnt_remove_now
    async def remove(self):
        self.generate_repo(enabled=False)
        return await run_cmd_and_log(
            self.REMOVE_SCRIPT, self.REMOVE_LOG_FILE_MASK
        )

_CPANEL_HARDENED_PHP_MSG = (
    "For EL9/Ubuntu20 cpanel servers use cPanel Profile to configure harden"
    " php.\nMore info:\n\t"
    " https://docs.cpanel.net/ea4/basics/the-ea-cpanel-tools-package-scripts/\n\t"  # noqa: E501
    " https://docs.cpanel.net/whm/software/easyapache-4-interface/"
)

class EaPHPCentosEL9(EaPHPCentos):
    MSG = _CPANEL_HARDENED_PHP_MSG

@ea4_only
    @AbstractFeature.raise_if_shouldnt_install_now
    async def install(self):
        self.generate_repo(enabled=True)
        return "Repo imunify360-ea-php-hardened activated.\n" + self.MSG

@ea4_only
    @AbstractFeature.raise_if_shouldnt_remove_now
    async def remove(self):
        self.generate_repo(enabled=False)
        return "Repo imunify360-ea-php-hardened removed.\n" + self.MSG

async def _check_installed_impl(self) -> bool:
        repo = configparser.ConfigParser()
        try:
            repo.read(self.REPO_FILE)
            enabled = repo[self.REPO_NAME]["enabled"] == "1"
        except Exception:
            enabled = False
        return enabled

class EaPHPUbuntuCpanel(AbstractFeature):
    REPO_FILE = "/etc/apt/sources.list.d/cloudlinux-ea4.list"
    REPO_URL = (
        "https://repo.cloudlinux.com/cloudlinux-ubuntu/cloudlinux-ea4/stable/"
    )
    REPO_NAME = "cloudlinux-ea4"
    MSG = _CPANEL_HARDENED_PHP_MSG
    INSTALL_LOG_FILE_MASK = "empty"
    REMOVE_LOG_FILE_MASK = "empty"

def _generate_repo(self):
        with open(self.REPO_FILE, "w") as f:
            f.write(f"deb {self.REPO_URL} ./\n")

def _remove_repo(self):
        try:
            os.remove(self.REPO_FILE)
        except FileNotFoundError:
            pass
        except OSError:
            logger.error("Can't delete %s", self.REPO_FILE)

@ea4_only
    @AbstractFeature.raise_if_shouldnt_install_now
    async def install(self):
        self._generate_repo()
        return f"Repo {self.REPO_NAME} activated.\n" + self.MSG

@ea4_only
    @AbstractFeature.raise_if_shouldnt_remove_now
    async def remove(self):
        self._remove_repo()
        return f"Repo {self.REPO_NAME} removed.\n" + self.MSG

@ea4_only
    async def status(self):
        return await super().status()

async def _check_installed_impl(self) -> bool:
        return os.path.isfile(self.REPO_FILE)

def _is_cloudlinux_release_installed() -> bool:
    return os.path.exists("/etc/cloudlinux-release")

def get_hardened_php_feature() -> Optional[AbstractFeature]:
    """
    :return: AbstractFeature subclass: feature that implements Hardened PHP
             installation for current environment.
    """
    has_cpanel = cPanel.is_installed()
    if (
        OsReleaseInfo.is_centos()
        or OsReleaseInfo.is_rhel()
        or OsReleaseInfo.is_oracle_linux()
        or OsReleaseInfo.is_almalinux()
        or OsReleaseInfo.is_rockylinux()
    ):
        if has_cpanel and int(distro.major_version()) >= 9:
            return EaPHPCentosEL9
        elif has_cpanel:
            return EaPHPCentos
        else:
            return HardenedPHPCentos
    if OsReleaseInfo.is_cloudlinux():
        if OsReleaseInfo.is_cloudlinux_solo():
            return HardenedPHPCloudLinuxSolo
        # CL regular
        return HardenedPHPCloudLinux
    if OsReleaseInfo.is_ubuntu() or OsReleaseInfo.is_debian():
        if has_cpanel:
            if _is_cloudlinux_release_installed():
                return HardenedPHPCloudLinux
            return EaPHPUbuntuCpanel
        return HardenedPHPUbuntu
    return None

[-] hardened_php.py [edit]
[+] __pycache__
[-] __init__.py [edit]
[+] ..