FileMgr

✏️ /opt/cloudlinux/venv/lib/python3.11/site-packages/clwpos/utils.py

✕ Tutup

# -*- 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 errno
import itertools
import hashlib
import logging
import os
import re
import shutil
import stat
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,
    cpinfo,
    is_admin,
    get_main_username_by_uid,
)
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,
)
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 clcommon.clwpos_lib import get_locale_from_envars

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:
        username = get_username_by_uid(os.geteuid())
        return pwd.getpwnam(username)

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",
            "if (defined('WP_CLI') && WP_CLI && !defined('WP_REDIS_IGNORED_GROUPS')) { define('WP_REDIS_IGNORED_GROUPS', ['user-queries']); }\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 _drop_to_user(username: str):
    """
    Run code under user's privileges if we are root; otherwise do nothing.
    Used by cleanup/migration routines which should not perform filesystem writes as root.
    """
    return drop_privileges(username) if not is_run_under_user() else contextlib.nullcontext()

def create_redis_cache_config(abs_wp_path: str) -> None:
    """
    Create config for redis-cache.
    Updates wp-config.php by inserting CloudLinux redis-cache section when needed.
    :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:
        absent_constants = {constant.name: constant.val for constant in RedisRequiredConstants}

wp_config_lines = wp_config.read(abs_wp_path)
        # Ensure we also add CLI-only ignored group when it is absent
        add_cli_directive = not any('WP_REDIS_IGNORED_GROUPS' in line for line in wp_config_lines)
        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 and not add_cli_directive:
                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 cleanup_legacy_wp_config_backups() -> None:
    """
    Cleanup for old customers:
    - Remove legacy backups created next to wp-config.php (docroot):
        <wp-path>/wp-config.php.backup
    - Remove old per-user backup directory:
        <home>/.clwpos/backups/wp-config
    """
    _log = logger or logging.getLogger(__name__)
    removed = skipped = 0

from clcommon.cpapi import cpusers, userdomains
    from clcommon.clwpos_lib import find_wp_paths

for username in cpusers():
        try:
            docroots = [dr for _, dr in userdomains(username)]
        except Exception as e:
            _log.debug("legacy wp-config backup cleanup: can't get docroots for %s: %s", username, e)
            continue

excludes = list(docroots)
        for docroot_path in docroots:
            try:
                with _drop_to_user(username):
                    wp_paths = list(find_wp_paths(docroot_path, excludes=excludes))
            except Exception as e:
                _log.debug("legacy wp-config backup cleanup: can't scan docroot %s for %s: %s",
                             docroot_path, username, e)
                continue

for wp_rel in wp_paths:
                wp_rel = "" if wp_rel in ("", ".", "./") else wp_rel
                if not isinstance(wp_rel, str):
                    skipped += 1
                    continue
                if os.path.isabs(wp_rel) or os.path.normpath(wp_rel).startswith(".."):
                    skipped += 1
                    _log.warning(
                        "legacy wp-config backup cleanup: suspicious wp_rel=%r for user=%s docroot=%s",
                        wp_rel, username, docroot_path
                    )
                    continue

abs_wp_path = os.path.join(docroot_path, wp_rel).rstrip("/")
                wp_config_path = str(wp_config.path(abs_wp_path))
                legacy_backup = f"{wp_config_path}.backup"

if not os.path.lexists(legacy_backup):
                    continue

try:
                    with _drop_to_user(username):
                        if not os.path.isabs(legacy_backup) or not os.path.normpath(legacy_backup).startswith(
                            os.path.normpath(docroot_path) + os.sep
                        ):
                            raise ValueError(f"refuse to delete outside docroot: {legacy_backup!r}")
                        os.unlink(legacy_backup)
                    removed += 1
                except Exception as e:
                    skipped += 1
                    _log.warning("legacy wp-config backup cleanup failed: %s: %s", legacy_backup, e)

try:
            home = pwd.getpwnam(username).pw_dir
        except KeyError:
            _log.debug("legacy wp-config backup cleanup: can't resolve home for %s", username)
            continue

if not home or home == "/" or not os.path.isabs(home):
            _log.warning("legacy wp-config backup cleanup: invalid home for %s: %r", username, home)
            continue

target = os.path.join(home, USER_WPOS_DIR, "backups", "wp-config")
        try:
            with _drop_to_user(username):
                if not os.path.lexists(target):
                    continue
                st = os.lstat(target)
                if stat.S_ISDIR(st.st_mode):
                    shutil.rmtree(target)
                else:
                    os.unlink(target)
        except OSError as e:
            if e.errno != errno.ENOENT:
                _log.warning("legacy wp-config backup cleanup: can't remove %s for %s: %s", target, username, e)
        except Exception as e:
            _log.warning("legacy wp-config backup cleanup: can't remove %s for %s: %s", target, username, e)

_log.info(
        "legacy wp-config backup cleanup done: removed=%d skipped=%d",
        removed, skipped
    )

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

def ensure_maxcache_available(is_json: bool):
    """
    Validate that MAx Cache is installed and its module is loaded
    (Apache and/or nginx).
    """
    apache_version_path = "/opt/cloudlinux/maxcache/.version"
    nginx_version_path = "/opt/cloudlinux/maxcache/.nginx-version"

apache_installed = os.path.isfile(apache_version_path)
    nginx_installed = os.path.isfile(nginx_version_path)

if not apache_installed and not nginx_installed:
        error_and_exit(is_json, {"result": _("MAx Cache is not installed.")})

apache_module_ok = False
    if apache_installed:
        try:
            proc = subprocess.run(
                ["apachectl", "-M"],
                capture_output=True,
                text=True,
                timeout=5,
            )
            output = (proc.stdout or "") + (proc.stderr or "")
            if "maxcache_module" in output:
                apache_module_ok = True
        except (FileNotFoundError, subprocess.SubprocessError, OSError):
            pass

nginx_module_ok = False
    if nginx_installed:
        nginx_module_ok = os.path.isfile(
            "/etc/nginx/conf.d/modules/ea-nginx-maxcache.conf"
        )

if not apache_module_ok and not nginx_module_ok:
        error_and_exit(
            is_json,
            {"result": _("MAx Cache module is not enabled (neither Apache nor nginx).")},
        )

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
    supported_features: 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({feature for feature in ALL_SUITES[suite].feature_set
                                      if feature in self.supported_features})
        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({feature for feature in ALL_SUITES[suite].feature_set
                                      if feature in self.supported_features})
        return _visible_features

@lru_cache(maxsize=None)
def get_suites_status_from_license_daemon_fallback():
    from clwpos.daemon import WposDaemon
    try:
        suites_statuses_by_license = daemon_communicate(
            {
                "command": WposDaemon.DAEMON_GET_SUPPORTED_SUITES_BY_LICENSE,
            }
        )
        # Skipping F841 linter issue, because it is better to keep the variable names as is to impove readability
        awp_premium_status, awp_cdn_status = (suites_statuses_by_license['accelerate_wp_premium'],  # noqa: F841
                                              suites_statuses_by_license['accelerate_wp_cdn'])
    except WposError:
        awp_premium_status, awp_premium_status = False, False
    return awp_premium_status, awp_premium_status

def is_feature_supported_by_license(suite):
    from clwpos.feature_suites import PremiumSuite, CDNSuitePro, CDNSuite
    if not os.geteuid():
        awp_premium_status, awp_cdn_status = get_suites_status_from_license()
    else:
        awp_premium_status, awp_cdn_status = get_suites_status_from_license_daemon_fallback()
    return {
        PremiumSuite.name: awp_premium_status,
        CDNSuite.name: awp_cdn_status,
        CDNSuitePro.name: awp_cdn_status,
    }.get(suite, True)

def get_supported_features():
    from .feature_suites import ALL_SUITES
    supported_features = set()
    for suite_name, suite_item in ALL_SUITES.items():
        for suite_feature in suite_item.features:
            if not suite_feature.IS_BILLABLE or is_feature_supported_by_license(suite_name):
                supported_features.add(suite_feature)
    return supported_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),
        supported_features=list(get_supported_features()),
        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
    )

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

return [suite.name for suite in suites]

def get_suites_status_from_license():
    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)
    return is_awp_premium_allowed, is_awp_cdn_allowed

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

non_overridable_fields = {'supported_features'}

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({k: v for k, v in configuration.items()
                                             if k not in non_overridable_fields})
            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,
                              encoding='utf-8',
                              errors='replace',
                              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,
                              encoding='utf-8',
                              errors='replace',
                              **kwargs)

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

def name_by_uid(uid):
    try:
        return get_username_by_uid(uid)
    except Exception:
        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}')
            # copy_owner to preserve existing permissions (e.g. root:linksafe) on file
            write_file_via_tempfile(''.join(new_ini_lines), full_path, 0o644, copy_owner=True)

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)
    # Avoid implicit daemon-reload in `enable` and do a single explicit reload when needed.
    retcode, stdout, stderr = run_command(['/usr/bin/systemctl', 'enable', '--no-reload', '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 _sha256_file(path: str) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()

def _files_identical(path_a: str, path_b: str) -> bool:
    """
    Best-effort content comparison. We use it to decide whether shipped systemd unit
    needs to be copied into /etc (and thus requires daemon-reload).
    """
    try:
        st_a = os.stat(path_a)
        st_b = os.stat(path_b)
        if st_a.st_size != st_b.st_size:
            return False
        return _sha256_file(path_a) == _sha256_file(path_b)
    except OSError:
        return False

def _systemd_service_is_enabled(service_name: str) -> bool:
    retcode, _, _ = run_command(['/usr/bin/systemctl', 'is-enabled', service_name],
                                return_full_output=True)
    return retcode == 0

def _systemd_unit_needs_daemon_reload(unit_name: str) -> Optional[bool]:
    """
    Returns:
    - True/False if systemd reports it explicitly
    - None if the property is unsupported/unknown (depends on systemd version)
    """
    retcode, stdout, _ = run_command(
        ['/usr/bin/systemctl', 'show', '-p', 'NeedDaemonReload', unit_name],
        return_full_output=True,
    )
    if retcode != 0:
        return None
    # systemd versions differ:
    # - newer: may support `--value` (we don't use it for compatibility)
    # - older: prints `NeedDaemonReload=yes|no` (or empty/absent)
    value = stdout.strip()
    if value == '':
        return None
    if '=' in value:
        _, rhs = value.split('=', 1)
        value = rhs.strip()
    value = value.lower()
    if value == '':
        return None
    return value == 'yes'

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
    """
    _log = logger or logging.getLogger(__name__)
    systemd_unit_file = '/etc/systemd/system/clwpos_monitoring.service'
    shipped_unit_file = '/usr/share/cloudlinux/clwpos_monitoring.service'

# If the unit file exists, keep it up to date, but avoid expensive daemon-reload
    # unless it is actually needed.
    if os.path.exists(systemd_unit_file):
        unit_name = 'clwpos_monitoring.service'
        needs_unit_update = not _files_identical(shipped_unit_file, systemd_unit_file)
        needs_enable = not _systemd_service_is_enabled(unit_name)
        need_reload_flag = _systemd_unit_needs_daemon_reload(unit_name)  # True/False/None
        needs_daemon_reload = needs_unit_update or (need_reload_flag is True)
        needs_daemon_reload_unknown = need_reload_flag is None and not needs_unit_update

daemon_exists = _is_monitoring_daemon_exists()
        _log.info(
            "install_monitoring_daemon: unit_exists=1 daemon_exists=%s needs_unit_update=%s "
            "needs_enable=%s need_daemon_reload=%s need_daemon_reload_unknown=%s",
            daemon_exists,
            needs_unit_update,
            needs_enable,
            needs_daemon_reload,
            needs_daemon_reload_unknown,
        )

retcode, stdout, stderr = 0, "", ""
        if needs_unit_update:
            # Copy shipped unit into /etc and perform enable+daemon-reload as required.
            _log.info("install_monitoring_daemon: updating unit file from shipped and reloading systemd")
            retcode, stdout, stderr = _update_clwpos_daemon_config_systemd(systemd_unit_file)
        else:
            if needs_enable:
                _log.info("install_monitoring_daemon: enabling %s (no-reload)", unit_name)
                retcode, stdout, stderr = run_command(
                    ['/usr/bin/systemctl', 'enable', '--no-reload', unit_name],
                    return_full_output=True,
                )
            if not retcode and needs_daemon_reload:
                _log.info("install_monitoring_daemon: running systemctl daemon-reload (NeedDaemonReload=yes)")
                retcode, stdout, stderr = run_command(
                    ['/usr/bin/systemctl', 'daemon-reload'],
                    return_full_output=True,
                )
            elif needs_daemon_reload_unknown:
                _log.info("install_monitoring_daemon: skipping daemon-reload (NeedDaemonReload unsupported/unknown)")

if retcode:
            return retcode, stdout, stderr
    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')

@cached_in_scope
def nginx_is_running() -> bool:
    return get_status_from_daemon('nginx')

@cached_in_scope
def apache2nginx_is_running() -> bool:
    return get_status_from_daemon('apache2nginx')

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_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()

@lru_cache(maxsize=None)
def is_user_owned_by_reseller(username: str, force_as_root=False):
    """
    Returns True/False whether target user is owned by reseller
    """
    if force_as_root or os.geteuid() == 0:
        user_owner = cpinfo(cpuser=username, keyls=('reseller',))[0][0]
        return not is_admin(user_owner)
    else:
        return is_user_owned_by_reseller_daemon_fallback(username)

@lru_cache(maxsize=None)
def is_user_owned_by_reseller_daemon_fallback(username: str):
    from clwpos.daemon import WposDaemon
    try:
        is_owned_by_reseller = daemon_communicate(
            {
                "command": WposDaemon.DAEMON_IS_USER_OWNED_BY_RESELLER
            }
        )["is_owned_by_reseller"]
    except WposError:
        return True
    return is_owned_by_reseller

def get_username_by_uid(uid):
    if os.geteuid() == 0:
        return get_main_username_by_uid(uid)
    else:
        pwd_list = ClPwd().get_pw_by_uid(uid)
        home_dirs = {}
        for user_pwd in pwd_list:
            home_dirs[user_pwd.pw_dir] = user_pwd.pw_name

sorted_dirs = sorted(home_dirs.keys(), key=lambda d: d.count('/'))

for main_home in sorted_dirs:
            if any(home.startswith(main_home + '/') for home in sorted_dirs if home != main_home):
                return home_dirs[main_home]
        return pwd.getpwuid(uid).pw_name

@lru_cache(maxsize=None)
def is_wp2_environment_safe():
    from clcommon.cpapi import is_wp2_environment

if os.geteuid() == 0:
        return is_wp2_environment()
    else:
        from clwpos.daemon import WposDaemon
        try:
            return daemon_communicate({"command": WposDaemon.DAEMON_IS_WP2_ENVIRONMENT})["is_wp2"]
        except WposError:
            return False  # Fallback if daemon query fails

Batal

⬅ site-packages 37 item

Nama	Tipe	Ukuran	Diubah	Perm	Aksi
📁 __pycache__	dir	—	09/07/26 03:10	0755	📂
📁 bin	dir	—	09/07/26 03:10	0755	📂
📁 cli_versions	dir	—	09/07/26 03:10	0755	📂
📁 feature_suites	dir	—	09/07/26 03:10	0755	📂
📁 hooks	dir	—	09/07/26 03:10	0755	📂
📁 migrations	dir	—	09/07/26 03:10	0755	📂
📁 object_cache	dir	—	09/07/26 03:10	0755	📂
📁 optimization_features	dir	—	09/07/26 03:10	0755	📂
📁 php	dir	—	09/07/26 03:10	0755	📂
📁 user	dir	—	09/07/26 03:10	0755	📂
🐍 __init__.py	py	928 B	18/06/26 22:26	0644	✏️ ⬇
🐍 billing.py	py	6.2 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 cl_wpos_exceptions.py	py	3.6 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 constants.py	py	5.6 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 create_user_uid_dirs.py	py	754 B	18/06/26 22:26	0644	✏️ ⬇
🐍 cron.py	py	2.1 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 daemon.py	py	39.5 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 daemon_base.py	py	2.8 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 daemon_config.py	py	621 B	18/06/26 22:26	0644	✏️ ⬇
🐍 daemon_redis_lib.py	py	11.9 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 daemon_subscription_handler.py	py	10.9 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 data_collector_utils.py	py	9.4 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 logsetup.py	py	4 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 papi.py	py	9.9 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 parse.py	py	2.1 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 redis_configuration_pid_file_cleaner.py	py	1 KB	18/06/26 22:26	0755	✏️ ⬇
🐍 report_generator.py	py	21.2 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 scoped_cache.py	py	1.3 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 socket_utils.py	py	6.8 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 stats.py	py	12 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 utils.py	py	63.7 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 whmcs_utils.py	py	9.5 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 wp_config.py	py	725 B	18/06/26 22:26	0644	✏️ ⬇
🐍 wp_utils.py	py	17.2 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 wpos_admin.py	py	72 KB	18/06/26 22:26	0644	✏️ ⬇
🐍 wpos_hooks.py	py	7.2 KB	18/06/26 22:26	0755	✏️ ⬇
🐍 wpos_req_scanner.py	py	4.4 KB	18/06/26 22:26	0644	✏️ ⬇

✏️ Rename

🔑 Ubah Permission