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

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

Editing: 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 os
import shutil
import struct
import sys
import time
import json
import pwd
import fcntl
from contextlib import contextmanager
from socket import socket, AF_UNIX, SOCK_STREAM
from typing import List, Tuple, Optional
from functools import wraps

from secureio import write_file_via_tempfile
from clcommon.cpapi.cpapiexceptions import NoDomain
from clcommon.lib.cledition import (
    is_cl_solo_edition
)
from cldetectlib import is_cpanel
from cllicenselib import check_license
from clcommon.cpapi import docroot
from clcommon.utils import run_command
from clwpos import gettext
from clwpos.cl_wpos_exceptions import (
    WposError,
    WPOSLicenseMissing,
    WpCliUnsupportedException,
    WpNotExists,
    WpConfigWriteFailed
)

from .logsetup import setup_logging

from clwpos.constants import (
    USER_WPOS_DIR,
    WPOS_DAEMON_SOCKET_FILE
)

from .socket_utils import pack_data_for_socket, read_unpack_response_from_socket
from .user.website_check.errors import RollbackException

logger = None

WP_CONFIG_SUFFIX = "wp-config.php"

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

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))
        else:
            print(json.dumps(data_dict, sort_keys=True))
    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_wpos_supported() -> bool:
    """
    Сheck if system environment is supported by WPOS
    """
    return is_cl_solo_edition(skip_jwt_check=True) and is_cpanel()

def create_clwpos_dir_if_not_exists(username):
    """
    Creates {homedir}/.clwpos directory if it's not exists
    """
    clwpos_dir = os.path.join(home_dir(username), USER_WPOS_DIR)
    if not os.path.isdir(clwpos_dir):
        os.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 get_pw(*, username: str = None):
    if username:
        return pwd.getpwnam(username)
    else:
        return pwd.getpwuid(os.geteuid())

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)
            response_dict = read_unpack_response_from_socket(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['context'])
            return response_dict
        except FileNotFoundError:
            raise WposError(gettext('CloudLinux WPOS 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)
    """
    socket_path = os.path.join(home_dir(), USER_WPOS_DIR, 'redis.sock')
    return ["// Start of CloudLinux generated section\n",
            "define('WP_REDIS_SCHEME', 'unix');\n",
            f"define('WP_REDIS_PATH', '{socket_path}');\n",
            "define('WP_REDIS_CLIENT', 'phpredis');\n",
            "define('WP_REDIS_PREFIX', 'CL');\n",
            "define('WP_REDIS_SELECTIVE_FLUSH', true);\n",
            "// End of CloudLinux generated section\n"]

def check_wp_config_existance(abs_wp_path: str, wp_config_path: str) -> None:
    """
    Check that wp-config.php exists inside Wordpress directory.
    :param abs_wp_path: absolute path to WordPress
    :param wp_config_path: absolute path to Wordpress config file
    :raises: WposError
    """
    if not os.path.exists(abs_wp_path):
        raise WpNotExists(abs_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 cache config from docroot's wp-config.php
    :param abs_wp_path: Absolute path to WordPress
    :raises: WposError
    """
    wp_config_path = os.path.join(abs_wp_path, WP_CONFIG_SUFFIX)
    check_wp_config_existance(abs_wp_path, wp_config_path)

try:
        with open(wp_config_path, "r") as wp_config:
            cleared_wp_config = [line for line in wp_config
                                 if line not in redis_cache_config_section()]

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 = os.path.join(abs_wp_path, WP_CONFIG_SUFFIX)
    check_wp_config_existance(abs_wp_path, 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)

redis_lines = (
            "define('WP_REDIS_SCHEME'",
            "define('WP_REDIS_PATH'",
            "define('WP_REDIS_CLIENT'",
            "define('WP_REDIS_PREFIX'",
            "define('WP_REDIS_SELECTIVE_FLUSH'",
        )

with open(wp_config_path, "r") as wp_config:
            wp_config_lines = wp_config.readlines()

for line in wp_config_lines:
            if any(redis_line in line for redis_line in redis_lines):
                break
        else:
            updated_config = [
                wp_config_lines[0],
                *redis_cache_config_section(),
                *wp_config_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 demote(uid, gid):
    """
    Set user's real uid and gid to specified ones.
    Checking equality of real and effective uids is needed
    because this function may be used by root
    with effective uid dropped to user's uid.
    In that case it is needed to set effective uid back to 0 first.
    """

def func():
        real_uid = os.getuid()
        eff_uid = os.geteuid()

if real_uid != eff_uid:
            os.seteuid(real_uid)

os.setgid(gid)
        os.setuid(uid)

return func

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):
    """
    Takes lock on file
    """
    lock_type = fcntl.LOCK_EX | fcntl.LOCK_NB
    try:
        lock_fd = open(path, "a+")
        for _ in range(10):
            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):
    """
    Usage:
    with acquire_lock(path):
       ... do something with files ...
    """
    lock_fd = lock_file(resource_path + '.lock')
    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()

def is_conflict_modules_installed(php_version, module):
    """
    Checks <module> enabled
    """
    from cpanel import php_bin
    path = php_bin(php_version)
    out = run_command([path, '-m'], env_data={})
    if module in out.split('\n'):
        return True
    return False

def wp_cli_compatibility_check(php_path: str):
    """
    Ensures wp-cli is compatible, e.g some
    php modules may prevent stable work
    """
    dangerous_module = 'snuffleupagus'
    if 'ea-php74' in php_path and is_conflict_modules_installed('ea-php74', dangerous_module):
        raise WpCliUnsupportedException(message=gettext('Seems like ea-php74 %(module)s module is '
                                                        'enabled. It may cause instabilities while managing '
                                                        'object_cache. Disable it and try again'),
                                        context={'module': dangerous_module})

def supported_php_handlers() -> List[str]:
    """
    Return list of supported handlers according to edition
    """
    supported = ['php-fpm']
    if not is_cl_solo_edition(skip_jwt_check=True):
        supported.append('lsapi')
    return supported

Name	Type	Actions
__pycache__	Directory
user	Directory
__init__.py	File
cl_wpos_exceptions.py	File
constants.py	File
daemon.py	File
daemon_base.py	File
daemon_main.py	File
daemon_redis_lib.py	File
logsetup.py	File
optimization_modules.py	File
parse.py	File
report_generator.py	File
socket_utils.py	File
utils.py	File
wpos_admin.py	File
wpos_req_scanner.py	File