Na}{

Path: ~//opt/alt/python37/lib/python3.7/site-packages/

File Content: `cpanel.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/LICENSE.TXT

from __future__ import absolute_import

import json
import os
import pwd
import re
import subprocess
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
from typing import Optional, Dict, Tuple, Union, List
from pkg_resources import parse_version

from clcommon.cpapi import userdomains
from clcommon.clwpos_lib import find_wp_paths, get_wp_cache_plugin
from clcommon.lib.cledition import is_cl_solo_edition
from secureio import write_file_via_tempfile

from clwpos.constants import (
    WP_CLI_EXTENSIONS,
    WP_CLI_ENABLE_ALL_FUNCTIONS,
    RedisRequiredConstants,
    EA_PHP_PREFIX,
    CAGEFSCTL,
    CLSOP_ZIP_PATH
)
from clwpos.cl_wpos_exceptions import WposError, PhpBrokenException, WpCliCommandError
from clwpos.daemon import WposDaemon
from clwpos.logsetup import setup_logging, ADMIN_LOGFILE_PATH, USER_LOGFILE_PATH
from clwpos import gettext as _


from clwpos.utils import (
    WposUser,
    is_run_under_user,
    check_domain,
    home_dir,
    clear_redis_cache_config,
    create_redis_cache_config,
    daemon_communicate,
    PHP,
    wp_cli_compatibility_check,
    run_in_cagefs_if_needed,
    create_pid_file,
    user_name,
    user_uid,
    litespeed_is_running
)

_logger = setup_logging(__name__)

BASE_CPANEL_EA_PHP_DIR = '/opt/cpanel'


def _get_php_handler(vhost: str) -> str:
    result = uapi("php_get_domain_handler", {"type": "vhost", "vhost": vhost})
    return result["php_handler"]


def _get_doc_roots_info() -> dict:
    user = pwd.getpwuid(os.geteuid()).pw_name
    result = {}
    for domain, doc_root in userdomains(user):
        result.setdefault(doc_root, []).append(domain)

    return result


def _add_wp_path_info(user_info: dict) -> dict:
    wp_paths = {}
    for doc_root, domains in user_info.items():
        # excludes only affects subpaths of doc_root
        excludes = list(user_info)
        item = {
            "domains": domains,
            "wp_paths": list(find_wp_paths(doc_root, excludes=excludes))
        }
        wp_paths[doc_root] = item
    return wp_paths


def _wp_info(doc_root: str, wp_path: str) -> dict:
    """Convert WP path to {"path": str, "version": str}"""
    absolute_wp_path = Path(doc_root, wp_path)
    version_file = list(absolute_wp_path.glob("wp-includes/version.php"))[0]
    result = subprocess.run(["/bin/grep", "-Po", "(?<=wp_version = ')[^']+", version_file], capture_output=True)
    wp_version = result.stdout.strip().decode()
    return {
        "path": wp_path,
        "version": wp_version,
    }


def _add_wp_info(user_info: dict) -> dict:
    for doc_root, doc_root_info in user_info.items():
        wp_paths = doc_root_info.pop("wp_paths")
        doc_root_info["wps"] = [_wp_info(doc_root, wp_path) for wp_path in wp_paths]
    return user_info


def _get_data_from_info_json(attribute: str) -> List:
    """
    Return attribute's value from info.json file.
    """
    from clwpos.feature_suites import get_admin_config_directory

    admin_config_dir = get_admin_config_directory(user_uid())
    info_json = os.path.join(admin_config_dir, "info.json")

    try:
        with open(info_json) as f:
            return json.load(f)[attribute]
    except (OSError, KeyError, json.JSONDecodeError) as e:
        _logger.exception("Error during reading of \"info.json\" file: %s", e)
        return []


def _php_get_vhost_versions():
    """
    @return: [
        {
            "account": "rm3",
            "account_owner": "root",
            "documentroot": "/home/example/public_html",
            "homedir": "/home/rm3",
            "is_suspended": 0,
            "main_domain": 1,
            "php_fpm": 1,
            "php_fpm_pool_parms": {
                "pm_max_children": 5,
                "pm_max_requests": 20,
                "pm_process_idle_timeout": 10
            },
            "phpversion_source": [
                {
                    "domain": "example.com",
                    "system_default": 1
                }
            ],
            "version": "ea-php72",
            "vhost": "otherchars.rm3.tld"
        }
    ]
    """
    try:
        return daemon_communicate({"command": WposDaemon.DAEMON_PHP_GET_VHOST_VERSIONS_COMMAND})["data"]
    except WposError:
        return _get_data_from_info_json("vhost_versions")


def _php_get_installed_versions():
    """
    @return: [
        "ea-php74"
    ]
    """
    try:
        return daemon_communicate({"command": WposDaemon.DAEMON_PHP_GET_INSTALLED_VERSIONS_COMMAND})["data"]
    except WposError:
        return _get_data_from_info_json("installed_versions")


def php_info():
    """
    Returns php info, example:
    [{'vhost': 'sub.wposuser.com', 'account': 'stackoverflow',
    'phpversion_source': {'domain': 'sub.wposuser.com'},
    'version': 'ea-php80', 'account_owner': 'root', 'php_fpm': 1,
    'php_fpm_pool_parms': {'pm_process_idle_timeout': 10,
    'pm_max_requests': 20, 'pm_max_children': 5},
    'main_domain': 0, 'documentroot': '/home/stackoverflow/public_html',
    'homedir': '/home/stackoverflow'},
    ...................................................................]
    """
    result = _php_get_vhost_versions()
    for elem in result:
        elem["version"] = _normalized_php_version(PHP(elem["version"]))
    return result


def ea_php_ini_file_path(ini_name: str, php_version: str):
    """
    Builds path to <ini_name>.ini file
    """
    return Path(PHP(php_version).dir()).joinpath(f'root/etc/php.d/{ini_name}')


def get_supported_ea_php():
    """
    Looks through /opt/cpanel and gets installed phps
    """
    base_dir = Path(BASE_CPANEL_EA_PHP_DIR)
    minimal_supported = parse_version('ea-php74')
    supported = []
    for item in os.listdir(base_dir):
        if item.startswith('ea-php') and parse_version(item) >= minimal_supported:
            supported.append(item)
    return supported


def configure_redis_extension_for_ea():
    """
    Sets up redis if needed:
     - installing package
     - enables in .ini file
    """
    need_cagefs_update = False
    wait_child_process = bool(os.environ.get('CL_WPOS_WAIT_CHILD_PROCESS'))
    php_versions_redis_data = {
        php: _redis_extension_info(PHP(php)) for php in get_supported_ea_php()
    }
    php_versions_to_enable_redis = [
        php for php, redis_data in php_versions_redis_data.items()
        if not redis_data.get('is_present') or not redis_data.get('is_loaded')
    ]
    if not php_versions_to_enable_redis:
        return

    with create_pid_file(EA_PHP_PREFIX):
        for php in php_versions_to_enable_redis:
            redis_data = php_versions_redis_data.get(php)
            if not redis_data.get('is_present'):
                redis_package = f'{php}-php-redis'
                result = subprocess.run(['yum', '-y', 'install', redis_package],
                                        capture_output=True,
                                        text=True)
                if result.returncode != 0 and 'Nothing to do' not in result.stdout:
                    _logger.error('Failed to install package %s, due to reason: %s', redis_package,
                                  f'{result.stdout}\n{result.stderr}')
                    continue
                enable_redis_extension_for_ea(php)
                need_cagefs_update = True
            elif not redis_data.get('is_loaded'):
                enable_redis_extension_for_ea(php)
                need_cagefs_update = True

        if need_cagefs_update and wait_child_process and os.path.isfile(CAGEFSCTL):
            try:
                subprocess.run([CAGEFSCTL, '--check-cagefs-initialized'],
                               stdout=subprocess.DEVNULL,
                               stderr=subprocess.DEVNULL,
                               check=True)
            except subprocess.CalledProcessError:
                _logger.info('CageFS in unintialized, skipping force-update')
            else:
                subprocess.run([CAGEFSCTL, '--wait-lock', '--force-update'],
                               stdout=subprocess.DEVNULL,
                               stderr=subprocess.DEVNULL)


def enable_redis_extension_for_ea(php_version):
    """
    Enables (if needed) redis extension in .ini config
    """
    path = ea_php_ini_file_path('50-redis.ini', php_version)
    keyword = 'redis.so'
    if not os.path.exists(path):
        _logger.error('Redis extension config: %s is not found, ensure corresponding rpm package installed: %s',
                      str(path), f'{php_version}-php-redis')
        return
    with open(path) as f:
        extension_data = f.readlines()

    uncommented_pattern = re.compile(fr'^\s*extension\s*=\s*{keyword}')
    commented_pattern = re.compile(fr'^\s*;\s*extension\s*=\s*{keyword}')
    enabled_line = f'extension = {keyword}\n'
    was_enabled = False
    lines = []

    for line in extension_data:
        if uncommented_pattern.match(line):
            return
        if not was_enabled and commented_pattern.match(line):
            lines.append(enabled_line)
            was_enabled = True
        else:
            lines.append(line)
    if not was_enabled:
        lines.append(enabled_line)
    write_file_via_tempfile(''.join(lines), path, 0o644)


@lru_cache()
def _redis_extension_info(version: PHP) -> dict:
    is_present = bool(list(version.dir().glob("**/redis.so")))
    php_bin_path = version.bin()
    if os.geteuid() == 0:
        exec_func = subprocess.run
    else:
        exec_func = run_in_cagefs_if_needed

    is_loaded = exec_func(
        f'{php_bin_path} -m | /bin/grep redis', shell=True, executable='/bin/bash', env={}
    ).returncode == 0 if is_present else False

    return {
        "is_present": is_present,
        "is_loaded": is_loaded
    }


def _add_php(user_info: dict) -> dict:
    """
    Updates user_info dict with php data
    """
    result = php_info()
    for item in result:
        user_info[item["documentroot"]]["php"] = {
            "version": item["version"],
            "fpm": bool(item["php_fpm"]),
            "handler": _get_php_handler(item["vhost"])
        }

    return user_info


def _add_object_cache_info(user_info: dict) -> dict:
    """
    Search for 'object-cache.php' files in 'wp-content/plugins' directory
    in order to find what plugin is being used for object caching.
    """
    for doc_root, doc_root_info in user_info.items():
        for wp in doc_root_info["wps"]:
            plugin = get_wp_cache_plugin(Path(doc_root).joinpath(wp["path"]), "object-cache")
            wp["object_cache"] = plugin

    return user_info


def get_user_info() -> dict:
    """
    Collect info about user.
    @return {
        '/home/user/public_html': {
            'domains': ['domain.com'],
            'wps': [
                {
                    'path': 'wp_path_1',
                    'version': '5.7.2',
                    'object_cache': 'redis-cache'
                }
            ],
            'php': {
                'version': 'ea-php74',
                'handler': 'cgi',
                'redis_extension': False,
                'fpm': True
            }
        }
    }
    """
    user_info = _get_doc_roots_info()

    for func in (_add_wp_path_info, _add_wp_info, _add_php):
        user_info = func(user_info)

    return user_info


def _get_php_version(abs_wp_path: str) -> PHP:
    """Return PHP version."""
    result = php_info()

    items = []
    for item in result:
        if abs_wp_path.startswith(item["documentroot"]):
            items.append((item["documentroot"], item["version"]))
    items.sort(reverse=True)
    return items[0][1]


def _normalized_php_version(version: PHP) -> PHP:
    """
    PHP selector can replace path with symlink. It's a reason why we need normalization.
    """
    if not is_cl_solo_edition(skip_jwt_check=True):
        command = f"{version.bin()} -i " \
                  f" | /bin/grep 'Loaded Configuration File'" \
                  f" | /bin/grep -oE \"(alt|ea).*php[^/]*/\""
        result = run_in_cagefs_if_needed(command, shell=True, executable='/bin/bash', env={})

        if result.stderr and not result.stdout:
            raise PhpBrokenException(str(version.bin()), result.stderr)

        return PHP(result.stdout.strip().strip("/").replace("/", "-"))

    return version


def filter_php_versions_with_not_loaded_redis(php_versions: List[PHP]) -> List[PHP]:
    """
    Filter list of given php versions to find out
    for which redis extension is presented but not loaded.
    """
    php_versions_with_not_loaded_redis = []
    for version in php_versions:
        php_redis_info = _redis_extension_info(version)
        if not php_redis_info['is_loaded'] and php_redis_info['is_present']:
            php_versions_with_not_loaded_redis.append(version)
    return php_versions_with_not_loaded_redis


@lru_cache(maxsize=None)
def get_cached_php_installed_versions() -> List[PHP]:
    """
    List all installed php version on the system
    :return: installed php version
    """
    result = _php_get_installed_versions()
    return [PHP(version) for version in result]


@lru_cache(maxsize=None)
def get_cached_php_versions_with_redis_loaded() -> set:
    """
    List all installed php version on the system which has redis-extension enabled
    :return: installed php versions which has redis-extension
    """
    versions = get_cached_php_installed_versions()
    return {version for version in versions if _redis_extension_info(version)["is_loaded"]}


@lru_cache(maxsize=None)
def get_cached_php_versions_with_redis_present() -> set:
    """
    List all installed php version on the system which has redis-extension installed
    :return: installed php versions which has redis-extension installed
    """
    versions = get_cached_php_installed_versions()
    return {version for version in versions if _redis_extension_info(version)["is_present"]}


def uapi(function: str, input_parameters: Optional[Dict[str, str]] = None):
    input_parameters_as_list = [f"{key}={value}" for key, value in input_parameters.items()] if input_parameters else []
    result = run_in_cagefs_if_needed(
        ["/usr/bin/uapi", "--output=json", "LangPHP", function, *input_parameters_as_list], env={}
    )

    return json.loads(result.stdout)["result"]["data"]


def is_multisite(path: str) -> bool:
    marker = 'cl_multisite_detected'
    command = 'if ( is_multisite() ) { echo "%s"; }' % marker
    result = wordpress(path, 'eval', command)
    if isinstance(result, WordpressError):
        raise WposError(message=result.message, context=result.context)
    return marker in result


def wp_get_constant(wp_path: str, constant: str, raise_exception=False) -> Optional[str]:
    """
    Get:
    - defined constant value
    - None in case of error
    - empty string if no such constant found
    """
    command = "if (defined('%(const)s')) { echo %(const)s; }" % {'const': constant}
    result = wordpress(wp_path, 'eval', command)
    if isinstance(result, WordpressError):
        if raise_exception:
            raise WpCliCommandError(message=result.message, context=result.context)
        _logger.error('Error during get WP constant: %s', result)
        return None
    return result


def diagnose_redis_connection_constants(docroot: str, wordpress_path: str):
    """
    Check required constants for redis connection establishment
    """
    redis_schema = wp_get_constant(os.path.join(docroot, wordpress_path), RedisRequiredConstants.WP_REDIS_SCHEME.name,
                                   raise_exception=True)
    if not redis_schema and redis_schema != RedisRequiredConstants.WP_REDIS_SCHEME.value:
        raise WposError('WordPress constant "%(constant)s" is not defined or defined with wrong value %(value)s',
                        context={'constant': RedisRequiredConstants.WP_REDIS_SCHEME.name, 'value': redis_schema})
    socket = wp_get_constant(os.path.join(docroot, wordpress_path), RedisRequiredConstants.WP_REDIS_PATH.name,
                             raise_exception=True)
    if not socket:
        raise WposError('WordPress constant "%(constant)s" is not defined',
                        context={'constant': RedisRequiredConstants.WP_REDIS_PATH.name})
    if not os.path.exists(socket):
        raise WposError('Redis socket %(socket)s does not exist in the system',
                        context={'socket': socket})


def _get_saved_php_version(abs_wp_path: str) -> Optional[PHP]:
    """
    Get domain's php version from a previously saved file.
    """
    if not is_run_under_user():
        raise WposError('Internal Error. Contact CloudLinux support')
    php_file_id = os.environ.get("CLWPOS_PHP_FILE_ID")
    php_info_file = WposUser(user_name()).php_info.format(file_id=php_file_id)
    if not os.path.exists(php_info_file):
        return None
    try:
        with open(php_info_file) as f:
            _php_info = json.load(f)
    except (OSError, json.decoder.JSONDecodeError) as e:
        _logger.exception("Error during reading of \".php_info\" file: %s", e)
        return None

    php_versions = []
    for vhost_info in _php_info:
        if abs_wp_path.startswith(vhost_info["documentroot"]):
            php_versions.append((vhost_info["documentroot"], vhost_info["version"]))

    if not php_versions:
        return None

    return PHP(sorted(php_versions, reverse=True)[0][1])


def get_php_version(abs_wp_path: str) -> PHP:
    """
    Return php_version that will be used for calling wp-cli commands.
    If 'CLWPOS_USE_SAVED_PHP_VERSION' envar is defined,
    try to get this version from a previously saved file.
    """
    use_saved_php_version = bool(os.environ.get("CLWPOS_USE_SAVED_PHP_VERSION"))
    if use_saved_php_version:
        php_version = _get_saved_php_version(abs_wp_path) or _get_php_version(abs_wp_path)
    else:
        php_version = _get_php_version(abs_wp_path)
    return php_version


@dataclass
class WordpressError:
    message: str
    context: dict


def wordpress(path: str, command: str, subcommand: str, *args, env=None) -> Union[str, WordpressError]:
    """
    Helper to execute wp commands, for example
        wp --path=<path> plugin install redis-cache
        wp --path=<path> plugin activate redis-cache
        wp --path=<path> redis enable
        wp --path=<path> plugin deactivate redis-cache
        wp --path=<path> plugin uninstall redis-cache
    @return: stderr if error was happened.
    """
    php_version = get_php_version(path)
    php_bin_path = str(php_version.bin())
    if not os.path.exists(php_bin_path):
        _logger.exception("Error during wp-cli command execution \"%s\": "
                          "invalid path to binary file \"%s\"",
                          command, php_bin_path)
        return WordpressError(
            message=_("Error during resolving path to php binary file:\n"
                      "got non-existent path \"%(path)s\"."),
            context={"path": php_bin_path}
        )
    # [attention] compatibility check may raise WpCliUnsupportedException exception
    wp_cli_compatibility_check(php_bin_path)
    command_part = ["--path={}".format(path), command, subcommand, *args]
    full_command = [
        php_bin_path,
        *WP_CLI_EXTENSIONS,
        # explicitly drop PHP disable_functions directive
        # in order to avoid errors like
        # 'Error: Cannot do 'launch': The PHP functions `proc_open()` and/or `proc_close()` are disabled'
        # during plugin manipulations
        WP_CLI_ENABLE_ALL_FUNCTIONS,
        "/opt/clwpos/wp-cli",
        *command_part
    ]
    environment = env or {}
    try:
        output = run_in_cagefs_if_needed(full_command, check=True, env=environment)
    except subprocess.CalledProcessError as error:
        command = ' '.join(full_command)
        _logger.exception("Error during command execution: \n%s\n"
                          "stdout=%s\n"
                          "stderr=%s",
                          command, error.stdout, error.stderr)
        logger_path = ADMIN_LOGFILE_PATH if not os.getuid() else USER_LOGFILE_PATH.format(homedir='HOME')
        return WordpressError(
            message=_("Unexpected error happened during command execution: '%(command)s'.\n"
                      "Event is logged to file: '%(logger_path)s' with stdout and stderr recorded."),
            context={
                "command": command,
                "logger_path": logger_path
            }
        )

    return output.stdout


class DocRootPath(str):
    """This class represent path to doc_root."""
    pass


class DomainName(str):
    """This class represent domain name."""
    pass


def disable_without_config_affecting(
        arg: Union[DocRootPath, DomainName], wp_path: str, *, module: str,
) -> Optional[WordpressError]:
    """
    Deactivate and delete specified wordpress module.
    :param arg: user's docroot or domain
    :param wp_path: path to user's wordpress directory
    :param module: module on which to perform disable operations
    :return: error if error was happened else None
    """
    if isinstance(arg, DomainName):
        doc_root = check_domain(arg)[-1]
    elif isinstance(arg, DocRootPath):
        doc_root = Path(home_dir(), arg)
    else:
        raise ValueError("Invalid argument format")

    abs_wp_path = str(Path(doc_root).joinpath(wp_path).absolute())
    last_error = None

    errors = disable_module(module, abs_wp_path)
    if errors:
        last_error = errors[-1]

    return last_error


def enable_without_config_affecting(
        arg: Union[DocRootPath, DomainName], wp_path: str, *, module: str,
) -> Tuple[bool, Dict[str, Union[str, dict]]]:
    """
    Install and activate specified wordpress module.
    :param arg: user's docroot or domain
    :param wp_path: path to user's wordpress directory
    :param module: module on which to perform enable operations
    :return: tuple that consists of enabling status and details
    """
    if isinstance(arg, DomainName):
        __, doc_root = check_domain(arg)
    elif isinstance(arg, DocRootPath):
        doc_root = Path(home_dir(), arg)
    else:
        raise ValueError("Invalid argument format")

    wp_path = wp_path.lstrip("/")
    abs_wp_path = str(Path(doc_root).joinpath(wp_path).absolute())

    # try to install plugin
    try:
        install_module(module, abs_wp_path)
    except WposError as e:
        return False, dict(
            message=_("WordPress plugin installation failed. "
                      "Try again and contact your system administrator if issue persists."),
            details=e.message,
            context=e.context
        )

    # try to activate plugin
    try:
        enable_module(module, abs_wp_path)
    except WposError as e:
        disable_module(module, abs_wp_path)
        return False, dict(
            message=_("WordPress plugin activation failed. Changes were reverted and caching module is now disabled. "
                      "Try again and contact your system administrator if issue persists."),
            details=e.message,
            context=e.context
        )

    return True, {}


def reload_redis(uid: int = None):
    """
    Make redis reload via CLWPOS daemon
    :param uid: User uid (optional)
    """
    cmd_dict = {"command": "reload"}
    if uid:
        cmd_dict['uid'] = uid
    daemon_communicate(cmd_dict)


def enable_module(module: str, abs_wp_path: str):
    """
    Enable specified WP optimization feature.
    """
    from clwpos.optimization_features import OBJECT_CACHE_FEATURE, SITE_OPTIMIZATION_FEATURE

    module_func_map = {
        OBJECT_CACHE_FEATURE: enable_redis_object_cache,
        SITE_OPTIMIZATION_FEATURE: enable_site_optimization_plugin,
    }
    module_func_map[module](abs_wp_path)


def enable_redis_object_cache(abs_wp_path: str):
    """
    Enable redis-cache plugin for user.
    :param abs_wp_path: absolute path to wp site
    :return:
    """
    res = wordpress(abs_wp_path, "plugin", "activate", "redis-cache")
    if isinstance(res, WordpressError):
        raise WposError(message=res.message, context=res.context)

    res = wordpress(abs_wp_path, "redis", "enable")
    if isinstance(res, WordpressError):
        raise WposError(message=res.message, context=res.context)
    create_redis_cache_config(abs_wp_path)


def obtain_wp_cli_env(abs_wp_path: str):
    """
    Returns needed envars for wp-cli
    """
    env = None
    if os.path.exists(os.path.join(abs_wp_path, '.htaccess')):
        server_software = 'LiteSpeed' if litespeed_is_running() else 'Apache'
        env = {'SERVER_SOFTWARE': server_software}
    return env


def enable_site_optimization_plugin(abs_wp_path: str):
    """
    Enable cloudlinux-site-optimization plugin for user.
    :param abs_wp_path: absolute path to wp site
    :return:
    """
    env = obtain_wp_cli_env(abs_wp_path)
    res = wordpress(abs_wp_path, "plugin", "activate", "clsop", env=env)
    if isinstance(res, WordpressError):
        raise WposError(message=res.message, context=res.context)


def install_module(module: str, abs_wp_path: str):
    """
    Install specified WP optimization feature.
    """
    from clwpos.optimization_features import OBJECT_CACHE_FEATURE, SITE_OPTIMIZATION_FEATURE

    module_func_map = {
        OBJECT_CACHE_FEATURE: install_redis_cache,
        SITE_OPTIMIZATION_FEATURE: install_site_optimization_plugin,
    }
    module_func_map[module](abs_wp_path)


def install_redis_cache(abs_wp_path: str):
    """
    Install redis-cache plugin for user.
    :param abs_wp_path: absolute path to wp site
    :return:
    """
    res = wordpress(abs_wp_path, "plugin", "install", "redis-cache")
    if isinstance(res, WordpressError):
        raise WposError(message=res.message, context=res.context)


def install_site_optimization_plugin(abs_wp_path: str):
    """
    Install cloudlinux-site-optimization plugin for user.
    :param abs_wp_path: absolute path to wp site
    :return:
    """
    env = obtain_wp_cli_env(abs_wp_path)
    res = wordpress(abs_wp_path, "plugin", "install", CLSOP_ZIP_PATH, env=env)
    if isinstance(res, WordpressError):
        raise WposError(message=res.message, context=res.context)


def disable_module(module: str, abs_wp_path: str) -> List[WordpressError]:
    """
    Deactivate and delete specified WP optimization feature.
    """
    from clwpos.optimization_features import OBJECT_CACHE_FEATURE, SITE_OPTIMIZATION_FEATURE

    module_func_map = {
        OBJECT_CACHE_FEATURE: disable_object_cache,
        SITE_OPTIMIZATION_FEATURE: disable_site_optimization_plugin,
    }
    return module_func_map[module](abs_wp_path)


def disable_object_cache(abs_wp_path: str) -> List[WordpressError]:
    """
    Delete cloudlinux info from wp-config.php,
    deactivate and delete redis-cache plugin for user.
    :param abs_wp_path: absolute path to wp site
    :return: list of errors that occurred during command execution
    """
    errors = []
    if is_plugin_activated(abs_wp_path, 'redis-cache'):
        res = wordpress(abs_wp_path, "plugin", "deactivate", "redis-cache")
        if isinstance(res, WordpressError):
            errors.append(res)

    if not errors and is_plugin_installed(abs_wp_path, 'redis-cache'):
        # continue procedure further only if previous step succeeded
        res = wordpress(abs_wp_path, "plugin", "delete", "redis-cache")
        if isinstance(res, WordpressError):
            errors.append(res)

    if not errors:
        # cleanup constants in the end only if deactivation/deletion succeeded,
        # because it may impact on deactivating/deleting plugin
        try:
            clear_redis_cache_config(abs_wp_path)
        except WposError as err:
            _logger.exception(err)
            errors.append(WordpressError(err.message, err.context))
        except Exception as e:
            _logger.exception(e)
            errors.append(
                WordpressError(
                    message=_('Unexpected error happened while clearing cache: %(error)s'),
                    context=dict(error=str(e)))
            )

    return errors


def is_plugin_activated(abs_wp_path: str, plugin_name: str, wp_cli_env=None) -> bool:
    result = wordpress(abs_wp_path, "plugin", "is-active", plugin_name, env=wp_cli_env)
    return not isinstance(result, WordpressError)


def is_plugin_installed(abs_wp_path: str, plugin_name: str, wp_cli_env=None) -> bool:
    result = wordpress(abs_wp_path, "plugin", "is-installed", plugin_name, env=wp_cli_env)
    return not isinstance(result, WordpressError)


def disable_site_optimization_plugin(abs_wp_path: str) -> List[WordpressError]:
    """
    Deactivate and delete cloudlinux-site-optimization plugin for user.
    :param abs_wp_path: absolute path to wp site
    :return: list of errors that occurred during command execution
    """
    errors = []

    env = obtain_wp_cli_env(abs_wp_path)

    if is_plugin_activated(abs_wp_path, 'clsop', env):

        res = wordpress(abs_wp_path, "plugin", "deactivate", "clsop", env=env)
        if isinstance(res, WordpressError):
            errors.append(res)

    if not errors and is_plugin_installed(abs_wp_path, 'clsop', env):
        # continue procedure further only if previous step succeeded
        res = wordpress(abs_wp_path, "plugin", "uninstall", "clsop", env=env)
        if isinstance(res, WordpressError):
            errors.append(res)

    return errors

Edit Rename Chmod Delete

FILE FOLDER

Name	Size	Permission
Babel-2.9.1-py3.7.egg-info	---	0755
Beaker-1.11.0-py3.7.egg-info	---	0755
Jinja2-3.0.3-py3.7.egg-info	---	0755
Mako-1.1.0-py3.7.egg-info	---	0755
OpenSSL	---	0755
Paste-1.7.5.1-py3.7.egg-info	---	0755
PyJWT-1.7.1-py3.7.egg-info	---	0755
Tempita-0.5.1-py3.7.egg-info	---	0755
__pycache__	---	0755
_distutils_hack	---	0755
aiosignal	---	0755
aiosignal-1.2.0a0-py3.7.egg-info	---	0755
alembic	---	0755
alembic-0.8.3-py3.7.egg-info	---	0755
asn1crypto	---	0755
asn1crypto-0.22.0-py3.7.egg-info	---	0755
async_timeout	---	0755
async_timeout-4.0.2-py3.7.egg-info	---	0755
attr	---	0755
attrs-17.4.0-py3.7.egg-info	---	0755
babel	---	0755
beaker	---	0755
beautifulsoup4-4.5.1-py3.7.egg-info	---	0755
bs4	---	0755
certifi	---	0755
certifi-2018.4.16-py3.7.egg-info	---	0755
chardet	---	0755
chardet-3.0.4-py3.7.egg-info	---	0755
charset_normalizer	---	0755
charset_normalizer-2.0.12-py3.7.egg-info	---	0755
cl_dom_collector	---	0755
clcommon	---	0755
clconfig	---	0755
clconfigure	---	0755
cldashboard	---	0755
clevents	---	0755
cllicense	---	0755
cllimits	---	0755
cllimits_validator	---	0755
cllimitslib_v2	---	0755
clpackages	---	0755
clquota	---	0755
clselect	---	0755
clselector	---	0755
clsentry	---	0755
clsummary	---	0755
clveconfig	---	0755
clwizard	---	0755
clwpos	---	0755
cssselect	---	0755
cssselect-0.9.2-py3.7.egg-info	---	0755
distlib	---	0755
docopt-0.6.2-py3.7.egg-info	---	0755
filelock	---	0755
filelock-0.0.0-py3.7.egg-info	---	0755
future	---	0755
future-0.17.0-py3.7.egg-info	---	0755
html5lib	---	0755
html5lib-1.0.1-py3.7.egg-info	---	0755
idna	---	0755
idna-2.5-py3.7.egg-info	---	0755
importlib_metadata	---	0755
importlib_metadata-0.0.0-py3.7.egg-info	---	0755
jinja2	---	0755
jsonschema	---	0755
jsonschema-3.2.0-py3.7.egg-info	---	0755
jwt	---	0755
libfuturize	---	0755
libpasteurize	---	0755
lve_utils	---	0755
lvemanager	---	0755
lvestats	---	0755
mako	---	0755
nose	---	0755
nose-1.3.7-py3.7.egg-info	---	0755
packaging	---	0755
packaging-16.8-py3.7.egg-info	---	0755
past	---	0755
paste	---	0755
pip	---	0755
pip-20.2.4.dist-info	---	0755
pkg_resources	---	0755
platformdirs	---	0755
platformdirs-0.0.0-py3.7.egg-info	---	0755
ply	---	0755
ply-3.8-py3.7.egg-info	---	0755
prettytable-0.7.2-py3.7.egg-info	---	0755
pyOpenSSL-17.3.0-py3.7.egg-info	---	0755
pycparser	---	0755
pycparser-2.14-py3.7.egg-info	---	0755
pyparsing-2.1.10-py3.7.egg-info	---	0755
python_editor-0.4-py3.7.egg-info	---	0755
pytz	---	0755
pytz-2017.2-py3.7.egg-info	---	0755
raven	---	0755
raven-6.3.0-py3.7.egg-info	---	0755
requests	---	0755
requests-2.26.0-py3.7.egg-info	---	0755
schema-0.7.1-py3.7.egg-info	---	0755
sentry_sdk	---	0755
sentry_sdk-0.19.4-py3.7.egg-info	---	0755
setuptools	---	0755
setuptools-58.3.0.dist-info	---	0755
six-1.15.0-py3.7.egg-info	---	0755
ssa	---	0755
svgwrite	---	0755
svgwrite-1.3.0-py3.7.egg-info	---	0755
tempita	---	0755
typing_extensions-3.7.4.3-py3.7.egg-info	---	0755
urllib3	---	0755
urllib3-1.26.6-py3.7.egg-info	---	0755
vendors_api	---	0755
virtualenv	---	0755
virtualenv-20.13.0-py3.7.egg-info	---	0755
webencodings	---	0755
webencodings-0.5.1-py3.7.egg-info	---	0755
wmt	---	0755
zipp-0.0.0-py3.7.egg-info	---	0755
Paste-1.7.5.1-py3.7-nspkg.pth	534 bytes	0644
PySocks-1.5.7-py3.7.egg-info	322 bytes	0644
cl_proc_hidepid.py	4549 bytes	0644
clcontrollib.py	53127 bytes	0644
cldetectlib.py	18745 bytes	0644
cldiaglib.py	48675 bytes	0644
clhooklib.py	1432 bytes	0644
cli_utils.py	1775 bytes	0644
cllicenselib.py	8220 bytes	0644
clsetuplib.py	4422 bytes	0644
clsudo.py	15049 bytes	0644
cpanel.py	29687 bytes	0644
distlib-0.3.4-py3.7.egg-info	1162 bytes	0644
distutils-precedence.pth	152 bytes	0644
docopt.py	19946 bytes	0644
editor.py	2550 bytes	0755
lve_stats-2.0-py3.7.egg-info	185 bytes	0644
lveapi.py	19697 bytes	0644
lvectllib.py	94545 bytes	0644
lvestat.py	7106 bytes	0644
prettytable.py	54204 bytes	0644
pyparsing.py	229867 bytes	0644
remove_ubc.py	5670 bytes	0755
schema.py	29198 bytes	0644
secureio.py	17387 bytes	0644
six.py	34159 bytes	0644
socks.py	29952 bytes	0644
sockshandler.py	2913 bytes	0644
typing_extensions.py	83727 bytes	0644
zipp.py	8425 bytes	0644

File Content: cpanel.py

File Content: `cpanel.py`