.

. BSA HACKER
Server : Apache
System : Linux nusantara.hosteko.com 4.18.0-553.16.1.lve.el8.x86_64 #1 SMP Tue Aug 13 17:45:03 UTC 2024 x86_64
User : koperas1 ( 1254)
PHP Version : 7.4.33
Disable Function : NONE
Directory : /opt/cloudlinux/venv/lib64/python3.11/site-packages/clwpos/
Upload File :
Current File : //opt/cloudlinux/venv/lib64/python3.11/site-packages/clwpos/utils.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_lib.py - helper functions for clwpos utility

from __future__ import absolute_import

import contextlib
import dataclasses
import datetime
import itertools
import logging
import os
import re
import shutil
import struct
import sys
import time
import json
import pwd
import typing

import fcntl
import uuid
import subprocess
from dataclasses import dataclass, asdict, field
from glob import iglob
from enum import Enum
from gettext import gettext as _
from urllib.parse import (
    urlencode,
    urlparse,
    parse_qsl,
    urlunparse
)
from packaging.version import Version

import psutil
from contextlib import contextmanager
from functools import wraps, lru_cache
from pathlib import Path
from socket import socket, AF_UNIX, SOCK_STREAM
from typing import List, Tuple, Optional, Set, ContextManager
import platform

from secureio import write_file_via_tempfile, disable_quota
from clcommon.cpapi.cpapiexceptions import NoDomain
from clcommon.clpwd import ClPwd, drop_privileges
from clcommon.clcaptain import mkdir
from clcommon.lib.cledition import (
    is_cl_shared_pro_edition,
    CLEditionDetectionError
)
from clcommon.lib.jwt_token import read_jwt, decode_jwt
from clcommon.lib.consts import CLN_JWT_TOKEN_PATH, DEFAULT_JWT_ES_TOKEN_PATH
from jwt import PyJWTError, exceptions

from cllicenselib import check_license
from clcommon.cpapi import docroot, get_domain_login, get_server_ip, cpusers
from clcommon.utils import exec_utility, run_command, demote
from clwpos import gettext, wp_config
from clwpos.cl_wpos_exceptions import (
    WposError,
    WPOSLicenseMissing,
    WpCliUnsupportedException,
    WpNotExists,
    WpConfigWriteFailed,
    PhpBrokenException
)
from clcommon.ui_config import UIConfig
from clcommon.clcagefs import in_cagefs, _is_cagefs_enabled
from clcommon.const import Feature
from clcommon.cpapi import is_panel_feature_supported

from .logsetup import setup_logging

from clwpos.constants import (
    USER_WPOS_DIR,
    WPOS_DAEMON_SOCKET_FILE,
    CLCONFIG_UTILITY,
    RedisRequiredConstants,
    CAGEFS_ENTER_USER_BIN,
    CAGEFS_ENTER_UTIL,
    CLWPOS_OPT_DIR,
    ALT_PHP_PREFIX,
    EA_PHP_PREFIX,
    PLESK_PHP_PREFIX,
    USER_CLWPOS_CONFIG,
    PUBLIC_OPTIONS,
    SUITES_MARKERS,
    XRAY_MANAGER_UTILITY,
    XRAY_USER_SOCKET,
)

from .socket_utils import pack_data_for_socket, read_unpack_response_from_socket_client
from .user.website_check.errors import RollbackException
from clwpos.scoped_cache import cached_in_scope

if typing.TYPE_CHECKING:
    from clwpos.php.base import PHP

logger = None


def catch_error(func):
    """
    Decorator for catching errors
    """

    def func_wrapper(self, *args, **kwargs):
        global logger
        if logger is None:
            logger = setup_logging(__name__)
        try:
            return func(self, *args, **kwargs)
        except RollbackException as e:
            error_and_exit(self._is_json, {
                'context': e.context,
                'result': e.message,
                'issues': e.errors
            })
        except WposError as e:
            if isinstance(e, WPOSLicenseMissing):
                logger.warning(e)
            else:
                logger.exception(e)
            response = {'context': e.context, 'result': e.message, 'warning': e.warning}
            if e.details:
                response['details'] = e.details
            error_and_exit(self._is_json, response)
        except Exception as e:
            logger.exception(e)
            error_and_exit(self._is_json, {'context': {}, 'result': str(e)})

    return func_wrapper


class ExtendedJSONEncoder(json.JSONEncoder):
    """
    Makes it easier to use ENUMs and DATACLASSes in program,
    automatically converting them when json is printed.
    """
    def __init__(self, **kwargs):
        kwargs['ensure_ascii'] = False
        super().__init__(**kwargs)

    def default(self, obj):
        if isinstance(obj, Enum):
            return obj.value
        elif isinstance(obj, (datetime.date, datetime.datetime)):
            return obj.isoformat()
        elif isinstance(obj, Version):
            return str(obj)
        elif dataclasses.is_dataclass(obj):
            return dataclasses.asdict(obj)
        return json.JSONEncoder.default(self, obj)


def _print_dictionary(data_dict, is_json: bool = False, is_pretty: bool = False):
    """
    Print specified dictionary
    :param data_dict: data dictionary to print
    :param is_json: True - print in JSON, False - in text
    :param is_pretty: True - pretty json print, False - none (default)
    :return: None
    """
    if is_json:
        # Print as JSON
        if is_pretty:
            print(json.dumps(data_dict, indent=4, sort_keys=True, cls=ExtendedJSONEncoder))
        else:
            print(json.dumps(data_dict, sort_keys=True, cls=ExtendedJSONEncoder))
    else:
        # Print as text
        print(data_dict)


def error_and_exit(is_json: bool, message: dict, error_code: int = 1):
    """
    Print error and exit
    :param is_json:
    :param message: Dictionary with keys "result" as string and optional "context" as dict
    :param error_code: Utility return code on error
    """
    if 'warning' in message.keys() and not message.get('warning'):
        message.pop('warning')

    if is_json:
        message.update({"timestamp": time.time()})
        _print_dictionary(message, is_json, is_pretty=True)
    else:
        try:
            print(str(message["result"]) % message.get("context", {}))
        except KeyError as e:
            print("Error: %s [%s]" % (str(e), message))
    sys.exit(error_code)


def print_data(is_json: bool, data: dict, result="success"):
    """
    Output data wrapper
    :param is_json:
    :param data: data for output to stdout
    :param result:
    """
    if isinstance(data, dict):
        data.update({"result": result, "timestamp": time.time()})
    _print_dictionary(data, is_json, is_pretty=True)


def is_run_under_user() -> bool:
    """
    Detects is we running under root
    :return: True - user, False - root
    """
    return os.geteuid() != 0


def is_shared_pro_safely(safely: bool):
    """
    Detecting of shared_pro edition depends on jwt token
    There are some cases when we do not fail if there are
    cases with decoding (e.g summary collection)
    """
    try:
        return is_cl_shared_pro_edition()
    except CLEditionDetectionError:
        if safely:
            return False
        else:
            raise


def is_wpos_supported() -> bool:
    """
    Сheck if system environment is supported by WPOS
    :return:
        True - CPanel/Plesk on Solo/ CL Shared Pro/ CL Admin
        False - else
    """
    # is_panel_feature_supported() already knows edition specific available features
    return is_panel_feature_supported(Feature.WPOS)


def create_clwpos_dir_if_not_exists(user_pw: pwd.struct_passwd):
    """
    Creates {homedir}/.clwpos directory if it's not exists
    """
    clwpos_dir = os.path.join(user_pw.pw_dir, USER_WPOS_DIR)
    if not os.path.isdir(clwpos_dir):
        mkdir(clwpos_dir, mode=0o700)


def get_relative_docroot(domain, homedir):
    dr = docroot(domain)[0]
    if not dr.startswith(homedir):
        raise WposError(f"docroot {dr} for domain {domain} should start with {homedir}")
    return dr[len(homedir):].lstrip("/")


def home_dir(username: str = None) -> str:
    pw = get_pw(username=username)
    return pw.pw_dir


def user_name() -> str:
    return get_pw().pw_name


def user_uid(*, username: str = None) -> int:
    return get_pw(username=username).pw_uid


def get_pw(*, username: str = None):
    if username:
        return pwd.getpwnam(username)
    else:
        return pwd.getpwuid(os.geteuid())


class WposUser:
    """
    Helper class to construct paths to user's WPOS dir and files inside it.
    """

    def __init__(self, username: str, homedir: str = None) -> None:
        self.name = username
        self.home_dir = home_dir(username) if homedir is None else homedir
        self.wpos_dir = os.path.join(self.home_dir, USER_WPOS_DIR)
        self.wpos_config = os.path.join(self.wpos_dir, USER_CLWPOS_CONFIG)
        self.redis_conf = os.path.join(self.wpos_dir, 'redis.conf')
        self.redis_socket = os.path.join(self.wpos_dir, 'redis.sock')
        self.php_info = os.path.join(self.wpos_dir, '.php_info-{file_id}')

    def __eq__(self, other):
        return self.name == other.name

    def __hash__(self):
        return hash(self.name)


def daemon_communicate(cmd_dict: dict) -> Optional[dict]:
    """
    Send command to CLWPOS daemon via socket
    :param cmd_dict: Command dictionary
    :return: Daemon response as dictionary, None - daemon data/socket error
    """
    bytes_to_send = pack_data_for_socket(cmd_dict)
    with socket(AF_UNIX, SOCK_STREAM) as s:
        try:
            s.connect(WPOS_DAEMON_SOCKET_FILE)
            s.sendall(bytes_to_send)
            # to not hang forever
            s.settimeout(120)
            response_dict = read_unpack_response_from_socket_client(s)
            if response_dict is None or not isinstance(response_dict, dict):
                raise WposError(
                    message=gettext('Unexpected response from daemon. '
                                    'Report this issue to your system administrator.'),
                    details=str(response_dict),
                    context={})
            if response_dict['result'] != 'success':
                raise WposError(message=gettext('Daemon was unable to execute the requested command.'),
                                details=response_dict['result'],
                                context=response_dict.get('context'))
            return response_dict
        except FileNotFoundError:
            raise WposError(gettext('CloudLinux AccelerateWP daemon socket (%(filename)s) not found. '
                                    'Contact your system administrator.'),
                            {'filename': WPOS_DAEMON_SOCKET_FILE})
        except (ConnectionError, OSError, IOError, AttributeError, struct.error, KeyError) as e:
            raise WposError(gettext('Unexpected daemon communication error.'), details=str(e))


def redis_cache_config_section() -> List[str]:
    """
    Construct list of lines (configuration settings)
    that should be in Wordpress config file to enable redis.
    Please note that deleting of the plugin would flush all keys related to the plugin (site) from redis.
    REDIS_PREFIX and SELECTIVE_FLUSH in wp-config.php would guarantee that plugin will not flush keys unrelated
    to this plugin (site)
    """
    disable_banners_value = "false"
    if get_server_wide_options().disable_object_cache_banners:
        disable_banners_value = "true"

    socket_path = os.path.join(home_dir(), USER_WPOS_DIR, 'redis.sock')
    prefix_uuid = uuid.uuid4()
    redis_prefix = RedisRequiredConstants.WP_REDIS_PREFIX
    redis_schema = RedisRequiredConstants.WP_REDIS_SCHEME
    redis_client = RedisRequiredConstants.WP_REDIS_CLIENT
    redis_flush = RedisRequiredConstants.WP_REDIS_SELECTIVE_FLUSH
    redis_graceful = RedisRequiredConstants.WP_REDIS_GRACEFUL
    disable_banners = RedisRequiredConstants.WP_REDIS_DISABLE_BANNERS

    return ["// Start of CloudLinux generated section\n",
            f"define('{redis_schema.name}', '{redis_schema.val}');\n",
            f"define('{RedisRequiredConstants.WP_REDIS_PATH.name}', '{socket_path}');\n",
            f"define('{redis_client.name}', '{redis_client.val}');\n",
            f"define('{redis_graceful.name}', '{redis_graceful.val}');\n",
            f"define('{redis_prefix.name}', '{redis_prefix.val}{prefix_uuid}');\n",
            f"define('{redis_flush.name}', {redis_flush.val});\n",
            f"define('{disable_banners.name}', {disable_banners_value});\n",
            "// End of CloudLinux generated section\n"]


def check_wp_config_existance(wp_config_path: str) -> None:
    """
    Check that wp-config.php exists inside Wordpress directory.
    :param wp_config_path: absolute path to Wordpress config file
    :raises: WposError
    """
    wp_path = os.path.dirname(wp_config_path)
    if not os.path.exists(wp_path):
        raise WpNotExists(wp_path)

    if not os.path.isfile(wp_config_path):
        raise WposError(message=gettext("Wordpress config file %(file)s is missing"),
                        context={"file": wp_config_path})


def clear_redis_cache_config(abs_wp_path: str) -> None:
    """
    Clear cloudlinux section with redis object cach config from docroot's wp-config.php
    :param abs_wp_path: Absolute path to WordPress
    :raises: WposError
    """
    wp_config_path = str(wp_config.path(abs_wp_path))
    check_wp_config_existance(wp_config_path)
    lines_to_filter = redis_cache_config_section()

    def __config_filter(line: str) -> bool:
        """
        Filter function that should delete CL config options from the `redis_cache_config_section()`
        """
        return line not in lines_to_filter and 'WP_REDIS_PREFIX' not in line

    try:
        wp_config_lines = wp_config.read(abs_wp_path)
        cleared_wp_config = list(filter(__config_filter, wp_config_lines))
        write_file_via_tempfile("".join(cleared_wp_config), wp_config_path, 0o600)
    except (OSError, IOError) as e:
        raise WpConfigWriteFailed(wp_config_path, e)


def create_redis_cache_config(abs_wp_path: str) -> None:
    """
    Create config for redis-cache.
    We use manual copy cause we want to preserve file metadata
    and permissions and also we could add some custom config editing in the future.
    :param abs_wp_path: absolute path to WordPress
    :raises: WposError
    """
    wp_config_path = str(wp_config.path(abs_wp_path))
    check_wp_config_existance(wp_config_path)

    try:
        backup_wp_config = f"{wp_config_path}.backup"
        if not os.path.isfile(backup_wp_config):
            shutil.copy(wp_config_path, backup_wp_config)

        absent_constants = {constant.name: constant.val for constant in RedisRequiredConstants}

        wp_config_lines = wp_config.read(abs_wp_path)
        cleaned_lines = []
        for line in wp_config_lines:
            absent_constants = {k: v for k, v in absent_constants.items() if f"define('{k}'" not in line}
            # nothing to do, all constants are already in conf
            if not absent_constants:
                return

            # cleanup existing consts, to rewrite all
            if not any(f"define('{redis_constant.name}'" in line for redis_constant in RedisRequiredConstants):
                cleaned_lines.append(line)

        updated_config = [
            cleaned_lines[0],
            *redis_cache_config_section(),
            *cleaned_lines[1:],
        ]
        write_file_via_tempfile("".join(updated_config), wp_config_path, 0o600)

    except (OSError, IOError) as e:
        raise WpConfigWriteFailed(wp_config_path, e)


def check_license_decorator(func):
    """Decorator to check for license validity
    """

    @wraps(func)
    def wrapper(*args, **kwargs):
        """License check wrapper"""
        if not check_license():
            raise WPOSLicenseMissing()
        return func(*args, **kwargs)

    return wrapper


def check_domain(domain: str) -> Tuple[str, str]:
    """
    Validates domain, determines it's owner and docroot or exit with error
    :param domain: Domain name to check
    :return: Tuple (username, docroot)
    """
    try:
        document_root, owner = docroot(domain)
        return owner, document_root
    except NoDomain:
        # No such domain
        raise WposError(message=gettext("No such domain: %(domain)s."), context={"domain": domain})


def lock_file(path: str, attempts: Optional[int]):
    """
    Try to take lock on file with specified number of attempts.
    """
    lock_type = fcntl.LOCK_EX
    if attempts is not None:
        # avoid blocking on lock
        lock_type |= fcntl.LOCK_NB
    try:
        lock_fd = open(path, "a+")
        for _ in range(attempts or 1):  # if attempts is None do 1 attempt
            try:
                fcntl.flock(lock_fd.fileno(), lock_type)
                break
            except OSError:
                time.sleep(0.3)
        else:
            raise LockFailedException(gettext("Another utility instance is already running. "
                                              "Try again later or contact system administrator "
                                              "in case if issue persists."))
    except IOError:
        raise LockFailedException(gettext("IO error happened while getting lock."))
    return lock_fd


class LockFailedException(Exception):
    """
    Exception when failed to take lock
    """
    pass


@contextmanager
def acquire_lock(resource_path: str, attempts: Optional[int] = 10):
    """
    Lock a file, than do something.
    Make specified number of attempts to acquire the lock,
    if attempts is None, wait until the lock is released.
    Usage:
    with acquire_lock(path, attempts=1):
       ... do something with files ...
    """
    lock_fd = lock_file(resource_path + '.lock', attempts)
    yield
    release_lock(lock_fd)


def release_lock(descriptor):
    """
    Releases lock file
    """
    try:
        # lock released explicitly
        fcntl.flock(descriptor.fileno(), fcntl.LOCK_UN)
    except IOError:
        # we ignore this cause process will be closed soon anyway
        pass
    descriptor.close()


@cached_in_scope
def wp_cli_compatibility_check(php_version: 'PHP'):
    """
    Ensures wp-cli is compatible, e.g some
    php modules may prevent stable work
    """
    dangerous_module = 'snuffleupagus'
    if 'ea-php74' == php_version.identifier \
            and php_version.is_extension_loaded(dangerous_module):
        raise WpCliUnsupportedException(message=gettext('Seems like ea-php74 %(module)s module is '
                                                        'enabled. It may cause instabilities while managing '
                                                        'Object Caching. Disable it and try again'),
                                        context={'module': dangerous_module})


def set_wpos_icon_visibility(hide: bool) -> Tuple[int, str]:
    """
    Call cloudlinux-config utility
    to hide/show WPOS icon in user's control panel interface.
    """
    params = [
        'set',
        '--data',
        json.dumps({'options': {'uiSettings': {'hideAccelerateWPApp': hide}}}),
        '--json',
    ]
    returncode, stdout = exec_utility(CLCONFIG_UTILITY, params)
    return returncode, stdout

def is_ui_icon_hidden(icon_name='hideAccelerateWPApp') -> bool:
    """
    Check the current state of WPOS icon in user's control panel interface
    """
    return UIConfig().get_param(icon_name, 'uiSettings')

def should_xray_user_agent_enabled(feature_visible):
    """
    1. xray utility exists = alt-php-xray package installed
    2. feature is visible
    """
    return all([os.path.exists(XRAY_MANAGER_UTILITY),
                feature_visible])

def should_xray_user_agent_disabled():
    """
    1. xray utility exists = alt-php-xray installed
    2. xray socket exists
    3. end-user plugin was not enabled by admin = hidden in UI
    """
    return all([os.path.exists(XRAY_MANAGER_UTILITY),
                    os.path.exists(XRAY_USER_SOCKET),
                    is_ui_icon_hidden(icon_name='hideXrayApp')])

@dataclass
class WHMCSServerWideOptions:
    allowed_suites: Optional[List] = field(default_factory=list)
    visible_suites: Optional[List] = field(default_factory=list)


@dataclass
class ServerWideOptions:
    """
    Options holder representing server-wide option
    available for reading for any user on server.

    Only can be changed by root.
    """
    show_icon: bool
    allowed_suites: List
    visible_suites: List
    supported_suites: List
    hidden_features: List

    whmcs_options: WHMCSServerWideOptions = field(default_factory=WHMCSServerWideOptions)
    disable_object_cache_banners: Optional[bool] = None
    disable_smart_advice_notifications: Optional[bool] = None
    disable_smart_advice_wordpress_plugin: Optional[bool] = None
    disable_smart_advice_reminders: Optional[bool] = None

    upgrade_url: Optional[str] = None
    upgrade_url_cdn: Optional[str] = None

    def get_upgrade_url_for_user(self, username, domain, feature='object_cache'):
        """
        Append some needed arguments to upgrade url to make it specific for user.
        Please pay attention that we add *customer_name* instead of system user,
        that may be different on plesk.
        """
        from clwpos.feature_suites import PremiumSuite, CDNSuitePro

        # we should keep all the features here because we have smart-advice
        # which displays upgrade links per-advice and those advices
        # may be for different features
        feature_to_suite = {
            **{feature: PremiumSuite.name for feature in PremiumSuite.primary_features},
            **{feature: CDNSuitePro.name for feature in CDNSuitePro.primary_features},
        }

        if feature not in feature_to_suite:
            return None

        target_url = None

        if feature in PremiumSuite.primary_features:
            if self.upgrade_url is None:
                return None
            target_url = self.upgrade_url
        if feature in CDNSuitePro.primary_features:
            if self.upgrade_url_cdn is None:
                return None
            target_url = self.upgrade_url_cdn

        if target_url is None:
            return None

        url_parts = list(urlparse(target_url))
        query = dict(parse_qsl(url_parts[4]))
        query.update({
            'username': get_domain_login(username, domain),
            'domain': domain,
            'server_ip': get_server_ip(),
            'm': 'cloudlinux_advantage',
            'action': 'provisioning',
            'suite': feature_to_suite[feature]
        })

        url_parts[4] = urlencode(query)

        return urlunparse(url_parts)

    @property
    def allowed_suites_list(self):
        return list(set(list(self.allowed_suites) + list(self.whmcs_options.allowed_suites)))

    @property
    def visible_suites_list(self):
        return list(set(list(self.visible_suites) + list(self.whmcs_options.visible_suites)))

    @property
    def allowed_features(self):
        # TODO: fix this circle import one day
        from .feature_suites import ALL_SUITES
        _allowed_features = set()
        for suite in self.allowed_suites_list:
            _allowed_features.update(ALL_SUITES[suite].feature_set)
        return _allowed_features

    @property
    def visible_features(self):
        from .feature_suites import ALL_SUITES
        _visible_features = set()
        for suite in self.visible_suites_list:
            _visible_features.update(ALL_SUITES[suite].feature_set)
        return _visible_features


def get_default_server_wide_options() -> ServerWideOptions:
    """
    Return default content of /opt/clwpos/public_config.json.
    This file is accessible by all users on server.
    """
    # circular import :(
    from .feature_suites import AWPSuite, PremiumSuite, CDNSuite, CDNSuitePro, SUPPORTED_SUITES

    is_icon_hidden = UIConfig().get_param('hideAccelerateWPApp', 'uiSettings')

    visible_suites = []
    allowed_suites = []

    # --allowed-for-all previously used marker files
    # to mark suites as enabled
    # we must keep that behaviour
    for suite in (PremiumSuite.name, AWPSuite.name, CDNSuite.name, CDNSuitePro.name):
        if not os.path.isfile(SUITES_MARKERS[suite]):
            continue
        visible_suites.append(suite)
        allowed_suites.append(suite)

    return ServerWideOptions(
        show_icon=not is_icon_hidden,
        allowed_suites=allowed_suites,
        visible_suites=visible_suites,
        supported_suites=list(SUPPORTED_SUITES),
        hidden_features=[]
    )

def get_supported_suites():
    """
    Get list of supported suites taking
    into account license and status on CLN.
    """
    from .feature_suites import (
        AWPSuite,
        PremiumSuite,
        CDNSuite,
        CDNSuitePro
    )

    # TODO: could we replace is_shared_pro_safely() with is_panel_feature_supported()?
    is_awp_premium_allowed = is_awp_cdn_allowed = is_shared_pro_safely(safely=True)
    if os.path.exists(CLN_JWT_TOKEN_PATH):
        jwt = _get_jwt_payload()

        is_awp_premium_allowed = jwt.get('is_awp_premium_allowed', is_awp_premium_allowed)
        is_awp_cdn_allowed = jwt.get('is_awp_cdn_allowed', is_awp_cdn_allowed)

    suites = itertools.compress(
        [AWPSuite, PremiumSuite, CDNSuite, CDNSuitePro],
        [True, is_awp_premium_allowed, is_awp_cdn_allowed, is_awp_cdn_allowed]
    )

    return [suite.name for suite in suites]


def _get_jwt_payload():
    """
    Read jwt, verify it and return payload.
    """
    token = read_jwt(CLN_JWT_TOKEN_PATH)
    try:
        jwt = decode_jwt(token, verify_exp=False)
    except PyJWTError as e:
        raise CLEditionDetectionError(f'Unable to detect edition from jwt token: {CLN_JWT_TOKEN_PATH}. '
                                      f'Please, make sure it is not broken, error: {e}')
    return jwt

def get_whcms_server_wide_options(server_wide_options) -> WHMCSServerWideOptions:
    if isinstance(server_wide_options.whmcs_options, WHMCSServerWideOptions):
        return server_wide_options.whmcs_options
    return WHMCSServerWideOptions(**server_wide_options.whmcs_options)

def get_server_wide_options() -> ServerWideOptions:
    """
    Gets server wide options which apply
    as defaults for all users
    """
    from .feature_suites import ALL_SUITES
    default_options = get_default_server_wide_options()
    if not os.path.isfile(PUBLIC_OPTIONS):
        return default_options

    with open(PUBLIC_OPTIONS, 'r') as f:
        content = f.read()

        try:
            configuration: dict = json.loads(content)

            # these two options have different way of merging: we
            # must sum them and keep only unique elements
            for option_to_merge in ['visible_suites', 'allowed_suites', 'supported_suites']:
                if option_to_merge not in configuration:
                    continue
                suites_from_config = configuration.pop(option_to_merge)
                suites_from_defaults = getattr(default_options, option_to_merge)
                # to filter out unknown suites from resulting structure
                # actually for downgrade cases, see AWP-272 for details
                merged_values = list(sorted(set(suites_from_defaults + list(set(
                    suites_from_config).intersection(set(ALL_SUITES))))))
                setattr(default_options, option_to_merge, merged_values)

            # the rest of the options just override their defaults
            default_options.__dict__.update(**configuration)
            default_options.whmcs_options = get_whcms_server_wide_options(default_options)

            # remove externally disabled suites from list
            try:
                server_suites_allowed = get_supported_suites()
            except PermissionError:
                # sometimes this function is called with user permissions
                # and we should handle error when trying to reach jwt token
                default_options.supported_suites = None
            else:
                for suite in default_options.supported_suites[:]:
                    if suite not in server_suites_allowed:
                        default_options.supported_suites.remove(suite)

            return default_options
        except json.decoder.JSONDecodeError as err:
            raise WposError(
                message=_("File is corrupted: Please, delete file %(config_file)s"
                          " or fix the line provided in details"),
                details=str(err),
                context={'config_file': PUBLIC_OPTIONS})


@contextmanager
def write_public_options() -> ContextManager[ServerWideOptions]:
    """Set icon visibility in clwpos public options file"""

    if not os.path.exists(CLWPOS_OPT_DIR):
        raise FileNotFoundError(
            f"Can't write public options as configuration directory {CLWPOS_OPT_DIR} does not exist"
        )

    public_config_data = get_server_wide_options()

    yield public_config_data

    with acquire_lock(PUBLIC_OPTIONS),\
            open(PUBLIC_OPTIONS, "w") as f:
        json.dump(asdict(public_config_data), f)


def run_in_cagefs_if_needed(command, **kwargs):
    """
    Wrapper for subprocess to enter cagefs
    do not enter cagefs if:
     - CloudLinux Solo
     - if process already started as user in cagefs
    """
    locale = get_locale_from_envars()
    if 'env' in kwargs and locale:
        kwargs['env']['LANG'] = locale

    logging.info('Executing command: %s with environment: %s',
                 str(command),
                 str(kwargs.get('env')))

    if in_cagefs() or not is_panel_feature_supported(Feature.CAGEFS):
        return subprocess.run(command,
                              text=True,
                              capture_output=True,
                              preexec_fn=demote(os.geteuid(), os.getegid()),
                              **kwargs)
    else:
        if os.geteuid() == 0:
            raise WposError(message=gettext('Internal error: command %s must not be run as root. '
                                            'Please contact support if you have questions: '
                                            'https://cloudlinux.zendesk.com') % command)
        if isinstance(command, str):
            with_cagefs_enter = CAGEFS_ENTER_UTIL + ' --no-io-and-memory-limit ' + command
        else:
            with_cagefs_enter = [CAGEFS_ENTER_UTIL, '--no-io-and-memory-limit'] + command
        return subprocess.run(with_cagefs_enter,
                              preexec_fn=demote(os.geteuid(), os.getegid()),
                              text=True,
                              capture_output=True,
                              **kwargs)


def uid_by_name(name):
    """
    Returns uid for user
    """
    try:
        return ClPwd().get_uid(name)
    except ClPwd.NoSuchUserException:
        return None

class PhpIniConfig:
    """
    Helper class to update extensions in php .ini files.
    """

    def __init__(self, php_version, custom_logger=None):
        self.php_version = php_version
        self.disabled_pattern = re.compile(r'^;\s*extension\s*=\s*(?P<module_name>\w+)\.so')
        self.enabled_pattern = re.compile(r'^\s*extension\s*=\s*(?P<module_name>\w+)\.so')
        self.extension = re.compile(r'^\s*;?\s*extension\s*=\s*(?P<module_name>\w+)\.so')
        self.logger = custom_logger or logging.getLogger(__name__)

        # for cagefs user location
        self.wildcard_ini_user_locations = (
            dict(path=f'/var/cagefs/*/*/etc/cl.php.d/{self.php_version.identifier}',
                 user=lambda path: path.split('/')[4]),
        )

    def _parse_extension_name(self, line):
        """
        Parse .so extensions safely
        """
        try:
            return line.split('=')[1].split('.so')[0]
        except Exception as e:
            self.logger.warning('Cannot parse extension name from line: %s, error: %s',
                                line,
                                str(e))
            return None

    def get_ini_content(self, ini_path):
        full_path = os.path.join(self.php_version.dir, ini_path)
        if not os.path.exists(full_path):
            return []

        with open(full_path) as f:
            ini_content = f.readlines()

        modules = []
        for ext in ini_content:
            # extension=igbinary.so -> igbinary
            raw_module_name = self._parse_extension_name(ext)
            if not raw_module_name:
                continue
            modules.append(raw_module_name)
        return modules


    def create_custom_ini(self, path: str, modules: List[str]):
        full_path = os.path.join(self.php_version.dir, path)

        # does not exist yet
        if not os.path.exists(full_path):
            self._write_modules(full_path, modules, exists=False)
        else:
            # overwrite
            self.enable_modules(path, modules)

    def remove_custom_ini(self, path, all_ini=None):
        if all_ini:
            full_path = os.path.join(self.php_version.dir, path)
            if os.path.exists(full_path):
                self.logger.debug(f'Custom ini to be removed: {full_path}')
                os.unlink(full_path)
        self.update_user_ini('acceleratewp.ini', [], remove=True)

    def update_user_ini(self, ini_filename, modules, remove=False):
        for location in self.wildcard_ini_user_locations:
            cagefs_paths = iglob(location['path'])
            for dir_path in cagefs_paths:
                try:
                    self._update_single_ini(location, dir_path, modules, ini_filename, remove)
                except Exception:
                    self.logger.exception('Error updating single acceleratewp.ini')
                    continue

    def _update_single_ini(self, location, dir_path, modules, ini_filename, remove=False):
        username = location['user'](dir_path)
        path = os.path.join(dir_path, ini_filename)

        with drop_privileges(username), \
                disable_quota():
            if remove:
                if os.path.exists(path):
                    self.logger.debug('Custom user ini: %s will be removed', path)
                    os.unlink(path)
            else:
                self._write_modules(path, modules, exists=os.path.exists(path))


    def _enabled_modules(self, path: str) -> Set[str]:
        """
        Return enabled modules.
        :param path: full path to .ini file
        """
        with open(path, 'r') as f:
            return {self.enabled_pattern.match(line).group('module_name') for line in f
                    if self.enabled_pattern.match(line) is not None}

    def _extensions_list(self, path):
        with open(path, 'r') as f:
            return {self.extension.match(line).group('module_name') for line in f
                    if self.extension.match(line) is not None}

    def enable_modules(self, path: str, modules: List[str]) -> bool:
        """
        Enable specified modules in .ini php file.
        :param path: path to .ini file related to php directory
        :param modules: list of modules that should be enabled
        """
        full_path = os.path.join(self.php_version.dir, path)
        if not os.path.exists(full_path):
            return False
        self.logger.debug(f'Enable such extensions: {modules}')
        modules_to_enable = set(modules)
        if modules_to_enable:
            self._write_modules(full_path, modules_to_enable)
        return True

    @staticmethod
    def _format_as_ini_ext(module):
        """
        redis -> extension=redis.so
        """
        return f'extension={module}.so\n'

    def _write_modules(self, full_path, modules_to_enable, exists=True):
        new_ini_lines = []
        self.logger.debug(f'Such extensions are required to be enabled: {modules_to_enable}')

        if exists:
            modules_to_enable = set(modules_to_enable)
            with open(full_path) as f:
                for line in f.readlines():
                    if any(self._format_as_ini_ext(ext) in line for ext in modules_to_enable):
                        self.logger.debug(f'Skip {line}, {modules_to_enable} will be added further')
                        continue
                    new_ini_lines.append(line)

        sorted_modules = sorted(modules_to_enable)

        # order matters for redis extension, it should be the last to load properly
        if 'redis' in sorted_modules:
            sorted_modules.sort(key=lambda x: x.endswith('redis'))

        for module in sorted_modules:
            extension_line = self._format_as_ini_ext(module)
            self.logger.debug(f'Appending lines to be written: {extension_line}')
            new_ini_lines.append(extension_line)

        if new_ini_lines:
            self.logger.debug(f'Path to write: {full_path}')
            write_file_via_tempfile(''.join(new_ini_lines), full_path, 0o644)

    def get_required_modules(self, path):
        """
        Reads <ext>.ini file and loads all required extensions
        """
        full_path = os.path.join(self.php_version.dir, path)
        if not os.path.exists(full_path):
            return []
        required_modules = list(self._extensions_list(full_path))
        self.logger.debug(f'Required extensions for {path} are: {required_modules}')
        return required_modules

    def disable_modules(self, path: str, modules: List[str]) -> bool:
        """
        Disable specified modules in .ini php file.
        :param path: path to .ini file related to php directory
        :param modules: list of modules that should be disabled
        """
        full_path = os.path.join(self.php_version.dir, path)
        if not os.path.exists(full_path):
            return False
        modules_to_disable = set(modules) & self._enabled_modules(full_path)
        if modules_to_disable:
            with open(full_path) as f:
                new_ini_lines = [self._disable_module(line, modules_to_disable)
                                 for line in f.readlines()]
            write_file_via_tempfile(''.join(new_ini_lines), full_path, 0o644)
        return True

    def _enable_module(self, line: str, modules_to_enable: Set[str]) -> str:
        """
        Search for disabled module in line, uncomment line to enable module.
        """
        match = self.disabled_pattern.match(line)
        if match is not None:
            module_name = match.group('module_name')
            if module_name in modules_to_enable:
                modules_to_enable.remove(module_name)
                return line.lstrip(';').lstrip()
        return line

    def _disable_module(self, line: str, modules_to_disable: Set[str]) -> str:
        """
        Search for enabled module in line, comment line to disable module.
        """
        match = self.enabled_pattern.match(line)
        if match is not None:
            module_name = match.group('module_name')
            if module_name in modules_to_disable:
                return f';{line}'
        return line


def _run_clwpos_as_user_in_cagefs(user=None):
    """
    All user-related actions must run inside of cagefs for security reasons.
    If solo just return because cagefs is only for shared and shared pro
    If root executed, we enter into user cagefs if user is pointed
    If not in cagefs and cagefs is enabeled for user enter into cagefs
    """
    if not is_panel_feature_supported(Feature.CAGEFS):
        return

    if not is_run_under_user():
        if user is None:
            raise WposError(message=gettext(
                "Internal Error: root enters into CageFS without specifying username"
                "Please contact support if you have questions: "
                "https://cloudlinux.zendesk.com"
            )
            )
        cmd = [CAGEFS_ENTER_USER_BIN,
               '--no-io-and-memory-limit',
               user] + sys.argv[:1] + sys.argv[3:]
    elif not in_cagefs() and _is_cagefs_enabled(user=user_name()):
        cmd = [CAGEFS_ENTER_UTIL, '--no-io-and-memory-limit'] + sys.argv

    else:
        return
    env = {'LANG': get_locale_from_envars()}
    logging.info('Executing command: %s with environment: %s', str(cmd), str(env))
    p = subprocess.Popen(cmd, stdout=sys.stdout, stdin=sys.stdin,
                         env=env)
    p.communicate()
    sys.exit(p.returncode)


class RedisConfigurePidFile:
    """
    Helper class that provides methods to work with
    pid files of php redis configuration processes.
    """

    def __init__(self, php_prefix: str) -> None:
        self._pid_file_name = f'{php_prefix}-cloudlinux.pid'
        self.path = Path(CLWPOS_OPT_DIR, self._pid_file_name)

    def create(self) -> None:
        with self.path.open('w') as f:
            f.write(str(os.getpid()))

    def remove(self) -> None:
        if self.path.is_file():
            self.path.unlink()

    def exists(self) -> bool:
        return self.path.is_file()

    @property
    def pid(self) -> int:
        if not self.exists():
            return -1
        with self.path.open() as f:
            try:
                return int(f.read().strip())
            except ValueError:
                pass
        return -1


@contextmanager
def create_pid_file(php_prefix: str):
    """
    Context manager for creating pid file of current process.
    Removes pid file on exit.
    """
    pid_file = RedisConfigurePidFile(php_prefix)
    try:
        pid_file.create()
        yield
    finally:
        pid_file.remove()


def is_php_redis_configuration_running(php_prefix: str) -> bool:
    """
    Find out if PHP redis configuration process is running.
    Based on looking for presence of pid files.
    For root also checks process existence.
    """
    pid_file = RedisConfigurePidFile(php_prefix)
    if os.geteuid() != 0:
        return pid_file.exists()
    try:
        process = psutil.Process(pid_file.pid)
        return 'enable_redis' in process.name()
    except (ValueError, psutil.NoSuchProcess):
        return False


def is_alt_php_redis_configuration_running() -> bool:
    """
    Find out if alt-PHP redis configuration process is running.
    """
    return is_php_redis_configuration_running(ALT_PHP_PREFIX)


def is_ea_php_redis_configuration_running() -> bool:
    """
    Find out if ea-PHP redis configuration process is running.
    """
    return is_php_redis_configuration_running(EA_PHP_PREFIX)


def is_plesk_php_redis_configuration_running() -> bool:
    """
    Find out if ea-PHP redis configuration process is running.
    """
    return is_php_redis_configuration_running(PLESK_PHP_PREFIX)


def is_redis_configuration_running() -> bool:
    """
    Find out if redis configuration process
    is running for any PHP (ea-php or alt-php).
    """
    return is_alt_php_redis_configuration_running() or \
           is_ea_php_redis_configuration_running() or \
           is_plesk_php_redis_configuration_running()


def update_redis_conf(new_user: WposUser, old_user: WposUser) -> None:
    """
    Replace user's wpos directory path in redis.conf.
    """
    with open(new_user.redis_conf) as f:
        redis_conf_lines = f.readlines()

    updated_lines = [
        line.replace(old_user.wpos_dir, new_user.wpos_dir) for line in redis_conf_lines
    ]
    write_file_via_tempfile(''.join(updated_lines), new_user.redis_conf, 0o600)


def update_wp_config(abs_wp_path: str, new_user: WposUser, old_user: WposUser) -> None:
    """
    Replace user's redis socket path in wp-config.php.
    """
    try:
        wp_config_lines = wp_config.read(abs_wp_path)
    except OSError as e:
        print('Error occurred during opening wp-config.php '
              f'located in path "{abs_wp_path}": {e}', file=sys.stderr)
        return

    updated_lines = [
        line.replace(old_user.redis_socket, new_user.redis_socket)
        if old_user.redis_socket in line else line
        for line in wp_config_lines
    ]
    write_file_via_tempfile(''.join(updated_lines), wp_config.path(abs_wp_path), 0o600)


def get_parent_pid() -> int:
    """
    Get parent process PID.
    """
    proc = psutil.Process(os.getpid())
    return proc.ppid()


def _is_monitoring_daemon_exists() -> bool:
    """
    Detect CL WPOS daemon presence in system
    :return: True - daemon works / False - No
    """
    # /sbin/service clwpos_monitoring status
    # retcode != 0 - clwpos_monitoring not running/not installed
    #         == 0 - clwpos_monitoring running
    returncode, _, _ = run_command(['/sbin/service', 'clwpos_monitoring', 'status'], return_full_output=True)
    if returncode != 0:
        return False
    return True


def _update_clwpos_daemon_config_systemd(systemd_unit_file) -> Tuple[int, str, str]:
    """
    Update systemd unit file and reload systemd
    """
    shutil.copy('/usr/share/cloudlinux/clwpos_monitoring.service', systemd_unit_file)
    retcode, stdout, stderr = run_command(['/usr/bin/systemctl', 'enable', 'clwpos_monitoring.service'],
                                          return_full_output=True)
    if not retcode:
        retcode, stdout, stderr = run_command(['/usr/bin/systemctl', 'daemon-reload'], return_full_output=True)
    return retcode, stdout, stderr


def _install_daemon_internal(systemd_unit_file: str, is_module_allowed_on_server: bool) -> Tuple[int, str, str]:
    """
    Install WPOS daemon to system and start it
    """
    retcode, stdout, stderr = 0, None, None
    if 'el6' in platform.release():
        retcode, stdout, stderr = run_command(['/sbin/chkconfig', '--add', 'clwpos_monitoring'],
                                              return_full_output=True)
    else:
        if is_module_allowed_on_server:
            # CL Shared Pro and module enabled
            # Update unit file and reload systemd - setup daemon
            retcode, stdout, stderr = _update_clwpos_daemon_config_systemd(systemd_unit_file)
    if not retcode:
        retcode, stdout, stderr = run_command(['/sbin/service', 'clwpos_monitoring', 'start'],
                                              return_full_output=True)
    return retcode, stdout, stderr


def install_monitoring_daemon(is_module_allowed_on_server: bool) -> Tuple[int, str, str]:
    """
    Install WPOS daemon to server if need:
        - if daemon already present - do nothing;
        - on CL Shared Pro install daemon if module allowed
    On solo and if /etc/systemd/system/clwpos_monitoring.service present it will be updated always
    We do not need restart installed daemon here, it's done in rpm_posttrans.sh
    :param is_module_allowed_on_server: True/False
    """
    systemd_unit_file = '/etc/systemd/system/clwpos_monitoring.service'
    # if from rpm_posttrans
    if os.path.exists(systemd_unit_file):
        # Update unit file and reload systemd
        _update_clwpos_daemon_config_systemd(systemd_unit_file)
    if _is_monitoring_daemon_exists():
        return 0, "", ""
    return _install_daemon_internal(systemd_unit_file, is_module_allowed_on_server)


def get_status_from_daemon(service):
    command_get_service_status_dict = {"command": f"get-{service}-status"}
    try:
        daemon_result = daemon_communicate(command_get_service_status_dict)
    except WposError:
        return False
    return daemon_result.get('status')


@cached_in_scope
def redis_is_running() -> bool:
    return get_status_from_daemon('redis')


@cached_in_scope
def litespeed_is_running() -> bool:
    return get_status_from_daemon('litespeed')


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

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

    try:
        with open(info_json) as f:
            return json.load(f)[attribute]
    except (KeyError, json.JSONDecodeError) as e:
        logging.exception("Error during reading of \"info.json\" file: %s", e)
        raise WposError(_("Failed to retrieve data about php version which is currently used. "
                          "Daemon is not available and cache data is malformed, please try again and"
                          " contact your administrator if the issue persists."))
    except FileNotFoundError:
        raise WposError(_("Failed to retrieve data about php version which is currently used. "
                          "Daemon is not available and cache data is not available. "
                          "Contact your administrator if the issue persists."))


def drop_permissions_if_needed(username):
    # there is no need to drop privileges if we are already
    # running as user, so we should handle this case
    # by using empty context instead
    context = drop_privileges
    if os.geteuid():
        context = contextlib.nullcontext

    return context(username)


def get_subscription_status(allowed_features: dict, suite: str, feature: str):
    from clwpos.daemon import WposDaemon

    subscription_status = 'active' if feature in allowed_features.get(suite) else 'no'
    try:
        is_pending = daemon_communicate({
            "command": WposDaemon.DAEMON_GET_UPGRADE_ATTEMPT_STATUS,
            "feature": feature
        })["pending"]
    except WposError:
        # in a rare situation when daemon is not active we
        # still would like to return list of modules
        # this is an old test-covered behavior that I would
        # not like to change now
        # it seems that in 99% of cases daemon must be active as we
        # start in when first module is enabled
        is_pending = False
    if is_pending:
        subscription_status = 'pending'
    return subscription_status


def jwt_token_check():
    """
    JWT token check. Mostly copied from cllib, but with some accelerate-wp tunes, including:
    - clsolo, cladmin tokens are now valid
    - no need to check for shared, because our tools just don't work on shared
    """
    success_flag, error_message, token_string = True, "OK", None
    try:
        token_string = read_jwt(DEFAULT_JWT_ES_TOKEN_PATH)
    except (OSError, IOError):
        return False, "JWT file {} read error".format(DEFAULT_JWT_ES_TOKEN_PATH), None

    try:
        decode_jwt(token_string)
    except exceptions.InvalidIssuerError:
        success_flag, error_message, token_string = False, "JWT token issuer is invalid", None
    except exceptions.ExpiredSignatureError:
        success_flag, error_message, token_string = False, "JWT token expired", None
    except exceptions.PyJWTError:
        success_flag, error_message, token_string = False, "JWT token format error", None
    return success_flag, error_message, token_string

def get_locale_from_envars():
    """
    Locale could be set via those envvars, let`s get them in same priority gettext does
    for envar in ('LANGUAGE', 'LC_ALL', 'LC_MESSAGES', 'LANG'):
    LANGUAGE = (unset),
    LC_ALL = (unset),
    LC_MESSAGES = "UTF-8",
    LANG = "uk_UA.UTF-8"
    """
    return (os.environ.get('LANGUAGE')
            or os.environ.get('LC_ALL')
            or os.environ.get('LC_MESSAGES')
            or os.environ.get('LANG')
            or 'en_US')

def get_accelerate_wp_version():
    # written in .spec
    version_file = '/usr/share/cloudlinux/accelerate-wp.version'
    if not os.path.exists(version_file):
        return None
    with open(version_file) as f:
        return f.read().strip()