✏️ Editing: cluseroptselect.py Read Only

# -*- coding: utf-8 -*-

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

from __future__ import absolute_import
from __future__ import print_function
from __future__ import division
import ast
import os
import base64
import re
import configparser

from builtins import map
from future.utils import iteritems

from .cluserextselect import ClUserExtSelect
from .clselectexcept import ClSelectExcept
from clcommon import clcaptain
from clcommon import clcagefs
from . import utils
from xml.sax.saxutils import unescape
from clcommon.utils import ExternalProgramFailed
from clcommon.php_conf_reader import PhpConfReader, PhpConfBaseException,\
    PhpConfReadError, PhpConfLoadException, PhpConfNoSuchAlternativeException

def _eval_bitwise_expr(expr):
    """Safely evaluate a bitwise expression containing only integers and |&~!^ operators."""
    tree = ast.parse(expr, mode='eval')

def _eval(node):
        if isinstance(node, ast.Expression):
            return _eval(node.body)
        if isinstance(node, ast.Constant) and isinstance(node.value, int):
            return node.value
        # ast.Num was removed in Python 3.14; keep the branch only when present
        # so the same code works on Python 3.6 (CL7), 3.9 (CL8/9) and 3.14+ (CL10+).
        if hasattr(ast, 'Num') and isinstance(node, ast.Num):
            return node.n
        if isinstance(node, ast.UnaryOp):
            if isinstance(node.op, (ast.Invert, ast.Not)):
                return ~_eval(node.operand) if isinstance(node.op, ast.Invert) else int(not _eval(node.operand))
            raise ValueError("unsupported unary operator")
        if isinstance(node, ast.BinOp):
            left = _eval(node.left)
            right = _eval(node.right)
            if isinstance(node.op, ast.BitOr):
                return left | right
            if isinstance(node.op, ast.BitAnd):
                return left & right
            if isinstance(node.op, ast.BitXor):
                return left ^ right
            raise ValueError("unsupported binary operator")
        raise ValueError("unsupported expression")

return _eval(tree)

class ClUserOptSelect(ClUserExtSelect):
    """
    Class for processing user options
    """
    OPTIONS_PATH = '/etc/cl.selector.conf.d/php.conf' if clcagefs.in_cagefs() else '/etc/cl.selector/php.conf'

def __init__(self, item='php', exclude_pid_list=None):
        ClUserExtSelect.__init__(self, item, exclude_pid_list)
        self._whitelist = {}
        self._user_excludes = set()
        self._html_escape_table = {" ": "&nbsp;", '"': "&quot;", "'": "&apos;",
                                   ">": "&gt;", "<": "&lt;", "&": "&amp;"}
        self._html_unescape_table = {v: k for k, v in iteritems(self._html_escape_table)}

def insert_options(self, user, version,
                       optset, decoder, append=False, quiet=True, create=True,
                       website=None):
        """
        Inserts supplied options into current ones
        @param optset: string
        @param decoder: string
        @param website: Website or None for per-domain configuration
        """
        options = {}
        if optset != '':
            options = self._process_option_string(
                optset=optset, decoder=decoder, expect_separator=True)
            options = self._remove_forbidden_options(options, version, quiet)

return utils.apply_for_at_least_one_user(
            self.insert_json_options,
            self._clpwd.get_names(self._clpwd.get_uid(user)),
            ClSelectExcept.UnableToSaveData,
            version, options, append, create, website
        )

def insert_json_options(self, user, version, options, append=False, create=True,
                            website=None):
        """
        Inserts supplied options into current ones
        @param user: string
        @param version: string
        @param options: object
        @param website: Website or None for per-domain configuration
        """
        self._check_user_in_cagefs(user)
        user_ini_path = self._compose_user_ini_path(user, version, website=website)
        (contents, extensions,
            extensions_data) = self._load_ini_contents(user_ini_path)
        contents = self._prepare_options_data(contents)
        if append:
            contents.update(options)
        else:
            contents = options
        options_set = self._compose_options_set(contents)
        if options_set:
            options_set = self._wrap_options(options_set)
        data = self._compose_output_data(
            options_set, extensions, extensions_data)
        # Convert 'no value' values of directives
        for idx in range(0, len(data)):
            line = data[idx]
            line_parts = line.split('=')
            if len(line_parts) != 2:
                continue
            if line_parts[1] == 'no value':
                # put empty string instead 'no value' to directive value
                data[idx] = line_parts[0] + '='
        self._write_to_file(
            user, '\n'.join(data).rstrip()+'\n', user_ini_path, create)
        self._reload_processes(user)
        self._backup_settings(user, version, options_set, create)

def bulk_insert_options(self, user, version, options, append=False, create=True,
                            website=None):
        """
        Handles multiple users with same uids
        @param website: Website or None for per-domain configuration
        """
        return utils.apply_for_at_least_one_user(
            self.insert_json_options,
            self._clpwd.get_names(self._clpwd.get_uid(user)),
            ClSelectExcept.UnableToSaveData,
            version, options, append, create, website
        )

def delete_options(self, user, version,
                       optset, decoder, quiet=True, website=None):
        """
        Deletes supplied options from current ones
        @param website: Website or None for per-domain configuration
        """
        return utils.apply_for_at_least_one_user(
            self._delete_user,
            self._clpwd.get_names(self._clpwd.get_uid(user)),
            ClSelectExcept.UnableToSaveData,
            optset, decoder, version, website
        )

def _delete_user(self, user, optset, decoder, version, website=None):
        options = self._process_option_string(
            optset=optset, decoder=decoder, expect_separator=False)

self._check_user_in_cagefs(user)

user_ini_path = self._compose_user_ini_path(user, version, website=website)
        (contents, extensions,
            extensions_data) = self._load_ini_contents(user_ini_path)

contents = self._prepare_options_data(contents)

for opt in options.keys():
            contents.pop(opt, None)

options_set = self._compose_options_set(contents)
        options_set = self._wrap_options(options_set)

data = self._compose_output_data(
            options_set, extensions, extensions_data)

self._write_to_file(
            user, '\n'.join(data).rstrip()+'\n', user_ini_path)

self._reload_processes(user)
        self._backup_settings(user, version, options_set)

def get_options(self, user, version=None, website=None):
        """
        Returns options summary for a user
        @param user: string
        @param version: string
        @param website: Website or None for per-domain configuration
        return: dict
        """
        if not version:
            version = self.get_version(user)[0]
        if version == 'native':
            raise ClSelectExcept.UnableToGetExtensions(version)
        self._get_ini_defaults(version)
        self._get_user_ini(user, version, website=website)
        return self._get_whitelist(version)

def reset_options(self, users=None, versions=None, website=None):
        """
        Deletes all custom options settings
        @param users: list
        @param versions: list
        @param website: Website or None for per-domain configuration
        """
        all_users = self.list_all_users()
        alternatives = self.get_all_alternatives_data()
        for version in alternatives.keys():
            if versions and version not in versions:
                continue
            for user in all_users:
                if users and user not in users:
                    continue
                try:
                    self.insert_options(user=user, version=version,
                       optset='', decoder='plain', append=False, quiet=True,
                       create=False, website=website)
                except ClSelectExcept.NotCageFSUser:
                    continue

def _prepare_options_data(self, contents):
        options = {}
        for item in contents:
            if item.strip() == "":
                continue
            if item.startswith(';>===') or item.startswith(';<==='):
                continue
            key, value = list(map((lambda x:x.strip()), item.split('=', 1)))
            if value == '':
                value = 'no value'
            options.update({key: value})
        return options

def _get_whitelist(self, version):
        """
        Returns whitelist data
        """
        if not self._whitelist:
            self._load_whitelist(version)
        return self._whitelist

def _load_whitelist(self, version):
        """
        Parses php config file (not php.ini!) and updates structure
        """
        # Get short_php_version_to_full map
        alternatives = self.get_all_alternatives_data()
        self._check_alternative(version, alternatives)
        if '.' not in version:
            raise ClSelectExcept.UnableToGetExtensions(version)
        # Short to full PHP version map. Example: {'4.4', '4.4.9'}
        php_versions = dict()
        for short_ver, ver_data in iteritems(alternatives):
            php_versions[short_ver] = ver_data['version']
        try:
            # Read config
            conf_reader = PhpConfReader(self.OPTIONS_PATH)
            php_conf_dict = conf_reader.get_config_for_selectorctl(version, php_versions)
            self._whitelist.update(php_conf_dict)
        except PhpConfNoSuchAlternativeException as e:
            raise ClSelectExcept.UnableToGetExtensions(e.php_version)
        except (PhpConfReadError, PhpConfLoadException, PhpConfBaseException) as e:
            raise ClSelectExcept.UnableToLoadData(self.OPTIONS_PATH, str(e))

def _handle_option_item(option_item, expect_separator=True):
        """
        Splits options data into key-value pair and returns it
        @param option_item: string
        @param expect_separator: bool
        @return: dict
        """
        if ':' in option_item:
            option_name, option_value = option_item.split(':', 1)
        else:
            if not expect_separator:
                option_name, option_value = option_item, ''
            else:
                raise ClSelectExcept.WrongData(
                    "Colon as a separator expected (%s)!" % (option_item,))
        return {option_name: option_value}
    _handle_option_item = staticmethod(_handle_option_item)

def _decoder(data, decoder='plain'):
        """
        Decodes option item
        @param data: string
        @param decoder: string
        @return: string
        """
        dispatcher = {
            'plain': (lambda x: x),
            'base64': (lambda x: base64.b64decode(x).decode())}
        try:
            return dispatcher[decoder](data)
        except KeyError:
            return dispatcher['plain'](data)
    _decoder = staticmethod(_decoder)

def _process_option_string(cls, optset, decoder='plain', expect_separator=True):
        """
        Wrapper around options parsing routines
        @param optset: string
        @param decoder: callback name
        @expect_separator: bool
        @return: dict
        """
        options = {}
        if optset:
            for option_item in optset.split(','):
                option_item = cls._decoder(option_item, decoder)
                options.update(
                    cls._handle_option_item(
                        option_item, expect_separator))
        return options
    _process_option_string = classmethod(_process_option_string)

def _remove_forbidden_options(self, options, version, quiet=True):
        """
        Check if all options to process are present in white list
        and removes forbidden ones or raise an exception
        @param options: dict
        @param quiet: bool
        @return: dict
        """
        whitelist = self._get_whitelist(version)
        if not set(options.keys()).issubset(set(whitelist.keys())):
            white_list_options = {}
            for opt_name, opt_value in iteritems(options):
                if opt_name not in whitelist:
                    if quiet:
                        continue
                    else:
                        raise ClSelectExcept.UnableToProcessOption(opt_name)
                white_list_options[opt_name] = opt_value
            options = white_list_options
        return options

def _compose_options_set(options):
        """
        Construct option item from key and value pair
        @param options: dict
        return: list
        """
        options_set = []
        for opt_name, opt_value in iteritems(options):
            options_set.append("%s=%s" % (opt_name, opt_value))
        return options_set
    _compose_options_set = staticmethod(_compose_options_set)

def _wrap_options(self, contents):
        """
        Adds identifying string before and after dataset
        @param contents: list
        """
        data = [';>=== Start of PHP Selector Custom Options ===']
        data.extend(contents)
        data.append(';<=== End of PHP Selector Custom Options =====')
        return data

def _compose_output_data(contents, extensions, extensions_data):
        """
        Construct output
        @param contents: list
        @param extensions: list
        @param extensions_data: dict
        return: list
        """
        data = []
        for item in extensions:
            data.extend(extensions_data[item])
            # Add two spacelines between each extension
            data.extend(["", ""])

data.extend(contents)
        return data
    _compose_output_data = staticmethod(_compose_output_data)

def _check_version(self, test, version):
        """
        Compares version in use and version required by PHP feature
        and return true if PHP feature satisfies
        """
        alternatives = self.get_all_alternatives_data()
        self._check_alternative(version, alternatives)
        if '.' not in version:
            raise ClSelectExcept.UnableToGetExtensions(version)
        v_array = list(map((lambda x: int(x)), alternatives[version]['version'].split('.')))
        # if test has 2 section, add third
        if len(test.split('.')) == 2:
            test += '.0'
        patt = re.compile(r'([<>=]{1,2})?(\d+\.\d+\.\d+)\.?')
        m = patt.match(test)
        if not m:
            raise ClSelectExcept.NoSuchAlternativeVersion(test)
        action = m.group(1)
        test = list(map((lambda x: int(x)), m.group(2).split('.')))
        version_int = v_array[0] << 11 | v_array[1] << 7 | v_array[2]
        test_int = test[0] << 11 | test[1] << 7 | test[2]
        if action == r'<' and version_int < test_int:
            return True
        if action == r'<=' and version_int <= test_int:
            return True
        if action == r'>' and version_int > test_int:
            return True
        if action == r'>=' and version_int >= test_int:
            return True
        if not action or action == r'=':
            version_int = v_array[0] << 11 | v_array[1] << 7
            test_int = test[0] << 11 | test[1] << 7
            if version_int == test_int:
                return True
        return False

def _get_php_error_tbl(self, php_ver):
        # http://php.net/manual/en/errorfunc.constants.php
        php_error_table = {
            1:     'E_ERROR',
            2:     'E_WARNING',
            4:     'E_PARSE',
            8:     'E_NOTICE',
            16:    'E_CORE_ERROR',
            32:    'E_CORE_WARNING',
            64:    'E_COMPILE_ERROR',
            128:   'E_COMPILE_WARNING',
            256:   'E_USER_ERROR',
            512:   'E_USER_WARNING',
            1024:  'E_USER_NOTICE',
            2048:  'E_STRICT'  # E_STRICT since PHP 5 but not included in E_ALL until PHP 5.4.0
        }
        if self._check_version('<5.2.0', php_ver):
            php_error_table[2047] = 'E_ALL'
        if self._check_version('>=5.2.0', php_ver):
            php_error_table[4096] = 'E_RECOVERABLE_ERROR'  # E_RECOVERABLE_ERROR since PHP 5.2.0
            if self._check_version('<5.3.0', php_ver):
                php_error_table[6143] = 'E_ALL'  # E_ALL 6143 in PHP 5.2.x
        if self._check_version('>=5.3.0', php_ver):
            php_error_table[8192] = 'E_DEPRECATED'        # E_DEPRECATED since PHP 5.3.0
            php_error_table[16384] = 'E_USER_DEPRECATED'  # E_USER_DEPRECATED since PHP 5.3.0
            if self._check_version('<5.4.0', php_ver):
                php_error_table[30719] = 'E_ALL'  # E_ALL 30719 in PHP 5.3.x
        if self._check_version('>=5.4.0', php_ver):
            php_error_table[32767] = 'E_ALL'  # E_ALL 32767 in PHP >= 5.4.x
        return php_error_table

def _php_string2error(self, str_, php_ver):
        """
        Convert php error level 'error-reporting' from string to code
        http://php.net/manual/ru/function.error-reporting.php
        #>>> ClUserOptSelect(item='php')._php_string2error('E_ALL & ~E_NOTICE', '5.4')
        32759
        #>>> ClUserOptSelect(item='php')._php_string2error('E_USER_ERROR | E_NOTICE', '5.4')
        264
        #>>> ClUserOptSelect(item='php')._php_string2error('E_ERROR | E_WARNING | E_PARSE | E_COMPILE_ERROR', '5.4')
        71
        #>>> ClUserOptSelect(item='php')._php_string2error('E_ERROR | INCORRECT', '5.4')  # incorrect variable 'INCORRECT'
        None
        #>>> ClUserOptSelect(item='php')._php_string2error('E_ERROR + E_WARNING', '5.4')   # incorrect operator '+'
        None
        :param str: error_reporting variable
        :return None|int: error_reporting error code; return None if can't convert
        """
        VALID_SYMBOLS = '0123456789|&~!^ '  # http://php.net/manual/en/errorfunc.constants.php
        php_error_table = self._get_php_error_tbl(php_ver)
        # replacing all constants to the numbers
        for code, name in iteritems(php_error_table):
            str_ = str_.replace(name, str(code))

# check if str_ has only valid symbols
        if set(str_).difference(set(VALID_SYMBOLS)):
            return None

try:
            error_code = int(_eval_bitwise_expr(str_))
        except (SyntaxError, ValueError, TypeError):
            return None
        return error_code

def _get_error_desc(self, value, version, range_):
        if not re.match(r'^-?\d{1,5}$', value):  # error-reporting code must be from 32767 to -32767
            return ''
        desc = []
        value = int(value)
        for error_string in range_:
            if self._php_string2error(error_string, php_ver=version) == value:
                return error_string

php_error_table = self._get_php_error_tbl(php_ver=version)
        for error in php_error_table:
            if (error & value) == error:
                desc.append(php_error_table[error])
        return r' | '.join(desc)

def _get_ini_defaults(self, version):
        """
        Gets PHP defaults (calls php -i)
        @param version: string
        """
        alternatives = self.get_all_alternatives_data()
        self._check_alternative(version, alternatives)
        whitelist = self._get_whitelist(version)
        if not os.path.isfile(alternatives[version]['data'][self._item]):
            raise ClSelectExcept.NoSuchAlternativeVersion(version)
        env_data = os.environ
        if ('SCRIPT_FILENAME' in env_data):
            script_path = '/usr/share/l.v.e-manager/utils/clinfo.php'
            if os.path.exists(script_path):
                env_data['SCRIPT_FILENAME'] = script_path
            cmd = [alternatives[version]['data'][self._item]]
        else:
            cmd = [alternatives[version]['data'][self._item], '-qi']
            env_data.pop('SERVER_SOFTWARE', None)
        env_data['PHP_FCGI_MAX_REQUESTS'] = '1'
        env_data['PHP_FCGI_CHILDREN'] = '0'
        env_data['ACCEPT_ENCODING'] = ''
        env_data['HTTP_ACCEPT_ENCODING'] = ''
        tag_pattern = re.compile(
            r'<tr[^>]*?><td[^>]*>(.*?)</td><td[^>]*>(.*?)</td>(?:<td[^>]*>(.*?)</td>)?</tr>')
        strip_pattern = re.compile(r'<[^>]*?>')
        cmd[1:1] = ['-d', 'opcache.enable=0',
                    '-d', 'zlib.output_compression=Off',
                    '-d', 'auto_append_file=none',
                    '-d', 'extension=mbstring.so',
                    '-d', 'auto_prepend_file=none',
                    '-d', 'disable_functions=none']
        output = utils.run_command(cmd, env_data)
        lines = tag_pattern.findall(output)
        # Directives which values are rewritten while execute CMD
        rewritten_directives = ['opcache.enable',
                                'zlib.output_compression',
                                'auto_append_file',
                                'extension',
                                'auto_prepend_file',
                                'disable_functions']

configuration_file = None
        for l in lines:
            directive = re.sub(strip_pattern, '', l[0])
            if 'Loaded Configuration File' in directive:
                s = re.sub(strip_pattern, '', (l[2] or l[1]))
                configuration_file = unescape(s, self._html_unescape_table).strip()
            if directive in whitelist:
                # convert html entries to string
                s = re.sub(strip_pattern, '', (l[2] or l[1]))
                value = unescape(s, self._html_unescape_table)
                if value == 'no value':
                    if ('default' in whitelist[directive] and
                            whitelist[directive]['default'] != ""):
                        continue
                    else:
                        whitelist[directive]['default'] = ""
                else:
                    if directive == 'error_reporting':
                        error_range = whitelist[directive]['range'].split(',')
                        value = self._get_error_desc(value, version, error_range)
                    whitelist[directive]['default'] = value
                # Because we rewrite directives from list above when execute cmd
                # we need to use default value from php.ini
                if directive in rewritten_directives and configuration_file:
                        whitelist[directive]['default'] = self._get_value_from_ini_file(configuration_file, directive)
        self._whitelist.update(whitelist)

def _get_user_ini(self, user, version, website=None):
        """
        Parses user ini file and updates
        values of existing data
        @param user: string
        @param website: Website or None for per-domain configuration
        """
        self._get_whitelist(version)
        user_ini_path = self._compose_user_ini_path(user, version, website=website)
        (contents, extensions,
            extensions_data) = self._load_ini_contents(user_ini_path)
        contents = self._prepare_options_data(contents)
        for key in contents:
            try:
                self._whitelist[key]['value'] = contents[key]
            except KeyError:
                continue

def _backup_settings(self, user, version, data, create=True):
        """
        On saving user settings keep backup on user homedir
        @param user: string
        @param version: string
        @param data: list
        """
        user_backup_path = os.path.join(
            self._clpwd.get_homedir(user), '.cl.selector')
        if not os.path.isdir(user_backup_path):
            try:
                clcaptain.mkdir(user_backup_path)
            except (OSError, ExternalProgramFailed) as e:
                raise ClSelectExcept.UnableToSaveData(user_backup_path, e)
        user_backup_file = os.path.join(
            user_backup_path, "alt_php%s.cfg" % version.replace('.', ''))
        # replace 'no value' in directive value to empty
        for idx in range(0, len(data)):
            line = data[idx]
            line_parts = line.split('=')
            if len(line_parts) == 2 and line_parts[1] == 'no value':
                data[idx] = line_parts[0] + '='
        self._write_to_file(
            user, '\n'.join(data), user_backup_file, create)

def backup_php_options(self, user):
        """
        rewrite php backup file with php options
        @param  user: string
        """
        self._check_user_in_cagefs(user)
        alternatives = self.get_all_alternatives_data()
        for version in alternatives.keys():
            user_ini_path = self._compose_user_ini_path(user, version)
            (contents, extensions,
                extensions_data) = self._load_ini_contents(user_ini_path)
            contents = self._prepare_options_data(contents)
            options_set = self._compose_options_set(contents)
            if options_set:
                options_set = self._wrap_options(options_set)
                self._backup_settings(user, version, options_set)

def _get_value_from_ini_file(self, configuration_file, directive):
        """
        get value from ini file
        Now used for getting default value for some php options,
        which we cannot get garanted
        :param configuration_file: ini file for reading
        :param directive: key name
        :return: value of key or ''
        """
        config = configparser.ConfigParser(interpolation=None, strict=False)
        try:
            config.read(configuration_file)
            return config['PHP'].get(directive)
        except (KeyError, PermissionError):
            raise ClSelectExcept.FileProcessError(configuration_file)

🔐 Sid Gifari File Manager Pro