✏️ Editing: wpos_user.py Read Only

# -*- 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
"""
Implementation of clwpos-user CLI utility
"""
import argparse
import os
import pwd
from dataclasses import dataclass, asdict
from itertools import chain
from typing import Dict, List, Optional, Tuple

from clcommon.clpwd import drop_user_privileges
from clcommon.const import Feature as ModuleFeature
from clcommon.cpapi import is_panel_feature_supported
from clwpos.optimization_features.features import WP_BROKEN_CORE_REPAIR_TIP
from packaging.version import parse as parse_version

from clwpos import constants
from clwpos import gettext as _
from clwpos.cl_wpos_exceptions import WposError
from clwpos.daemon import WposDaemon
from clwpos.data_collector_utils import (
    get_user_info,
)
from clwpos.feature_suites import (
    get_allowed_modules,
    get_allowed_features_dict
)
from clwpos.feature_suites.configurations import get_visible_modules, get_visible_features_dict
from clwpos.logsetup import (
    setup_logging,
    USER_LOGFILE_PATH,
)
from clwpos.object_cache.redis_utils import (
    reload_redis
)
from clwpos.optimization_features import (
    OBJECT_CACHE_FEATURE,
    CDN_FEATURE,
    ALL_OPTIMIZATION_FEATURES,
    CompatibilityIssue,
    Feature,
    UniqueId,
    convert_feature_list_to_interface,
    disable_without_config_affecting,
    enable_without_config_affecting,
    DomainName,
    DocRootPath
)
from clwpos.papi import get_subscriptions_by_pw, is_feature_hidden_server_wide
from clwpos.parse import ArgumentParser, CustomFormatter
from clwpos.scoped_cache import enable_caching
from clwpos.user.config import UserConfig, LicenseApproveStatus
from clwpos.user.progress_check import (
    CommandProgress,
    track_command_progress,
    update_command_progress,
)
from clwpos.user.redis_lib import RedisLibUser
from clwpos.user.website_check import post_site_check, RollbackException
from clwpos.utils import (
    USER_WPOS_DIR,
    check_license_decorator,
    print_data,
    catch_error,
    error_and_exit,
    daemon_communicate,
    check_domain,
    is_run_under_user,
    _run_clwpos_as_user_in_cagefs,
    is_redis_configuration_running,
    get_server_wide_options,
    get_pw,
)
from clwpos.wp_utils import (
    php_info,
    update_redis_disable_banners_constant,
    WordpressError,
    wordpress,
    obtain_wp_cli_env,
)

from clwpos.daemon_redis_lib import redis_socket_health_check

logger = setup_logging(__name__)
LSWS_INCOMPATIBLE_FEATURES = (OBJECT_CACHE_FEATURE, CDN_FEATURE)

parser = ArgumentParser(
    "/usr/bin/clwpos-user",
    "Utility for control CL AccelerateWP under user",
    formatter_class=CustomFormatter
)

def get_user(username):
    try:
        return get_pw(username=username)
    except KeyError:
        raise argparse.ArgumentTypeError("User '%s' does not exist" % username)

if not is_run_under_user():
    parser.add_argument('--user', default=None, required=True, type=get_user)
else:
    parser.add_argument('--user', default=get_pw(), type=RuntimeError, help=argparse.SUPPRESS)

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 ManagedFeature:
    """
    To track (docroot, website, feature) which was enabled/disabled
    """
    docroot: str
    website: str
    feature_name: str

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

def __init__(self):
        self._is_json = False
        self._opts = None

self.command_progress: Optional[CommandProgress] = None

@property
    def redis_socket_path(self) -> str:
        return os.path.join(self._opts.user.pw_dir, USER_WPOS_DIR, 'redis.sock')

@catch_error
    def run(self, argv):
        """
        Run command action
        :param argv: sys.argv[1:]
        :return: clwpos-user utility retcode
        """
        self._opts = self._parse_args(argv)

# If wp_path equal to ".", then convert it to "" to reduce future errors with non-valid wp_path
        if getattr(self._opts, 'wp_path', None) == UserConfig.INVALID_PATH_KEY:
            self._opts.wp_path = ''

self._is_json = True
        if not is_run_under_user():
            _run_clwpos_as_user_in_cagefs(self._opts.user.pw_name)
        else:
            _run_clwpos_as_user_in_cagefs()
        if not is_run_under_user() and not is_panel_feature_supported(ModuleFeature.CAGEFS):
            drop_user_privileges(self._opts.user.pw_name, effective_or_real=False, set_env=True)
        logger.info('Income request=%s', str(self._opts))
        result = getattr(self, self._opts.command.replace("-", "_"))()
        logger.info('Command finished with result=%s', str(result))
        print_data(self._is_json, result)

def _parse_args(self, argv):
        raise NotImplementedError

def _check_monitoring_daemon_and_exit(self, feature):
        """
        Ensures monitoring socket is present, otherwise - exit
        """
        if feature == OBJECT_CACHE_FEATURE \
                and 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},
                }
            )

def _check_redis_configuration_process_and_exit(self, feature):
        """
        Ensures redis configuration processes are not in progress.
        """
        if feature == OBJECT_CACHE_FEATURE and is_redis_configuration_running():
            error_and_exit(
                self._is_json,
                {
                    "result": _("Configuration of PHP redis extension is in progress. "
                                "Please wait until the end of this process to use AccelerateWP."),
                }
            )

def _is_redis_daemon_reload_needed(self, user_config: UserConfig, feature: Feature,
                                       before_config_modified=False) -> bool:
        """
        Determine if it is needed to send reload command to Wpos redis daemon.
        Such reload should be done:

1. Enabling object cache and redis is not alive for user
        2. Enabling object cache, but there are no successful sites (failed enabling)
        2. Disabling object cache for last site
        """
        command = self._opts.command
        if feature != OBJECT_CACHE_FEATURE:
            return False

disable_commands = [
            'disable',
            'disable-all'
        ]

sites_count_with_enabled_module = user_config.get_enabled_sites_count_by_modules([feature])

uid = os.geteuid()
        is_redis_alive = redis_socket_health_check(uid)

# disable last site, redis is still running -> expecting turning off
        turn_off_upon_disable_last_site = (command in disable_commands and
                                           is_redis_alive and
                                           sites_count_with_enabled_module == 0)

# enabling, but redis is not running yet
        turn_on_upon_enabling = (command == 'enable' and
                                 not is_redis_alive)

# enabling, but no actual sites with feature
        turn_off_upon_failed_enabling = (before_config_modified and
                                         command == 'enable' and
                                         sites_count_with_enabled_module == 0)

logger.info('Check whether to reload redis for '
                    'user=%s, '
                    'command=%s, '
                    'is_redis_alive=%s, '
                    'before config modified status=%s,'
                    'sites_with_enabled_cache_left=%s',
                    str(uid),
                    command,
                    str(is_redis_alive),
                    str(before_config_modified),
                    str(sites_count_with_enabled_module))

if turn_on_upon_enabling or turn_off_upon_failed_enabling or turn_off_upon_disable_last_site:
            logger.info('Redis is needed to be reloaded for user=%s', str(uid))
            return True

return False

@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'])
            handlers.add(item['handler'])
        return versions, handlers

def collect_general_issues(self, module_name: str):
        """
        Collects general (not depending on worpress/docroot setup)
        server misconfigurations/incompatibilities with WPOS
        """
        issues = []
        return issues

@update_command_progress
    def collect_docroot_issues(self, module: Feature, doc_root_info: Dict, php_info: List, visible_features: List):
        """
        Collects incompatibilities related to docroot (non-supported handler, etc)
        """
        if doc_root_info['php_version'] is None:
            return [
                CompatibilityIssue(
                    unique_id='PHP_MISCONFIGURED',
                    header=_("Unable to detect php configuration"),
                    description=_("We were unable to detect php configuration for this website. "
                                  "This might happen if there is a misconfiguration in control panel."),
                    fix_tip=_("Please try again later and contact your administrator"
                              "if the issue persists."),
                    telemetry=dict(debug_data=doc_root_info)
                )
            ]

issues = []
        issues.extend(self._check_ambiguous_php_config(doc_root_info, php_info))

additional_issues = module.collect_docroot_issues(
            doc_root_info, visible_features=visible_features)
        issues.extend(additional_issues)

return issues

def _check_ambiguous_php_config(self, doc_root_info, php_info):
        """
        Checks whether docroot has multiple php configs
        for the same path. This usually happens on cPanel when
        user or admin selects different php versions for primary domain
        and alias which points at the same location.
        """
        issues = []

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)
                    },
                    unique_id=UniqueId.PHP_MISCONFIGURATION,
                    telemetry=dict(
                        reason='PHP_VERSION_BADLY_CONFIGURED',
                        domains=domains_per_docroot
                    )
                )
            )

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)
                    },
                    unique_id=UniqueId.PHP_MISCONFIGURATION,
                    telemetry=dict(
                        reason='PHP_HANDLER_BADLY_CONFIGURED',
                        domains=domains_per_docroot
                    )
                )
            )

return issues

@update_command_progress
    def collect_wordpress_issues(
        self, module_name: Feature, wordpress_info: Dict, docroot: str, module_is_enabled: bool
    ):
        """
        Collects incompatibilities related to wordpress setup (conflicting plugin enabled, etc)
        """
        issues = []

# Implicitly assume a module's version string is properly formatted
        minimum_supported_wp_version = parse_version(module_name.minimum_supported_wp_version())
        wp_version = wordpress_info.get("version")

if not wp_version:
            issues.append(
                CompatibilityIssue(
                    header=_('Failed to determine WordPress version'),
                    description=_(
                        'Optimization feature couldn\'t be applied because '
                        'WordPress installation at %(wp_path)s seems to be '
                        'damaged or misconfigured.\n'
                        'Details: %(error_details)s',
                    ),
                    fix_tip=WP_BROKEN_CORE_REPAIR_TIP,
                    context=dict(
                        wp_path=docroot,
                        error_details=wordpress_info.get(
                            'version_missing_reason',
                            # We expect version is None -> version_missing_reason is not None,
                            #  but no guaranties on other side.
                            # FIXME: turn "version" into proper algebraic type once the code
                            #  is properly typed
                            'wp-include/version.php file is invalid',
                        ),
                    ),
                    unique_id=UniqueId.MISCONFIGURED_WORDPRESS,
                    telemetry=dict(minimum_supported_wp_version=str(minimum_supported_wp_version)),
                )
            )
        elif wp_version < minimum_supported_wp_version:
            issues.append(
                CompatibilityIssue(
                    header=_('Unsupported WordPress version'),
                    description=_('Optimization feature 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=str(minimum_supported_wp_version),
                        wp_version=str(wp_version)
                    ),
                    unique_id=UniqueId.UNCOMPATIBLE_WORDPRESS_VERSION,
                    telemetry=dict(
                        minimum_supported_wp_version=str(minimum_supported_wp_version),
                        wp_version=str(wp_version)
                    )
                ))

issues.extend(module_name.collect_wordpress_issues(wordpress_info, docroot, module_is_enabled))
        return issues

@parser.argument('--listen', action='store_true', required=True)
    @parser.argument('--domain',  required=False, default=None)
    @parser.argument('--wp-path', required=False, default=None)
    @parser.argument('--feature',  required=True)
    @parser.argument('--ignore-errors',  required=False, default=False, action='store_true')
    @parser.argument('--advice-id', type=int, required=False, default=None)
    @parser.command()
    def subscription(self):

if any((self._opts.domain, self._opts.wp_path is not None)) \
                and not all((self._opts.domain, self._opts.wp_path is not None)):
            error_and_exit(self._is_json, {'result': "Both domain and wp_path arguments "
                                                     "are requited if one of them is set"})

features = self._opts.feature.split(',')
        for f in features:
            if f not in [feature.to_interface_name() for feature in ALL_OPTIMIZATION_FEATURES]:
                error_and_exit(self._is_json, {'result': "Unsupported feature passed"})

daemon_communicate({
            'command': WposDaemon.DAEMON_REGISTER_UPGRADE_ATTEMPT,
            'domain': self._opts.domain,
            'wp_path': self._opts.wp_path,
            # comma separated str: "object_cache,critical_css,.."
            'feature': self._opts.feature,
            'ignore_errors': self._opts.ignore_errors,
            'advice_id': self._opts.advice_id,
        })

return {}

@parser.argument('--status', '-s', help='Status of license agreements', action='store_true')
    @parser.argument('--approve', '-a', help='Approve license agreement')
    @parser.argument('--text', '-t', help='Read and approve license agreement')
    @parser.command()
    def agreement(self):
        if self._opts.status:
            uc = UserConfig(self._opts.user)

result = {}
            for feature in ALL_OPTIMIZATION_FEATURES:
                result[feature.NAME] = uc.get_license_approve_status(feature).name

return {'licenses': result}
        elif self._opts.approve:
            feature = Feature(self._opts.approve)

uc = UserConfig(self._opts.user)
            uc.approve_license_agreement(feature)
            return {
                'status': "License agreement accepted"
            }
        elif self._opts.text:
            feature = Feature(self._opts.text)

if not feature.HAS_LICENSE_TERMS:
                return {
                    'result': "NO_LICENSE_TERMS"
                }

return {
                'text': open(feature.LICENSE_TERMS_PATH).read()
            }
        else:
            raise NotImplementedError('Unknown command')

def _get_for_domain(self, docroot, doc_root_info, visible_modules, general_feat_issues, docroots_features_issues):
        """
        Function to get information about features and wp for a domain.
        :return: list of dictionaries
        "wps_out": [
                {
                    "features": {
                        "accelerate_wp": {
                            "enabled": false,
                            "visible": true
                        },
                        "cdn": {
                            "enabled": false,
                            "visible": false
                        },
                        ...
                    },
                    "path": "",
                    "version": "x.x.x"
                }
            ]
        """
        wps_out = []
        wps_to_process = []

# Check if wp_path is specified and exists in the list of WPs
        wp_path_specified = hasattr(self._opts, 'wp_path') and self._opts.wp_path
        wp_path_found = False

for wp in doc_root_info["wps"]:
            if wp_path_specified:
                if wp["path"] == self._opts.wp_path:
                    wps_to_process.append(wp)
                    wp_path_found = True
                    break
            else:
                wps_to_process.append(wp)

# If wp_path was specified but not found, process all WPs
        if wp_path_specified and not wp_path_found:
            wps_to_process = doc_root_info["wps"]
        for wp in wps_to_process:
            feats = {}
            for feat in ALL_OPTIMIZATION_FEATURES:
                if feat in get_server_wide_options().hidden_features:
                    logger.warning('Optimization feature "%s" is skipped, because it is hidden by administrator', feat)
                    continue
                is_enabled = self._is_enabled(doc_root_info["domains"][0], wp["path"], feat)
                module_info = {"enabled": is_enabled, "visible": feat in visible_modules}

if doc_root_info['php_version'] is None:
                    wordpress_issues = []
                else:
                    wordpress_issues = self.collect_wordpress_issues(
                        feat, wp, docroot, module_is_enabled=is_enabled
                    )
                issues = [
                    *general_feat_issues[feat],
                    *docroots_features_issues[docroot][feat],
                    *wordpress_issues,
                ]

if issues:
                    module_info.update({"issues": [
                        issue.dict_repr for issue in issues
                    ]})

feats[feat.to_interface_name()] = module_info

wps_out += [
                {
                    "path": wp["path"],
                    "version": wp["version"] or "UNKNOWN",
                    "features": feats,
                },
            ]
        return wps_out

@track_command_progress
    def _get(self):
        allowed_features_dict = get_allowed_features_dict(self._opts.user.pw_uid)

visible_features_dict = get_visible_features_dict(self._opts.user.pw_uid)
        visible_features = list(chain(*visible_features_dict.values()))

hidden_features_set = set(get_server_wide_options().hidden_features)

converted_allowed_features = {
            feature_set: convert_feature_list_to_interface([
                feature for feature in features if feature not in hidden_features_set
            ])
            for feature_set, features in allowed_features_dict.items()
        }

converted_visible_features = {
            feature_set: convert_feature_list_to_interface([
                feature for feature in features if feature not in hidden_features_set
            ])
            for feature_set, features in visible_features_dict.items()
        }

with enable_caching():
            try:
                phpinfo = php_info()
            except WposError:
                if visible_features:
                    raise
                # no modules allowed and the fact that we cant retrieve php_info
                # probably means that Wpos was never allowed for user
                return {"docroots": [], "allowed_features": converted_allowed_features,
                        "visible_features": converted_visible_features}

self.command_progress.update()

user_info = self._get_user_info()
            visible_modules = get_visible_modules(self._opts.user.pw_uid)

self.command_progress.recalculate_number_of_total_stages(user_info)
            self.command_progress.update()

# Collect issues which depends only on a feature
            general_feat_issues = {}
            for feat in ALL_OPTIMIZATION_FEATURES:
                if feat in get_server_wide_options().hidden_features:
                    logger.warning('Optimization feature "%s" is skipped, because it is hidden by administrator', feat)
                    continue

general_feat_issues[feat] = self.collect_general_issues(feat)

# Collect issues which depends on a docroot and feature
            docroots_features_issues = {
                docroot: {
                    feat: self.collect_docroot_issues(
                        feat,
                        doc_root_info,
                        phpinfo,
                        visible_features,
                    )
                    for feat in general_feat_issues.keys()
                }
                for docroot, doc_root_info in user_info.items()
            }

# Collect WordPress installation issues for each feature and gather result
            docroots = []

# Filter docroots if a domain is specified
            docroots_to_process = user_info.items()
            if hasattr(self._opts, 'domain') and self._opts.domain:
                filtered_docroots = [(docroot, doc_root_info) for docroot, doc_root_info in user_info.items() 
                                    if self._opts.domain in doc_root_info['domains']]
                if not filtered_docroots:
                    error_message = _("Domain '%(domain)s' not found in any available docroots")
                    formatted_message = error_message % {"domain": self._opts.domain}
                    raise WposError(
                        message=formatted_message,
                        context={"domain": self._opts.domain}
                    )
                docroots_to_process = filtered_docroots

for docroot, doc_root_info in docroots_to_process:
                wps_out = self._get_for_domain(docroot, doc_root_info, visible_modules,
                                               general_feat_issues, docroots_features_issues)
                docroots += [{
                    **doc_root_info,
                    "wps": wps_out,
                    # convert output to old defined format
                    "php_version": doc_root_info["php_version"].identifier
                    if doc_root_info["php_version"] else None,
                }]

return {
            "docroots": docroots,
            "allowed_features": converted_allowed_features,
            "visible_features": converted_visible_features,
            **self._get_billing_status(),
            **self._get_redis_status()
        }

def _get_billing_status(self):
        subscriptions = get_subscriptions_by_pw(self._opts.user)

return {
            "subscription": subscriptions,
            "upgrade_url": {
                feature.lower(): self._get_upgrade_url(feature.lower())
                for feature in {OBJECT_CACHE_FEATURE.NAME, CDN_FEATURE.NAME}
            },
        }

def _get_redis_status(self):
        return {
            "used_memory": RedisLibUser(self.redis_socket_path).get_redis_used_memory(),
            "max_cache_memory": UserConfig(self._opts.user).get_config().get(
                "max_cache_memory", UserConfig.DEFAULT_MAX_CACHE_MEMORY
            )
        }

def _get_upgrade_url(self, feature: str) -> str | None:
        try:
            upgrade_url = daemon_communicate({
                "command": WposDaemon.DAEMON_GET_UPGRADE_LINK_COMMAND,
                "feature": feature
            })["upgrade_url"]
        except WposError:
            upgrade_url = None
        return upgrade_url

@parser.argument('--website', type=str, help='Website to scan', required=True)
    @parser.command()
    def scan(self):
        """
        ATTENTION: --website = actually domain, -> pass domain name instead of website !!!
        """
        u_id = self._opts.user.pw_uid
        allowed_modules, visible_modules = get_allowed_modules(u_id), get_visible_modules(u_id)

try:
            phpinfo = php_info()
        except WposError:
            if visible_modules:
                raise
            phpinfo = None

user_info = self._get_user_info()

wps = {}
        for module_name in ALL_OPTIMIZATION_FEATURES:
            general_issues = self.collect_general_issues(module_name)

if module_name not in visible_modules:
                general_issues.append(
                    CompatibilityIssue(
                        header=_('Feature was not made visible'),
                        description=_('Optimization feature was not made visible to user'),
                        fix_tip=_('Contact Administrator to allow using optimization feature'),
                        unique_id=UniqueId.FEATURE_NOT_MADE_VISIBLE,
                        telemetry=dict(reason='FEATURE_NOT_MADE_VISIBLE')
                    )
                )

if is_feature_hidden_server_wide(module_name):
                general_issues.append(
                    CompatibilityIssue(
                        header=_('Feature is hidden server wide'),
                        description=_('Optimization feature is hidden in server wide settings'),
                        fix_tip=_('Contact Administrator to make non-hidden'),
                        unique_id=UniqueId.FEATURE_HIDDEN_SERVER_WIDE,
                        telemetry=dict(reason='FEATURE_HIDDEN_SERVER_WIDE')
                    )
                )

for docroot, doc_root_info in user_info.items():
                if self._opts.website not in doc_root_info["domains"]:
                    continue

if module_name in visible_modules:
                    docroot_issues = self.collect_docroot_issues(
                        module_name, doc_root_info, phpinfo, allowed_modules)

for wp in doc_root_info["wps"]:
                    is_enabled = self._is_enabled(doc_root_info["domains"][0], wp["path"], module_name)

incompatibilities = []
                    if is_enabled:
                        wps.setdefault(wp['path'], []).append({
                            'type': UniqueId.CLOUDLINUX_MODULE_ALREADY_ENABLED,
                            'context': dict(),
                            'advice_type': module_name.NAME
                        })

elif module_name in visible_modules:
                        if doc_root_info['php_version'] is None:
                            wordpress_issues = []
                        else:
                            wordpress_issues = self.collect_wordpress_issues(
                                module_name, wp, docroot, module_is_enabled=is_enabled)

incompatibilities = [*general_issues, *docroot_issues, *wordpress_issues]
                    else:
                        incompatibilities = [*general_issues]

wps.setdefault(wp['path'], []).extend([
                        {
                            'type': issue.unique_id,
                            'context': issue.telemetry,
                            'advice_type': module_name.NAME
                        }
                        for issue in incompatibilities if isinstance(issue, CompatibilityIssue)
                    ])

# convert output to old defined format
        for document_root in user_info.values():
            if not document_root["php_version"]:
                continue
            document_root["php_version"] = document_root["php_version"].identifier

return {
            "issues": wps
        }

def _maxcache(self):
        action = 'enable' if getattr(self._opts, 'enable', False) else 'disable'
        success_key = 'maxcache_enabled' if action == 'enable' else 'maxcache_disabled'
        domain_filter = getattr(self._opts, 'domain', None)

user_pw = self._opts.user if hasattr(self._opts.user, 'pw_name') else get_pw()
        username = user_pw.pw_name

summary = {
            success_key: [],
            'error': [],
            'accelerate_wp_plugin_needs_to_be_updated': [],
            'accelerate_wp_plugin_not_installed': [],
        }

user_info = self._get_user_info()
        wp_cli_env = obtain_wp_cli_env()

for docroot, info in user_info.items():
            domains = info.get('domains') or []
            domain = domains[0] if domains else None
            if domain_filter and domain_filter not in domains:
                continue
            for wp in info.get('wps', []):
                wp_path = (wp.get('path') or '').strip("/")
                label_domain = domain or username
                label_path = wp_path or '/'
                site_label = f"{username}:{label_domain}:{label_path}"

abs_wp_path = os.path.join(docroot, wp_path) if wp_path else docroot
                if not os.path.isfile(os.path.join(abs_wp_path, 'wp-content/plugins/clsop/clsop.php')):
                    summary['accelerate_wp_plugin_not_installed'].append(site_label)
                    continue

try:
                    result = wordpress(abs_wp_path, 'accelerate-wp', 'maxcache', action, env=wp_cli_env)
                except Exception:
                    logger.exception("MAx Cache %s failed for %s", action, site_label)
                    summary['error'].append(site_label)
                    continue

if isinstance(result, WordpressError):
                    err_text = f"{result.message} {result.context}"
                    if 'not a registered' in err_text.lower():
                        summary['accelerate_wp_plugin_needs_to_be_updated'].append(site_label)
                    else:
                        summary['error'].append(site_label)
                else:
                    summary[success_key].append(site_label)

if domain_filter and not any(domain_filter in item for sublist in summary.values() for item in sublist):
            # domain was requested but not found in user info -> treat as not installed
            summary['accelerate_wp_plugin_not_installed'].append(f"{username}:{domain_filter}:/")

return summary

def _get_user_info(self, docroot: str = None):
        def reformat_php_data(info):
            """"""
            php = info.pop("php", {})
            info["php_version"] = php.get("version")
            info["php_handler"] = php.get("handler")

if docroot is None:
            user_info = get_user_info()
            for doc_root_info in user_info.values():
                reformat_php_data(doc_root_info)
        else:
            user_info = get_user_info()[docroot]
            reformat_php_data(user_info)

return user_info

def get_current_issues_by_docroot(
            self, module_name: Feature, wordpress_path: str, docroot: str, is_module_enabled: bool
    ):
        """
        Obtains issues for special docroot and wordpress
        """
        wordpress_issues = []
        allowed_modules = get_allowed_modules(self._opts.user.pw_uid)
        general_issues = self.collect_general_issues(module_name)
        phpinfo = php_info()
        user_info_by_docroot = self._get_user_info(docroot)
        docroot_issues = self.collect_docroot_issues(module_name, user_info_by_docroot, phpinfo, allowed_modules)
        for wp in user_info_by_docroot["wps"]:
            if wp['path'] != wordpress_path:
                continue
            wordpress_issues = self.collect_wordpress_issues(module_name, wp, docroot,
                                                             module_is_enabled=is_module_enabled)
        return [issue.dict_repr for issue in (*general_issues, *docroot_issues, *wordpress_issues)]

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

@parser.argument('--max_cache_memory', help='Maximum cache memory to use in MB',
                     type=MaxCacheMemory, default=constants.DEFAULT_MAX_CACHE_MEMORY)
    @parser.command()
    @check_license_decorator
    def set(self):
        uc = UserConfig(self._opts.user)
        # 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

def _disable_all(self):
        def _disable_single(docroot, site, feature_name):
            last_error = disable_without_config_affecting(
                DocRootPath(doc_root), site, module=feature_name
            )
            if last_error:
                raise WposError(
                    'Error(s) "%(errors)s" during disabling plugin in WordPress',
                    context={'errors': last_error}
                    )
            uc.disable_module(domain=None, wp_path=site, module=feature_name, relative_docroot=docroot)

uc = UserConfig(self._opts.user)
        all_enabled_features = uc.enabled_modules()
        deactivated_objects = []
        for doc_root, site, feature_name in all_enabled_features:
            try:
                _disable_single(doc_root, site, feature_name)
                deactivated_objects.append(
                    ManagedFeature(docroot=doc_root, website=site, feature_name=feature_name)
                )
            except Exception:
                logger.exception("Cannot disable feature for: %s", str((doc_root, site, feature_name)))
                continue

if self._is_redis_daemon_reload_needed(uc, feature=Feature(feature_name)):
                try:
                    reload_redis(skip_last_reload_time='yes')
                except Exception:
                    logger.exception("Failed to reload redis")

return {"disabled": [asdict(item) for item in deactivated_objects]}

@track_command_progress
    def _disable(self):
        username, doc_root = check_domain(self._opts.domain)
        wp_path = self._opts.wp_path.strip("/")

if '..' in wp_path.split('/'):
            error_and_exit(
                self._is_json,
                {"result": "Invalid wp_path: must not contain '..' components"}
            )

abs_wp_path = os.path.realpath(os.path.join(doc_root, wp_path))
        if not abs_wp_path.startswith(os.path.realpath(doc_root) + '/') \
                and abs_wp_path != os.path.realpath(doc_root):
            error_and_exit(
                self._is_json,
                {"result": "Invalid wp_path: resolved path is outside the document root"}
            )

uc = UserConfig(self._opts.user)
        errors = []

feature_name = self._opts.feature.optimization_feature()

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

self._check_monitoring_daemon_and_exit(feature=feature_name)

self.command_progress.update()

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

if last_error:
            logger.error(last_error.message % last_error.context)
            errors.append(last_error)
        else:
            # do not change config values in case if disable reported errors
            try:
                uc.disable_module(self._opts.domain, wp_path, feature_name)
            except Exception as e:
                logger.exception("unable to disable module in config")
                errors.append(e)

self.command_progress.update()

if self._is_redis_daemon_reload_needed(uc, feature=feature_name):
            try:
                reload_redis(skip_last_reload_time='yes')
            except WposError as e:
                s_details = e.details % e.context
                logger.exception("CLWPOS daemon error: '%s'; details: '%s'", e.message, s_details)
                errors.append(e)
            except Exception as e:
                logger.exception("unable to reload cache backend")
                errors.append(e)

self.command_progress.update()

is_module_enabled = self._is_enabled(self._opts.domain, wp_path, feature_name)
        is_module_visible = feature_name in get_visible_modules(self._opts.user.pw_uid)

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

if is_module_enabled:
            if last_error:
                raise WposError(
                    message=_("Optimization feature 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=self._opts.user.pw_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:
            response = {
                "feature": {
                    "enabled": is_module_enabled,
                    "visible": is_module_visible
                },
                "used_memory": RedisLibUser(self.redis_socket_path).get_redis_used_memory()
            }
            if last_error:
                response.update(
                    {
                        "warning": _("Optimization feature 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=self._opts.user.pw_dir)
                        }
                    }
                )
            else:
                issues = self.get_current_issues_by_docroot(feature_name, wp_path, doc_root, is_module_enabled)
                if issues:
                    response["feature"]["issues"] = issues

return response

@parser.argument('--feature', type=str, help='List of AccelerateWP optimization features separated by commas '
                                                 'on which to perform an action',
                     choices=ALL_OPTIMIZATION_FEATURES, default=OBJECT_CACHE_FEATURE)
    @parser.argument('--action', type=str, help='Action to perform on AccelerateWP optimization features',
                     required=True)
    @parser.command()
    def do_action(self):
        """
        Perform action on selected AccelerateWP features
        """
        action = self._opts.action
        feature = self._opts.feature
        # 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](feature)  # perform action on module

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

@check_license_decorator
    @track_command_progress
    def _enable(self):
        """
        Enable object_cache for user with compliance with end-user spec.
        :return:
        """
        wp_path = self._opts.wp_path.strip("/")

if '..' in wp_path.split('/'):
            error_and_exit(
                self._is_json,
                {"result": "Invalid wp_path: must not contain '..' components"}
            )

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

make_web_check = not any([self._opts.ignore_errors, self._opts.skip_dns_check])

# bypass to allow approvements in UI which has checkbox
        # asking "do you agree" right before we install the feature
        if self._opts.approve_license_terms:
            uc.approve_license_agreement(feature_name)

if uc.get_license_approve_status(feature_name) == LicenseApproveStatus.NOT_APPROVED:
            raise WposError('License approve required to use this feature. '
                            'Open AccelerateWP plugin in your control panel, enable feature and '
                            'accept terms and conditions to proceed with installation.')

included_features = feature_name.included_optimization_features()

is_module_enabled = all([self._is_enabled(self._opts.domain, wp_path, feature)
                                 for feature in included_features])
        if is_module_enabled:
            return {"warning": _("The %(feature)s optimization feature is already enabled on "
                                 "the domain %(domain)s. Nothing to be done."),
                    "context": {"domain": self._opts.domain, "feature": self._opts.feature}}

if feature_name in get_server_wide_options().hidden_features:
            error_and_exit(
                self._is_json,
                {
                    "result": _("Optimization feature %(feature)s is hidden by administrator."),
                    "context": {"feature": self._opts.feature},
                }
            )

is_feature_visible = all([feature in get_visible_modules(self._opts.user.pw_uid) for feature in included_features])
        is_feature_allowed = all([feature in get_allowed_modules(self._opts.user.pw_uid) for feature in included_features])
        # check that enabling of module is allowed by admin
        if not is_feature_allowed:
            if is_feature_visible:
                from clwpos.daemon import WposDaemon
                error_and_exit(
                    self._is_json,
                    {
                        "result": _("PAYMENT_REQUIRED"),
                        "upgrade_url": daemon_communicate({
                            "command": WposDaemon.DAEMON_GET_UPGRADE_LINK_COMMAND,
                            'feature': feature_name
                        })["upgrade_url"],
                        # keep it here so UI can still highlight it
                        "context": {"feature": self._opts.feature},
                    }
                )
            else:
                error_and_exit(
                    self._is_json,
                    {
                        "result": _("Usage of the optimization feature %(feature)s is prohibited by admin."),
                        "context": {"feature": self._opts.feature},
                    }
                )

self._check_redis_configuration_process_and_exit(feature=feature_name)
        self._check_monitoring_daemon_and_exit(feature=feature_name)

self.command_progress.update()

with enable_caching():
            # check that user's wordpress fits to requirements
            issues = self.get_current_issues_by_docroot(
                feature_name, wp_path, doc_root, is_module_enabled)
            if issues:
                error_and_exit(
                    self._is_json,
                    {'feature': {'enabled': is_module_enabled, 'issues': issues},
                     'result': _('Website "%(domain)s/%(wp_path)s" has compatibility '
                                 'issues and optimization feature cannot be enabled.'),
                     'context': {'domain': self._opts.domain, 'wp_path': wp_path}},
                )

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

self.command_progress.update()
            #
            # # redis object cache plugin checks connectivity to redis during activation
            if self._is_redis_daemon_reload_needed(uc, feature=feature_name, before_config_modified=True):
                reload_redis(uid=pwd.getpwnam(username).pw_uid, force='yes', skip_last_reload_time='yes')

self.command_progress.update()

# try to enable module with wp-cli without adding info into user's wpos config
            ok, enable_failed_info = enable_without_config_affecting(
                DomainName(self._opts.domain),
                wp_path,
                module=feature_name,
                ignore_errors=self._opts.ignore_errors
            )
        if not ok:
            # reload once again, because plugin activation failed
            if self._is_redis_daemon_reload_needed(uc, feature=feature_name, before_config_modified=True):
                reload_redis(uid=pwd.getpwnam(username).pw_uid, skip_last_reload_time='yes')
            if 'result' in enable_failed_info:
                error_and_exit(self._is_json, message=enable_failed_info)

raise WposError(**enable_failed_info)

for feature in included_features:
            uc.enable_module(self._opts.domain, wp_path, feature)

self.command_progress.update()

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

self.command_progress.update()

is_module_enabled = all([self._is_enabled(self._opts.domain, wp_path, feature)
                                 for feature in included_features])
        with enable_caching():
            issues = self.get_current_issues_by_docroot(feature_name, wp_path, doc_root, is_module_enabled)

module_data = {
            'enabled': is_module_enabled
        }
        if issues:
            module_data.update({
                'issues': issues
            })

return {
            'feature': module_data,
            'used_memory': RedisLibUser(self.redis_socket_path).get_redis_used_memory()
        }

def _website_check(self, abs_wp_path, domain, uc, wp_path, module, 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, module)
                reload_redis(skip_last_reload_time='yes')
                Feature(module).disable(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)
                },
            )

@parser.command()
    def get_progress(self):
        """
        Return current progress of the currently performed command.
        Provides amount of:
        - total stages - how many stages the command execution
          process is divided into,
        - completed stages - how many stages have been already completed.
        """
        return CommandProgress.get_status()

def _object_cache_banner(self):
        disable_object_cache_banners = False
        if self._opts.disable:
            disable_object_cache_banners = True

if self._opts.all:
            user_info = self._get_user_info()

for docroot, doc_root_info in user_info.items():
                for wp in doc_root_info["wps"]:
                    self._object_cache_banner_toggle(
                        disable_object_cache_banners, docroot, wp["path"], doc_root_info["domains"][0]
                    )

return {}
        elif self._opts.domain:
            domain = self._opts.domain
            try:
                username, docroot = check_domain(domain)
            except WposError:
                raise
            self._object_cache_banner_toggle(
                disable_object_cache_banners, docroot, self._opts.wp_path, self._opts.domain
            )
            return {}
        else:
            raise NotImplementedError('Unknown command')

def _object_cache_banner_toggle(self, disable_banner: bool, abs_wp_path: str, wp_path: str, domain: str):
        module_enabled = self._is_enabled(domain, wp_path, OBJECT_CACHE_FEATURE)

if module_enabled is False:
            constant_value = None
        else:
            constant_value = str(disable_banner).lower()

abs_wp_path = os.path.join(abs_wp_path, wp_path)
        # abs_wp_path is <docroot>+<wp_name>
        res = update_redis_disable_banners_constant(abs_wp_path, constant_value)

if isinstance(res, WordpressError):
            raise WposError(message=res.message, context=res.context)

🔐 Sid Gifari File Manager Pro