Yu1709$!%yuyu!$%wutu!5=126+

                
____  ___       .__                 
\   \/  /  _____|  |   ______  _  __
 \     /  /  ___/  |  /  _ \ \/ \/ /
 /     \  \___ \|  |_(  <_> )     / 
/___/\  \/____  >____/\____/ \/\_/  
      \_/     \/    @ligmakokz

Editing: wpos_user.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

# wpos_user.py - work code for clwpos-user utility

from __future__ import absolute_import

import argparse
import json
import os
from enum import Enum

from typing import Dict, List, Tuple

from pkg_resources import parse_version
from dataclasses import dataclass, asdict, field

from clcommon.clpwd import drop_user_privileges
from clcommon.utils import is_litespeed_running

import cpanel
from clwpos.cl_wpos_exceptions import WposError
from clwpos.parse import ArgumentParser, CustomFormatter
from clwpos.optimization_modules import OBJECT_CACHE_MODULE, get_allowed_modules, ALL_OPTIMIZATION_MODULES
from clwpos.user.config import UserConfig
from clwpos.user.redis_lib import RedisLibUser
from clwpos import gettext as _

from clwpos.utils import (
    USER_WPOS_DIR,
    check_license_decorator,
    print_data,
    catch_error,
    error_and_exit,
    daemon_communicate,
    home_dir,
    user_name,
    check_domain,
    is_run_under_user,
    is_conflict_modules_installed,
    supported_php_handlers
)

from clwpos.logsetup import setup_logging, USER_LOGFILE_PATH, init_wpos_sentry_safely
from clwpos import constants

from clwpos.user.website_check import post_site_check, RollbackException
from cpanel import wordpress, WordpressError
from lve_utils.sentry import init_lve_utils_sentry_client
from cldetectlib import is_cpanel

logger = setup_logging(__name__)

parser = ArgumentParser(
    "/usr/bin/clwpos-user",
    "Utility for control CL WPOS under user",
    formatter_class=CustomFormatter
)
if not is_run_under_user():
    parser.add_argument('--user', default=None, required=True)

class MaxCacheMemory(int):
    """
    Class to validate format and values of cache memory setted by user.
    """
    def __new__(cls, *args, **kwargs):
        try:
            instance = super(MaxCacheMemory, cls).__new__(cls, *args, **kwargs)
        except ValueError:
            raise argparse.ArgumentTypeError("invalid value type, must be integer")

min_memory = constants.MINIMUM_MAX_CACHE_MEMORY
        max_memory = constants.MAXIMUM_MAX_CACHE_MEMORY

if not min_memory <= instance <= max_memory:
            raise argparse.ArgumentTypeError(
                f"value must be in range: [{min_memory}, {max_memory}]")

return instance

@dataclass
class Issue:
    """
    Generic class for keeping compatibility/misconfiguration issues
    """
    header: str
    description: str
    fix_tip: str

context: Dict[str, str] = field(default_factory=dict)

@property
    def dict_repr(self):
        return asdict(self)

@dataclass
class CompatibilityIssue(Issue):
    """
    For compatibility issues
    """
    type: str = 'incompatibility'

@dataclass
class MisconfigurationIssue(Issue):
    """
    For misconfiguration issues
    """
    type: str = 'misconfiguration'

class PluginStatus(Enum):
    UNINSTALLED = 'uninstalled'
    ACTIVE = 'active'
    INACTIVE = 'inactive'

def get_wp_plugin_status(wordpress_abs_path, plugin_name) -> PluginStatus:
    """
    Get information about WordPress plugin current status.

:param wordpress_abs_path:
        absolute path to wordpress installation
    :param plugin_name:
        name of plugin as it listed in plugins directory
    :return:
        PluginStatus
    """
    plugins_data = wordpress(wordpress_abs_path, 'plugin', 'list', '--name=%s' % plugin_name, '--json')
    if isinstance(plugins_data, WordpressError):
        raise WposError(message=plugins_data.message, context=plugins_data.context)

try:
        response = json.loads(plugins_data)
    except (ValueError, TypeError, json.JSONDecodeError) as e:
        raise WposError(
            message=_('Malformed plugins information received from wp-cli. '
                      'Raw response is %(response)s.'),
            context={'response': plugins_data},
            details=str(e)
        )

# in case of missing plugin wp-cli returns empty dict
    if not response:
        return PluginStatus.UNINSTALLED

# in any other case we get list of one element with parameters
    return PluginStatus(response[0]['status'])

def redis_is_running() -> bool:
    command_get_redis_status_dict = {"command": "get-redis-status"}
    try:
        daemon_result = daemon_communicate(command_get_redis_status_dict)
    except WposError:
        return False
    if 'status' in daemon_result:
        return daemon_result['status']
    return False

class CloudlinuxWposUser(object):
    """
    Class for run clwpos-user utility commands
    """
    COMMAND_RELOAD_DICT = {"command": "reload"}

def __init__(self):
        self._is_json = False
        self._opts = None
        self._wp_config_suffix = "wp-config.php"
        self._is_redis_running = None
        self._supported_handlers = None
        self._supported_php_versions = None
        self._php_versions_with_redis_not_loaded = None
        init_wpos_sentry_safely(logger)

@property
    def redis_socket_path(self):
        return os.path.join(home_dir(), USER_WPOS_DIR, 'redis.sock')

@property
    def supported_handlers(self):
        if self._supported_handlers is None:
            self._supported_handlers = supported_php_handlers()
        return self._supported_handlers

@property
    def is_redis_running(self):
        if self._is_redis_running is None:
            self._is_redis_running = redis_is_running()
        return self._is_redis_running

@property
    def supported_php_versions(self):
        """
        List of php version installed on the system with redis-extension enabled
        and version >= minimum supported version
        :return:
        """
        if self._supported_php_versions is None:
            php_with_redis = cpanel.get_cached_php_versions_with_redis_loaded()
            self._supported_php_versions = [php for php in php_with_redis
                                            if self.get_digits_from_version(php) >= constants.MINIMUM_SUPPORTED_PHP]
        return self._supported_php_versions

@property
    def php_versions_with_redis_present(self):
        """
        List of php version installed on the system with redis-extension enabled
        and version >= minimum supported version
        :return:
        """
        if self._php_versions_with_redis_not_loaded is None:
            php_with_redis = cpanel.get_cached_php_versions_with_redis_present()
            self._php_versions_with_redis_not_loaded = [php for php in php_with_redis
                                                        if self.get_digits_from_version(php)
                                                        >= constants.MINIMUM_SUPPORTED_PHP]
        return self._php_versions_with_redis_not_loaded

@staticmethod
    def get_digits_from_version(version: str) -> int:
        """

:param version: version of php "ea-php56", "alt-php74"
        :return: digits extracted from version
        """
        version = version.replace("ea-php", "")
        version = version.replace("alt-php", "")
        return int(version)

@catch_error
    def run(self, argv):
        """
        Run command action
        :param argv: sys.argv[1:]
        :return: clwpos-user utility retcode
        """
        self._parse_args(argv)
        if not is_run_under_user():
            drop_user_privileges(self._opts.user, effective_or_real=False, set_env=True)

result = getattr(self, self._opts.command.replace("-", "_"))()
        print_data(self._is_json, result)

def _parse_args(self, argv):
        """
        Parse command line arguments
        :param argv: sys.argv[1:]
        """
        self._opts = parser.parse_args(argv)
        self._is_json = True

def _check_monitoring_daemon_and_exit(self):
        """
        Ensures monitoring socket is present, otherwise - exit
        """
        if not os.path.exists(constants.WPOS_DAEMON_SOCKET_FILE):
            error_and_exit(
                self._is_json,
                {
                    "result": _("Unable to find monitoring daemon socket %(daemon_file)s. Please, contact your "
                                "system administrator to ensure service %(service_name)s is currently running"),
                    "context": {"daemon_file": constants.WPOS_DAEMON_SOCKET_FILE,
                                "service_name": constants.MONIROTING_SERVICE},
                }
            )

@staticmethod
    def _get_php_versions_handlers_pair_for_docroot(domains_per_docroot: List, php_data: List) -> Tuple:
        """
        Returns pair (all_php_versions, all_php_handlers) for domains in
        docroot
        """
        versions, handlers = set(), set()
        for item in php_data:
            domain = item['vhost']
            if domain not in domains_per_docroot:
                continue
            versions.add(item['version'])
            if item['php_fpm']:
                handlers.add('php-fpm')
            else:
                handlers.add(cpanel._get_php_handler(domain))
        return versions, handlers

@staticmethod
    def collect_general_issues():
        """
        Collects general (not depending on worpress/docroot setup)
        server misconfigurations/incompatibilities with WPOS
        """
        issues = []
        if is_litespeed_running():
            issues.append(
                CompatibilityIssue(
                    header=_('Litespeed is unsupported'),
                    description=_('Litespeed webserver is running on your server, '
                                  'it is not currently supported by WPOS.'),
                    fix_tip=_('Please ask your system administrator to switch webserver to Apache '
                              'using the following instruction: %(switch_lsws_link)s '
                              'or keep watching our blog: %(blog_link)s '
                              'for news about LiteSpeed support.'),
                    context=dict(
                        switch_lsws_link='https://www.interserver.net/tips/kb/switch-apache-litespeed/',
                        blog_link='https://blog.cloudlinux.com/'
                    )
                ).dict_repr)
        return issues

def collect_docroot_issues(self, doc_root_info: Dict, php_info: List):
        """
        Collects incompatibilities related to docroot (non-supported handler, etc)
        """
        supported_handlers = self.supported_handlers
        issues = []
        incompatible_php_modules = {}

if is_cpanel():
            domains_per_docroot = doc_root_info['domains']
            versions, handlers = self._get_php_versions_handlers_pair_for_docroot(domains_per_docroot, php_info)
            if len(versions) > 1:
                issues.append(
                    CompatibilityIssue(
                        header=_('Different PHP versions for domains'),
                        description=_('Those domains: %(domains)s are in same docroot, '
                                      'but using different PHP version'),
                        fix_tip=_('Set or ask your system administrator to set same PHP version on those domains'),
                        context={
                            'domains': ', '.join(domains_per_docroot)
                        }
                    ).dict_repr
                )
            if len(handlers) > 1:
                issues.append(
                    CompatibilityIssue(
                        header=_('Different PHP handlers for domains'),
                        description=_('Those domains: %(domains)s are in same docroot, '
                                      'but using different PHP handlers'),
                        fix_tip=_('Set or ask your system administrator to set same PHP handler on those domains'),
                        context={
                            'domains': ', '.join(domains_per_docroot)
                        }
                    ).dict_repr
                )
        if doc_root_info["php"]["handler"] not in supported_handlers:
            issues.append(
                CompatibilityIssue(
                    header=_('Unsupported PHP handler'),
                    description=_('Website uses unsupported PHP handler. Currently supported '
                                  'handler(s): %(supported_handlers)s.'),
                    fix_tip=_('Please, set or ask your system administrator to set one of the '
                              'supported PHP handlers for the domain: %(supported_handlers)s. '
                              'Or keep watching our blog: %(blog_url)s for supported handlers list updates.'),
                    context={
                        'supported_handlers': ", ".join(supported_handlers),
                        'blog_url': 'https://blog.cloudlinux.com/'
                    }
                ).dict_repr
            )
        incompatible_module = 'snuffleupagus'
        php_version = doc_root_info['php']['version']
        if incompatible_php_modules.get(php_version) == incompatible_module or \
                is_conflict_modules_installed(php_version, incompatible_module):
            incompatible_php_modules[php_version] = incompatible_module
            issues.append(
                CompatibilityIssue(
                    header=_('Unsupported PHP module is loaded'),
                    description=_('Incompatible PHP module "%(incompatible_module)s" is currently used.'),
                    fix_tip=_('Please, disable or remove "%(incompatible_module)s" PHP extension.'),
                    context=dict(incompatible_module=incompatible_module)
                ).dict_repr)
        if php_version not in self.supported_php_versions:
            if self.supported_php_versions:
                fix_tip = _('Please, set or ask your system administrator to set one of the '
                            'supported PHP version: %(compatible_versions)s for the domain.')
            else:
                fix_tip = _('Please, ask your system administrator to setup at least one of the '
                            'recommended PHP version in accordance with docs (%(docs_url)s).')
            issues.append(
                CompatibilityIssue(
                    header=_('Incompatible PHP version'),
                    description=_('Incompatible PHP version %(php_version)s is currently used.'),
                    fix_tip=fix_tip,
                    context=dict(php_version=php_version, compatible_versions=', '.join(self.supported_php_versions),
                                 docs_url="https://user-docs.solo.cloudlinux.com/wpos-plugin")
                ).dict_repr)
        if php_version in self.php_versions_with_redis_present:
            if self.supported_php_versions:
                fix_tip = _('Please, set or ask your system administrator to set one of the '
                            'supported PHP version: %(compatible_versions)s for the domain.')
            else:
                fix_tip = _('Please, ask your system administrator to setup at least one of the '
                            'recommended PHP version in accordance with docs (%(docs_url)s).')
            issues.append(
                CompatibilityIssue(
                    header=_('Redis extension is not loaded'),
                    description=_('Redis PHP extension is required for optimization module, but not loaded for '
                                  'selected PHP version: %(php_version)s.'),
                    fix_tip=fix_tip,
                    context=dict(php_version=php_version, compatible_versions=', '.join(self.supported_php_versions),
                                 docs_url="https://user-docs.solo.cloudlinux.com/wpos-plugin")
                ).dict_repr
            )

return issues

@staticmethod
    def _is_our_roc_plugin(wp_path: str) -> bool:
        """
        Checks that WP's WP_REDIS_PATH is defined and contains our path
        """
        wp_redis_path = cpanel.wp_get_constant(wp_path, 'WP_REDIS_PATH')
        if wp_redis_path and "/.clwpos/redis.sock" in wp_redis_path:
            return True
        return False

def check_installed_roc_plugin(self, wp_path: str) -> bool:
        """
        Checks that ROC plugin was installed by us, or not exists at all
        Returns False if the plugin was found but our config modifications was not
        """
        if os.path.exists(os.path.join(wp_path, 'wp-content', 'plugins', 'redis-cache')):
            if not self._is_our_roc_plugin(wp_path):
                return False
        return True

def collect_wordpress_issues(self, wordpress: Dict, docroot: str, check_module_compatibility: bool):
        """
        Collects incompatibilities related to wordpress setup (conflicting plugin enabled, etc)
        """
        issues = []
        minimum_supported_wp_version = '3.7'
        if parse_version(wordpress["version"]) < parse_version(minimum_supported_wp_version):
            issues.append(
                CompatibilityIssue(
                    header=_('Unsupported WordPress version'),
                    description=_('Optimization module is incompatible with '
                                  'WordPress version %(wp_version)s being used.'),
                    fix_tip=_('The minimal supported WordPress version is %(minimum_supported_wp_version)s. '
                              'Upgrade your WordPress installation or reinstall it from the scratch.'),
                    context=dict(
                        minimum_supported_wp_version=minimum_supported_wp_version,
                        wp_version=wordpress["version"]
                    )
                ).dict_repr)

if check_module_compatibility:
            if wordpress["object_cache"] != "redis-cache":
                issue = self._get_wp_plugin_compatibility_issues(docroot, wordpress)
                if issue:
                    issues.append(issue)

if not self.is_redis_running:
                issues.append(
                    MisconfigurationIssue(
                        header=_('Redis is not running'),
                        description=_('Object cache module is enabled, but redis process is not running.'),
                        fix_tip=_('Please refresh page, redis will start automatically in 5 minutes. '
                                  'If the issue persists - contact your system administrator and report this issue')
                    ).dict_repr
                )
        detected_wp_plugin = wordpress.get('object_cache')
        if detected_wp_plugin == "Unknown":
            issues.append(
                CompatibilityIssue(
                    header=_('Conflicting object cache plugin enabled'),
                    description=_('Unknown custom object cache plugin is already enabled'),
                    fix_tip=_('Disable WordPress plugin that is being used for the object caching '
                              'using the WordPress settings.')
                ).dict_repr)
        elif detected_wp_plugin == "w3-total-cache":
            issues.append(
                CompatibilityIssue(
                    header=_('Object Cache module of W3 Total Cache plugin is incompatible'),
                    description=_('WordPress website already has Object Cache feature enabled '
                                  'with caching backend configured by the the W3 Total Cache plugin.'),
                    fix_tip=_('Deactivate Object Cache in W3 Total Cache plugin settings.'),
                    context=dict()
                ).dict_repr)
        elif detected_wp_plugin not in (None, "redis-cache"):
            issues.append(
                CompatibilityIssue(
                    header=_('Conflicting object cache plugin enabled'),
                    description=_('The "%(detected_wp_plugin)s" plugin conflicts with the '
                                  '"Redis Object Cache" that is required by optimization module.'),
                    fix_tip=_('Deactivate Object Cache in plugin settings or '
                              'completely turn off conflicting plugin using WordPress administration page'),
                    context=dict(detected_wp_plugin=detected_wp_plugin)
                ).dict_repr)
        if not self.check_installed_roc_plugin(os.path.join(docroot, wordpress['path'])):
            issues.append(
                CompatibilityIssue(
                    header=_('Another Redis Object Cache plugin is installed'),
                    description=_('Non CloudLinux Redis Object Cache is installed for the website'),
                    fix_tip=_('Uninstall Redis Object Cache plugin using WordPress administration page')
                ).dict_repr)

try:
            multisite = cpanel.is_multisite(os.path.join(docroot, wordpress["path"]))
            if multisite:
                issues.append(
                    CompatibilityIssue(
                        header=_('WordPress Multisite mode is enabled'),
                        description=_('WordPress uses the Multisite mode which is currently not supported.'),
                        fix_tip=_('Install or configure WordPress in the single-site mode.')
                    ).dict_repr)
        except WposError as e:
            issues.append(
                CompatibilityIssue(
                    header=_('Unexpected WordPress error'),
                    description=_('Unable to detect if the WordPress installation has the Multisite mode enabled '
                                  'mode due to unexpected error. '
                                  '\n\n'
                                  'Technical details:\n%(error_message)s.\n'
                                  '\nMost likely WordPress installation is not working properly.'),
                    fix_tip=_('If this is only one issue, please check that your website is working properly – '
                              'try to run the specified command to find any obvious '
                              'errors in the WordPress configuration. '
                              'Otherwise, try to fix other issues first - it may help to resolve this issue as well.'),
                    context=dict(
                        error_message=e.message % e.context
                    )
                ).dict_repr)
        return issues

def _get_wp_plugin_compatibility_issues(self, docroot, wordpress):
        """
        Get issue that relates to currently installed redis-cache
        plugin or None if everything is ok
        """
        try:
            plugin_status = get_wp_plugin_status(
                wordpress_abs_path=os.path.join(docroot, wordpress["path"]),
                plugin_name='redis-cache')
        except WposError as e:
            return CompatibilityIssue(
                header=_('Unexpected WordPress error'),
                description=_(
                    'Unable to detect the WordPress plugins '
                    'due to unexpected error. '
                    '\n\n'
                    'Technical details:\n%(error_message)s.\n'
                    '\nMost likely WordPress installation is not working properly.'
                ),
                fix_tip=_(
                    'Check that your website is working properly – '
                    'try to run the specified command to find any obvious '
                    'errors in the WordPress configuration. '
                    'Otherwise, try to fix other issues first - '
                    'it may help to resolve this issue as well.'
                ),
                context=dict(
                    error_message=e.message % e.context
                )
            ).dict_repr

if plugin_status == PluginStatus.INACTIVE:
            return MisconfigurationIssue(
                header=_('"Redis Object Cache" plugin is turned off'),
                description=_('Object cache module is enabled, but the '
                              '"Redis Object Cache" plugin is turned off. Caching does not work'),
                fix_tip=_('Activate the Redis Object Cache plugin in the Wordpress admin page and '
                          'enable Object Cache Drop-in in the Redis Object Cache plugin settings. '
                          'As alternative, just turn the caching module off and on again.')
            ).dict_repr
        elif plugin_status == PluginStatus.ACTIVE:
            return MisconfigurationIssue(
                header=_('The Object Cache Drop-in not installed'),
                description=_('The Object Cache Drop-In is not activated. Caching does not work'),
                fix_tip=_('Activate the Object Cache using the Redis Object Cache plugin '
                          'settings page. '
                          'As alternative, just turn the caching module off and on again.')
            ).dict_repr
        elif plugin_status == PluginStatus.UNINSTALLED:
            return MisconfigurationIssue(
                header=_('"Redis Object Cache" plugin is not installed'),
                description=_('The "Redis Object Cache" WordPress plugin is not installed. '
                              'Caching does not work'),
                fix_tip=_('Turn the caching module off and on again. '
                          'Contact your administrator if the issue persists.')
            ).dict_repr
        else:
            raise WposError(_('Unexpected plugin status: %(status)s'), context=dict(status=plugin_status))

@parser.argument("--module", help="modules list separated by comma",
                     choices=ALL_OPTIMIZATION_MODULES, type=str, default=OBJECT_CACHE_MODULE)
    @parser.command(help="Output WPOS user config (with Redis status and Redis current memory)")
    def get(self):
        user_info = cpanel.get_user_info()
        php_info = cpanel.get_php_info()

general_issues = self.collect_general_issues()

for docroot, doc_root_info in user_info.items():
            if doc_root_info["php"]["fpm"]:
                doc_root_info["php"]["handler"] = "php-fpm"
            docroot_issues = self.collect_docroot_issues(doc_root_info, php_info)

new_wps = []
            for wp in doc_root_info["wps"]:
                enabled = self._is_enabled(doc_root_info["domains"][0], wp["path"], OBJECT_CACHE_MODULE)
                wordpress_issues = self.collect_wordpress_issues(wp, docroot, check_module_compatibility=enabled)

redis_cache_module = {
                    "enabled": enabled
                }
                issues = [*general_issues, *docroot_issues, *wordpress_issues]
                if issues:
                    redis_cache_module.update({
                        'issues': issues
                    })

new_wps.append(
                    {
                        "path": wp["path"],
                        "version": wp["version"],
                        "modules": {"object_cache": redis_cache_module},
                    }
                )
            doc_root_info["wps"] = new_wps

php = doc_root_info.pop("php")
            doc_root_info["php_version"] = php["version"]
            doc_root_info["php_handler"] = php["handler"]

return {
            "docroots": list(user_info.values()),
            "allowed_modules": get_allowed_modules(),
            "used_memory": RedisLibUser(self.redis_socket_path).get_redis_used_memory(),
            "max_cache_memory": UserConfig(user_name()).get_config().get(
                "max_cache_memory", UserConfig.DEFAULT_MAX_CACHE_MEMORY
            )
        }

def get_current_issues_by_docroot(self, wordpress_path: str, docroot: str, is_module_enabled: bool):
        """
        Obtains issues for special docroot and wordpress
        """
        wordpress_issues = []
        general_issues = self.collect_general_issues()
        php_info = cpanel.get_php_info()
        user_info_by_docroot = cpanel.get_user_info()[docroot]
        if user_info_by_docroot["php"]["fpm"]:
            user_info_by_docroot["php"]["handler"] = "php-fpm"
        docroot_issues = self.collect_docroot_issues(user_info_by_docroot, php_info)
        for wp in user_info_by_docroot["wps"]:
            if wp['path'] != wordpress_path:
                continue
            wordpress_issues = self.collect_wordpress_issues(wp, docroot,
                                                             check_module_compatibility=is_module_enabled)
        return [*general_issues, *docroot_issues, *wordpress_issues]

def _is_enabled(self, domain: str, wp_path: str, module: str) -> bool:
        uc = UserConfig(user_name())
        return uc.is_module_enabled(domain, wp_path, module) and module in get_allowed_modules()

@parser.argument('--max_cache_memory', help='Maximum cache memory to use in MB',
                     type=MaxCacheMemory, default=constants.DEFAULT_MAX_CACHE_MEMORY)
    @parser.command(help='Set parameters of global modules in WPOS user config')
    @check_license_decorator
    def set(self):
        uc = UserConfig(user_name())
        # TODO: this method must probably UPDATE config and not REPLACE it completely
        params = {"max_cache_memory": f"{self._opts.max_cache_memory}mb"}
        uc.set_params(params)
        daemon_communicate(self.COMMAND_RELOAD_DICT)
        return params

@parser.argument("--module", help="modules list separated by comma", type=str, default="object_cache")
    @parser.argument('--wp-path', help='Path to user\'s wordpress', type=str, default='')
    @parser.argument('--domain', help='User\'s wordpress domain', type=str, required=True)
    @parser.command(help='Disables and uninstalls module on wordpress')
    def disable(self):
        username, doc_root = check_domain(self._opts.domain)
        wp_path = self._opts.wp_path.lstrip("/")
        uc = UserConfig(user_name())
        errors = []

if not uc.is_module_enabled(self._opts.domain, wp_path):
            return {"warning": _('Module %(module)s is already disabled on the domain %(domain)s. Nothing to be done.'),
                    "context": {"domain": self._opts.domain, "module": self._opts.module}}

self._check_monitoring_daemon_and_exit()

last_error = cpanel.disable_without_config_affecting(
            cpanel.DomainName(self._opts.domain), self._opts.wp_path)

if last_error:
            logger.error(last_error.message % last_error.context)
            errors.append(last_error)

try:
            uc.disable_module(self._opts.domain, self._opts.wp_path)
        except Exception as e:
            logger.exception("unable to disable module in config")
            errors.append(e)

try:
            daemon_communicate(self.COMMAND_RELOAD_DICT)
        except Exception as e:
            logger.exception("unable to reload cache backend")
            errors.append(e)

enabled = self._is_enabled_and_exist_on_system(self._opts.domain, self._opts.wp_path, self._opts.module)

try:
            last_error = errors.pop(-1)
        except IndexError:
            last_error = None

if enabled:
            if last_error:
                raise WposError(
                    message=_("Optimization module disabling failed because one or more steps reported error. "
                              "Caching is still active, but may work unstable. Try disabling it again. "
                              "Contact your system administrator if this issue persists. "
                              "Detailed information you can find in log file '%(log_path)s'"),
                    details=last_error.message,
                    context={
                        'log_path': USER_LOGFILE_PATH.format(homedir=home_dir()),
                        **getattr(last_error, 'context', {})
                    }
                )
            else:
                error_and_exit(
                    self._is_json,
                    {"result": _("WordPress caching module is still enabled, but no errors happened. "
                                 "Try again and contact your system administrator if this issue persists.")},
                )
        else:
            if last_error:
                return {
                    "warning": _("Optimization module disabled, but one or more steps reported error. "
                                 "Detailed information you can find in log file '%(log_path)s'"),
                    "context": {
                        'log_path': USER_LOGFILE_PATH.format(homedir=home_dir())
                    }
                }
            else:
                current_module_status = self._is_enabled(self._opts.domain, wp_path, OBJECT_CACHE_MODULE)
                issues = self.get_current_issues_by_docroot(wp_path, doc_root, current_module_status)
                module_data = {
                    'enabled': current_module_status
                }
                if issues:
                    module_data.update({
                        'issues': issues
                    })
                return {
                    'module': module_data,
                    "used_memory": RedisLibUser(self.redis_socket_path).get_redis_used_memory()
                }

def _is_enabled_and_exist_on_system(self, domain, wp_path, module):
        for doc_root in self.get()["docroots"]:
            if domain in doc_root["domains"]:
                for wp in doc_root["wps"]:
                    if wp["path"] == wp_path:
                        return wp["modules"][module]["enabled"]

return False

@parser.argument('--module', type=str, help='List of WPOS modules separated by commas'
                                                ' on which to perform an action',
                     choices=ALL_OPTIMIZATION_MODULES, default=OBJECT_CACHE_MODULE)
    @parser.argument('--action', type=str, help='Action to perform on WPOS modules', required=True)
    @parser.command(help='Perform action on WPOS modules')
    def do_action(self):
        """
        Perform action on selected WPOS modules.
        """
        action = self._opts.action
        module = self._opts.module
        # dictionary where key - action, value - function that provides that action
        action_to_func = {"purge": self.redis_purge}
        if action not in action_to_func:
            error_and_exit(
                self._is_json,
                {
                    "result": f'Invalid action "{action}", currently only "purge" action is supported'
                },
            )

return action_to_func[action](module)  # perform action on module

def redis_purge(self, *args):
        """
        Clean entire redis cache for user.
        """
        return RedisLibUser(self.redis_socket_path).purge_redis()

@parser.argument(
        "--ignore-errors",
        help="ignore site check results after plugin install and enable",
        action="store_true",
    )
    @parser.argument("--wp-path", help="path to user's wordpresses", type=str, default="")
    @parser.argument("--domain", help="User's wordpress domain", type=str, required=True)
    @parser.argument("--module", help="Modules to enable", type=str, required=False,
                     choices=ALL_OPTIMIZATION_MODULES, default=OBJECT_CACHE_MODULE)
    @parser.command(help="Installs and enables module on wordpress")
    @check_license_decorator
    def enable(self):
        """
        Enable object_cache for user with compliance with end-user spec.
        :return:
        """
        domain = self._opts.domain
        wp_path = self._opts.wp_path.lstrip("/")
        module = self._opts.module
        ignore_errors = self._opts.ignore_errors

username, doc_root = check_domain(self._opts.domain)

abs_wp_path = os.path.join(doc_root, wp_path)
        uc = UserConfig(username)

if self._is_enabled(domain, wp_path, OBJECT_CACHE_MODULE):
            return {"warning": _("The %(module)s module is already enabled on "
                                 "the domain %(domain)s. Nothing to be done."),
                    "context": {"domain": domain, "module": module}}

self._check_monitoring_daemon_and_exit()

compatibilities = self.get()

# check that enabling of module is allowed by admin
        if module not in compatibilities["allowed_modules"]:
            error_and_exit(
                self._is_json,
                {
                    "result": _("Usage of the %(module)s module is prohibited by admin."),
                    "context": {"module": module},
                }
            )

# check that user's wordpress fits to requirements
        ok, requirements_failed_info = cpanel.requirements_check(
            compatibilities, wp_path, domain, OBJECT_CACHE_MODULE)
        if not ok:
            error_and_exit(self._is_json, requirements_failed_info)

# check user's site before installation
        if not ignore_errors:
            self._website_check(abs_wp_path, domain, uc, wp_path)

# try enable module with wp-cli without adding info into user's wpos config
        ok, enable_failed_info = cpanel.enable_without_config_affecting(
            cpanel.DomainName(domain),
            wp_path,
            module=module,
        )
        if not ok:
            raise WposError(**enable_failed_info)

uc.enable_module(domain, wp_path)
        try:
            cpanel.reload_redis()
        except Exception:
            uc.disable_module(domain, wp_path)
            cpanel.rollback_object_cache(abs_wp_path)
            raise

# check user's site after installation and enabling
        if not ignore_errors:
            self._website_check(abs_wp_path, domain, uc, wp_path, rollback=True)

current_module_status = self._is_enabled(domain, wp_path, OBJECT_CACHE_MODULE)
        issues = self.get_current_issues_by_docroot(wp_path, doc_root, current_module_status)
        module_data = {
            'enabled': current_module_status
        }
        if issues:
            module_data.update({
                'issues': issues
            })
        return {
            'module': module_data,
            'used_memory': RedisLibUser(self.redis_socket_path).get_redis_used_memory()
        }

def _website_check(self, abs_wp_path, domain, uc, wp_path, rollback=False):
        """
        Performs website availability checks and raises
        errors in case of problems

:param abs_wp_path:
            absolute path to wordpress installation
        :param domain:
            domain that wordpress installation belongs to
        :param uc:
            user config instance
        :param wp_path:
            relative path to wordpress installation
        :param rollback:
            whether to roll redis and plugin changes back
        :return:
        """

try:
            post_site_check(domain, wp_path, abs_wp_path)
        except WposError as e:
            if rollback:
                uc.disable_module(domain, wp_path)
                cpanel.reload_redis()
                cpanel.rollback_object_cache(abs_wp_path)

if isinstance(e, RollbackException):
                raise

error_and_exit(
                is_json=True,
                message={
                    "context": {},
                    "result": _("Unexpected error occurred during plugin installation for WordPress. "
                                "Try again and contact your system administrator if the issue persists."),
                    "details": str(e)
                },
            )

Name	Type	Actions
__pycache__	Directory
website_check	Directory
config.py	File
redis_lib.py	File
wpos_user.py	File