Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified
📁 ..	-	2026-03-27 02:38
📄 __init__.py from __future__ import absolute_import from __future__ import print_function from __future__ import division from .clselect import ClSelect from .clextselect import ClExtSelect from .cluserselect import ClUserSelect from .cluserextselect import ClUserExtSelect from .cluseroptselect import ClUserOptSelect from .clselectexcept import ClSelectExcept from .clselectprint import clprint __all__ = [ "ClSelect", "ClExtSelect", "ClUserSelect", "ClUserExtSelect", "ClUserOptSelect", "ClSelectExcept", "clprint" ]	536 B	2026-02-26 15:33
📁 __pycache__	-	2026-03-21 15:56
📁 baseclselect	-	2026-03-27 02:38
📄 clextselect.py # -- 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 import collections import configparser import os import operator from clcommon import clcagefs from .clselect import ClSelect from .clselectexcept import ClSelectExcept from .clselectprint import clprint from . import utils # dependencies modulse dict. Example { 'ext_name': 'ext1' } - ext1 depends from ext_name depend_modules_dict = dict() class ClExtSelect(ClSelect): CONFLICTS_PATH = ( '/etc/cl.selector.conf.d/php.extensions.conflicts' if clcagefs.in_cagefs() else '/etc/cl.selector/php.extensions.conflicts' ) SYSTEM_ALT_PATH = '/opt/alt' def __init__(self, item='php'): ClSelect.__init__(self, item) self._conflicts = [] # Sets in _get_enabled_extensions method # True - extension list was read from native php built-ins # False - extension list was read from /etc/cl.selector/defaults.cfg self._use_default_exts_from_native_php = False def _is_disabled_extention(self, ext_name): return ext_name in self._hidden_extensions def enable_extensions(self, version, ext_list): """ Adds extensions to default list of extensions for a version """ alternatives = self.get_all_alternatives_data() self._check_alternative(version, alternatives) defaults_contents = self._process_ini_file( self.DEFAULTS_PATH, (self._item, version), self._add_extensions, ext_list, action = 'enable_extentions') self._write_to_file( '\n'.join(defaults_contents), ClExtSelect.DEFAULTS_PATH) def replace_extensions(self, version, ext_list): """ Replaces extensions to default list of extensions for a version. Writes/updates /etc/cl.selector/defaults.cfg file :param version: alt-php version to process :param ext_list: list extensions to set as defaults for the version """ alternatives = self.get_all_alternatives_data() self._check_alternative(version, alternatives) defaults_contents = self._process_ini_file( self.DEFAULTS_PATH, (self._item, version), self._replace_extensions, ext_list) self._write_to_file( '\n'.join(defaults_contents), self.DEFAULTS_PATH) def disable_extensions(self, version, ext_list): """ Removes extensions from default list of extensions for a version :param version: alt-php version to process :param ext_list: comma separated extensions list to delete """ alternatives = self.get_all_alternatives_data() self._check_alternative(version, alternatives) defaults_contents = self._process_ini_file( self.DEFAULTS_PATH, (self._item, version), self._del_extensions, ext_list, action = 'disable_extentions') self._write_to_file('\n'.join(defaults_contents), self.DEFAULTS_PATH) def list_extensions(self, version): """ Returns list of extensions marking built-ins and enabled ones Also replaces mysqli->nd_mysqli in defaults.cfg for new installations according to LVEMAN-1399 :param version: php version :return Tuple: (extension_name, extension_state) extension_state: None -- built-in extension False/True -- disabled/enabled extension """ ext_mysqli_name = 'mysqli' ext_nd_mysqli_name = 'nd_mysqli' ext_list_to_write = list() is_need_to_write_defaults = False alternatives = self.get_all_alternatives_data() self._check_alternative(version, alternatives) # Get extensions list from /etc/cl.selector/defaults.cfg for supplied verson or # list of built-in extesions for native if version does not present in defaults.cfg enabled_extensions = self._get_enabled_extensions(version) # Get extension list for version as_built_in = self._get_builtins(version) try: # Get extensions list for version - list of files # /opt/alt/phpXX/etc/php.d.all/*.ini without .ini extension # Without dependencies analysis as_extensions = self._load_extensions_list(version) except ClSelectExcept.UnableToGetExtensions: as_extensions = [] # ['bz2', 'calendar'] -> ('bz2', None), ('calendar', None) all_extensions = list(map((lambda i: (i, None)), as_built_in)) for ext in as_extensions: status = False if (ext in enabled_extensions) and (ext not in as_built_in): status = True ext_set = set([(ext, True), (ext, False), (ext, None)]) if not set(all_extensions).intersection(ext_set): # add ext and its status to result list # LVEMAN-1399: # If defaults modules was taken from native php built-ins # replace mysqli to nd_mysqli (if it exist) and set status True to it if ext == ext_mysqli_name and self._use_default_exts_from_native_php\ and ext_nd_mysqli_name in as_extensions: all_extensions.append((ext_mysqli_name, False)) all_extensions.append((ext_nd_mysqli_name, True)) # After module replacement we need to write new list as defaults is_need_to_write_defaults = True # add nd_mysqli to list for write ext_list_to_write.append(ext_nd_mysqli_name) else: all_extensions.append((ext, status)) if status: # if module enabled, add it to list for write ext_list_to_write.append(ext) # all_extensions example: [('bz2', None), ('calendar', None), (u'zip', True), (u'zmq', False)] # If module replacement occures, write list to defaults.cfg if is_need_to_write_defaults: self.replace_extensions(version, ext_list_to_write) return tuple(sorted(all_extensions, key=operator.itemgetter(0))) def _get_enabled_extensions(self, version): """ Returns list of enabled extensions for a version """ try: # reads extensions list from /etc/cl.selector/defaults.cfg data = self._dh.get( "%s%s" % (self._item, version), 'modules') self._use_default_exts_from_native_php = False return list(map((lambda i: i.strip()), data.split(','))) except (configparser.NoSectionError, configparser.NoOptionError): self._use_default_exts_from_native_php = True return self._get_builtins('native') def _add_extensions(self, section_info, section, data, trace=True): """ Adds 'modules' option to section or extends it @param section_info: tuple (item and version) @param section: list @param data: list @return: list """ section_header = self._make_section_header(section_info) if len(section) == 0 or section_header != section[0]: return section midx = None modules = [] alt_path = self._compose_alt_path(section_info[1]) for idx in range(len(section)): if section[idx].startswith('modules'): midx = idx break if midx: modules_string = section[midx][section[midx].find('=')+1:].strip() modules.extend( list(map((lambda i: i.strip()), modules_string.split(',')))) modules.extend(data) modules = self._filter_out_conflicts(modules) resolved_modules = self._include_dependencies(modules, alt_path) modules_string = 'modules = %s' % (','.join(sorted(resolved_modules))) if midx: section[midx] = modules_string else: section.append(modules_string) return self._smooth_data(section) def _replace_extensions(self, section_info, section, data, trace=True): """ Adds 'modules' option to section or extends it @param section_info: tuple (item and version). Example: ('php', '5.2') @param section: list. Modules from /etc/cl.selecto/defaults.cfg for supplied php version Example: ['[php5.2]', 'modules = bcmath,dom,gd,imap,json,mcrypt,mysql,mysqli,phar,posix,sockets,uuid,wddx,xmlreader,zip', '', ''] @param data: list: Modules list to set from command line @:param trace: ????, Currently not using, always True @return: list """ global depend_modules_dict section_header = self._make_section_header(section_info) if len(section) == 0 or section_header != section[0]: return section midx = None alt_path = self._compose_alt_path(section_info[1]) for idx in range(len(section)): if section[idx].startswith('modules'): midx = idx break modules = data[:] if trace: resolved_modules = set() modules = self._filter_out_conflicts(modules) for mod in modules: include_dep = self._include_dependencies([mod], alt_path) if len(include_dep) != 1: # Dependencies found, add them to depend_modules_dict # Add deps to dict depend_modules_dict.update({dep_module: mod for dep_module in include_dep if dep_module != mod}) resolved_modules.update(include_dep) modules_string = 'modules = %s' % (','.join(sorted(resolved_modules))) else: modules_string = 'modules = %s' % (','.join(sorted(data))) if midx: section[midx] = modules_string else: section.append(modules_string) # Cleanup dependency list - remove from dependencies all modules, present in command line modules = depend_modules_dict.copy() for dep_module in modules.keys(): if dep_module in data: del depend_modules_dict[dep_module] return self._smooth_data(section) def _del_extensions(self, section_info, section, data, trace=True): """ Deletes items in data list from section list @param section_info: tuple (item and version) @param section: list @param data: list of extension names to delete @return: list """ section_header = self._make_section_header(section_info) if len(section) == 0 or section_header != section[0]: return section midx = None alt_path = self._compose_alt_path(section_info[1]) for idx in range(len(section)): if section[idx].startswith('modules'): midx = idx break if not midx: return section modules_string = section[midx][section[midx].find('=')+1:].strip() modules = set(map((lambda i: i.strip()), modules_string.split(','))) resolved_modules = modules.copy() for item in set(data): if item not in modules: continue rest_of_modules = modules.difference([item]) if self._is_dependency(item, rest_of_modules, alt_path): continue resolved_modules.discard(item) resolved_modules = self._include_dependencies(resolved_modules, alt_path) modules_string = 'modules = %s' % (','.join(sorted(resolved_modules))) section[midx] = modules_string return self._smooth_data(section) def _is_dependency(cls, ext, modules, alt_path): """ Checks if module in modules dependent on ext and returns true or false @param ext: Module to check @param modules: set of names of installed modules @param alt_path: Path to alt-php ini dir: /opt/alt/phpXX/etc/php.d.all @return: bool. True if ext present in dependencies of any module in modules list """ global depend_modules_dict for mod in modules: dependencies = cls._get_dependencies(mod, alt_path) if ext in dependencies: depend_modules_dict[ext] = mod return True return False _is_dependency = classmethod(_is_dependency) def _compose_alt_path(self, version): """ Composes and returns path for alternatives """ return os.path.join( self.SYSTEM_ALT_PATH, "%s%s" % (self._item, version.replace('.', '')), "etc", "%s.d.all" % (self._item,)) def _include_dependencies(cls, ext_list, alt_path, data=None): """ Includes dependencies into extensions list and update data dict if present @param ext_list: list @param alt_path: string @param data: dict @return: list """ in_section = False result_ext_list = [] handled = set() q = collections.deque(ext_list) while q: ext = q.popleft() if ext in handled: continue handled.add(ext) ext_path = os.path.join(alt_path, f'{ext}.ini') try: f = open(ext_path) file_contents = [] pending_contents = [] for line in f: if line.startswith('extension') or line.startswith('zend_extension'): ext_name = cls._single_out_extension(ext, line) if ext_name != ext and ext_name not in handled: q.appendleft(ext_name) continue file_contents.append(f';---{ext}---') in_section = True file_contents.extend(pending_contents) pending_contents = [] if not (line.startswith(';') or line.startswith('\n')): if in_section: file_contents.append(line.rstrip()) else: pending_contents.append(line.rstrip()) f.close() if data is not None and ext not in data: data[ext] = file_contents # Adding to the beggining of the result list due to LVEMAN-504 result_ext_list.insert(0, ext) except (OSError, IOError): continue return result_ext_list _include_dependencies = classmethod(_include_dependencies) def _get_dependencies(cls, ext, alt_path): """ Checks if an extension has dependencies and if so returns them Otherwise returns None @param ext: string @return: set """ dependencies = set() ext_path = os.path.join(alt_path, "%s.ini" % (ext,)) try: f = open(ext_path) for line in f: if line.startswith('extension'): ext_name = cls._single_out_extension(ext, line) if ext_name != ext: dependencies.add(ext_name) return dependencies except (OSError, IOError): return dependencies _get_dependencies = classmethod(_get_dependencies) def _single_out_extension(ext, line): """ Singles out and returns extension from line """ quirks = {'ixed': 'sourceguardian'} if '/' in line: ext_name = line[line.rfind('/')+1:].strip() else: ext_name = line[line.find('=')+1:].strip(' "') if '.' in ext_name: ext_name = ext_name[:ext_name.find('.')] if '-' in ext_name: ext_name = ext_name[:ext_name.rfind('-')] if ext_name in quirks: ext_name = quirks[ext_name] elif ext in ext_name: ext_name = ext elif ('_' in ext and ''.join(map((lambda i: i.capitalize()), ext.split('_'))) == ext_name): ext_name = ext return ext_name _single_out_extension = staticmethod(_single_out_extension) def _filter_out_conflicts(self, ext_list): """ Removes from extensions list conflicting ones """ if not self._conflicts: self._load_conflicting_extensions() clean_set = set() for ext in ext_list: if self._is_not_conflicting(ext, clean_set) and \ not self._is_disabled_extention(ext): clean_set.add(ext) #else: #clprint.print_diag( # 'text', # {'status': 'WARN', # 'message': '%s skipped as conflicting (%s)' % (ext, str(clean_set))}) return clean_set def _is_not_conflicting(self, ext, clean_set): """ Checks extension against conflicting sets """ for conflict in self._conflicts: if ext in conflict: if len(clean_set.copy().intersection(conflict)) != 0: return False return True def _load_conflicting_extensions(self): """ Loads conflicting extensions from file and saves'em as list of sets """ conflicts = utils.read_file_as_string(self.CONFLICTS_PATH) for line in conflicts.splitlines(): if ',' not in line: continue conflict_set = set(map((lambda i: i.strip()), line.split(','))) if len(conflict_set) < 2: continue self._conflicts.append(conflict_set) def _load_extensions_list(self, version): """ Loads alternative extensions list for a version @param version: string """ alt_path = self._compose_alt_path(version) try: alt_extensions = [] for filename in os.listdir(alt_path): if not filename.endswith('.ini'): continue extension = filename[:filename.find('.ini')] if extension in self._hidden_extensions: continue alt_extensions.append(extension) return sorted(alt_extensions) except OSError: raise ClSelectExcept.UnableToGetExtensions(version) @staticmethod def _print_dependencies_info(dependens_info): """ Prints info @param ext: string @param data: list """ for (i, ext) in dependens_info: clprint.print_diag( 'text', {'status': 'WARN', 'message': '%s enabled as dependency (%s)' % (i, ext)}) @staticmethod def get_dependencies_list(ext, data, ext_list): """ Get array of dependenses [(ext, depending ext)] @param ext: string @param data: list """ if not data: return [] diff = set(data).difference([ext]) return [(i, ext) for i in diff if i not in ext_list] @staticmethod def get_conflicts_info(init_list, processed_set): return list(set(init_list).difference(processed_set)) @staticmethod def _print_conflicts_info(diff): """ Prepares data for printing conflicts if any @param init_list: list @param processed_set: set """ if diff: for i in diff: clprint.print_diag( 'text', {'status': 'WARN', 'message': '%s skipped as conflicting' % (i,)})	19.61 KB	2026-02-26 15:33
📄 clpassenger.py # -- 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 fcntl import pwd import syslog from datetime import datetime from future.utils import iteritems from future.moves import configparser as ConfigParser import io import logging import os import re import subprocess from clcommon import clcaptain, utils from clcommon.cpapi import userdomains from clcommon.utils import get_file_system_in_which_file_is_stored_on from clcommon.utils import get_file_lines, write_file_lines from clcommon.utils import mod_makedirs from clquota import QuotaWrapper, NoSuchUserException, InsufficientPrivilegesException, IncorrectLimitFormatException, \ GeneralException, NoSuchPackageException, QuotaDisabledException from lveapi import PyLve, PyLveError from secureio import set_user_perm, set_root_perm from typing import Dict, Union # NOQA from .clselectexcept import ClSelectExcept from .utils import file_readlines, file_write, s_partition from .utils import get_abs_rel, mkdir_p, file_read, file_writelines from .utils import get_using_realpath_keys, realpaths_are_equal # logger for clpassenger module logger = logging.getLogger(__name__) logger.setLevel(logging.ERROR) # disable output of logs to console null_handler = logging.StreamHandler(open('/dev/null', 'w')) logger.addHandler(null_handler) HTACCESS_BEGIN = '# DO NOT REMOVE. CLOUDLINUX PASSENGER CONFIGURATION BEGIN' HTACCESS_END = '# DO NOT REMOVE. CLOUDLINUX PASSENGER CONFIGURATION END' RACK_PATH = 'config.ru' RACK_TEMPLATE = r'''app = proc do \|env\| message = "It works!\n" version = "Ruby %s\n" % RUBY_VERSION response = [message, version].join("\n") [200, {"Content-Type" => "text/plain"}, [response]] end run app ''' RESTART_PATH = 'tmp/restart.txt' WSGI_PATH = 'passenger_wsgi.py' WSGI_TEMPLATE = r'''import os import sys sys.path.insert(0, os.path.dirname(__file__)) def application(environ, start_response): start_response('200 OK', [('Content-Type', 'text/plain')]) message = 'It works!\n' version = 'Python %s\n' % sys.version.split()[0] response = '\n'.join([message, version]) return [response.encode()] ''' APPJS_PATH = 'app.js' APPJS_TEMPLATE = r'''var http = require('http'); var server = http.createServer(function(req, res) { res.writeHead(200, {'Content-Type': 'text/plain'}); var message = 'It works!\n', version = 'NodeJS ' + process.versions.node + '\n', response = [message, version].join('\n'); res.end(response); }); server.listen(); ''' def drop_root_perm(user): userpwd = pwd.getpwnam(user) set_user_perm(userpwd.pw_uid, userpwd.pw_gid, exit = False) def get_config_lock(config_path, mode): try: conf_file = open(config_path, mode, errors='surrogateescape') fcntl.flock(conf_file.fileno(), fcntl.LOCK_EX) return conf_file except IOError: return None def release_lock(lock_file): try: lock_file.close() except: pass def write_config(user, config_path, config): """ Write config with locking. Drop permissions if method called as root. """ permissions_dropped = False # we must drop & restore permissions only when we # call this method as root, otherwise we can get unexpected # behavior when we drop permission on higher level, # and restore them here, in deep tree of method calls if os.getegid() == 0 or os.geteuid() == 0: drop_root_perm(user) permissions_dropped = True config_file = None try: check_and_createdir(config_path) config_file = get_config_lock(config_path, 'r') file_content = io.StringIO() config.write(file_content) clcaptain.write(config_path, file_content.getvalue()) # clcaptain raises exception from clcommon.utils.ExternalProgramFailes, IOError # exception ClSelectExcept.ExternalProgramFailed has the same name but different except (IOError, OSError, ClSelectExcept.UnableToSaveData, utils.ExternalProgramFailed) as e: syslog.syslog(syslog.LOG_WARNING, "Can't write {}: {}".format(config_path, e)) finally: release_lock(config_file) if permissions_dropped: set_root_perm(exit=False) def check_and_createdir(path): user_backup_path = os.path.dirname(path) if not os.path.isdir(user_backup_path): try: clcaptain.mkdir(user_backup_path) # clcaptain raises exception from clcommon.utils.ExternalProgramFailes # exception ClSelectExcept.ExternalProgramFailed has the same name but different except (OSError, ClSelectExcept.ExternalProgramFailed, utils.ExternalProgramFailed) as e: raise ClSelectExcept.UnableToSaveData(user_backup_path, e) def get_htaccess_cache_path(user): userpwd = pwd.getpwnam(user) return os.path.join(userpwd.pw_dir, '.cl.selector', 'htaccess_cache') def _get_info_about_htaccess_cache_file(path_to_file): # type: (str) -> Dict """ Get info (stat, first n symbols and file system in which file is stored) about htaccess_cache file """ time_format = '%Y-%m-%d %H:%M:%S' number_of_symbols = 100 file_info = {} if os.path.exists(path_to_file): try: file_stat = os.stat(path_to_file) file_info['file_size'] = file_stat.st_size file_info['gid'] = file_stat.st_gid file_info['uid'] = file_stat.st_uid file_info['permissions'] = oct(file_stat.st_mode) file_info['last_access'] = datetime.fromtimestamp(file_stat.st_atime).strftime(time_format) file_info['last_modification'] = datetime.fromtimestamp(file_stat.st_mtime).strftime(time_format) # Not necessary to read file and get file system if it's empty if file_info['file_size'] == 0: return file_info try: with open(path_to_file, 'r') as f: file_info['first_symbols'] = f.read(number_of_symbols) # move to n symbol from end file f.seek(-number_of_symbols, 2) file_info['last_symbols'] = f.read(number_of_symbols) except (OSError, IOError) as err: file_info['error'] = 'We cannot get first and last %s symbols from "%s" file. Exception: %s' % ( number_of_symbols, path_to_file, err, ) file_info['file_system'] = get_file_system_in_which_file_is_stored_on(path_to_file)['details'] except (OSError, IOError) as err: file_info['error'] = 'We cannot get info about "%s" file. Exception: %s' % ( path_to_file, err, ) return file_info def _get_user_lve_limits(user_uid): # type: (int) -> Union[Dict[int, int, int, int, int, int, int], Dict[str]] """ Getting user lve limits for logging those for next debug """ result = dict() try: py_lve = PyLve() py_lve.initialize() user_limits = py_lve.lve_info(user_uid) result['cpu'] = user_limits.ls_cpu / user_limits.ls_cpu_weight result['pmem'] = user_limits.ls_memory_phy result['vmem'] = user_limits.ls_memory result['io'] = user_limits.ls_io result['iops'] = user_limits.ls_iops result['ep'] = user_limits.ls_enters result['nproc'] = user_limits.ls_nproc except PyLveError as err: result['error'] = 'We cannot get lve limits for user with uid "%s". Exception: %s' % ( user_uid, err, ) return result def _get_user_quota_limits(user_uid): # type: (int) -> Union[Dict[str, str, str], Dict[str]] """ Getting user quota limits for logging those for next debug """ result = dict() user_uid = str(user_uid) try: quota_wrapper = QuotaWrapper() user_quotas = quota_wrapper.get_user_limits(user_uid)[user_uid] result = user_quotas except ( NoSuchUserException, NoSuchPackageException, InsufficientPrivilegesException, GeneralException, IncorrectLimitFormatException, QuotaDisabledException, IOError, OSError ) as err: result['error'] = 'We cannot get quota limits for user with uid "%s". Exception: %s' % ( user_uid, err, ) return result def _log_debug_info_about_user_and_config_file(user, config_path, error): # type: (str, str, Exception) -> None """ Logging info (lve & quota limits) about user and info (stat info, first & last n symbols) about config file """ file_info = _get_info_about_htaccess_cache_file(config_path) debug_info = dict() debug_info['config_file_info'] = file_info debug_info['user_info'] = dict() try: user_uid = pwd.getpwnam(user).pw_uid except KeyError as err: debug_info['user_info']['error'] = 'User "%s" does not exist. Exception: %s' % ( user, err, ) user_uid = None if user_uid is not None: debug_info['user_info']['lve_limits'] = dict() debug_info['user_info']['lve_limits'].update(_get_user_lve_limits(user_uid)) debug_info['user_info']['quota_limits'] = dict() debug_info['user_info']['quota_limits'].update(_get_user_quota_limits(user_uid)) logger.exception(error, exc_info=True, extra=debug_info) def read_config(user): config = ConfigParser.RawConfigParser(strict=False) config_path = get_htaccess_cache_path(user) config_file = get_config_lock(config_path, 'r') if config_file is not None: try: config.readfp(config_file) # LU-1035 except (IOError, OSError) as err: # Logging additional information for next debug _log_debug_info_about_user_and_config_file(user, config_path, err) # LU-1032 except (ConfigParser.ParsingError, ConfigParser.MissingSectionHeaderError): _unlink(config_path) syslog.syslog(syslog.LOG_WARNING, "Config {} is broken.".format(config_path)) # if cought ParsingError - return Empty config config = ConfigParser.RawConfigParser(strict=False) finally: release_lock(config_file) return config, config_path def get_htaccess_cache(user, doc_root): config, _ = read_config(user) if config.has_section(doc_root): try: htaccess_list = config.get(doc_root, 'htaccess_list').split(',') return htaccess_list except ConfigParser.NoOptionError: return None return None def write_htaccess_cache(user, doc_root, data): data = data.split('\n') data = list(filter(bool, data)) config, config_path = read_config(user) if not config.has_section(doc_root): config.add_section(doc_root) config.set(doc_root, 'htaccess_list', ','.join(data)) write_config(user, config_path, config) def update_htaccess_cache(user, path_to_file, doc_root): config, config_path = read_config(user) if config.has_section(doc_root): htaccess_list = config.get(doc_root, 'htaccess_list').split(',') else: config.add_section(doc_root) config.set(doc_root, 'htaccess_list', '') htaccess_list = [] if path_to_file not in htaccess_list: htaccess_list.append(path_to_file) htaccess_list = list(filter(bool, htaccess_list)) config.set(doc_root, 'htaccess_list', ','.join(htaccess_list)) write_config(user, config_path, config) def remove_passenger_lines_from_htaccess(htaccess_filename): """ Removes clpassenger lines from .htaccess to stop application :param htaccess_filename: Application .htaccess path :return: None """ lines = file_readlines(htaccess_filename, errors='surrogateescape') new_lines = [] in_config = False for line in lines: if line.startswith(HTACCESS_BEGIN): in_config = True if line.startswith(HTACCESS_END): in_config = False continue if not in_config: new_lines.append(line) # write new .htaccess new_lines = rm_double_empty_lines(new_lines) file_writelines(htaccess_filename, new_lines, 'w', errors='surrogateescape') def configure(user, directory, alias, interpreter, binary, populate=True, action=None, doc_root=None, startup_file=APPJS_PATH, passenger_log_file=None): """ Configure passenger application :param user: name of unix user :param directory: name of dir in user home :param alias: alias of application :param interpreter: interpreter which execute application :param binary: binary of interpreter that execute application :param populate: True if application have to be be populated :param action: action with apllication. can be transit or None :param doc_root: doc_root :param startup_file: start application file :param passenger_log_file: Passenger log filename to write to app's .htaccess :return: None """ abs_dir, _ = get_abs_rel(user, directory) if os.path.exists(abs_dir) and not os.path.isdir(abs_dir): raise ClSelectExcept.WebAppError( 'Destination exists and it is not a directory') if interpreter not in ('python', 'ruby', 'nodejs'): raise ClSelectExcept.InterpreterError( "Unsupported interpreter ('%s')" % interpreter) user_summary = summary(user) try: app_summary = get_using_realpath_keys(user, directory, user_summary) except KeyError: if doc_root is None: raise ClSelectExcept.NoSuchApplication( 'No such application (or application not configured) "%s"' % directory) else: if action != 'transit': exists_dir = app_summary['directory'] raise ClSelectExcept.WebAppError("Specified directory already used by '%s'" % exists_dir) if not doc_root: doc_root = app_summary['docroot'] # Alias, which is empty, means that user passed uri equaled to doc root # and we don't want to normalize the alias, because normalized empty # alias is point and that alias doesn't work in htaccess if alias != '': alias = os.path.normpath(alias) abs_alias, _ = get_abs_rel(user, os.path.join(doc_root, alias)) htaccess = os.path.join(abs_alias, '.htaccess') htaccess_needs_update = True if os.path.exists(htaccess): htaccess_raw = file_read(htaccess, errors='surrogateescape') if HTACCESS_BEGIN in htaccess_raw: for item in user_summary.values(): # The condition allows to detect common part of aliases # For details see commit message item_alias = os.path.normpath(item['alias']) + os.sep if os.path.dirname(os.path.commonprefix([item_alias, alias + os.sep])) != '': exists_dir = item['directory'] if exists_dir != abs_dir: raise ClSelectExcept.WebAppError( "Specified alias is already used by the other " "application: '%s'. Please, specify another application url." % exists_dir) else: # Do not write to .htaccess, it is already correct htaccess_needs_update = False lines = htaccess_raw.splitlines() else: lines = [] if htaccess_needs_update: lines.append('') lines.append(HTACCESS_BEGIN) lines.append('PassengerAppRoot "%s"' % abs_dir) lines.append('PassengerBaseURI "/%s"' % alias) lines.append('Passenger%s "%s"' % (interpreter.title(), binary)) # for some reason autodetect of `app.js` is not working if interpreter == 'nodejs': lines.append('PassengerAppType node') lines.append('PassengerStartupFile %s' % startup_file) # append PassengerAppLogFile directive if need if passenger_log_file and interpreter in ('python', 'nodejs'): lines.append('PassengerAppLogFile "%s"' % passenger_log_file) lines.append(HTACCESS_END) lines = rm_double_empty_lines(lines) mkdir_p(abs_alias) file_writelines(htaccess, ('%s\n' % line for line in lines), errors='surrogateescape') update_htaccess_cache(user, htaccess, doc_root) if populate: # Also creates startup_file populate_app(user, directory, interpreter, startup_file=startup_file) def fix_homedir(user): for domain_alias, data in iteritems(_summary(user)): _, alias = domain_alias old_home = os.path.commonprefix((data['directory'], data['binary'])) _, _, directory = s_partition(data['directory'], old_home) # old python selector has binary as file # and get_abs_rel does realpath() # while new selector binary is symlink # and realpath works wrong binary_dir = os.path.dirname(data['binary']) binary_name = os.path.basename(data['binary']) _, _, _binary = s_partition(binary_dir, old_home) binary = os.path.join(get_abs_rel(user, _binary)[0], binary_name) htaccess_path = data['htaccess'] _unconfigure(htaccess_path) configure(user, directory, alias, data['interpreter'], binary, doc_root=data['docroot']) def move(user, directory, old_alias, new_alias, old_doc_root=None, new_doc_root=None): app_data = get_using_realpath_keys(user, directory, summary(user)) old_doc_root = old_doc_root or app_data['docroot'] new_doc_root = new_doc_root or old_doc_root old_abs_alias = os.path.join(old_doc_root, old_alias) old_htaccess = os.path.join(old_abs_alias, '.htaccess') new_abs_alias = os.path.join(new_doc_root, new_alias) new_htaccess = os.path.join(new_abs_alias, '.htaccess') if not realpaths_are_equal(user, old_htaccess, new_htaccess): _unconfigure(old_htaccess) lines = file_readlines(old_htaccess, errors='surrogateescape') open(old_htaccess, 'w').close() file_writelines(new_htaccess, lines, 'a', errors='surrogateescape') update_htaccess_cache(user, new_htaccess, new_doc_root) def purge(user): for directory in summary(user): unconfigure(user, directory) def populate_app(user, directory, interpreter, startup_file=APPJS_PATH): """ Populate application :param user: name of unix user :param directory: application path in user's home :param interpreter: interpreter which run application :param startup_file: main application file :return: None """ abs_dir, rel_dir = get_abs_rel(user, directory) app_public = os.path.join(abs_dir, 'public') app_tmp = os.path.join(abs_dir, 'tmp') mkdir_p(app_public) mkdir_p(app_tmp) app_configru = os.path.join(abs_dir, RACK_PATH) app_wsgi = os.path.join(abs_dir, WSGI_PATH) app_js = os.path.join(abs_dir, startup_file) configru_installed = os.path.isfile(app_configru) wsgi_installed = os.path.isfile(app_wsgi) appjs_installed = os.path.isfile(app_js) if configru_installed: configru_unchanged = file_read(app_configru) == RACK_TEMPLATE if wsgi_installed: wsgi_unchanged = file_read(app_wsgi) == WSGI_TEMPLATE if appjs_installed: appjs_unchanged = file_read(app_js) == APPJS_TEMPLATE if interpreter == 'python': if not wsgi_installed: file_write(app_wsgi, WSGI_TEMPLATE) if configru_installed and configru_unchanged: _unlink(app_configru) _unlink(app_js) elif interpreter == 'ruby': if not configru_installed: file_write(app_configru, RACK_TEMPLATE, 'w') if wsgi_installed and wsgi_unchanged: _unlink(app_wsgi) _unlink(app_js) elif interpreter == 'nodejs': if not appjs_installed: # add ability to specify startup path # like 'not/existing/subdir/app.js' dir_path = os.path.dirname(app_js) if not os.path.isdir(dir_path): mod_makedirs(dir_path, 0o755) file_write(app_js, APPJS_TEMPLATE) if appjs_installed and appjs_unchanged: _unlink(app_configru) _unlink(app_wsgi) restart(user, directory) def _unlink(path): try: os.unlink(path) except OSError: pass def _find_htaccess_files(doc_root): p = subprocess.Popen(['/bin/find', doc_root, '-name', '.htaccess'], stdout=subprocess.PIPE, stderr=subprocess.PIPE) # Process each line as bytes and attempt decoding to UTF-8 clean_lines = [] for line in p.stdout: try: decoded_line = line.decode('utf-8') clean_lines.append(decoded_line.strip()) except UnicodeDecodeError: # Skip lines that cannot be decoded to UTF-8 continue return '\n'.join(clean_lines) def _summary(user, userdomains_data=None): # TODO PTCLLIB-132 # it's using cache info about users domains # this mechanism should be removed after implementing of caching on DA for userdomains cpapi method domain_docroot_pairs = userdomains(user) if userdomains_data is None else userdomains_data domain_alias_docroot = [] for domain, doc_root in domain_docroot_pairs: if doc_root is None: continue htaccess_cache = get_htaccess_cache(user, doc_root) if not htaccess_cache: stdoutdata = _find_htaccess_files(doc_root) write_htaccess_cache(user, doc_root, stdoutdata) htaccess_cache = get_htaccess_cache(user, doc_root) if htaccess_cache is None: # if write_htaccess_cache was unsuccessful, we would still get None here continue for ht_path in htaccess_cache: if ht_path: alias = os.path.dirname(ht_path) domain_alias_docroot.append((domain, alias, doc_root)) return _htaccess_summary(user, domain_alias_docroot) def _htaccess_summary(user, domain_alias_docroot): summ = {} for domain, alias, doc_root in domain_alias_docroot: htaccess = os.path.join(alias, '.htaccess') try: htaccess_raw = file_read(htaccess, errors='surrogateescape') except (IOError, OSError): continue approot = re.search( '^PassengerAppRoot\s+"?(?P<directory>.+?)"?$', htaccess_raw, re.MULTILINE) if not approot: continue interpreter = re.search( '^Passenger(?P<interpreter>Python\|Ruby\|Nodejs)\s+' '"?(?P<binary>.+?)"?$', htaccess_raw, re.MULTILINE) if not interpreter: continue alias_abs, _ = get_abs_rel(user, alias) doc_root_abs, _ = get_abs_rel(user, doc_root) _, _, alias = s_partition(alias_abs, doc_root_abs) alias = alias.lstrip(os.sep) domain_alias = (domain, alias,) # detect what alias belongs domain appuri = re.search('^PassengerBaseURI\s+"?(?P<appuri>.+?)"?$', htaccess_raw, re.MULTILINE) if appuri and not compare_aliases(appuri.groupdict()['appuri'], alias): continue summ[domain_alias] = { 'htaccess': htaccess, 'domain': domain, 'docroot': doc_root, 'directory': approot.groupdict()['directory'], 'interpreter': interpreter.groupdict()['interpreter'].lower(), 'binary': interpreter.groupdict()['binary'], } return summ def compare_aliases(alias1, alias2): return os.path.normpath(alias1.strip('/')) == os.path.normpath(alias2.strip('/')) #FIXME: Need join/rewrite "summary" and "_summary" functions def summary(user, userdomains_data=None): summ_result = {} for domain_alias, value in iteritems(_summary(user, userdomains_data=userdomains_data)): domain, alias = domain_alias app_root = value['directory'] try: _, directory = get_abs_rel(user, app_root) except ClSelectExcept.WrongData: syslog.syslog( syslog.LOG_WARNING, '{} is broken, directory {} is not in user\'s home.'.format( os.path.join(alias, '.htaccess'), app_root )) continue value['alias'] = alias try: app_summary = get_using_realpath_keys(user, directory, summ_result) except KeyError: value['domains'] = [domain] summ_result[directory] = value else: # add domains key if directory has multiple domains if 'domains' not in app_summary: app_summary['domains'] = [] else: app_summary['domains'].append(domain) return summ_result def unconfigure(user, directory): app_data = get_using_realpath_keys(user, directory, summary(user)) htaccess = app_data['htaccess'] _unconfigure(htaccess) def _unconfigure(htaccess): htaccess_raw = file_read(htaccess, errors='surrogateescape') lines = htaccess_raw.splitlines() new_lines = [] in_config = False for line in lines: if line == HTACCESS_BEGIN: in_config = True continue if line == HTACCESS_END: in_config = False continue if in_config: continue new_lines.append(line) lines = rm_double_empty_lines(new_lines) file_writelines(htaccess, ('%s\n' % line for line in lines), 'w', errors='surrogateescape') def iter_path(root, sub): for p in sub.split(os.sep): root = os.path.join(root, p) yield root def restart(user, directory): abs_dir, _ = get_abs_rel(user, directory) if not os.path.exists(abs_dir): raise ClSelectExcept.MissingApprootDirectory("Missing directory %(abs_dir)s" % {'abs_dir': abs_dir}) tmp_dir = os.path.join(abs_dir, 'tmp') if not os.path.exists(tmp_dir): os.mkdir(tmp_dir) app_restart = os.path.join(abs_dir, RESTART_PATH) # imitation system 'touch' if os.path.exists(app_restart): os.utime(app_restart, None) else: open(app_restart, 'a').close() def rm_double_empty_lines(lines): _lines = [] empty_line = True for line in lines: if line.strip(): empty_line = False elif empty_line: continue else: empty_line = True _lines.append(line) if empty_line: return _lines[:-1] return _lines	26.52 KB	2026-02-26 15:33
📄 clselect.py # -- 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 os import re import sys from glob import glob from builtins import map from future.moves import configparser as ConfigParser from .clselectexcept import ClSelectExcept, BaseClSelectException from .clselectprint import clprint from clcommon import clcagefs class ClSelect(object): BASE_ETC_CL_SELECTOR = "/etc/cl.selector.conf.d" if clcagefs.in_cagefs() else "/etc/cl.selector" CONFIG_PATH = f"{BASE_ETC_CL_SELECTOR}/selector.conf" DEFAULTS_PATH = f"{BASE_ETC_CL_SELECTOR}/defaults.cfg" DEFAULT_PHP_PATH = "/usr/bin/php" NATIVE_CONF = f"{BASE_ETC_CL_SELECTOR}/native.conf" USER_CONF = f"{BASE_ETC_CL_SELECTOR}/user.conf" CONFIGS_DIR = f"{BASE_ETC_CL_SELECTOR}/php.extensions.d/" _CAGFSCTL = "/usr/sbin/cagefsctl" # Cache PHP index file CACHEFILE_DIR = "/var/lve" CACHEFILE_CAGEFS_DIR = "/var/lve/php.dat.d" CACHEFILE_PHP_PATTERN = "/php%s.dat" CACHEFILE_PATTERN = CACHEFILE_DIR + CACHEFILE_PHP_PATTERN CACHEFILE_CAGEFS_PATTERN = CACHEFILE_CAGEFS_DIR + CACHEFILE_PHP_PATTERN CACHEFILE_PHP_NATIVE_PATTERN = "/php_native_ver.dat" CACHEFILE_NATIVE_VER_PATTERN = CACHEFILE_DIR + CACHEFILE_PHP_NATIVE_PATTERN CACHEFILE_NATIVE_VER_CAGEFS_PATTERN = CACHEFILE_CAGEFS_DIR + CACHEFILE_PHP_NATIVE_PATTERN @staticmethod def check_multiphp_system_default_version(): if clcagefs.in_cagefs(): return try: from clcagefslib.selector.configure import multiphp_system_default_is_ea_php, selector_modules_must_be_used except ImportError: raise BaseClSelectException("CageFS not installed.") if not multiphp_system_default_is_ea_php() and not selector_modules_must_be_used(): raise BaseClSelectException( "system default PHP version is alt-php. PHP Selector is disabled. Use cPanel MultiPHP manager instead." ) @staticmethod def work_without_cagefs(): return os.path.exists(ClSelect.USER_CONF) def __init__(self, item="php"): self._item = item self._dh = self._get_default_config_handler() self._selector_contents = {} self._native_contents = {} self._hidden_extensions = set() self._native_version = None self.without_cagefs = ClSelect.work_without_cagefs() self._load_config_files() def check_requirements(self): # check that selectorctl and cagefsctl are available utilities = [ ( lambda: os.path.exists(self._get_native_path("cli")), ClSelectExcept.NativeNotInstalled(self._get_native_path("cli")), ), (lambda: self._is_in_cagefs() or os.path.exists(self._CAGFSCTL), ClSelectExcept.MissingCagefsPackage()), ] for predicate, error in utilities: if predicate(): continue raise error def _load_config_files(self): for filename in glob(os.path.join(self.CONFIGS_DIR, ".cfg")): self._load_config_file(filename) def _load_config_file(self, filepath): dh = ConfigParser.SafeConfigParser(interpolation=None, strict=False) try: dh.read(filepath) except ConfigParser.Error as e: raise ClSelectExcept.FileProcessError(filepath, message="Config is malformed, error: %s" % str(e)) try: self._hidden_extensions.update(dh.get("extensions", "hide_extensions").split(",")) except (ConfigParser.NoSectionError, ConfigParser.NoOptionError): pass def list_alternatives(self): """ Returns alternatives summary as tuple :rtype: tuple """ alternatives = self.get_all_alternatives_data() list_of_alternatives = [] for alt in sorted(alternatives.keys()): try: list_of_alternatives.append((alt, alternatives[alt]["version"], alternatives[alt]["data"][self._item])) except KeyError: continue return tuple(list_of_alternatives) def get_all_alternatives_data(self): """ Returns dict of all selector config contents. If no data loads them :return: {'4.4': {'version': '4.4.9', 'data': {'lsphp': '/opt/alt/php44/usr/bin/lsphp', 'php.ini': '/opt/alt/php44/etc/php.ini', 'php': '/opt/alt/php44/usr/bin/php-cgi', 'php-cli': '/opt/alt/php44/usr/bin/php'}}} :rtype: dict """ if not self._selector_contents: try: self._load_alternatives_config() except (ClSelectExcept.ConfigNotFound, ClSelectExcept.WrongConfigFormat): return {} return self._selector_contents def get_alternatives_data(self, version): """ Returns selector config contents of certain version as dict. If no data loads them @param version: string, selector version @return: dict """ if not self._selector_contents: self._load_alternatives_config() try: return {version: self._selector_contents[version]} except KeyError: raise ClSelectExcept.NoSuchAlternativeVersion(version) def get_version(self, show_native_version=False): """ Gets default selector version """ alternatives = self.get_all_alternatives_data() try: version = self._dh.get("versions", self._item) return (version, alternatives[version]["version"], alternatives[version]["data"][self._item]) except (ConfigParser.NoSectionError, KeyError): return self._compose_native_info(show_native_version) def set_version(self, version): """ Sets default selector version """ alternatives = self.get_all_alternatives_data() self._check_alternative(version, alternatives) defaults_contents = self._process_ini_file( self.DEFAULTS_PATH, ("versions",), self._add_or_change_option, (self._item, version) ) self._write_to_file("\n".join(defaults_contents), self.DEFAULTS_PATH) def enable_version(self, version): """ Removes disabled state from version """ alternatives = self.get_all_alternatives_data() self._check_alternative(version, alternatives) defaults_contents = self._process_ini_file( self.DEFAULTS_PATH, (self._item, version), self._remove_option, "state" ) self._write_to_file("\n".join(defaults_contents), self.DEFAULTS_PATH) def disable_version(self, version): """ Marks a vesrion as disabled """ try: default_version = self._dh.get("versions", self._item) except (ConfigParser.NoSectionError, ConfigParser.NoOptionError): default_version = "native" if version == default_version: raise ClSelectExcept.VersionModificationBlocked( f"Cannot disable the default version '{version}'. Please set another version as default first." ) alternatives = self.get_all_alternatives_data() self._check_alternative(version, alternatives) defaults_contents = self._process_ini_file( self.DEFAULTS_PATH, (self._item, version), self._add_or_change_option, ("state", "disabled") ) self._write_to_file("\n".join(defaults_contents), self.DEFAULTS_PATH) def is_version_enabled(self, version): """ Method that allows you to check if some version is enabled in config. E.g. is_version_enabled('5.4') -> True :rtype: bool """ return not self._dh.has_option("%s%s" % (self._item, version), "state") def get_summary(self, show_native_version=False): """ Returns state of alternatives @return: tuple[version, tuple[isEnabled, isDefault]] """ alternatives = self.get_all_alternatives_data() native_info = self._compose_native_info(show_native_version) summary = {"native": {"enabled": True, "default": False}} alt_versions = sorted(alternatives.keys()) + ["native"] for version in alt_versions: if version not in summary: summary[version] = {} summary[version]["enabled"] = self.is_version_enabled(version) summary[version]["default"] = False try: default_version = self._dh.get("versions", self._item) except (ConfigParser.NoSectionError, ConfigParser.NoOptionError): default_version = "native" try: summary[default_version]["default"] = True except KeyError: raise ClSelectExcept.NoSuchAlternativeVersion(default_version) summary[native_info[0]] = summary.pop("native") alt_versions.remove("native") alt_versions.append(native_info[0]) for idx in range(len(alt_versions)): v = alt_versions[idx] alt_versions[idx] = (v, (summary[v]["enabled"], summary[v]["default"])) return tuple(alt_versions) def _load_alternatives_config(self): """ Parses selector config file and fills an instance config dict. Example: {'4.4': {'version': '4.4.9', 'data': {'lsphp': '/opt/alt/php44/usr/bin/lsphp', 'php.ini': '/opt/alt/php44/etc/php.ini', 'php': '/opt/alt/php44/usr/bin/php-cgi', 'php-cli': '/opt/alt/php44/usr/bin/php'}}} :raises ClSelectExcept.ConfigNotFound: :raises ClSelectExcept.WrongConfigFormat: """ try: f = open(self.CONFIG_PATH) for line in f: stripped_line = line.strip() if stripped_line == "": continue (item, short_version, long_version, item_path) = stripped_line.split() if self._item not in item: continue if short_version not in self._selector_contents: self._selector_contents[short_version] = {} self._selector_contents[short_version]["version"] = long_version if "data" not in self._selector_contents[short_version]: self._selector_contents[short_version]["data"] = {} self._selector_contents[short_version]["data"][item] = item_path if not self._selector_contents: raise ClSelectExcept.ConfigNotFound(None, message="alt-php packages not found") except (OSError, IOError) as e: raise ClSelectExcept.ConfigNotFound( "Cannot read %s: %s. Native used" % (self.CONFIG_PATH, e), message="alt-php packages not found" ) except ValueError: raise ClSelectExcept.WrongConfigFormat(self.CONFIG_PATH) def _get_default_config_handler(self, path=None): """ Gets ConfigParser handler for future use """ dh = ConfigParser.ConfigParser(interpolation=None, strict=False) dh.optionxform = str if path: dh.read(path) else: dh.read(self.DEFAULTS_PATH) return dh def _check_alternative(version, alternatives): if version != "native" and version not in alternatives: raise ClSelectExcept.NoSuchAlternativeVersion(version) _check_alternative = staticmethod(_check_alternative) def _make_section_header(section_info): """ Gets section header data tuple and returns ini section header string @param section_info: tuple @return: string """ section_fmt = "[%s]" % "".join(["%s"] len(section_info)) return section_fmt % section_info _make_section_header = staticmethod(_make_section_header) def _smooth_data(data): """ Removes empty lines from list and appends newline if missing """ data = list(filter((lambda i: i != ""), data)) if not data or data[-1] != "\n": data.append("\n") return data _smooth_data = staticmethod(_smooth_data) def _process_ini_file(self, path, section_info, function, data, trace=True, action=None): """ Parses ini file by sections, calls supplied callable to modify section is question, returns file as list of strings """ contents = [] no_section_contents = [] section = [] in_section = False found = False has_default = False section_header = self._make_section_header(section_info) try: f = open(path) for line in f: line = line.strip() if line.startswith("["): in_section = True if "[versions]" in line: has_default = True if section_header == line: found = True if len(no_section_contents) != 0: contents.extend(no_section_contents) no_section_contents = [] contents.extend(function(section_info, section, data, trace)) section = [line] continue if in_section: section.append(line) else: no_section_contents.append(line) contents.extend(function(section_info, section, data, trace)) f.close() except (OSError, IOError): pass if not has_default and "[versions]" not in section_header: default = ["[versions]", "%s = native" % self._item, ""] default.extend(contents) contents = default if not found: try: build_in = self._get_builtins("native") except ClSelectExcept.UnableToGetExtensions: pass # skip if native version not installed if action == "disable_extentions": contents.extend( function(section_info, [section_header, "modules = " + ",".join(build_in)], data, trace) ) elif action == "enable_extentions": data.extend(build_in) contents.extend(function(section_info, [section_header], data, trace)) else: contents.extend(function(section_info, [section_header], data, trace)) return contents def _get_php_binary_path(self, version): """ Retrives path to php binary for supplied version :param version: php version to retrive path :return: path to php binary. If alternative version not found native php binary path returned """ item = "%s-cli" % self._item alternatives = self.get_all_alternatives_data() try: path = alternatives[version]["data"][item] except KeyError: path = self._get_native_path(suffix="cli") return path def get_all_php_binaries_paths(self): """ Retrives paths to php binary for all versions :return: Dictionary version -> path. Example: { '5.2': '/opt/alt/php52/usr/bin/php', '5.3': '/opt/alt/php53/usr/bin/php', 'native': '/usr/bin/php' } """ alternatives = self.get_all_alternatives_data() paths_dict = {"native": self._get_php_binary_path("native")} for version in alternatives.keys(): paths_dict[version] = self._get_php_binary_path(version) return paths_dict def _is_in_cagefs(self): """ Check if running inside CageFS :return: bool """ return clcagefs.in_cagefs() def _read_php_cache_file(self, version): """ Retrives contents of cache file for supplied php version :param version: PHP version to read file :return: file contents """ if self._is_in_cagefs(): filename = self.CACHEFILE_CAGEFS_PATTERN % version else: filename = self.CACHEFILE_PATTERN % version with open(filename, "r") as f: return f.read() def _get_builtins(self, version): """ Gets php extensions from the /var/lve/phpX.X.dat cache file, which contains list of modules that are either compiled-in or enabled in /opt/alt/phpXX/etc/php.ini config file """ builtins = [] # Read php output from cache file for approptiate php version try: output = self._read_php_cache_file(version) except (OSError, IOError): raise ClSelectExcept.UnableToGetExtensions(version) # Section that contains list of modules usually starts and ends with # [PHP Modules] and [Zend Modules] headers respectively start_pattern, end_pattern = "[PHP Modules]", "[Zend" start_index, end_index = output.find(start_pattern), output.find(end_pattern) start_index = 0 if start_index == -1 else start_index + len(start_pattern) modules_list = output[start_index:end_index] module_pattern = re.compile(r"\w") for ext in modules_list.split("\n"): if not module_pattern.match(ext): continue module = "_".join(re.split("\s+", ext.lower())) if module not in self._hidden_extensions: builtins.append(module) return builtins def _remove_option(self, section_info, section, data, trace=True): """ Adds 'modules' option to section or extends it @param section_info: tuple (item and version) @param section: list @param data: string @return: list """ section_header = self._make_section_header(section_info) if len(section) == 0 or section_header != section[0]: return section return self._smooth_data(list(filter((lambda i: not i.startswith(data)), section))) def _add_or_change_option(self, section_info, section, data, trace=True): """ Adds 'modules' option to section or extends it @param section_info: tuple @param section: list @param data: tuple @return: list """ section_header = self._make_section_header(section_info) if len(section) == 0 or section_header != section[0]: return section oidx = None for idx in range(len(section)): if section[idx].startswith(data[0]): oidx = idx break option = "%s = %s" % data if oidx: section[oidx] = option else: section.append(option) return self._smooth_data(section) def _write_to_file(self, file_contents, file_path): """ Saves data to file """ try: f = open(file_path, "w") f.write("%s\n" % file_contents) f.close() except (OSError, IOError) as e: raise ClSelectExcept.UnableToSaveData(file_path, e) def _get_native_path(self, suffix=None): """ Returns path for native interpreter """ suffixes = {"cli": "-cli", "ini": ".ini", "fpm": "-fpm"} item = self._item if suffix and suffix in suffixes: item = "%s%s" % (self._item, suffixes[suffix]) if not self._native_contents: self._load_native_contents(self._item) if item in self._native_contents: path = self._native_contents[item] if os.path.isfile(path) and not os.path.islink(path): return path return self._native_contents[self._item] def _load_native_contents(self, value): """ Parses native.conf file and loads it as dict, for example: {'php-fpm': '/usr/local/sbin/php-fpm', 'php.ini': '/usr/local/lib/php.ini', 'php': '/usr/bin/php', 'php-cli': '/usr/bin/php'} """ try: f = open(self.NATIVE_CONF) for line in f: if line.startswith("#"): continue if value not in line: continue if "=" not in line: continue item, path = list(map((lambda x: x.strip()), line.split("="))) self._native_contents[item] = path if value not in self._native_contents: self._native_contents[value] = self.DEFAULT_PHP_PATH f.close() except (OSError, IOError): self._native_contents[value] = self.DEFAULT_PHP_PATH def _compose_native_info(self, show_version=False): if not show_version: return "native", "native", self._get_native_path() native_version = self.get_native_version(verbose=False) if native_version: return ("native (%s)" % (native_version[0],), "native (%s)" % (native_version[1],), self._get_native_path()) return "native", "native", self._get_native_path() def get_native_version(self, verbose=True): if self._native_version: return self._native_version version_pattern = re.compile(r"PHP\s+(?P<full>(?P<short>\d+\.\d+)\.\d+)") # Read php output from cache file for approptiate php version try: f = open(self.CACHEFILE_NATIVE_VER_PATTERN, "r") data = f.read() f.close() except (OSError, IOError) as e: if verbose: clprint.print_diag("text", {"status": "ERROR", "message": str(e)}) return None for line in data.splitlines(): m = version_pattern.match(line) if m: short, full = m.group("short"), m.group("full") self._native_version = (short, full) return short, full return None	21.96 KB	2026-02-26 15:33
📄 clselectctl.py # -- 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 os import pwd import re from collections import namedtuple import logging import traceback import cldetectlib as detect from typing import List # NOQA from future.utils import iteritems from . import clpassenger from clcommon.cpapi import cpusers, getCPName from clcommon.clpwd import drop_privileges, ClPwd from clcommon.cpapi.plugins.cpanel import is_no_php_binaries_on_cpanel from .clselectexcept import ClSelectExcept from .clselectnodejs.node_manager import NodeManager from .clselectpython.python_manager import PythonManager from .utils import get_abs_rel from typing import Optional, List # NOQA DISABLED_DIRECTORY_NAMES = ['public_html', 'rubyvenv', 'virtualenv', '.cl.selector', '.cagefs', 'www', 'nodevenv'] ApplicationSummary = namedtuple('ApplicationSummary', ('user', 'version', 'doc_root', 'interpreter', 'app_status')) InterpreterSummary = namedtuple('InterpreterSummary', ('version', 'version_full', 'enabled', 'installed')) def check_directory(directory): if ' ' in directory: raise ValueError('Directory should not contain spaces') if os.path.isabs(directory): raise ValueError('Directory should be relative to user\'s home') if directory[-1] == '/': raise ValueError('Directory should not contain a trailing slash') if directory in DISABLED_DIRECTORY_NAMES: raise ValueError('Directory "%s" not allowed' % directory) all_valid = re.match(r'[-\.\w\/]+$', directory) if all_valid is None: raise ValueError('Directory name contains invalid characters') def get_alias(alias): for c in ('#', '?', './'): if c in alias: raise ValueError('Alias is not valid') # root aliases if alias in ('.', ''): alias = '/' return alias.strip('/') def get_directory(prefix): return prefix.replace('_', '/').replace('//', '_') def get_prefix(directory): return directory.replace('_', '__').replace('/', '_') def get_user(user): if not user: current_euid = os.geteuid() user = pwd.getpwuid(current_euid).pw_name if user == 'root': raise ValueError('User parameter must be specified if current user is root') return user def _verify_application(interpreter, app_root, binary_path=None): """ Application is valid only if binary and app_root exists :param binary_path: path to binary in virtual environment :param app_root: path to root directory of application :return: result of checking :rtype: bool """ if interpreter in ('nodejs', 'python'): return os.path.isdir(app_root) return os.path.isfile(binary_path) and os.path.isdir(app_root) def server_applications_summary(interpreter): # type: (str) -> List[ApplicationSummary] """Find and return all apps for given interpreter on server""" domains_data = None applications = [] # LVEMAN-1408 euid = os.geteuid() for user in cpusers(): try: if detect.is_da(): # NB: python will import this only 1 time from clcommon.cpapi.plugins.directadmin import userdomains domains_list = userdomains(user, as_root=True), # called as root to save time on right's escalated call domains_data = list( filter( None, map( lambda x: x[0] if x else None, # take only domain from (domain, docroot) tuples domains_list, ) ) ) # drop permissions in order not to with drop_privileges(user): # LVEMAN-1408: Force check user existence after drop user rights pwd.getpwnam(user) applications.extend( _user_applications_short_summary(user, interpreter, domains_data)) except ClPwd.NoSuchUserException: continue except KeyError: # LVEMAN-1408: this may be thrown by pwd.getpwnam in clpassenger.py module. # It uses python pwd library instead ClPwd # Send debug message to sentry logger = logging.getLogger(__name__) logger.error("DEBUG: user %s present in panel but doesn't exist in system. Process euid=%d. Trace: %s at %s" % (user, euid, traceback.format_exc(), str(traceback.extract_stack()))) return applications def _user_applications_short_summary(user, interpreter, domains_docroots_data=None): """ Return generator with all applications for given user and interpreter. To increase performance, only a small part of the information about the applications is given. :param user: name of panel user :param interpreter: name of interpreter (python, ruby, etc) :param domains_docroots_data: total data about users domains :return: Generator[ApplicationSummary] """ userdomains_data = domains_docroots_data if interpreter in ('nodejs', 'python'): try: if interpreter == 'nodejs': from .clselectnodejs.apps_manager import ApplicationsManager else: from .clselectpython.apps_manager import ApplicationsManager config_data = ApplicationsManager().get_user_config_data(user) except ClSelectExcept.WrongData: return for app, data in iteritems(config_data): try: app_root, _ = get_abs_rel(user, app) except ClSelectExcept.WrongData: continue if _verify_application(interpreter, app_root): yield ApplicationSummary( user=user, version=data['%s_version' % interpreter], doc_root=data['domain'], interpreter=interpreter, app_status=data['app_status'] ) else: for dummy, data in iteritems(clpassenger.summary(user, userdomains_data=userdomains_data)): if data['interpreter'] != interpreter: continue # return only really existing applications binary, app_root = data['binary'], data['directory'] if _verify_application(interpreter, app_root, binary): it_version = os.path.basename(os.path.dirname(os.path.dirname(binary))) yield ApplicationSummary(user=user, version=it_version, doc_root=data['docroot'], interpreter=data['interpreter'], app_status='started') def get_default_version(interpreter): """Return default version for given interpreter """ if interpreter == 'nodejs': return NodeManager().default_version if interpreter == 'python': return PythonManager().default_version return None def _iter_interpreters(interpreter): """Return generator for interpreters all versions""" if interpreter == 'nodejs': interpreters = NodeManager().get_summary(installed_interpreters_only=True) elif interpreter == 'python': interpreters = PythonManager().get_summary(installed_interpreters_only=True) else: raise NotImplementedError() for version, version_info in iteritems(interpreters['available_versions']): enabled = version_info['status'] == 'enabled' if interpreter == 'nodejs': # for nodejs 6.16.0 -> 6 major_version = str(int(version.split('.')[0])) else: # for python 2.7.16 -> 2.7 major_version = '.'.join(version.split('.')[:2]) yield InterpreterSummary(version=major_version, version_full=version, enabled=enabled, installed=True) def _iter_php_interpreters(): """ Return generator with all PHP versions on server. :return: Generator[InterpreterSummary] """ from .clselect import ClSelect php = ClSelect() versions = php.list_alternatives() for version, full_version, _ in versions: yield InterpreterSummary( version, full_version, installed=True, enabled=php.is_version_enabled(version)) def interpreter_versions_short_summary(interpreter): # type: (str) -> List[InterpreterSummary] """Find and return all versions for given interpreter on server""" if interpreter == 'php': # if there is no installed php - no sense to collect statistics if getCPName() == 'cPanel' and is_no_php_binaries_on_cpanel(): return [] return list(_iter_php_interpreters()) elif interpreter == 'ruby': from .clselectruby.interpreters import interpreters elif interpreter in ('python', 'nodejs'): return list(_iter_interpreters(interpreter)) else: raise NotImplementedError it_list = [] for it in interpreters(): # for now, both python and ruby interpreters cannot be disabled it_list.append(InterpreterSummary(it.version, it.version_full, installed=True, enabled=True)) return it_list	9.15 KB	2026-02-26 15:33
📄 clselectctlnodejsuser.py # -- 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 print_function from __future__ import absolute_import from __future__ import division import os import pwd import requests import requests.packages.urllib3.exceptions as urllib_exceptions # pylint: disable=E0401 import shutil from requests.exceptions import ConnectionError from future.utils import iteritems from past.builtins import basestring from . import clpassenger from . import clselectctl from . import utils from clcommon.cpapi import userdomains from clcommon.utils import mod_makedirs from .clselectexcept import ClSelectExcept from .clselectnodejs.apps_manager import ApplicationsManager from .clselectnodejsuser import environments, interpreters from .baseclselect import APP_STARTED_CONST, APP_STOPPED_CONST from .clselectprint import clprint DEFAULT_STARTUP_FILE = "app.js" DEFAULT_APP_STATE = APP_STARTED_CONST def _create_environment(user, directory, version, env_name=None, destroy_first=False): prefix = _get_prefix(user, directory) if not env_name: env_name = version environment = environments.Environment(env_name, user, prefix) if not environment.exists() or destroy_first: try: interpreter = interpreters.interpreters(key="version")[version] except KeyError: raise ClSelectExcept.NoSuchAlternativeVersion(version) environment.create(interpreter, destroy_first=destroy_first) return environment def _get_environment(user, directory, app_summary=None): prefix = _get_prefix(user, directory) if app_summary is None: user_summary = clpassenger.summary(user) try: app_summary = utils.get_using_realpath_keys(user, directory, user_summary) except KeyError: raise ClSelectExcept.NoSuchApplication( 'No such application (or application not configured) "%s"' % directory ) binary = app_summary["binary"] env_name = os.path.basename(os.path.dirname(os.path.dirname(binary))) environment = environments.Environment(env_name, user, prefix) return environment def _get_prefix(user, directory): _, rel_dir = utils.get_abs_rel(user, directory) return os.path.join(environments.DEFAULT_PREFIX, rel_dir) def _ensure_version_enabled(version, user): """ Check whether particular interpreter version is enabled and raises exception if not :param user: user to include in exception """ from .clselectnodejs import node_manager if not node_manager.NodeManager().is_version_enabled(version): raise ClSelectExcept.UnableToSetAlternative(user, version, "version is not enabled") def _get_existing_nesting_app_for(new_app_directory, user, apps_manager=None): """ Return None if new_app_directory is not nested inside an existing user application, and the name of conflicting application otherwise """ if apps_manager is None: apps_manager = ApplicationsManager() abs_app_path, _ = utils.get_abs_rel(user, new_app_directory) full_config = apps_manager.get_user_config_data(user) for existing_app in full_config.keys(): # In case home directory was symlinked after config was written existing_app_abs_path, _ = utils.get_abs_rel(user, existing_app) if (abs_app_path + os.sep).startswith(existing_app_abs_path + os.sep): return existing_app return None def create( user, directory, alias, version=None, doc_root=None, app_mode=None, env_vars=None, startup_file=None, domain_name=None, apps_manager=None, passenger_log_file=None, ): """ Create application :param user: unix user name :param directory: application path in user's home (app-root) :param alias: alias (app-uri) :param version: version of interpreter :param doc_root: doc_root :param app_mode: application mode for nodejs :param env_vars: dict with enviroment variables :param startup_file: main application file :param domain_name: domain name :param apps_manager: Application Manager. Class that responsible for gathering and writing information about applications :param passenger_log_file: Passenger log filename to write to app's .htaccess :return: None """ if apps_manager is None: apps_manager = ApplicationsManager() if version is None: raise ClSelectExcept.WrongData("Not passed version as argument") _ensure_version_enabled(version, user) if startup_file is None: startup_file = DEFAULT_STARTUP_FILE if env_vars is None: env_vars = {} if app_mode is None: app_mode = "production" conflicting_app = _get_existing_nesting_app_for(directory, user, apps_manager) if conflicting_app is not None: raise ClSelectExcept.BusyApplicationRoot(conflicting_app) alias = clselectctl.get_alias(alias) environment = _create_environment(user, directory, version) binary = environment.interpreter().binary clpassenger.configure( user, directory, alias, apps_manager.INTERPRETER, binary, doc_root=doc_root, startup_file=startup_file, passenger_log_file=passenger_log_file, ) # webapp was started clpassenger.restart(user, directory) if not domain_name: # if domain name not defined - try to detemine it summary_data = clpassenger.summary(user) app_summary = utils.get_using_realpath_keys(user, directory, summary_data) domain_name = app_summary["domain"] app_data = { "nodejs_version": version, "domain": domain_name, "app_uri": alias, "app_status": DEFAULT_APP_STATE, "startup_file": startup_file, "app_mode": app_mode, "config_files": [], "env_vars": env_vars, } if passenger_log_file: app_data["passenger_log_file"] = passenger_log_file apps_manager.add_app_to_config(user, directory, app_data) try: apps_manager.add_env_vars_for_htaccess(user, directory, env_vars, doc_root) except Exception as err: clprint.print_diag("text", {"status": "ERROR", "message": str(err)}) def _create_symlink_to_node_modules(username, environment_path, app_root): """ Creates symlink to app's node_modules in app_root :param username: user name :param environment_path: Path to app's virtual environment :param app_root: Application root :return: None """ # This function logic is duplicated in npm wrapper, but we still need it here too. See commit # message of LVEMAN-1335 message = ( "Cloudlinux NodeJS Selector demands to store node modules for application " "in separate folder (virtual environment) pointed by symlink " 'called "node_modules". That`s why application should not contain folder/file ' "with such name in application root" ) link_name = os.path.join(pwd.getpwnam(username).pw_dir, app_root, "node_modules") if os.path.islink(link_name): try: os.remove(link_name) except OSError: raise ClSelectExcept.RemoveSymlinkError("Can`t remove symlink %(link_name)s" % {"link_name": link_name}) elif os.path.exists(link_name): # link_name not a old symlink, but exists raise ClSelectExcept.SymlinkError(message) link_to = os.path.join(environment_path, "lib/node_modules") if not os.path.exists(link_to): mod_makedirs(link_to, 0o775) try: os.symlink(link_to, link_name) except OSError: raise ClSelectExcept.CreateSymlinkError("Error creating symlink. " + message) def destroy(user, app_directory, doc_root, apps_manager=None): """ Destroy web app with specified directory for specified user :param user: username :param app_directory: application directory :param doc_root: Document root for selected domain :param apps_manager: Application Manager. Class that responsible for gathering and writing information about applications :return: None """ if apps_manager is None: apps_manager = ApplicationsManager() # get app state app_config = apps_manager.get_app_config(user, app_directory) # remove env from htaccess try: # if domain is already removed we shouldn't do anything if doc_root is not None: apps_manager.add_env_vars_for_htaccess(user, app_directory, None, doc_root) except Exception as err: clprint.print_diag("text", {"status": "ERROR", "message": str(err)}) if app_config.get("app_status") == APP_STOPPED_CONST: # stopped app user_home = pwd.getpwnam(user).pw_dir # Remove app's nodenv, e.g. /home/cltest1/nodevenv/test_app nodevenv_path = "/".join([user_home, "nodevenv", app_directory]) try: # Remove app's nodenv, e.g. /home/cltest1/nodevenv/test_app shutil.rmtree(nodevenv_path) except OSError: pass # if domain is already removed we shouldn't do anything if doc_root is not None: # Remove NodeJS lines from .htaccess htaccess_filename = apps_manager.get_htaccess_by_appdir(user, app_directory, doc_root) clpassenger.remove_passenger_lines_from_htaccess(htaccess_filename) # remove app from node-selector.json file apps_manager.remove_app_from_config(user, app_directory) # Update application status in passenger try: clpassenger.restart(user, app_directory) except ClSelectExcept.MissingApprootDirectory: pass return # Moved before removing an app from node-selector.json because in case if # app_root is absent we still want to remove venv dir of an application. # The method didn't anyhing if app_root is absent and skip other commands # including removal of a venv dir prefix = _get_prefix(user, app_directory) abs_dir, _ = utils.get_abs_rel(user, prefix) try: # Remove app's nodenv, e.g. /home/cltest1/nodevenv/test_app shutil.rmtree(abs_dir) except OSError: pass user_summary = clpassenger.summary(user) # remove app from node-selector.json file app_in_config = apps_manager.remove_app_from_config(user, app_directory) try: utils.get_using_realpath_keys(user, app_directory, user_summary) except KeyError: # if app was existed and app's dir is not exists, we skip all further actions if app_in_config: return None else: raise ClSelectExcept.WrongData('No such application (or application not configured) "%s"' % app_directory) # remove app from passenger clpassenger.unconfigure(user, app_directory) # Clear .htaccess try: clpassenger.restart(user, app_directory) except ClSelectExcept.MissingApprootDirectory: pass def start(user, app_directory, doc_root, apps_manager=None): """ Starts web app with specified directory for specified user :param user: username :param app_directory: application directory :param doc_root: Document root for selected domain :param apps_manager: Application Manager. Class that responsible for gathering and writing information about applications :return: None """ if apps_manager is None: apps_manager = ApplicationsManager() app_config = apps_manager.get_app_config(user, app_directory) if app_config is None: raise ClSelectExcept.WrongData('No such application (or application not configured) "%s"' % app_directory) # Do nothing if application already started if app_config.get("app_status") == APP_STARTED_CONST: return # Create .htaccess file for the application apps_manager.update_htaccess_file(user, app_directory, doc_root) # Save new application status in user's config apps_manager.set_app_status(user, app_directory, APP_STARTED_CONST) # Update application status in passenger clpassenger.restart(user, app_directory) def restart(user, app_directory, doc_root, apps_manager=None): """ Restarts web app with specified directory for specified user :param user: username :param app_directory: application directory :param doc_root: Document root for selected domain :param apps_manager: Application Manager. Class that responsible for gathering and writing information about applications :return: None """ if apps_manager is None: apps_manager = ApplicationsManager() app_config = apps_manager.get_app_config(user, app_directory) if app_config is None: raise ClSelectExcept.WrongData('No such application (or application not configured) "%s"' % app_directory) # If application was stopped - start it if app_config.get("app_status") == APP_STOPPED_CONST: start(user, app_directory, doc_root) else: clpassenger.restart(user, app_directory) def stop(user, app_directory, doc_root, apps_manager=None): """ Stops web app with specified directory for specified user :param user: username :param app_directory: application directory :param doc_root: Document root for selected domain :param apps_manager: Application Manager. Class that responsible for gathering and writing information about applications :return: None """ if apps_manager is None: apps_manager = ApplicationsManager() app_config = apps_manager.get_app_config(user, app_directory) if app_config is None: raise ClSelectExcept.WrongData('No such application (or application not configured) "%s"' % app_directory) # Do nothing if application is not started if app_config.get("app_status") == APP_STOPPED_CONST: return htaccess_filename = apps_manager.get_htaccess_by_appdir(user, app_directory, doc_root, app_config) # Remove NodeJS lines from .htaccess clpassenger.remove_passenger_lines_from_htaccess(htaccess_filename) # Save new application status in user's config apps_manager.set_app_status(user, app_directory, APP_STOPPED_CONST) # Update application status in passenger clpassenger.restart(user, app_directory) def check_response_from_webapp(domain, alias, action=None): """ Check response from user's webapp before and after calling action. Also compare both responses :param domain: domain associated with webapp :param alias: URI associated with webapp :param action: called action, that make something with webapp: install modules, transit it, etc :return: None """ app_is_inaccessible_before = 'Web application is inaccessible by its address "%s". The operation wasn\'t performed.' app_is_inaccessible_after = ( "The operation was performed, but check availability of application has failed. " 'Web application is inaccessible by its address "%s" after the operation.' ) app_is_broken = ( "The operation was performed, but check availability of application has failed. " 'Web application responds, but its return code "%s" or ' 'content type before operation "%s" doesn\'t equal to contet type after operation "%s".' ) requests.packages.urllib3.disable_warnings(urllib_exceptions.InsecureRequestWarning) # pylint: disable=E1101 if not callable(action): raise ClSelectExcept.WrongData("Wrong action for calling in checking webapp") webapp_url = "https://{domain}/{alias}".format( domain=domain, alias=alias, ) # for hiding from the module of Apache `mod_security` headers = { "User-Agent": "Mozilla/5.0 (X11; U; Linux i686; en-US; rv:1.9.2.13) " "Gecko/20101209 CentOS/3.6-2.el5.centos Firefox/3.6.13" } try: # Applies only to wildcard domains because '*' allows accessing URLs like # Example: http://{anything}.domain/nodejs_app. webapp_url = utils.transform_wildcard_url(webapp_url) request = requests.get(webapp_url, verify=False, headers=headers) except ConnectionError: # site is not available by https protocol webapp_url = webapp_url.replace("https://", "http://") try: request = requests.get(webapp_url, verify=False, headers=headers) except ConnectionError: raise ClSelectExcept.WebAppError(app_is_inaccessible_before % webapp_url) before_mime_type = request.headers.get("Content-Type") before_status = request.headers.get("Status") action() try: request = requests.get(webapp_url, verify=False, headers=headers) except ConnectionError: raise ClSelectExcept.WebAppError(app_is_inaccessible_after % webapp_url) after_mime_type = request.headers.get("Content-Type") after_status = request.headers.get("Status") # assume that app is broken if: # it's response Content-Type or Status code (if first code wasn't 500) changed # if last Status code was 500 (internal error) if ( before_mime_type.lower() != after_mime_type.lower() or (before_status != after_status and before_status is not None and before_status[:3] != "500") or (after_status is not None and after_status[:3] == "500") ): raise ClSelectExcept.WebAppError(app_is_broken % (after_status, before_mime_type, after_mime_type)) def _get_info_about_webapp(app_summary=None, user=None): """ Get info (alias and domain) about user's web application :param app_summary: dict -> summary info about user's web application :param user: str -> name of unix user :return: tuple -> (alias, domain) """ if app_summary is None: raise ClSelectExcept.WrongData("Was passed incorrect summary info about application") if user is None: raise ClSelectExcept.WrongData("Was passed incorrect name of user") alias = app_summary["alias"] user_domains = userdomains(user) found_domains = [ domain for domain, doc_root in user_domains if utils.realpaths_are_equal(user, doc_root, app_summary["docroot"]) ] if len(found_domains) == 0: raise ClSelectExcept.WrongData("Can not found suitable domain for application") app_domain = found_domains[0] return alias, app_domain def install(user, directory, extension="-", skip_web_check=False, apps_manager=None): """ Install nodejs extension to user's webapp :param user: name os unix user :param directory: directory with webapp (app-root) :param extension: name and version of extension :param skip_web_check: skip check web application after change it's properties :param apps_manager: Application Manager. Class that responsible for gathering and writing information about applications :return: None """ if apps_manager is None: apps_manager = ApplicationsManager() user_config = apps_manager.get_user_config_data(user) try: app_data = utils.get_using_realpath_keys(user, directory, user_config) except KeyError: raise ClSelectExcept.WrongData("Record about application {} is absent".format(directory)) if app_data["app_status"] != APP_STARTED_CONST: skip_web_check = True else: alias = app_data["app_uri"] app_domain = app_data["domain"] nodejs_version = app_data["nodejs_version"] cwd, _ = utils.get_abs_rel(user, directory) environment = _create_environment(user, directory, nodejs_version) def action(): # npm install environment.extension_install(extension=extension, cwd=cwd) clpassenger.restart(user, directory) if not skip_web_check: try: check_response_from_webapp( domain=app_domain, alias=alias, action=action, ) except ClSelectExcept.WebAppError as err: raise ClSelectExcept.WebAppError("An error occured during installation of modules. %s" % err) else: action() def summary(user): summ = {} for directory, data in iteritems(clpassenger.summary(user)): if data["interpreter"] != ApplicationsManager.INTERPRETER: continue environment = _get_environment(user, directory, data).as_deepdict() summ[directory] = { "domain": data["domain"], "alias": data["alias"], "environment": environment["name"], "interpreter": environment["interpreter"], } # add only list with additions domains if "domains" in data and len(data["domains"]) > 1: summ[directory]["domains"] = data["domains"] return summ def validate_env_vars(env_vars): if type(env_vars) is not dict: raise TypeError for item in env_vars: if not isinstance(env_vars[item], basestring): raise TypeError return env_vars	20.66 KB	2026-02-26 15:33
📄 clselectctlphp.py #!/opt/cloudlinux/venv/bin/python3 -bb # -- 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 print_function from __future__ import absolute_import from __future__ import division import subprocess import sys import getopt import os import clcommon import traceback import simplejson as json import shutil import pwd from builtins import map from .clselect import ClSelect from .clextselect import ClExtSelect, depend_modules_dict from .cluserselect import ClUserSelect from .cluserextselect import ClUserExtSelect from .cluseroptselect import ClUserOptSelect from .clselectprint import clprint from clcommon import clcagefs, ClPwd from clcommon.const import Feature from clcommon.cpapi import is_panel_feature_supported from clcommon.sysctl import SysCtlConf, SYSCTL_CL_CONF_FILE from clcommon.utils import is_ubuntu, run_command, ExternalProgramFailed from .utils import make_symlink, safely_resolve_username_and_doc_root from .clselectexcept import ClSelectExcept, BaseClSelectException from .models import Website # This is the oldest version with all bad things API_0 = 0 # This version replaces hardcoded interpreter name in utility output with more # generic key names that will be correctly understood by cloudlinux-selector API_1 = 1 # Path to cagefs command CAGEFSCTL_COMMAND = "/usr/sbin/cagefsctl" def usage(): print(" -v \| --version : Specify alternative version") print(" -u \| --user : Username") print(" --domain : Domain name (for per-domain configuration)") print(" -l \| --list : List alternatives for interpreter") print(" -G \| --list-extensions : List global set of extensions for alternative") print(" -S \| --summary : List summary of alternatives") print(" -s \| --user-summary : List user summary of alternatives") print(" -C \| --current : Show currently selected alternative") print(" -c \| --user-current : Show currently selected alternative for a user") print(" -B \| --set-current : Set alternative as global default") print(" -b \| --set-user-current : Set alternative as user default") print(" -Y \| --enable-alternative : Enable alternative globally") print(" -N \| --disable-alternative : Disable alternative globally") print(" -E \| --enable-extensions : Enable comma-separated list of extensions globally for a version") print(" -D \| --disable-extensions : Disable comma-separated list of extensions globally for a version") print(" -R \| --replace-extensions : Replace extensions with comma-separated list of extensions ") print(" for a version globally") print(" -e \| --enable-user-extensions : Enable comma-separated list of extensions for a user") print(" -d \| --disable-user-extensions : Disable comma-separated list of extensions for a user") print(" -r \| --replace-user-extensions : Replace user extensions with comma-separated list of extensions") print(" -t \| --reset-user-extensions : Replace user extensions with version default extensions") print(" -g \| --list-user-extensions : List enabled extensions for a user. With key --all shows all extensions") print(" -a \| --all : Show all extensions") print(" -p \| --print-summary : If specified along with setting an alternative prints user summary") print(" -V \| --show-native-version : Shows native version while showing summary or selected version") print(" -L \| --list-users : List users who use a specified alternative") print(" -T \| --change-to-version : Changes to a specified version all users who have a certain version") print(" -k \| --add-options : Add comma-separated list options for a user") print(" -m \| --replace-options : Replace user options with comma-separated list of options") print(" -x \| --delete-options : Delete comma-separated list options for a user") print(" -Q \| --base64 : Expects data as comma-separated base64-encoded string") print(" -q \| --quiet : Suppress errors messages for wrong input") print(" -P \| --print-options : Prints user options. By default prints as plain text") print(" --print-options-safe : Prints user options. By default prints as plain text (safe strings)") print(" --apply-symlinks-rules : Recreate symlinks to php extensions for all users ") print(" based on /etc/cl.selector/symlinks.rules file") print(" --exclude-pid-list : Specify list of PIDs of processes that should not be signaled by SIGHUP") print(" -j \| --json : Print data as JSON") print(" -w \| --csv : Print data as CSV") print(" -W \| --perl : Print data as perl structure") print(" --api-version : Integer, representing specific api version to use. ") print(" Defaults to {}".format(API_0)) print(" -z \| --reset-options : Deletes all user custom options. Range can be narrowed with user ") print(" or version options") print(" --update-backup : write settings to backup") print(" --apply-global-php-ini : use with 0, 1 or 2 arguments from the list: error_log, date.timezone") print(" without arguments applies all global php options including two above") print(" --setup-without-cagefs : setup PHP Selector without CageFS") print(" --revert-to-cagefs : revert to default setup of PHP Selector (with CageFS)") print(" --for-all-users : applies specified action for all users in CageFS. ") print(" Available only for enable/disable user extensions") def print_error_and_exit(message, prefix=None): """ Prints to stderr @param message: string """ fmt = "%s\n" if prefix: fmt = "%s:%s\n" % (prefix, "%s") sys.stderr.write(fmt % message) sys.exit(1) def check_args_presence(): """ Checks presence of command line arguments and exits with usage info if missing """ if len(sys.argv) == 1: print_error_and_exit("Command line arguments expected. For help use '-h' or '--help' options") def get_name_modifier(version): """ """ BASE_ALT_DIR = "/opt/alt/php" ver = version.replace(".", "") name_modifier_file = BASE_ALT_DIR + ver + "/name_modifier" name_modifier = "" if os.path.isfile(name_modifier_file): try: name_modifier = open(name_modifier_file, "r").readline().strip() except (OSError, IOError): return "" return name_modifier def letter_to_status(letter): if letter == "-": return "disabled" elif letter == "e": return "enabled" def format_summary(data, format="text", api_version=API_0): if api_version == API_0: available_versions_key = "PHPConfiguration" default_version_key = "defaultPHPversion" else: available_versions_key = "available_versions" default_version_key = "default_version" states = ["e", "d", "s"] text_lines = [] json_dict = { available_versions_key: [], } for alt in data: row_length = len(alt[1]) fmt = " ".join(["%s"] * (row_length + 1)) row_data = [alt[0]] row_data.extend(list(map((lambda i: ((alt[1][i] and states[i]) or "-")), range(row_length)))) # pylint: disable=range-builtin-not-iterating name_modifier = get_name_modifier(alt[0]) if name_modifier != "": fmt = " ".join(["%s"] * (row_length + 2)) row_data.append(name_modifier) if format == "text": text_lines.append(fmt % tuple(row_data)) if "d" in row_data: json_dict[default_version_key] = row_data[0] json_dict[available_versions_key].append( { "version": row_data[0], "status": letter_to_status(row_data[1]), "name_modifier": name_modifier, } ) if format == "json": return json.dumps(json_dict) elif format == "text": return "\n".join(text_lines) def print_summary(data, format="text", api_version=API_0): """ Prints alternatives summary """ data = format_summary(data, format, api_version) print(data) def check_params(config, param_list): """ Check that config has param_list and this params not None """ for param in param_list: if param not in config or config[param] == None: print_error_and_exit("Error: %s must be specified" % param) def ext_letter_to_status(letter): if letter == "~": return "build-in" elif letter == "+": return "enabled" else: return "disabled" def fill_descriptions(tmp_list): descr_file = "/etc/cl.selector.conf.d/phpextdesc.txt" if clcagefs.in_cagefs() else "/etc/cl.selector/phpextdesc.txt" with open(descr_file) as f: desct_content = f.readlines() full_desct_dict = {} for line in desct_content: line_parts = line.split("=") full_desct_dict[line_parts[0]] = line_parts[1].strip() for item in tmp_list: try: item["description"] = full_desct_dict[item["name"]] except KeyError: # skip extention without description pass return tmp_list def print_json_status_ok(): result_dict = {"status": "ok"} print(json.dumps(result_dict)) def get_cpanel_user(): """ Return user (name of cpanel account) for PHP Selector without CageFS feature """ if os.path.isfile(ClSelect.USER_CONF): with open(ClSelect.USER_CONF, "r") as f: return f.read().strip() for user in ClPwd().get_user_dict(): if os.path.exists("/var/cpanel/users/" + user): return user return None def set_cpanel_user(user): """ Set user (name of cpanel account) for PHP Selector without CageFS feature """ with open(ClSelect.USER_CONF, "w") as f: f.write(user) os.chmod(ClSelect.USER_CONF, 0o644) def switch_linksafe(enable=False): if not is_panel_feature_supported(Feature.LVE): return new_conf = "/etc/sysctl.d/cloudlinux-linksafe.conf" if os.path.isfile(new_conf): conf = new_conf else: conf = SYSCTL_CL_CONF_FILE sysctl_cfg = SysCtlConf(config_file=conf) sysctl_cfg.set("fs.protected_symlinks_create", str(int(enable))) sysctl_cfg.set("fs.protected_hardlinks_create", str(int(enable))) if subprocess.call("sysctl --system &>/dev/null", shell=True, executable="/bin/bash") != 0: print("Error while executing: sysctl --system") sys.exit(1) def modify_search_path_in_bashrc(user, homedir, obj=None, add=True): """ Add path to PATH variable in ~/.bashrc :param user: name of user :type user: string :param homedir: path to home directory :type homedir: string :param obj: instance of ClUserSelect class :type obj: instance of ClUserSelect class :param add: add path to .bashrc when True, remove otherwise :type obj: bool """ if obj is None: obj = ClUserSelect("php") cur_user = obj._change_uid(user) # On Ubuntu we are using ~/.profile # On CL - ~/.bashrc if is_ubuntu(): bashrc = homedir + "/.profile" else: bashrc = homedir + "/.bashrc" line = "PATH=$HOME/" + ClUserSelect.SELECTOR2_DIR + ":$HOME/.cl.selector:$PATH" with open(bashrc, "r") as f: found = line + "\n" in f if add: if not found: with open(bashrc, "a") as f: f.write("\n" + line + "\n") else: if found: clcommon.utils.delete_line_from_file(bashrc, line) obj._restore_uid(cur_user) def restore_settings_from_backup(user, homedir, uid, alt_dirs, obj=None): """ Restore (apply) settings for PHP Selector from backup :param user: name of user :type user: string :param homedir: path to user's home directory :type homedir: string :param uid: user's uid :type uid: int :param alt_dirs: list of alt-php directories like ['php51', 'php52'] :type alt_dirs: list :param obj: instance of ClUserSelect class :type obj: ClUserSelect object """ def cleanup(): # Delete unneeded files shutil.rmtree(os.path.join("/var/cagefs", str(uid)[-2:]), True) shutil.rmtree("/usr/share/cagefs/etc", True) shutil.rmtree("/usr/share/cagefs/etc.new", True) if obj is None: obj = ClUserSelect("php") base_dest_path = homedir + "/.cl.selector" cleanup() # Generate alt_php.ini for all versions using cagefsctl if ( subprocess.call("/usr/sbin/cagefsctl --silent --force-update-etc " + user, shell=True, executable="/bin/bash") != 0 ): sys.exit(1) cur_user = obj._change_uid(user) # Copy generated alt_php.ini files to new location in user's home directory base_src_path = os.path.join("/var/cagefs", str(uid)[-2:], user, "etc", "cl.php.d") for alt_dir in alt_dirs: src_path = base_src_path + "/alt-" + alt_dir + "/alt_php.ini" dest_path = base_dest_path + "/alt_" + alt_dir + ".ini" shutil.copy(src_path, dest_path) # Select php version from backup (or default when backup does not exist) obj.set_version_from_backup(user) shutil.rmtree(homedir + "/.cagefs", True) obj._restore_uid(cur_user) cleanup() def disable_cagefs_service(): if os.path.isfile("/usr/bin/systemctl"): subprocess.run("/usr/bin/systemctl disable cagefs", shell=True, executable="/bin/bash") subprocess.run("/usr/bin/systemctl stop cagefs", shell=True, executable="/bin/bash") subprocess.run("/usr/bin/systemctl mask cagefs", shell=True, executable="/bin/bash") else: subprocess.run("/sbin/service cagefs stop &> /dev/null", shell=True, executable="/bin/bash") subprocess.run("/sbin/chkconfig cagefs off", shell=True, executable="/bin/bash") def enable_cagefs_service(): if os.path.isfile("/usr/bin/systemctl"): subprocess.run("/usr/bin/systemctl unmask cagefs", shell=True, executable="/bin/bash") subprocess.run("/usr/bin/systemctl enable cagefs", shell=True, executable="/bin/bash") subprocess.run("/usr/bin/systemctl start cagefs", shell=True, executable="/bin/bash") else: subprocess.run("/sbin/chkconfig cagefs on", shell=True, executable="/bin/bash") subprocess.run("/sbin/service cagefs start &> /dev/null", shell=True, executable="/bin/bash") def setup_without_cagefs(args): """ Setup PHP Selector without CageFS """ sys.path.append("/usr/share/cagefs") try: import cagefslib except ImportError: print("Error: CageFS is not installed") sys.exit(1) # alt-php versions are installed ? alt_dirs = cagefslib.get_alt_dirs() if not alt_dirs: print("alt-php not found") sys.exit(1) # detect cpanel user if args: user = args[0] set_cpanel_user(user) else: user = get_cpanel_user() if not user: print("Error: failed to detect cpanel account. Please specify name of an account as argument:") print("selectorctl --setup-without-cagefs USER") sys.exit(1) if not os.path.exists(ClSelect.USER_CONF): set_cpanel_user(user) # disable linksafe protection switch_linksafe() pw = pwd.getpwnam(user) homedir = pw.pw_dir # create symlinks to user's alt_php.ini files for alt_dir in alt_dirs: alt_path = "/opt/alt/" + alt_dir + "/link/conf/alt_php.ini" user_path = homedir + "/.cl.selector/alt_" + alt_dir + ".ini" make_symlink(user_path, alt_path) obj = ClUserSelect("php") restore_settings_from_backup(user, homedir, pw.pw_uid, alt_dirs, obj) obj.create_selector_symlinks(user) modify_search_path_in_bashrc(user, homedir, obj) disable_cagefs_service() # kill user's processes in LVE subprocess.run( "/usr/sbin/lvectl destroy " + str(pw.pw_uid) + " &>/dev/null; /usr/sbin/lvectl apply " + str(pw.pw_uid) + " &>/dev/null", shell=True, executable="/bin/bash", ) def revert_to_cagefs(): """ Revert to default PHP Selector setup with CageFS """ if not os.path.exists(ClSelect.USER_CONF): print('PHP Selector is in default mode already ("with CageFS" mode)') sys.exit(1) sys.path.append("/usr/share/cagefs") try: import cagefslib except ImportError: print("Error: CageFS is not installed") sys.exit(1) # alt-php versions are installed ? alt_dirs = cagefslib.get_alt_dirs() if not alt_dirs: print("alt-php not found") sys.exit(1) switch_linksafe(enable=True) # delete symlinks to user's alt_php.ini files for alt_dir in alt_dirs: alt_path = "/opt/alt/" + alt_dir + "/link/conf/alt_php.ini" if os.path.islink(alt_path): os.unlink(alt_path) user = get_cpanel_user() if not user: print("Error: failed to detect user") sys.exit(1) pw = pwd.getpwnam(user) homedir = pw.pw_dir obj = ClUserSelect("php") modify_search_path_in_bashrc(user, homedir, obj, add=False) # remove config file (file-switch) for "without CageFS" mode os.unlink(ClSelect.USER_CONF) # Generate alt_php.ini for all versions using cagefsctl subprocess.run("/usr/sbin/cagefsctl --silent --force-update-etc " + user, shell=True, executable="/bin/bash") enable_cagefs_service() # kill user's processes in LVE subprocess.run( "/usr/sbin/lvectl destroy " + str(pw.pw_uid) + " &>/dev/null; /usr/sbin/lvectl apply " + str(pw.pw_uid) + " &>/dev/null", shell=True, executable="/bin/bash", ) def apply_global_php_ini(args): """ Apply "global" php.ini settings to all alt-php versions :param args: list of command line parameters (names of php.ini options) :type args: list """ sys.path.append("/usr/share/cagefs") try: import cagefslib import cagefsreconfigure except ImportError: print("Error: CageFS is not installed") sys.exit(1) # alt-php versions are installed ? if cagefslib.get_alt_versions(): cagefsreconfigure.replace_alt_settings(options=args) def _check_depencies_and_print_message(print_format, print_message): if len(depend_modules_dict): # Warning - blocked modules present modules_list = list() for module_name, dep_module in depend_modules_dict.items(): modules_list.append(" '%s' is required for '%s'" % (module_name, dep_module)) clprint.print_diag(print_format, {"status": "WARN", "message": print_message + ",".join(modules_list)}) def get_extensions(interpreter, version, fmt="text"): ext_list = ClExtSelect(interpreter).list_extensions(version) return parse_extensions(ext_list, version, fmt) def parse_extensions(ext_list, version, fmt): json_list = [] for ext in ext_list: action = "~" if ext[1] is True: action = "+" elif ext[1] is False: action = "-" if fmt == "text": json_list.append((action, ext[0])) else: json_list.append({"name": ext[0], "description": "", "state": ext_letter_to_status(action)}) json_list = fill_descriptions(json_list) if fmt == "json": result_dict = {"version": version, "extensions": json_list} return result_dict return json_list def get_cagefs_users(): """ Return list of users that are in CageFS If CageFS is not installed or initialized throws exception and prints it :return: """ not_installed_msg = "No such file or directory" not_initialized_msg = "CageFS is not initialized" try: users = run_command([CAGEFSCTL_COMMAND, "--list-enabled"]).strip() if users == "": return [] return users.split("\n")[1:] # First element shows number of users except ExternalProgramFailed as e: if not_installed_msg in str(e): print_error_and_exit("ERROR: CageFS not installed.") elif not_initialized_msg in str(e): print_error_and_exit( 'Error: CageFS is not initialized. Use "/usr/sbin/cagefsctl --init" to initialize CageFS' ) print_error_and_exit(e) def verify_options_in_cagefs(options, supported_options): """ Verifies if the provided options are supported in CageFS and appends the current user option if user-related options are absent """ input_options = {opt for opt, _ in options} if not input_options & set(supported_options): print_error_and_exit("Error: This command is not supported in CageFS") if "--user" not in input_options and "-u" not in input_options: current_user_name = pwd.getpwuid(os.geteuid()).pw_name options.append(("--user", current_user_name)) def _get_website_from_domain(user, domain): """ Resolve domain to docroot and create a Website object. Returns None if domain is not provided. """ if not domain: return None try: _, doc_root = safely_resolve_username_and_doc_root(user, domain) return Website(domain=domain, docroot=doc_root) except (ClSelectExcept, BaseClSelectException): # If domain resolution fails, return None (will use per-user config) return None def main(): config = {} config["interpreter"] = "php" config["version"] = None config["show-all"] = False config["print-summary"] = False config["show-native-version"] = False config["decoder"] = "plain" config["quiet"] = False config["format"] = "text" config["api-version"] = API_0 config["domain"] = None actions = {} exclude_pid_list = [] check_args_presence() try: opts, args = getopt.getopt( sys.argv[1:], "hi:lSsCcB:Y:N:E:D:R:v:Gu:b:ge:d:r:atpVLT:k:m:x:QqPjwWz", [ "help", "setup-without-cagefs", "revert-to-cagefs", "interpreter=", "list", "summary", "user-summary", "current", "user-current", "set-current=", "enable-alternative=", "disable-alternative=", "enable-extensions=", "disable-extensions=", "replace-extensions=", "version=", "list-extensions", "user=", "domain=", "set-user-current=", "list-user-extensions", "enable-user-extensions=", "disable-user-extensions=", "replace-user-extensions=", "all", "reset-user-extensions", "print-summary", "show-native-version", "list-users", "change-to-version=", "add-options=", "replace-options=", "delete-options=", "base64", "apply-symlinks-rules", "quiet", "print-options", "print-options-safe", "json", "csv", "perl", "api-version=", "reset-options", "update-backup", "apply-global-php-ini", "exclude-pid-list=", "for-all-users", ], ) except getopt.GetoptError: usage() sys.exit(1) opts_supported_in_cagefs = [ "-l", "--list", "-s", "--user-summary", "-c", "--user-current", "-b", "--set-user-current", "-e", "--enable-user-extensions", "-d", "--disable-user-extensions", "-r", "--replace-user-extensions", "-t", "--reset-user-extensions", "-g", "--list-user-extensions", "-k", "--add-options", "-m", "--replace-options", "-x", "--delete-options", "-P", "--print-options", "--print-options-safe", "-z", "--reset-options", ] if clcagefs.in_cagefs(): verify_options_in_cagefs(opts, opts_supported_in_cagefs) for o, a in opts: if o in ["-h", "--help"]: usage() sys.exit(0) elif o in ["--exclude-pid-list"]: for pid in a.split(","): try: exclude_pid_list.append(int(pid)) except ValueError: continue elif o in ["--apply-symlinks-rules"]: actions["apply-symlinks-rules"] = True elif o in ["--setup-without-cagefs"]: setup_without_cagefs(args) sys.exit(0) elif o in ["--revert-to-cagefs"]: revert_to_cagefs() sys.exit(0) elif o in ("--apply-global-php-ini",): apply_global_php_ini(args) sys.exit(0) elif o in ["-l", "--list"]: actions["list-alternatives"] = True elif o in ["-S", "--summary"]: actions["show-summary"] = True elif o in ["-s", "--user-summary"]: actions["show-user-summary"] = True elif o in ["-C", "--current"]: actions["show-current"] = True elif o in ["-c", "--user-current"]: actions["show-user-current"] = True elif o in ["-a", "--all"]: config["show-all"] = True elif o in ["-v", "--version"]: config["version"] = a elif o in ["-d", "--domain"]: config["domain"] = a elif o in ["-u", "--user"]: config["user"] = a clpwd = ClPwd() if ClSelect.work_without_cagefs(): uid = clpwd.get_uid(a) else: users = a.split(",") user_list = list() try: if len(users) == 1: uid = clpwd.get_uid(a) if os.geteuid() == 0: for user in clpwd.get_names(uid): ClUserSelect().cagefs_copy_etc(user) else: for user in users: if user not in user_list: user_list += clpwd.get_names(clpwd.get_uid(user)) config["user"] = ",".join(user_list) except ClPwd.NoSuchUserException as e: sys.stderr.write(str(e) + "\n") sys.exit(1) elif o in ["--domain"]: config["domain"] = a elif o in ["-B", "--set-current"]: actions["set-current"] = a elif o in ["-b", "--set-user-current"]: actions["set-user-current"] = a elif o in ["-Y", "--enable-alternative"]: actions["enable-alternative"] = a elif o in ["-N", "--disable-alternative"]: actions["disable-alternative"] = a elif o in ["-G", "--list-extensions"]: actions["list-extensions"] = True elif o in ["-g", "--list-user-extensions"]: actions["list-user-extensions"] = True elif o in ["-E", "--enable-extensions"]: actions["enable-extensions"] = a elif o in ["-D", "--disable-extensions"]: actions["disable-extensions"] = a elif o in ["-R", "--replace-extensions"]: actions["replace-extensions"] = a elif o in ["-e", "--enable-user-extensions"]: actions["enable-user-extensions"] = a elif o in ["-d", "--disable-user-extensions"]: actions["disable-user-extensions"] = a elif o in ["-r", "--replace-user-extensions"]: actions["replace-user-extensions"] = a elif o in ["-t", "--reset-user-extensions"]: actions["reset-user-extensions"] = True elif o in ["-p", "--print-summary"]: config["print-summary"] = True elif o in ["-V", "--show-native-version"]: config["show-native-version"] = True elif o in ["-L", "--list-users"]: actions["list-users"] = True elif o in ["-T", "--change-to-version"]: actions["change-to-version"] = a elif o in ["-k", "--add-options"]: actions["add-options"] = a elif o in ["-m", "--replace-options"]: actions["replace-options"] = a elif o in ["-x", "--delete-options"]: actions["delete-options"] = a elif o in ["-Q", "--base64"]: config["decoder"] = "base64" elif o in ["-q", "--quiet"]: config["quiet"] = True elif o in ["-P", "--print-options"]: actions["print-options"] = True elif o in ["--print-options-safe"]: actions["print-options-safe"] = True elif o in ["-j", "--json"]: config["format"] = "json" elif o in ["-w", "--csv"]: config["format"] = "csv" elif o in ["--api-version"]: config["api-version"] = int(a) elif o in ["-W", "--perl"]: config["format"] = "perl" elif o in ["-z", "--reset-options"]: actions["reset-options"] = True elif o in ["--update-backup"]: actions["update-backup"] = True elif o in ["--for-all-users"]: if "user" in config: print_error_and_exit( "--for-all-users and --user options are mutually" " exclusive options and cannot be used simultaneously." "\nUse --for-all-user OR --user instead." ) users = get_cagefs_users() if not users: print_error_and_exit("No changes were made: there are no users with cagefs enabled ") clpwd = ClPwd() user_list = list() for user in users: if user not in user_list: user_list += clpwd.get_names(clpwd.get_uid(user)) config["user"] = ",".join(user_list) if len(actions) != 1: if len(actions) == 0 and config["show-native-version"]: try: print(ClSelect(config["interpreter"]).get_native_version()[0]) except TypeError: pass else: print_error_and_exit("Wrong set of options", "ERROR") try: # check if we are able to do anything before actually parsing options # this two exceptions check for native version inside # this is done because selectorctl is called in alt-php spec in cycle # and we should avoid printing lot of messages there if "set-user-current" not in actions and "show-user-current" not in actions: ClSelect().check_requirements() if "list-alternatives" in actions: if config["format"] != "json": for alt in ClSelect(config["interpreter"]).list_alternatives(): print("%s\t%s\t%s" % (alt)) else: alternatives_dict = {"status": "ok", "data": []} for alt in ClSelect(config["interpreter"]).list_alternatives(): alternatives_dict["data"].append({"short": alt[0], "full": alt[1], "path": alt[2]}) print(json.dumps(alternatives_dict)) elif "show-summary" in actions: data = ClSelect(config["interpreter"]).get_summary(config["show-native-version"]) print_summary(data, config["format"], config["api-version"]) elif "show-current" in actions: print("%s\t%s\t%s" % ClSelect(config["interpreter"]).get_version(config["show-native-version"])) elif "set-current" in actions: ClSelect(config["interpreter"]).set_version(actions["set-current"]) if config["format"] == "json": print_json_status_ok() elif "enable-alternative" in actions: if "user" in config and config["user"]: raise ClSelectExcept.VersionModificationBlocked( "The 'enable-alternative' option is used to enable configurations globally." " The 'user' option can't be used with it." ) else: ClSelect(config["interpreter"]).enable_version(actions["enable-alternative"]) if config["format"] == "json": print_json_status_ok() elif "disable-alternative" in actions: if "user" in config and config["user"]: raise ClSelectExcept.VersionModificationBlocked( "The 'disable-alternative' option is used to disable configurations globally." " The 'user' option can't be used with it." ) else: ClSelect(config["interpreter"]).disable_version(actions["disable-alternative"]) if config["format"] == "json": print_json_status_ok() elif "list-extensions" in actions: check_params(config, ("interpreter", "version")) ext_list = get_extensions(config["interpreter"], config["version"], config["format"]) if config["format"] == "text": for item in ext_list: print("%s %s" % item) elif config["format"] == "json": print(json.dumps(ext_list)) elif "enable-extensions" in actions: check_params(config, ("interpreter", "version")) ClExtSelect(config["interpreter"]).enable_extensions( config["version"], list(map((lambda i: i.strip()), actions["enable-extensions"].split(","))) ) elif "disable-extensions" in actions: check_params(config, ("interpreter", "version")) ClExtSelect(config["interpreter"]).disable_extensions( config["version"], list(map((lambda i: i.strip()), actions["disable-extensions"].split(","))) ) if len(depend_modules_dict): # Warning - blocked modules present _check_depencies_and_print_message(config["format"], "Modules left by dependencies:") elif "replace-extensions" in actions: check_params(config, ("interpreter", "version")) ClExtSelect(config["interpreter"]).replace_extensions( config["version"], list(map((lambda i: i.strip()), actions["replace-extensions"].split(","))) ) if len(depend_modules_dict): # Warning - blocked modules present _check_depencies_and_print_message(config["format"], "Modules left/added by dependencies:") elif config["format"] == "json": print_json_status_ok() elif "show-user-summary" in actions: check_params(config, ("interpreter", "user")) data = ClUserSelect(config["interpreter"], exclude_pid_list).get_summary( config["user"], config["show-native-version"] ) print_summary(data) elif "show-user-current" in actions: check_params(config, ("interpreter", "user")) print( "%s\t%s\t%s" % ClUserSelect(config["interpreter"], exclude_pid_list).get_version( config["user"], config["domain"], config["show-native-version"] ) ) elif "apply-symlinks-rules" in actions: check_params(config, ("interpreter",)) ClUserSelect(config["interpreter"], exclude_pid_list).apply_symlinks_rules() elif "set-user-current" in actions: check_params(config, ("interpreter", "user")) # hack for alt-php spec where we read and re-apply # php version for each user in system # in order not to bump deps, we just silently # ignore version set requests for 'native' # (which should be set for all users on server # because web ui does not work) try: ClSelect().check_requirements() except ClSelectExcept.NativeNotInstalled: if actions["set-user-current"] != "native": raise exit(0) # we intentionally use first user cause set_version has workaround for multiple same uid users # LVEMAN-1670 user = clpwd.get_names(uid)[0] c = ClUserSelect(config["interpreter"], exclude_pid_list) data = c.set_version( user, actions["set-user-current"], config["domain"], config["print-summary"], config["show-native-version"] ) c.clean_crui_images(clpwd.get_names(uid)) if config["print-summary"]: print_summary(data) elif "list-user-extensions" in actions: check_params(config, ("interpreter", "user")) website = _get_website_from_domain(config["user"], config["domain"]) if config["show-all"]: for ext in ClUserExtSelect(config["interpreter"], exclude_pid_list).list_all_extensions( config["user"], config["version"], website=website ): action = "-" if ext[1] is None: action = "~" elif ext[1] is True: action = "+" print("%s %s" % (action, ext[0])) else: for ext in ClUserExtSelect(config["interpreter"], exclude_pid_list).list_enabled_extensions( config["user"], config["version"], website=website ): print(ext[0]) elif "enable-user-extensions" in actions: check_params(config, ("interpreter", "version", "user")) ClSelect.check_multiphp_system_default_version() users = config["user"].split(",") for user in users: c = ClUserExtSelect(config["interpreter"], exclude_pid_list) website = _get_website_from_domain(user, config["domain"]) c.bulk_enable_extensions( user=user, version=config["version"], ext_list=list(map((lambda i: i.strip()), actions["enable-user-extensions"].split(","))), check_ext=True, website=website, ) c.clean_crui_images(users) elif "disable-user-extensions" in actions: check_params(config, ("interpreter", "version", "user")) ClSelect.check_multiphp_system_default_version() users = config["user"].split(",") for user in users: c = ClUserExtSelect(config["interpreter"], exclude_pid_list) website = _get_website_from_domain(user, config["domain"]) c.bulk_disable_extensions( user, config["version"], list(map((lambda i: i.strip()), actions["disable-user-extensions"].split(","))), website=website, ) c.clean_crui_images(users) elif "replace-user-extensions" in actions: check_params(config, ("interpreter", "version", "user")) ClSelect.check_multiphp_system_default_version() user = clpwd.get_names(uid)[0] c = ClUserExtSelect(config["interpreter"], exclude_pid_list) website = _get_website_from_domain(user, config["domain"]) c.replace_extensions( user, config["version"], list(map((lambda i: i.strip()), actions["replace-user-extensions"].split(","))), website=website ) c.clean_crui_images(clpwd.get_names(uid)) elif "reset-user-extensions" in actions: check_params(config, ("interpreter", "version", "user")) ClSelect.check_multiphp_system_default_version() user = clpwd.get_names(uid)[0] c = ClUserExtSelect(config["interpreter"], exclude_pid_list) website = _get_website_from_domain(user, config["domain"]) extensions = c.reset_extensions(user, config["version"], website=website) c.clean_crui_images(clpwd.get_names(uid)) print(",".join(extensions)) elif "list-users" in actions: check_params(config, ("interpreter", "version")) users = ClUserSelect(config["interpreter"], exclude_pid_list).list_users(config["version"]) print(",".join(users)) elif "change-to-version" in actions: check_params(config, ("interpreter", "version")) ClUserSelect(config["interpreter"], exclude_pid_list).change_to_version( actions["change-to-version"], config["version"] ) elif "add-options" in actions: check_params(config, ("interpreter", "version", "user")) ClSelect.check_multiphp_system_default_version() user = clpwd.get_names(uid)[0] c = ClUserOptSelect(config["interpreter"], exclude_pid_list) c.insert_options(user, config["version"], actions["add-options"], config["decoder"], True, config["quiet"]) c.clean_crui_images(clpwd.get_names(uid)) if config["format"] == "json": clprint.print_data(config["format"], {}) elif "replace-options" in actions: check_params(config, ("interpreter", "version", "user")) ClSelect.check_multiphp_system_default_version() user = clpwd.get_names(uid)[0] c = ClUserOptSelect(config["interpreter"], exclude_pid_list) c.insert_options( user, config["version"], actions["replace-options"], config["decoder"], False, config["quiet"] ) c.clean_crui_images(clpwd.get_names(uid)) if config["format"] == "json": clprint.print_data(config["format"], {}) elif "delete-options" in actions: check_params(config, ("interpreter", "version", "user")) ClSelect.check_multiphp_system_default_version() user = clpwd.get_names(uid)[0] c = ClUserOptSelect(config["interpreter"], exclude_pid_list) c.delete_options(user, config["version"], actions["delete-options"], config["decoder"], config["quiet"]) c.clean_crui_images(clpwd.get_names(uid)) if config["format"] == "json": clprint.print_data(config["format"], {}) elif "reset-options" in actions: user = None version = None if "user" in config: user = config["user"].split(",") if config["version"]: version = config["version"].split(",") c = ClUserOptSelect(config["interpreter"], exclude_pid_list) c.reset_options(user, version) c.clean_crui_images(user) if config["format"] == "json": clprint.print_data(config["format"], {}) elif "print-options" in actions: check_params(config, ("interpreter", "user")) clprint.print_data( config["format"], ClUserOptSelect(config["interpreter"], exclude_pid_list).get_options(config["user"], config["version"]), ) elif "print-options-safe" in actions: check_params(config, ("interpreter", "user")) clprint.print_data( config["format"], ClUserOptSelect(config["interpreter"], exclude_pid_list).get_options(config["user"], config["version"]), escape=True, ) elif "update-backup" in actions: clpwd = ClPwd() for user in clpwd.get_user_dict().keys(): try: ClUserSelect()._check_user_in_cagefs(user) ClUserSelect()._backup_settings(user) ClUserOptSelect().backup_php_options(user) except ClSelectExcept.NotCageFSUser: pass # SKIP user with disabled cagefs except ClSelectExcept.UnableToSaveData as e: if not config["quiet"]: clprint.print_diag(config["format"], {"status": "ERROR", "message": str(e)}) pass # SKIP user with errors except ClSelectExcept.NativeNotInstalled as e: clprint.print_diag( config["format"], {"status": "WARNING", "message": str(e), "details": e.details, "context": e.context} ) sys.exit(1) except BaseClSelectException as e: clprint.print_diag( config["format"], {"status": "ERROR", "message": str(e), "details": e.details, "context": e.context} ) sys.exit(1) except (KeyError, UnboundLocalError): print_error_and_exit("Incomplete or incorrect set of arguments") except Exception as e: msg = traceback.format_exc() clprint.print_diag(config["format"], {"status": "ERROR", "message": msg}) if __name__ == "__main__": main()	45.25 KB	2026-02-26 15:33
📄 clselectctlpython.py # -- 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 print_function from __future__ import absolute_import from __future__ import division import getopt import glob import os import re import shutil import sys import pwd import fnmatch import secureio import traceback import requests import requests.packages.urllib3.exceptions as urllib_exceptions # pylint: disable=E0401 from packaging.version import Version from requests.exceptions import ConnectionError from future.utils import iteritems from . import clpassenger from . import clselectctl from . import utils import cldetectlib as detect from clcommon.cpapi import userdomains, docroot, cpusers, CP_NAME from clcommon.cpapi.cpapiexceptions import NoDomain, NotSupported from clcommon.utils import mod_makedirs from .clselectpython.apps_manager import ApplicationsManager, PythonAppFormatVersion, get_venv_rel_path from .clselectexcept import ClSelectExcept, BaseClSelectException from .clselectprint import clprint from .clselectpythonuser import extensions, environments, interpreters from .baseclselect import APP_STARTED_CONST, APP_STOPPED_CONST DEFAULT_STARTUP_FILE = "passenger_wsgi.py" DEFAULT_APP_STATE = "started" DEFAULT_APP_ENTRYPOINT = "application" WSGI_TEMPLATE = r"""import imp import os import sys sys.path.insert(0, os.path.dirname(__file__)) wsgi = imp.load_source('wsgi', '%s') application = wsgi.%s """ WSGI_TEMPLATE_PY12_PLUS = r"""import importlib.machinery import importlib.util import os import sys sys.path.insert(0, os.path.dirname(__file__)) def load_source(modname, filename): loader = importlib.machinery.SourceFileLoader(modname, filename) spec = importlib.util.spec_from_file_location(modname, filename, loader=loader) module = importlib.util.module_from_spec(spec) loader.exec_module(module) return module wsgi = load_source('wsgi', '%s') application = wsgi.%s """ _CONFIGURABLE_WSGI_TEMPLATE = r"""import os import sys sys.path.insert(0, os.path.dirname(__file__)) def %s(environ, start_response): start_response('200 OK', [('Content-Type', 'text/plain')]) message = 'It works!\n' version = 'Python v' + sys.version.split()[0] + '\n' response = '\n'.join([message, version]) return [response.encode()] """ WSGI_PATTERN = re.compile("imp\.load_source$.+, '(?P<script>.+)'$\s+application.+\.(?P<callable>.+)") WSGI_PATTERN_PY12_PLUS = re.compile( r"wsgi = load_source$'wsgi', '(?P<script>.+)'$\s+" r"application = wsgi\.(?P<callable>.+)" ) def usage(): print( "Warning: selectorctl utility for --interpreter python is deprecated. " "Only update-interpreter option is still possible to use. " "\nPlease, use cloudlinux-selector utility to manage your python applications" ) print(" -v \| --version VERSION : Specify alternative version") print(" -u \| --user USERNAME : Username") print(" --domain DOMAIN : Domain (or subdomain); users main domain as default") print(" -l \| --list : List alternatives for interpreter") print(" -G \| --list-extensions : List global set of packages") print(" -K \| --list-extensions-version : List version set of packages") print(" -s \| --user-summary : List user summary of webapps") print(" -y \| --create-webapp : Create user webapp") print(" \| --update-interpreter : Update binary python in virtual environment; you can use with ") print(" --user=* for all users; application name as default * (all app)") print(" -n \| --destroy-webapp : Destroy user webapp") print(" -f \| --relocate-webapp : Change webapp directory (files must be moved manually)") print(" -F \| --transit-webapp : Change webapp domain/alias") print(" -Z \| --restart-webapp : Restart webapp") print(" -t \| --stop-webapp : Stop webapp") print(" -c \| --user-current : Show currently selected alternative for user webapp") print(" -b \| --set-user-current : Set alternative as user webapp default") print(" -e \| --enable-user-extensions PACKAGES : Install comma-separated list of packages for user webapp") print(" -d \| --disable-user-extensions PACKAGES : Uninstall comma-separated list of packages for user webapp") print(" : Use '-' (minus) for all packages") print(" -r \| --replace-user-extensions PACKAGES : Update comma-separated list of packages for user webapp") print(" : Use '-' (minus) for all packages") print(" --recreate-virtualenv : Recreate python virtual environment") print(" --freeze-requirements : Freeze virtual environment installed packages") print(" -g \| --list-user-extensions : List installed packages for user webapp") print(" -o \| --setup-wsgi WSGI[:CALLABLE] : Setup wsgi") print(" -p \| --print-summary : If specified along with setting an alternative ") print(" prints user summary") print(" -j \| --json : Print data as JSON") print(" \| --verbose : show detail information about action") def _create_environment(user, directory, version, env_name=None, use_existing=True): """ Create new environment (or use existing if use_existing=True) :param user: unix user name that app belongs to :param directory: app root :param version: python version :param env_name: DEPRECATED :param use_existing: do not raise exception if env already exists """ _, prefix = get_venv_rel_path(user, directory) if not env_name: env_name = version environment = environments.Environment(env_name, user, prefix) if environment.exists() and not use_existing: raise ClSelectExcept.EnvironmentAlreadyExists(environment.path) if not environment.exists(): try: interpreter = interpreters.interpreters(key="version")[version] except KeyError: raise ClSelectExcept.NoSuchAlternativeVersion(version) environment.create(interpreter) return environment def _get_environment(user, directory, app_summary=None, apps_manager=None): if apps_manager is None: apps_manager = ApplicationsManager() try: if app_summary: binary = app_summary["binary"] else: binary = apps_manager.get_binary_path(user, directory, pwd.getpwnam(user).pw_dir) except ClSelectExcept.ConfigMissingError: # this method is called in many places, # there could be case when config is still absent or empty app_summary = clpassenger.summary(user).get(directory) if app_summary is None: raise ClSelectExcept.WebAppError('No such application (or application not configured) "%s"' % directory) binary = app_summary["binary"] env_name = os.path.basename(os.path.dirname(os.path.dirname(binary))) _, prefix = get_venv_rel_path(user, directory) environment = environments.Environment(env_name, user, prefix) return environment def create( user, directory, alias, version=None, doc_root=None, env_vars=None, startup_file=None, domain_name=None, entry_point=None, apps_manager=None, passenger_log_file=None, ): """ Create new python application :param user: unix user name :param directory: application path in user's home (app-root) :param alias: alias (app-uri) :param version: version of interpreter :param doc_root: doc_root :param env_vars: dict with environment variables :param startup_file: main application file :param domain_name: domain name :param entry_point: application entry point :param apps_manager: Application Manager. Class that responsible for gathering and writing information about applications :param passenger_log_file: Passenger log filename to write to app's .htaccess :return: None """ if apps_manager is None: apps_manager = ApplicationsManager() if version is None: raise ClSelectExcept.WrongData("Not passed version as argument") alias = clselectctl.get_alias(alias) user_summary = clpassenger.summary(user) try: app_summary = utils.get_using_realpath_keys(user, directory, user_summary) except KeyError: pass else: raise ClSelectExcept.AppRootBusy(app_summary["directory"]) environment = _create_environment(user, directory, version, use_existing=False) binary = environment.interpreter().binary clpassenger.configure( user, directory, alias, apps_manager.INTERPRETER, binary, doc_root=doc_root, passenger_log_file=passenger_log_file, ) clpassenger.restart(user, directory) # create new startup file with supplied name if entry_point: content = _CONFIGURABLE_WSGI_TEMPLATE % entry_point else: content = clpassenger.WSGI_TEMPLATE if startup_file and startup_file != DEFAULT_STARTUP_FILE: # Custom startup file, use existing mechanism to set new startup file and entrypoint startup_file_name = _get_full_path_to_startup_file(user, directory, startup_file) # to be able set /some/not/existing/dir for startup file dir_name = os.path.dirname(startup_file_name) if not os.path.exists(dir_name): mod_makedirs(dir_name, 0o755) # save new startup file utils.file_write(startup_file_name, content) # Set startup file and entrypoint (if need) if entry_point: setup_wsgi(user, directory, startup_file_name, entry_point) else: # No startup file provided, use default startup_file = DEFAULT_STARTUP_FILE # Set new startup_file and/or entrypoint # build startup_file full path - setup_wsgi function requirement startup_file_name = _get_full_path_to_startup_file(user, directory, startup_file) # save new startup file utils.file_write(startup_file_name, content) if entry_point: # Set new entrypoint setup_wsgi(user, directory, startup_file_name, entry_point) if not env_vars: env_vars = {} if not entry_point: # if entrypoint not supplied, use default in config. # We don't need to set it in the application because it the same with passenger settings entry_point = DEFAULT_APP_ENTRYPOINT # Add application to user's config file if not domain_name: # if domain name not defined - try to determine it summary_data = clpassenger.summary(user) app_summary = utils.get_using_realpath_keys(user, directory, summary_data) domain_name = app_summary["domain"] # TODO: use namedtuple or just class instead of plain dicts app_data = { "python_version": version, "app_version": PythonAppFormatVersion.STABLE, "domain": domain_name, "app_uri": alias, "app_status": DEFAULT_APP_STATE, "startup_file": startup_file, "config_files": [], "env_vars": env_vars, "entry_point": entry_point, } if passenger_log_file: app_data["passenger_log_file"] = passenger_log_file apps_manager.add_app_to_config(user, directory, app_data) try: apps_manager.add_env_vars_for_htaccess(user, directory, env_vars, doc_root) except Exception as err: clprint.print_diag("text", {"status": "ERROR", "message": str(err)}) def _ensure_version_enabled(version, user): """ Check whether particular interpreter version is enabled and raises exception if not :param user: user to include in exception """ from .clselectpython import python_manager if not python_manager.PythonManager().is_version_enabled(version): raise ClSelectExcept.UnableToSetAlternative(user, version, "version is not enabled") def check_response_from_webapp(domain, alias, action=None): """ Check response from user's webapp before and after calling action. Also compare both responses :param domain: domain associated with webapp :param alias: URI associated with webapp :param action: called action, that make something with webapp: install modules, transit it, etc :return: None """ app_is_inaccessible_before = 'Web application is inaccessible by its address "%s". The operation wasn\'t performed.' app_is_inaccessible_after = ( "The operation was performed, but check availability of application has failed. " 'Web application is inaccessible by its address "%s" after the operation.' ) app_is_broken = ( "The operation was performed, but check availability of application has failed. " 'Web application responds, but its return code "%s" or ' 'content type before operation "%s" doesn\'t equal to contet type after operation "%s".' ) requests.packages.urllib3.disable_warnings(urllib_exceptions.InsecureRequestWarning) # pylint: disable=E1101 if not callable(action): raise ClSelectExcept.WrongData("Wrong action for calling in checking webapp") webapp_url = "https://{domain}/{alias}".format( domain=domain, alias=alias, ) try: webapp_url = utils.transform_wildcard_url(webapp_url) request = requests.get(webapp_url, verify=False) except ConnectionError: # site is not available by https protocol webapp_url = webapp_url.replace("https://", "http://") try: request = requests.get(webapp_url, verify=False) except ConnectionError: raise ClSelectExcept.WebAppError(app_is_inaccessible_before % webapp_url) before_mime_type = request.headers.get("Content-Type") before_status = request.headers.get("Status") action() try: request = requests.get(webapp_url, verify=False) except ConnectionError: raise ClSelectExcept.WebAppError(app_is_inaccessible_after % webapp_url) after_mime_type = request.headers.get("Content-Type") after_status = request.headers.get("Status") # assume that app is broken if: # it's response Content-Type or Status code (if first code wasn't 500) changed # if last Status code was 500 (internal error) if ( before_mime_type.lower() != after_mime_type.lower() or (before_status != after_status and before_status is not None and before_status[:3] != "500") or (after_status is not None and after_status[:3] == "500") ): raise ClSelectExcept.WebAppError(app_is_broken % (after_status, before_mime_type, after_mime_type)) def _get_info_about_webapp(app_summary=None, user=None): """ Get info (alias and domain) about user's web application :param app_summary: dict -> summary info about user's web application :param user: str -> name of unix user :return: tuple -> (alias, domain) """ if app_summary is None: raise ClSelectExcept.WrongData("Was passed incorrect summary info about application") if user is None: raise ClSelectExcept.WrongData("Was passed incorrect name of user") alias = app_summary["alias"] user_domains = userdomains(user) found_domains = [ domain for domain, doc_root in user_domains if utils.realpaths_are_equal(user, doc_root, app_summary["docroot"]) ] if len(found_domains) == 0: raise ClSelectExcept.WrongData("Can not found suitable domain for application") app_domain = found_domains[0] return alias, app_domain def destroy(user, app_directory, doc_root, apps_manager=None): """ Destroy web app with specified directory for specified user :param user: username :param app_directory: application directory :param doc_root: Document root for selected domain :param apps_manager: Application Manager. Class that responsible for gathering and writing information about applications :return: None """ if apps_manager is None: apps_manager = ApplicationsManager() # get app state app_config = apps_manager.get_app_config(user, app_directory) # remove env from htaccess try: # if domain is already removed we shouldn't do anything if doc_root is not None: apps_manager.add_env_vars_for_htaccess(user, app_directory, None, doc_root) except Exception as err: clprint.print_diag("text", {"status": "ERROR", "message": str(err)}) if app_config is not None and app_config.get("app_status") == APP_STOPPED_CONST: # stopped app user_home = pwd.getpwnam(user).pw_dir # Remove app's virtualenv, e.g. /home/cltest1/virtualenv/test_app # TODO: Account for possible symlinks in LU-832 virtual_env_path = os.path.join(user_home, get_venv_rel_path(user, app_directory)[1]) try: shutil.rmtree(virtual_env_path) except OSError: pass # if domain is already removed we shouldn't do anything if doc_root is not None: # Remove passenger lines from .htaccess htaccess_filename = apps_manager.get_htaccess_by_appdir(user, app_directory, doc_root) clpassenger.remove_passenger_lines_from_htaccess(htaccess_filename) # remove app from python-selector.json file apps_manager.remove_app_from_config(user, app_directory) # Update application status in passenger clpassenger.restart(user, app_directory) return # Moved before removing an app from node-selector.json because in case if # app_root is absent we still want to remove venv dir of an application. # The method didn't anyhing if app_root is absent and skip other commands # including removal of a venv dir _, prefix = get_venv_rel_path(user, app_directory) abs_dir, _ = utils.get_abs_rel(user, prefix) try: # Remove app's virtualenv, e.g. /home/cltest1/virtualenvnv/test_app shutil.rmtree(abs_dir) except OSError: pass user_summary = clpassenger.summary(user) # remove app from python-selector.json file app_in_config = apps_manager.remove_app_from_config(user, app_directory) try: utils.get_using_realpath_keys(user, app_directory, user_summary) except KeyError: # if app was existed and app's dir is not exists, we skip all further actions if app_in_config: return None else: raise ClSelectExcept.WrongData('No such application (or application not configured) "%s"' % app_directory) # remove app from passenger clpassenger.unconfigure(user, app_directory) # Clear .htaccess try: clpassenger.restart(user, app_directory) except ClSelectExcept.MissingApprootDirectory: pass def fix_homedir(user): user = clselectctl.get_user(user) envs, _ = utils.get_abs_rel(user, environments.DEFAULT_PREFIX) for env in glob.glob(os.path.join(envs, "", "")): for script in glob.glob(os.path.join(env, "bin", "")): # we should not even try to read symlinks if os.path.islink(script): continue old_script = utils.file_read(script, "rb") new_script = [] encoded_env = env.encode() # in fact, this is only for activate scripts if b"VIRTUAL_ENV" in old_script: for line in old_script.splitlines(): if b"VIRTUAL_ENV" in line: # matches both cases with single and double quotes # example: VIRTUAL_ENV="<some_path>" VIRTUAL_ENV='<some_path>' line = re.sub( rb"VIRTUAL_ENV[^\\]([\"\'])(?P<old_path>.+)\1$", lambda m: m.group(0).replace(m.group("old_path"), encoded_env), line, ) new_script.append(line) # change all shebangs to python # TODO: what about shebangs to global python? elif old_script.startswith(b"#!"): new_script = old_script.splitlines() new_script[0] = b"#!%s/bin/python" % encoded_env if new_script: utils.file_writelines(script, (b"%s\n" % line for line in new_script), "wb") def install(user, directory, extension=None, requirements_path=None, skip_web_check=False, apps_manager=None): """ Install defined extension to python web application :param requirements_path: path to the text file with list of python modules for installation :param user: name of owner of web application :param directory: application's root :param extension: name of installed module :param skip_web_check: skip check web application after change it's properties :param apps_manager: Application Manager. Class that responsible for gathering and writing information about applications :return: None """ if apps_manager is None: apps_manager = ApplicationsManager() user = clselectctl.get_user(user) user_config = apps_manager.get_user_config_data(user) try: app_data = utils.get_using_realpath_keys(user, directory, user_config) except KeyError: raise ClSelectExcept.WrongData("Record about application {} is absent".format(directory)) if app_data["app_status"] != APP_STARTED_CONST: skip_web_check = True environment = _get_environment(user, directory, apps_manager=apps_manager) def action(): if extension is not None: environment.extension_install(extension) elif requirements_path is not None: environment.extension_install_requirements(os.path.join(directory, requirements_path)) if not skip_web_check: try: check_response_from_webapp( domain=app_data["domain"], alias=app_data["app_uri"], action=action, ) except ClSelectExcept.WebAppError as err: raise ClSelectExcept.WebAppError("An error occured during installation of modules. %s" % err) else: action() def list_extensions(user, directory): user = clselectctl.get_user(user) environment = _get_environment(user, directory) return environment.extensions() def restart(user, app_directory, doc_root=None, apps_manager=None): """ Restarts web app with specified directory for specified user :param user: username :param app_directory: application directory :param doc_root: Document root for selected domain :param apps_manager: Application Manager. Class that responsible for gathering and writing information about applications :return: None """ if apps_manager is None: apps_manager = ApplicationsManager() app_config = apps_manager.get_app_config(user, app_directory) if app_config is not None and app_config.get("app_status") == APP_STOPPED_CONST: # If application was stopped - start it # doc_root is None during 'update-interpreter' command if doc_root is not None: start(user, app_directory, doc_root) else: apps_summary = clpassenger.summary(user) if app_directory not in apps_summary: raise ClSelectExcept.WrongData('No such application (or application not configured) "%s"' % app_directory) clpassenger.restart(user, app_directory) def stop(user, app_directory, doc_root, apps_manager=None): """ Stops web app with specified directory for specified user :param user: username :param app_directory: application directory :param doc_root: Document root for selected domain :param apps_manager: Application Manager. Class that responsible for gathering and writing information about applications :return: None """ if apps_manager is None: apps_manager = ApplicationsManager() app_config = apps_manager.get_app_config(user, app_directory) if app_config is None: raise ClSelectExcept.WrongData('No such application (or application not configured) "%s"' % app_directory) # Do nothing if application already stopped if app_config.get("app_status") == APP_STOPPED_CONST: return htaccess_filename = apps_manager.get_htaccess_by_appdir(user, app_directory, doc_root, app_config) # Remove passenger lines from .htaccess clpassenger.remove_passenger_lines_from_htaccess(htaccess_filename) # Save new application status in user's config apps_manager.set_app_status(user, app_directory, APP_STOPPED_CONST) # Update application status in passenger clpassenger.restart(user, app_directory) def start(user, app_directory, doc_root, apps_manager=None): """ Starts web app with specified directory for specified user :param user: username :param app_directory: application directory :param doc_root: Document root for selected domain :param apps_manager: Application Manager. Class that responsible for gathering and writing information about applications :return: None """ if apps_manager is None: apps_manager = ApplicationsManager() app_config = apps_manager.get_app_config(user, app_directory) if app_config is None: raise ClSelectExcept.WrongData('No such application (or application not configured) "%s"' % app_directory) # Do nothing if application already started if app_config.get("app_status") == APP_STARTED_CONST: return # Create .htaccess file for the application apps_manager.update_htaccess_file(user, app_directory, doc_root) # Save new application status in user's config apps_manager.set_app_status(user, app_directory, APP_STARTED_CONST) # Update application status in passenger clpassenger.restart(user, app_directory) # Update wsgi file. It can be changed while app was stopped startup_file = app_config.get("startup_file") if startup_file != DEFAULT_STARTUP_FILE: startup_file_full_path = _get_full_path_to_startup_file( user, app_directory, app_config.get("startup_file"), ) setup_wsgi(user, app_directory, startup_file_full_path, app_config.get("entry_point")) def setup_wsgi(user, directory, wsgi_path, callable_object=None): user = clselectctl.get_user(user) abs_dir, _ = utils.get_abs_rel(user, directory) apps_summary = clpassenger.summary(user) if directory not in apps_summary: raise ClSelectExcept.WrongData('No such application (or application not configured) "%s"' % directory) environment = _get_environment(user, directory) version = environment.interpreter().version if Version(version) >= Version("3.12"): template = WSGI_TEMPLATE_PY12_PLUS else: template = WSGI_TEMPLATE if os.path.isabs(wsgi_path): if not os.path.isfile(wsgi_path): raise ClSelectExcept.WrongData('No such WSGI script "%s"' % wsgi_path) wsgi_path = utils.s_partition(wsgi_path, abs_dir)[2].strip(os.path.sep) if not wsgi_path: raise ClSelectExcept.WrongData("WSGI script is not in webapp directory") else: raise ClSelectExcept.WrongData("Path to WSGI script must be absolute") app_wsgi = os.path.join(abs_dir, clpassenger.WSGI_PATH) # Set new callable object if need if not callable_object: callable_object = DEFAULT_APP_ENTRYPOINT utils.file_write(app_wsgi, template % (wsgi_path, callable_object), "w") # Restart application clpassenger.restart(user, directory) def has_virtenv_dir(user): """ return True if user home dir has virtualenv dir """ homedir = pwd.getpwnam(user).pw_dir path = os.path.join(homedir, "virtualenv") return os.path.isdir(path) def summary(user): user = clselectctl.get_user(user) summ = {} if not has_virtenv_dir(user): return summ for directory, data in iteritems(clpassenger.summary(user)): if data["interpreter"] != ApplicationsManager.INTERPRETER: continue try: environment = _get_environment(user, directory, data).as_deepdict() app_wsgi = os.path.join(data["directory"], clpassenger.WSGI_PATH) wsgi_conf = utils.file_read(app_wsgi) version = environment.interpreter().version if Version(version) >= Version("3.12"): pattern = WSGI_PATTERN_PY12_PLUS else: pattern = WSGI_PATTERN match = pattern.search(wsgi_conf) if match: groupdict = match.groupdict() wsgi_path = groupdict["script"] callable_object = groupdict["callable"] else: wsgi_path = callable_object = "" env_info = { "alias": data["alias"], "domain": data["domain"], "environment": environment["name"], "interpreter": environment["interpreter"], "extensions": environment["extensions"], "wsgi": {"script": wsgi_path, "callable": callable_object}, } # add only list with additions domains if "domains" in data and len(data["domains"]) > 1: env_info["domains"] = data["domains"] except Exception as e: env_info = {"error": str(e)} summ[directory] = env_info return summ def _get_full_path_to_startup_file(user, directory, startup_file): # type: (str, str, str) -> str """ :param user: name of system user :param directory: name of dir with python applicatoin :param startup_file: name of startup file of python application Get full path to startup file of python application """ abs_dir, rel_dir = utils.get_abs_rel(user, directory) return os.path.join(abs_dir, startup_file) def uninstall(user, directory, extension): user = clselectctl.get_user(user) environment = _get_environment(user, directory) environment.extension_uninstall(extension) def update(user, directory, extension): user = clselectctl.get_user(user) environment = _get_environment(user, directory) environment.extension_update(extension) # TODO: We should remove that method completely, because this script already doesn't caused directly def main(): try: opts, args = getopt.getopt( sys.argv[1:], "hi:v:u:lGsynfFZtcbe:d:r:go:pjK:", [ "help", "interpreter=", "version=", "user=", "domain=", "list", "list-extensions", "user-summary", "create-webapp", "update-interpreter", "destroy-webapp", "relocate-webapp", "transit-webapp", "restart-webapp", "start-webapp", "stop-webapp", "user-current", "set-user-current", "enable-user-extensions=", "disable-user-extensions=", "replace-user-extensions=", "recreate-virtualenv", "freeze-requirements", "list-user-extensions", "setup-wsgi=", "print-summary", "json", "verbose", "list-extensions-version=", ], ) except getopt.GetoptError as err: sys.stderr.write(str(err)) usage() sys.exit(1) ext_list = "" fmt = "text" verbose = False print_summary = False user = None domain = None action = "" version = None if not opts: usage() sys.exit(1) allowed_opts = ("--update-interpreter",) if not any(item[0] in allowed_opts for item in opts): clprint.print_diag( fmt, { "status": "ERROR", "message": "Unsupported option was passed. Currently, selectorctl utility " "supports only update-interpreter option. " "Please, use cloudlinux-selector utility for python " "interpreter", }, ) sys.exit(1) for o, a in opts: if o in ("-i", "--interpreter"): pass elif o in ("-l", "--list"): action = "list" elif o in ("-y", "--create-webapp"): action = "create-webapp" elif o == "--update-interpreter": action = "update-interpreter" elif o in ("-n", "--destroy-webapp"): action = "destroy-webapp" elif o in ("-f", "--relocate-webapp"): action = "relocate-webapp" elif o in ("-F", "--transit-webapp"): action = "transit-webapp" elif o in ("-Z", "--restart-webapp"): action = "restart-webapp" elif o in ("-S", "--start-webapp"): action = "start-webapp" elif o in ("-t", "--stop-webapp"): action = "stop-webapp" elif o in ("-c", "--user-current"): action = "user-current" elif o in ("-b", "--set-user-current"): action = "set-user-current" elif o in ("-g", "--list-user-extensions"): action = "list-user-extensions" elif o in ("-e", "--enable-user-extensions"): action = "enable-user-extensions" ext_list = a elif o in ("--recreate-virtualenv",): action = "recreate-virtualenv" ext_list = a elif o in ("--freeze-requirements",): action = "freeze-requirements" ext_list = a elif o in ("-s", "--user-summary"): action = "user-summary" elif o in ("-j", "--json"): fmt = "json" elif o == "--verbose": verbose = True elif o in ("-r", "--replace-user-extensions"): action = "replace-user-extensions" ext_list = a elif o in ("-d", "--disable-user-extensions"): action = "disable-user-extensions" ext_list = a elif o in ("-v", "--version"): version = a elif o in ("-G", "--list-extensions"): action = "list-extensions" elif o in ("-u", "--user"): user = a.strip("'\"") # for deleting \' and \" elif o == "--domain": domain = a elif o in ("-o", "--setup-wsgi"): action = "setup-wsgi" wsgi_path, _, callable_object = utils.s_partition(a, ":") elif o in ("-p", "--print-summary"): print_summary = True elif o in ("-K", "--list-extensions-version"): action = "list-extensions-version" ext_list = a else: sys.stderr.write("unhandled option") sys.exit(1) if action == "": sys.stderr.write("ERROR:you must provide option for interpreter python") sys.exit(1) if action in ( "create-webapp", "destroy-webapp", "relocate-webapp", "transit-webapp", "restart-webapp", "start-webapp", "stop-webapp", "enable-user-extensions", "list-user-extensions", "replace-user-extensions", "disable-user-extensions", "setup-wsgi", "user-current", "set-user-current", ): if not args: sys.stderr.write("webapp must be specified") sys.exit(3) if ext_list == "-": if action == "enable-user-extensions": sys.stderr.write("installation of all extensions is not possible") sys.exit(4) elif ext_list: _exts = [_f for _f in ext_list.split(",") if _f] if domain: try: doc_root, user_ = docroot(domain) except NoDomain: clprint.print_diag(fmt, {"status": "ERROR", "message": 'No such domain: "%s"' % domain}) sys.exit(1) except NotSupported: clprint.print_diag(fmt, {"status": "ERROR", "message": "Python selector not supported for %s" % CP_NAME}) sys.exit(1) if not user: # we can obtain user name for domain user = user_ elif user != user_: clprint.print_diag( fmt, {"status": "ERROR", "message": 'domain "%s" is not owned by the user "%s"' % (domain, user)} ) sys.exit(1) else: # commands that need user specified if action in ( "create-webapp", "destroy-webapp", "relocate-webapp", "transit-webapp", "restart-webapp", "start-webapp", "stop-webapp", "enable-user-extensions", "list-user-extensions", "replace-user-extensions", "disable-user-extensions", "setup-wsgi", "user-summary", "update-interpreter", "user-current", "set-user-current", ): try: user = clselectctl.get_user(user) except ValueError as e: clprint.print_diag(fmt, {"status": "ERROR", "message": str(e)}) sys.exit(1) if user and not action == "update-interpreter": try: domain_list = userdomains(user) except NotSupported: clprint.print_diag( fmt, {"status": "ERROR", "message": "Python selector not supported for %s" % CP_NAME} ) sys.exit(1) _, doc_root = domain_list[0] # get document root for main domain try: error = 0 ok = 0 result = {} if action == "list-extensions": raise DeprecationWarning("%s is deprecated in old python selector" % action) elif action == "list-extensions-version": result = extensions.ExtensionInfo().list_extensions_version(_exts) ext_list = "" elif action == "list": result = interpreters.interpreters_dict("version") elif action == "user-summary": result = summary(user) elif action == "create-webapp": create(user, args[0], args[1], version, doc_root=doc_root) elif action == "update-interpreter": app_name = "" # all applications as default if len(args) == 1: app_name = args[0].strip("'\"") update_interpreter(user=user, app_name=app_name, version=(version,), verbose=verbose) elif action == "destroy-webapp": destroy(user, args[0], doc_root) elif action == "relocate-webapp": raise DeprecationWarning("%s is deprecated in old python selector" % action) elif action == "transit-webapp": new_doc_root = None if domain: # remove app to new domain if domain present new_doc_root = doc_root raise DeprecationWarning("%s is deprecated in old python selector" % action) elif action == "restart-webapp": # username, app-dir restart(user, args[0], doc_root) elif action == "start-webapp": # username, app-dir, doc_root start(user, args[0], doc_root) elif action == "stop-webapp": # username, app-dir, doc_root stop(user, args[0], doc_root) elif action == "user-current": raise DeprecationWarning("%s is deprecated in old python selector" % action) elif action == "set-user-current": raise DeprecationWarning("%s is deprecated in old python selector" % action) elif action == "list-user-extensions": result = list_extensions(user, args[0]) elif action == "setup-wsgi": # --setup-wsgi=<file_path:callable> # args[0] -- app_root # wsgi_path -- file_path # callable_object -- callable setup_wsgi(user, args[0], wsgi_path, callable_object) elif action == "recreate-virtualenv": for env in environments.environments(): env.recreate() elif action == "freeze-requirements": for env in environments.environments(): env.pip_freeze() else: alias = args[0] if ext_list == "-": _exts = list_extensions(user, alias) for extension in _exts: try: if action == "enable-user-extensions": install(user, alias, extension) elif action == "disable-user-extensions": uninstall(user, alias, extension) elif action == "replace-user-extensions": update(user, alias, extension) result.update({extension: {"status": "OK"}}) ok += 1 except (ValueError, ClSelectExcept.ExternalProgramFailed) as err: result.update({extension: {"status": "ERROR", "message": str(err)}}) error += 1 except (BaseClSelectException, extensions.PyPIRpcApiError) as err: clprint.print_diag(fmt, {"status": "ERROR", "message": str(err)}) sys.exit(1) except NotSupported: clprint.print_diag(fmt, {"status": "ERROR", "message": "Python selector not supported for %s" % CP_NAME}) sys.exit(1) except Exception as err: msg = traceback.format_exc() clprint.print_diag(fmt, {"status": "ERROR", "message": msg}) sys.exit(1) if not result and print_summary: result = summary(user) if error and ok: status = "PARTIAL" exit_status = 2 elif error: status = "ERROR" exit_status = 5 else: if ext_list and ok < 2: if print_summary: clprint.print_data(fmt, summary(user)) else: clprint.print_data(fmt, {}) else: clprint.print_data(fmt, result) sys.exit(0) message = "\n".join("%s: %s" % (k, v.get("message", v.get("status", ""))) for k, v in iteritems(result)) clprint.print_diag(fmt, {"status": status, "message": message}) sys.exit(exit_status) def _glob_users(user=""): """ Return filtered control panels users :param user: :return: """ return [user_name for user_name in cpusers() if fnmatch.fnmatch(user_name, user)] def update_interpreter(user="", version=("2.7", "3.3", "3.4"), app_name="*", verbose=True): if detect.is_plesk(): sys.stderr.write("WARNING: Update interpreter with Python selector not supported on %s, skipped\n" % CP_NAME) return alt_interpreters_dict = interpreters.interpreters(key="version") # obtain data about installed alt-python is_da_panel = detect.detect_DA_admin() apps_manager = ApplicationsManager() for user_name in _glob_users(user): if is_da_panel and utils.user_should_be_skipped(user_name): continue if not has_virtenv_dir(user_name): continue # change uid/gid to user_name user_pwd = pwd.getpwnam(user_name) real_uid = os.getuid() real_gid = os.getgid() os.setregid(user_pwd.pw_gid, real_uid) os.setreuid(user_pwd.pw_uid, real_gid) # turns off quota secureio.set_capability() # get users apps from selector json config for app_name_, data in iteritems(apps_manager.get_user_config_data(user_name)): if not fnmatch.fnmatch(app_name_, app_name): # filtering application name continue try: environment = _get_environment(user_name, app_name_, apps_manager=apps_manager) env_python_ver = environment.interpreter().version if env_python_ver not in version: # filtering virtual environment python version continue if verbose: print('Updating user: "%s"; application: "%s"...' % (user_name, app_name_)) update_result = environment.update_python_interpreter( backup=True, verbose=verbose, force=False, _alt_interpreters_dict=alt_interpreters_dict ) if update_result: # restart application only if updating success try: # do not worry, stopped apps will not be started # see restart method for details restart(user_name, app_name_) except (OSError, IOError) as e: if verbose: print('Failed to restart application "%s"; %s' % (app_name_, str(e))) except ClSelectExcept.InterpreterError as e: if verbose: print('Failed updating user: "%s"; application: "%s"; %s' % (user_name, app_name_, str(e))) # restore original uid/gid os.setuid(real_uid) os.setgid(real_gid)	44.75 KB	2026-02-26 15:33
📄 clselectctlruby.py # -- 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 print_function from __future__ import absolute_import from __future__ import division import getopt import glob import os import shutil import sys import traceback from future.utils import iteritems from . import clpassenger from . import clselectctl from . import utils from .clselectprint import clprint from .clselectexcept import ClSelectExcept, BaseClSelectException from .clselectruby import environments, extensions, interpreters from clcommon.cpapi import userdomains, docroot, CP_NAME from clcommon.cpapi.cpapiexceptions import NoDomain, NotSupported INTERPRETER = 'ruby' def usage(): print(' -v \| --version VERSION : Specify alternative version') print(' -u \| --user USERNAME : Username') print(' --domain DOMAIN : Domain (or subdomain); users main domain as default') print(' -l \| --list : List alternatives for interpreter') print(' -G \| --list-extensions : List global set of packages') print(' -K \| --list-extensions-version : List version set of packages') print(' -s \| --user-summary : List user summary of webapps') print(' -y \| --create-webapp : Create user webapp') print(' -n \| --destroy-webapp : Destroy user webapp') print(' -f \| --relocate-webapp : Change webapp directory (files must be moved manually)') print(' -F \| --transit-webapp : Change webapp domain/alias') print(' -Z \| --restart-webapp : Restart webapp') print(' -c \| --user-current : Show currently selected alternative for user webapp') print(' -b \| --set-user-current : Set alternative as user webapp default') print(' -e \| --enable-user-extensions GEMS : Install comma-separated list of gems for user webapp') print(' -d \| --disable-user-extensions GEMS : Uninstall comma-separated list of gems for user webapp') print(" : Use '-' (minus) for all gems") print(' -r \| --replace-user-extensions GEMS : Update comma-separated list of gems for user webapp') print(" : Use '-' (minus) for all gems") print(' -g \| --list-user-extensions : List installed gems for user webapp') print(' -p \| --print-summary : If specified along with setting an alternative prints user summary') print(' -j \| --json : Print data as JSON') def _create_environment(user, directory, version, env_name=None): prefix = _get_prefix(user, directory) if not env_name: env_name = version environment = environments.Environment(env_name, user, prefix) if not environment.exists(): try: interpreter = interpreters.interpreters(key='version')[version] except KeyError: raise ClSelectExcept.NoSuchAlternativeVersion(version) environment.create(interpreter) return environment def _get_environment(user, directory, app_summary=None): prefix = _get_prefix(user, directory) if app_summary is None: user_summary = clpassenger.summary(user) app_summary = user_summary.get(directory) if not app_summary: raise ClSelectExcept.NoSuchApplication('No such application (or application not configured) "%s"' % directory) binary = app_summary['binary'] env_name = os.path.basename(os.path.dirname(os.path.dirname(binary))) environment = environments.Environment(env_name, user, prefix) return environment def _get_prefix(user, directory): _, rel_dir = utils.get_abs_rel(user, directory) return os.path.join(environments.DEFAULT_PREFIX, clselectctl.get_prefix(rel_dir)) def create(user, directory, alias, version=None, env_name=None, doc_root=None): if version is None: raise ClSelectExcept.WrongData('Not passed version as argument') user = clselectctl.get_user(user) clselectctl.check_directory(directory) alias = clselectctl.get_alias(alias) environment = _create_environment(user, directory, version, env_name) binary = environment.interpreter().binary clpassenger.configure(user, directory, alias, INTERPRETER, binary, doc_root=doc_root) clpassenger.restart(user, directory) def current(user, directory, version=None, env_name=None): user = clselectctl.get_user(user) old_environment = _get_environment(user, directory) if not version: return {old_environment.name: old_environment.interpreter().as_dict()} old_extensions = set(old_environment.extensions()) new_environment = _create_environment(user, directory, version, env_name) new_extensions = set(new_environment.extensions()) for extension in new_extensions - old_extensions: try: new_environment.extension_uninstall(extension) except ClSelectExcept.ExternalProgramFailed: pass for extension in old_extensions - new_extensions: try: new_environment.extension_install(extension) except ClSelectExcept.ExternalProgramFailed: pass app_summary = clpassenger.summary(user)[directory] alias = app_summary['alias'] doc_root = app_summary['docroot'] htaccess_path = app_summary['htaccess'] binary = new_environment.interpreter().binary clpassenger._unconfigure(htaccess=htaccess_path) clpassenger.configure(user, directory, alias, INTERPRETER, binary, doc_root=doc_root) clpassenger.restart(user, directory) def destroy(user, directory): user = clselectctl.get_user(user) prefix = _get_environment(user, directory).prefix abs_dir, _ = utils.get_abs_rel(user, prefix) try: shutil.rmtree(abs_dir) except OSError: pass clpassenger.unconfigure(user, directory) try: clpassenger.restart(user, directory) except ClSelectExcept.MissingApprootDirectory: pass def install(user, directory, extension): user = clselectctl.get_user(user) environment = _get_environment(user, directory) environment.extension_install(extension) def list_extensions(user, directory): user = clselectctl.get_user(user) environment = _get_environment(user, directory) return environment.extensions() def relocate(user, old_directory, new_directory, fmt): if '/' in new_directory: raise ClSelectExcept.WrongData('You cannot move appication to subdir') user = clselectctl.get_user(user) clselectctl.check_directory(new_directory) old_abs, old_rel = utils.get_abs_rel(user, old_directory) new_abs, new_rel = utils.get_abs_rel(user, new_directory) old_user_summary = clpassenger.summary(user) if new_directory in old_user_summary: raise ClSelectExcept.WebAppError("Specified directory already used by '%s'" % new_abs) if old_directory not in old_user_summary: raise ClSelectExcept.WrongData("No such application (or application not configured) \"%s\"" % old_directory) htaccess_path = old_user_summary[old_directory]['htaccess'] doc_root = old_user_summary[old_directory]['docroot'] alias = old_user_summary[old_directory]['alias'] env_name = _get_environment(user, old_directory).name old_prefix = clselectctl.get_prefix(old_rel) new_prefix = clselectctl.get_prefix(new_rel) _old_envs, _ = utils.get_abs_rel(user, os.path.join( environments.DEFAULT_PREFIX, old_prefix)) old_envs = os.path.join(_old_envs, '') _new_envs, _ = utils.get_abs_rel(user, os.path.join( environments.DEFAULT_PREFIX, new_prefix)) new_envs = os.path.join(_new_envs, '') old_prompt = '(' + old_rel + ':' new_prompt = '(' + new_rel + ':' shutil.move(old_envs, new_envs) for activate in glob.glob(os.path.join(new_envs, '', 'bin', '')): old_activate = utils.file_read(activate) if old_prompt in old_activate: new_activate = old_activate.replace(old_prompt, new_prompt) utils.file_write(activate, new_activate, 'w') if not os.path.exists(new_abs): os.rename(old_abs, new_abs) prefix = _get_prefix(user, new_directory) environment = environments.Environment(env_name, user, prefix) binary = environment.interpreter().binary clpassenger._unconfigure(htaccess=htaccess_path) clpassenger.configure(user, new_directory, alias, INTERPRETER, binary, doc_root=doc_root) clpassenger.restart(user, new_directory) def restart(user, directory): user = clselectctl.get_user(user) apps_summary = clpassenger.summary(user) if directory not in apps_summary: raise ClSelectExcept.WrongData("No such application (or application not configured) \"%s\"" % directory) clpassenger.restart(user, directory) def summary(user): user = clselectctl.get_user(user) summ = {} for directory, data in iteritems(clpassenger.summary(user)): if data['interpreter'] != INTERPRETER: continue environment = _get_environment(user, directory, data).as_deepdict() summ[directory] = { 'domain': data['domain'], 'alias': data['alias'], 'environment': environment['name'], 'interpreter': environment['interpreter'], 'extensions': environment['extensions'], } # add only list with additions domains if "domains" in data and len(data["domains"]) > 1: summ[directory]["domains"] = data["domains"] return summ def transit(user, directory, alias, doc_root=None): user = clselectctl.get_user(user) apps_summary = clpassenger.summary(user) if directory not in apps_summary: raise ClSelectExcept.WrongData("No such application (or application not configured) \"%s\"" % directory) old_app_summary = apps_summary[directory] old_alias = old_app_summary['alias'] old_doc_root = old_app_summary['docroot'] new_alias = clselectctl.get_alias(alias) environment = _get_environment(user, directory) binary = environment.interpreter().binary clpassenger.configure(user, directory, new_alias, INTERPRETER, binary, True, 'transit', doc_root=doc_root) clpassenger.move(user, directory, old_alias, new_alias, old_doc_root=old_doc_root, new_doc_root=doc_root) clpassenger.restart(user, directory) def uninstall(user, directory, extension): user = clselectctl.get_user(user) environment = _get_environment(user, directory) environment.extension_uninstall(extension) def update(user, directory, extension): user = clselectctl.get_user(user) environment = _get_environment(user, directory) environment.extension_update(extension) def main(): try: opts, args = getopt.getopt( sys.argv[1:], 'hi:v:u:lGsynfFZcbe:d:r:gpjK:', [ 'help', 'interpreter=', 'version=', 'user=', 'domain=', 'list', 'list-extensions', 'user-summary', 'create-webapp', 'destroy-webapp', 'relocate-webapp', 'transit-webapp', 'restart-webapp', 'user-current', 'set-user-current', 'enable-user-extensions=', 'disable-user-extensions=', 'replace-user-extensions=', 'list-user-extensions', 'print-summary', 'json', 'list-extensions-version=' ]) except getopt.GetoptError as err: sys.stderr.write(str(err)) usage() sys.exit(1) ext_list = '' fmt = 'text' print_summary = False user = None domain = None action = '' version = None if not opts: usage() sys.exit(1) for o, a in opts: if o in ('-i', '--interpreter'): pass elif o in ('-l', '--list'): action = 'list' elif o in ('-y', '--create-webapp'): action = 'create-webapp' elif o in ('-n', '--destroy-webapp'): action = 'destroy-webapp' elif o in ('-f', '--relocate-webapp'): action = 'relocate-webapp' elif o in ('-F', '--transit-webapp'): action = 'transit-webapp' elif o in ('-Z', '--restart-webapp'): action = 'restart-webapp' elif o in ('-c', '--user-current'): action = 'user-current' elif o in ('-b', '--set-user-current'): action = 'set-user-current' elif o in ('-g', '--list-user-extensions'): action = 'list-user-extensions' elif o in ('-e', '--enable-user-extensions'): action = 'enable-user-extensions' ext_list = a elif o in ('-s', '--user-summary'): action = 'user-summary' elif o in ('-j', '--json'): fmt = 'json' elif o in ('-r', '--replace-user-extensions'): action = 'replace-user-extensions' ext_list = a elif o in ('-d', '--disable-user-extensions'): action = 'disable-user-extensions' ext_list = a elif o in ('-v', '--version'): version = a elif o in ('-G', '--list-extensions'): action = 'list-extensions' elif o in ('-u', '--user'): user = a elif o == '--domain': domain = a elif o in ('-p', '--print-summary'): print_summary = True elif o in ('-K', '--list-extensions-version'): action = 'list-extensions-version' ext_list = a else: sys.stderr.write('unhandled option') sys.exit(1) if action == '': sys.stderr.write('ERROR:you must provide option for interpreter ruby') sys.exit(1) if action in ('create-webapp', 'destroy-webapp', 'relocate-webapp', 'transit-webapp', 'restart-webapp', 'enable-user-extensions', 'list-user-extensions', 'replace-user-extensions', 'disable-user-extensions', ): if not args: sys.stderr.write('webapp must be specified') sys.exit(3) if domain: try: doc_root, user_ = docroot(domain) except NoDomain: clprint.print_diag(fmt, {'status': 'ERROR', 'message': 'No such domain: "%s"' % domain}) sys.exit(1) except NotSupported: clprint.print_diag(fmt, {'status': 'ERROR', 'message': 'Ruby selector not supported for %s' % CP_NAME}) sys.exit(1) if not user: # we can obtain user name for domain user = user_ elif user != user_: clprint.print_diag(fmt, {'status': 'ERROR', 'message': 'domain %s is not owned by the user %s' % (domain, user)}) sys.exit(1) elif user and not domain: try: domain_list = userdomains(user) except NotSupported: clprint.print_diag(fmt, {'status': 'ERROR', 'message': 'Ruby selector not supported for %s' % CP_NAME}) sys.exit(1) _, doc_root = domain_list[0] # get document root for main domain if ext_list == '-': if action == 'enable-user-extensions': sys.stderr.write('installlation of all extensions is not possible') sys.exit(4) elif ext_list: _exts = [_f for _f in ext_list.split(',') if _f] try: error = 0 ok = 0 result = {} if action == 'list-extensions': result = extensions.ExtensionInfo().list_extensions_cached() elif action == 'list-extensions-version': result = extensions.ExtensionInfo().list_extensions_version(_exts) ext_list = '' elif action == 'list': result = interpreters.interpreters_dict('version') elif action == 'user-summary': result = summary(user) elif action == 'create-webapp': create(user, args[0], args[1], version, doc_root=doc_root) elif action == 'destroy-webapp': destroy(user, args[0]) elif action == 'relocate-webapp': relocate(user, args[0], args[1], fmt) elif action == 'transit-webapp': new_doc_root = None if domain: # remove app to new domain if domain present new_doc_root = doc_root transit(user, args[0], args[1], doc_root=new_doc_root) elif action == 'restart-webapp': restart(user, args[0]) elif action == 'user-current': result = current(user, args[0]) elif action == 'set-user-current': current(user, args[0], version) elif action == 'list-user-extensions': result = list_extensions(user, args[0]) else: alias = args[0] if ext_list == '-': _exts = list_extensions(user, alias) for extension in _exts: try: if action == 'enable-user-extensions': install(user, alias, extension) elif action == 'disable-user-extensions': uninstall(user, alias, extension) elif action == 'replace-user-extensions': update(user, alias, extension) result.update({extension: {'status': 'OK'}}) ok += 1 except (ValueError, ClSelectExcept.ExternalProgramFailed) as err: result.update( {extension: {'status': 'ERROR', 'message': str(err)}}) error += 1 except BaseClSelectException as err: clprint.print_diag(fmt, {'status': 'ERROR', 'message': str(err)}) sys.exit(1) except Exception as err: msg = traceback.format_exc() clprint.print_diag(fmt, {'status': 'ERROR', 'message': msg}) sys.exit(1) if not result and print_summary: result = summary(user) if error and ok: status = 'PARTIAL' exit_status = 2 elif error: status = 'ERROR' exit_status = 5 else: if ext_list and ok < 2: if print_summary: clprint.print_data(fmt, summary(user)) else: clprint.print_data(fmt, {}) else: clprint.print_data(fmt, result) sys.exit(0) message = '\n'.join( '%s: %s' % (k, v.get('message', v.get('status', ''))) for k, v in iteritems(result)) clprint.print_diag(fmt, {'status': status, 'message': message}) sys.exit(exit_status)	18.59 KB	2026-02-26 15:33
📄 clselectexcept.py # -- 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 print_function from __future__ import division from __future__ import absolute_import import traceback from clcommon import FormattedException from clcommon.cpapi import ( getCPName, PLESK_NAME, CPANEL_NAME, DIRECTADMIN_NAME ) class BaseClSelectException(FormattedException): """Base class for all cl-select exceptions""" def __init__(self, message, details=None, context=None): super(BaseClSelectException, self).__init__( dict(message=message, details=details, context=context or {})) class ClSelectExcept(object): """ Custom exceptions class """ class ConfigNotFound(BaseClSelectException): def __init__(self, filename, message=None): if message is None: message = "Config file (%s) not found" % (filename,) super(ClSelectExcept.ConfigNotFound, self).__init__(message) class UnableToLoadData(BaseClSelectException): def __init__(self, filename, reason): message = "Could not load data from %s: %s" % (filename, reason) super(ClSelectExcept.UnableToLoadData, self).__init__(message) class BusyApplicationRoot(BaseClSelectException): def __init__(self, application_root): message = 'The application cannot be located inside of already ' \ 'existing one: {}. Please, remove the already existing ' \ 'application or specify the other \'application root\' path.'.format(application_root) super(ClSelectExcept.BusyApplicationRoot, self).__init__(message) class WrongData(BaseClSelectException): def __init__(self, message=None, context=None): if not message: message = "Error in data supplied" super(ClSelectExcept.WrongData, self).__init__(message, context=context) class WrongConfigFormat(BaseClSelectException): def __init__(self, filename): message = "Config file (%s) has errors" % filename super(ClSelectExcept.WrongConfigFormat, self).__init__(message) class NoSuchAlternativeVersion(BaseClSelectException): def __init__(self, version): message = 'No such alternative version (%s)' % version super(ClSelectExcept.NoSuchAlternativeVersion, self).__init__(message) class EnvironmentAlreadyExists(BaseClSelectException): def __init__(self, path): message = 'Virtual environment already exists (%s). ' \ 'Probably it is used by another application. ' \ 'Please, remove directory if it is no longer needed.' % path super(ClSelectExcept.EnvironmentAlreadyExists, self).__init__(message) class AppRootBusy(BaseClSelectException): def __init__(self, path): message = 'Specified directory already used by \'%s\'' % path super(ClSelectExcept.AppRootBusy, self).__init__(message) class UnableToProcessOption(BaseClSelectException): def __init__(self, data): message = 'Cannot process option (%s)' % data super(ClSelectExcept.UnableToProcessOption, self).__init__(message) class UnableToSaveData(BaseClSelectException): def __init__(self, filename, message): message = "Unable to save data to %s (%s)" % (filename, message) super(ClSelectExcept.UnableToSaveData, self).__init__(message) class ExternalProgramFailed(BaseClSelectException): def __init__(self, message, kwargs): super(ClSelectExcept.ExternalProgramFailed, self).__init__(message, kwargs) class NoSuchUser(BaseClSelectException): def __init__(self, user): message = "No such user (%s)" % user super(ClSelectExcept.NoSuchUser, self).__init__(message) class NoSuchExtension(BaseClSelectException): def __init__(self, ext_list, all_extensions): extensions = ', '.join(ext_list) extensions_aval = ', '.join(all_extensions) message = "No such extension(s): %s; available extensions is: %s" % (extensions, extensions_aval) super(ClSelectExcept.NoSuchExtension, self).__init__(message) class NoUserSelector(BaseClSelectException): def __init__(self, user): message = "No selector dir for user %s" % user super(ClSelectExcept.NoUserSelector, self).__init__(message) class UnableToSetAlternative(BaseClSelectException): def __init__(self, user, version, message): message = "Unable to set version %s for user %s: (%s)" % (version, user, message) super(ClSelectExcept.UnableToSetAlternative, self).__init__(message) class UnableToChangeToAnotherUser(BaseClSelectException): def __init__(self, user, message): message = "Unable to change to user %s (%s)" % (user, message) super(ClSelectExcept.UnableToChangeToAnotherUser, self).__init__(message) class NotCageFSUser(BaseClSelectException): def __init__(self, user): message = "User %s not in CageFS" % user super(ClSelectExcept.NotCageFSUser, self).__init__(message) class UnableToGetExtensions(BaseClSelectException): def __init__(self, version): message = 'Cannot get extensions list for version "%(version)s". ' \ 'Please, make sure that native php works fine and run ' \ '"%(setup_command)s" to configure PHP selector.' super(ClSelectExcept.UnableToGetExtensions, self).__init__(message, context={ 'version': version, 'setup_command': "cloudlinux-selector setup --interpreter=php --json" }) class NativeNotInstalled(BaseClSelectException): def __init__(self, native_php_path): cp_name = getCPName() message = 'Native php version is not installed ("%(native_php_path)s" binary is missing), ' \ 'thus CloudLinux PHP selector is not able to work. ' \ 'Please, install native php and run "%(setup_command)s" to configure PHP selector.' context = dict( control_panel=cp_name, native_php_path=native_php_path, setup_command="cloudlinux-selector setup --interpreter=php --json" ) if getCPName() == PLESK_NAME: context['http_docs'] = 'https://support.plesk.com/hc/en-us/articles/115004177974-' \ 'How-to-install-PHP-by-OS-vendor-in-Plesk-for-Linux' message = f'{message} PHP installation instructions for %(control_panel)s: %(http_docs)s' elif getCPName() == CPANEL_NAME: context['http_docs'] = 'https://documentation.cpanel.net/display/EA4/EasyApache+4+and+' \ 'the+ea-php-cli+Package' message = f'{message} For %(control_panel)s you should install `ea-php-cli` package: %(http_docs)s' elif getCPName() == DIRECTADMIN_NAME: context['http_docs'] = 'https://help.directadmin.com/item.php?id=345' message = f'{message} For %(control_panel)s you follow this instruction ' \ f'to build native php: %(http_docs)s' super(ClSelectExcept.NativeNotInstalled, self).__init__( message, context=context) class MissingCagefsPackage(BaseClSelectException): def __init__(self): link = "http://docs.cloudlinux.com/index.html?cagefs_installation.html" super(ClSelectExcept.MissingCagefsPackage, self).__init__( message="Required package `%(package)s` is not installed or broken. " "Please install or reinstall it. Details: %(link)s.", context=dict( link=link, package='cagefs' ) ) class SelectorException(BaseClSelectException): def __init__(self, message): super(ClSelectExcept.SelectorException, self).__init__(message) class InterpreterError(BaseClSelectException): def __init__(self, message, context=None): super(ClSelectExcept.InterpreterError, self).__init__( message, context=context or {}) class WebAppError(BaseClSelectException): def __init__(self, message): super(ClSelectExcept.WebAppError, self).__init__(message) class EmptyParam(BaseClSelectException): def __init__(self, message): message = "%s not specified" % message super(ClSelectExcept.EmptyParam, self).__init__(message) class NoSuchApplication(BaseClSelectException): def __init__(self, message): super(ClSelectExcept.NoSuchApplication, self).__init__(message) class FileProcessError(BaseClSelectException): def __init__(self, filename, message=None): message = "File (%s): %s" % (filename, message) super(ClSelectExcept.FileProcessError, self).__init__(message) class CreateSymlinkError(BaseClSelectException): def __init__(self, message): super(ClSelectExcept.CreateSymlinkError, self).__init__(message) class RemoveSymlinkError(BaseClSelectException): def __init__(self, message): super(ClSelectExcept.RemoveSymlinkError, self).__init__(message) class SymlinkError(BaseClSelectException): def __init__(self, message): super(ClSelectExcept.SymlinkError, self).__init__(message) class MissingApprootDirectory(BaseClSelectException): def __init__(self, message): super(ClSelectExcept.MissingApprootDirectory, self).__init__(message) class ConfigMissingError(BaseClSelectException): def __init__(self, message): super(ClSelectExcept.ConfigMissingError, self).__init__(message) class VersionModificationBlocked(BaseClSelectException): def __init__(self, message): super(ClSelectExcept.VersionModificationBlocked, self).__init__(message)	10.22 KB	2026-02-26 15:33
📁 clselectnodejs	-	2026-03-27 02:38
📁 clselectnodejsuser	-	2026-03-27 02:38
📁 clselectphp	-	2026-03-27 02:38
📁 clselectphpuser	-	2026-03-27 02:38
📄 clselectprint.py # -- 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 print_function from __future__ import division from __future__ import absolute_import import csv import simplejson import sys from xml.sax.saxutils import escape as _escape from past.builtins import basestring # noqa from future.utils import iteritems HTML_ESCAPE_TABLE = { '"': """, "'": "'" } def escape_string(data): if isinstance(data, basestring): return _escape(data, HTML_ESCAPE_TABLE) elif isinstance(data, (tuple, list)): new_data = [] for value in data: new_data.append(escape_string(value)) return new_data elif isinstance(data, dict): new_dict = {} for k, v in iteritems(data): new_dict[k] = escape_string(v) return new_dict return data def validate_json_message(data): # copies message from 'details' to 'message' # in case there is no 'message' in json if not 'message' in data: data['message'] = data['details'] class clprint(object): def print_data(cls, fmt, data, escape=None): """ Dispatches data to corresponing routine for printing @param fmt: string @param data: dict """ dispatcher = { 'json': cls.print_json, 'perl': cls.print_perl, 'csv': cls.print_csv, 'text': cls.print_text } try: dispatcher[fmt](data, escape=escape) except KeyError: dispatcher['text'](data) print_data = classmethod(print_data) def print_diag(cls, fmt, data): """ Dispatches data to corresponing routine for printing @param fmt: string @param data: dict """ dispatcher = { 'json': cls.print_diag_json, 'perl': cls.print_diag_perl, 'csv': cls.print_diag_csv, 'text': cls.print_diag_text } try: dispatcher[fmt](data) except KeyError: dispatcher['text'](data) print_diag = classmethod(print_diag) def print_csv(data, escape=None): """ Prints data as comma separated values @param data: dict """ csv_out = csv.writer(sys.stdout, quoting=csv.QUOTE_ALL) for opt in sorted(data.keys()): flat_array = [opt] for key, value in iteritems(data[opt]): if escape: value = escape_string(value) flat_array.extend([key, value]) csv_out.writerow(flat_array) print_csv = staticmethod(print_csv) def print_diag_csv(data): """ Prints diagnostic messages as comma separated values @param data: dict """ validate_json_message(data) csv_out = csv.writer(sys.stdout, quoting=csv.QUOTE_ALL) csv_out.writerow([data['status'], data['message']]) print_diag_csv = staticmethod(print_diag_csv) def print_json(data, escape=None): """ Prints data as JSON @param data: dict """ if data: if escape: data = escape_string(data) print(simplejson.dumps({ 'status': 'OK', 'data': data})) else: print(simplejson.dumps({'status': 'OK'})) print_json = staticmethod(print_json) def print_diag_json(data): """ Prints diagnostic messages as JSON @param data: dict """ print(simplejson.dumps(data)) print_diag_json = staticmethod(print_diag_json) def print_text(data, escape=None): """ Prints data as plain text @param data: dict """ for opt in sorted(data.keys()): print("TITLE:%s" % (opt,)) for key, v in iteritems(data[opt]): if escape: v = escape_string(v) print('%s:%s' % (key.upper(), v)) print('') print_text = staticmethod(print_text) @staticmethod def print_diag_text(data): """ Prints diagnostic messages as plain text @param data: dict """ validate_json_message(data) print("%s:%s" % (data['status'], data['message']), file=sys.stderr) if data.get('details'): print("Details:", file=sys.stderr) print(data.get('details', '') % data.get('context', {}), file=sys.stderr) def print_diag_perl(data): """ Prints diagnostic messages as perl data structure @param data: dict """ validate_json_message(data) print("{status=>%s,message=>%s}" % (data['status'], data['message'])) print_diag_perl = staticmethod(print_diag_perl) def print_perl(data, escape=None): """ Prints data as perl data structure @param data: dict """ out = [] for opt in sorted(data.keys()): structure = [] structure.append("title=>'%s'" % (opt, )) for k, v in iteritems(data[opt]): if escape: v = escape_string(v) structure.append("%s=>'%s'" % (k, v)) out.append("{%s}" % (','.join(structure))) print('[%s]' % (','.join(out),)) print_perl = staticmethod(print_perl)	5.39 KB	2026-02-26 15:33
📁 clselectpython	-	2026-03-27 02:38
📁 clselectpythonuser	-	2026-03-27 02:38
📁 clselectruby	-	2026-03-27 02:38
📄 clselectstatistics.py #!/opt/cloudlinux/venv/bin/python3 -bb # 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 print_function from __future__ import division from __future__ import absolute_import from collections import defaultdict from future.utils import iteritems from cldetectlib import get_suEXEC_status, get_suPHP_status from cllimits import CageFs, CageFsException from clcommon.cpapi import get_domains_php_info, get_system_php_info, getCPName from .cluserselect import ClUserSelect from .clselect import ClSelect as ClSelectPhp from .clselectctl import interpreter_versions_short_summary, server_applications_summary from typing import Optional, Dict # NOQA def _iter_versions(interpreter): """ Return list of InterpreterSummary objects :rtype: list[clselectctl.InterpreterSummary] """ return interpreter_versions_short_summary(interpreter) def iter_server_applications(interpreter): """ Return list of ApplicationSummary objects :rtype: list[clselectctl.ApplicationSummary] """ return server_applications_summary(interpreter) def get_versions_statistics(interpreter): # type: (str) -> Optional[Dict] interpreters_versions = _iter_versions(interpreter) if interpreters_versions is not None: return { it.version: { 'full_version': it.version_full, 'enabled': it.enabled, 'installed': it.installed } for it in interpreters_versions } def _is_acceptable_php_handler_cpanel(handler): """ Handler is supported in the following cases: - handler is suphp or lsapi with suexec or suphp (check for last two is in the _get_php_selector_usage method) - handler is cgi or fcgi (only with suexec, otherwise scripts do not run in cagefs) - handler is None (possible when php package was removed and we can`t detect handler, it is set to None) """ return handler in {'suphp', 'lsapi'} or \ handler in {'cgi', 'fcgi'} and get_suEXEC_status() or handler is None def _get_php_selector_domains_for_cpanel(): """ Return domains that are using php selector. 1. Skip domains whose version is not system default (php selector replaces only system default version binary with symlink to alt-php) 2. Take only domains that use supported handlers. """ vhosts_php_info = get_domains_php_info() default_php_id = get_system_php_info()['default_version_id'] phpselector_domains = defaultdict(set) for domain, domain_info in iteritems(vhosts_php_info): if domain_info['php_version_id'] != default_php_id: continue if _is_acceptable_php_handler_cpanel(domain_info['handler_type']): phpselector_domains[domain_info['username']].add(domain) return dict(phpselector_domains) def _get_php_selector_domains_for_plesk(): """ Return domains that meet panel specific requirements which will allow php selector to work when the common requirements are met. """ allowed_handlers = ('cgi', 'fastcgi') def is_allowed_version(handler_id): lsphp_vendor_version = 'x-httpd-lsphp-custom' if 'lsphp' in handler_id: return handler_id == lsphp_vendor_version return True domains_php_info = get_domains_php_info() result = defaultdict(set) for domain, info in iteritems(domains_php_info): if (info['handler_type'] in allowed_handlers and is_allowed_version(info['php_version_id'])): result[info['username']].add(domain) return dict(result) def _get_php_selector_domains_for_da(): """ Return domains that meet panel specific requirements which will allow php selector to work when the common requirements are met. """ allowed_handlers = ('fastcgi', 'suphp', 'cgi', 'lsphp') domains_php_info = get_domains_php_info() result = defaultdict(set) for domain, php_info in iteritems(domains_php_info): if php_info['handler_type'] in allowed_handlers: result[php_info['username']].add(domain) return dict(result) def _get_php_vhosts_for_current_panel(): """ Detect current control panel and get list of php selector domains for it Return None if control panel is not supported """ panel = getCPName() if panel == 'cPanel': return _get_php_selector_domains_for_cpanel() elif panel == 'DirectAdmin': return _get_php_selector_domains_for_da() elif panel == 'Plesk': return _get_php_selector_domains_for_plesk() else: # php selector is not officially # supported on another control panels return None def get_php_selector_usage(): """Get users and domains that use php selector""" cagefs = CageFs() if not cagefs.is_cagefs_present(): return None # check if cagefs is initialized try: cagefs._load_info() except CageFsException: return None # either suexec or suphp is required to move user into cagefs if not any((get_suEXEC_status(), get_suPHP_status())): return None # ignore unsupported control panels php_vhosts = _get_php_vhosts_for_current_panel() if php_vhosts is None: return None php = ClUserSelect() domains_by_php_version = defaultdict(set) users_by_php_version = defaultdict(set) for user, user_domains in iteritems(php_vhosts): if not cagefs.get_user_status(user): continue version = php.get_version(user, show_native_version=False)[0] domains_by_php_version[version].update(user_domains) users_by_php_version[version].add(user) return { 'domains_by_php_version': domains_by_php_version, 'users_by_php_version': users_by_php_version } def get_native_version_safe(): """ Safely get native version, or None if not set """ php = ClSelectPhp() v = php.get_native_version(verbose=False) # tuple[version, full_version] \| None if v is None: return v return v[0] def get_mode_of_php_selector(): # type: () -> str """ Get state of PHP selector: without CageFS or normal """ return 'without_cagefs' if ClSelectPhp().work_without_cagefs() else 'normal' def get_default_php_version(): # type: () -> str """ Get default version of PHP selector """ # yep, get_version() returns default(!) version, don't ask me 'why' return ClSelectPhp().get_version()[0]	6.51 KB	2026-02-26 15:33
📄 cluserextselect.py # -- 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 os from clcommon import clcagefs from .clextselect import ClExtSelect from .cluserselect import ClUserSelect from .clselectexcept import ClSelectExcept from .clselectprint import clprint from .utils import apply_for_at_least_one_user from .models import Website try: from clcagefslib.webisolation.jail_utils import get_website_id except ImportError: get_website_id = None class ClUserExtSelect(ClUserSelect, ClExtSelect): USER_INI = 'alt_php.ini' def __init__(self, item='php', exclude_pid_list=None): ClUserSelect.__init__(self, item, exclude_pid_list) self._alt_extensions = None self._user_extensions = [] self._conflicts = [] def list_enabled_extensions(self, user, version=None, website=None): """ Returns enabled user extensions for a version as a tuple @param user: string @param version: string @param website: Website or None - website info for per-domain configuration @return: tuple """ self._check_user_in_cagefs(user) if version is None: version = self.get_version(user, domain=website.domain if website else None)[0] if version == 'native': raise ClSelectExcept.UnableToGetExtensions(version) return tuple(map((lambda i: (i, True)), sorted(self._get_enabled_extensions(user, version, website=website)))) def list_all_extensions(self, user, version=None, website=None): """ Returns as a tuple all user extensions for a version, marking enabled and disabled ones @param user: string @param version: string @param website: Website or None - website info for per-domain configuration @return: tuple """ self._check_user_in_cagefs(user) if version is None: version = self.get_version(user, domain=website.domain if website else None)[0] if version == 'native': raise ClSelectExcept.UnableToGetExtensions(version) user_extensions = set(self._get_enabled_extensions(user, version, website=website)) builtin_extensions = set(self._get_builtins(version)) php_d_all_extensions = set(self._get_all_extensions(version)) all_extensions = builtin_extensions \| php_d_all_extensions extensions = [] for ext in sorted(all_extensions): enabled = False if ext in builtin_extensions: enabled = None elif ext in user_extensions: enabled = True extensions.append((ext, enabled)) return tuple(extensions) def _get_enabled_extensions(self, user, version, website=None): """ Returns list of enabled user extensions @param user: string @param version: string @param website: Website or None - website info for per-domain configuration @return: list """ if len(self._user_extensions) == 0: self._load_user_extensions(user, version, website=website) return self._user_extensions def _get_all_extensions(self, version): """ Returns list of all extensions for a version except compiled-in ones @param user: string @param version: string @return: list """ if self._alt_extensions is None: self._alt_extensions = self._load_extensions_list(version) return self._alt_extensions def _check_extensions(self, ext_list, version): """ validation extensions name @param ext_list: list @param version: string """ all_extensions = self._get_all_extensions(version) bad_extensions = set(ext_list).difference(set(all_extensions)) if bad_extensions: raise ClSelectExcept.NoSuchExtension(ext_list=bad_extensions, all_extensions=all_extensions) def bulk_enable_extensions(self, user, version, ext_list, check_ext=False, website=None): return self.bulk_handle_extensions(user, self.enable_extensions, version, ext_list, check_ext=check_ext, website=website) def bulk_handle_extensions(self, user, func, args, kwargs): return apply_for_at_least_one_user( func, self._clpwd.get_names(self._clpwd.get_uid(user)), ClSelectExcept.UnableToSaveData, args, **kwargs ) def enable_extensions(self, user: str, version: str, ext_list: list[str], check_ext: bool = False, website: Website \| None = None) -> None: """ Adds extensions to user php.ini """ self._check_user_in_cagefs(user) if check_ext: self._check_extensions(ext_list=ext_list, version=version) user_ini_path = self._compose_user_ini_path(user, version, website=website) alt_path = self._compose_alt_path(version) contents, extensions, extensions_data = self._load_ini_contents(user_ini_path) resulting_extensions = list(extensions_data.keys()) + ext_list ext_list_without_conflicts = self._filter_out_conflicts(resulting_extensions) _conflicts_info = ClExtSelect.get_conflicts_info(resulting_extensions, ext_list_without_conflicts) self._print_conflicts_info(_conflicts_info) extensions = [ext for ext in extensions + ext_list if ext in ext_list_without_conflicts] resolved_dependencies = [] for extension in extensions: resolved = [ext for ext in self._include_dependencies([extension], alt_path, extensions_data) if ext not in resolved_dependencies] dependence_info = ClExtSelect.get_dependencies_list(extension, resolved, ext_list) self._print_dependencies_info(dependence_info) resolved_dependencies.extend(resolved) for ext in resolved_dependencies: contents.extend( self._smooth_data(extensions_data[ext])) contents = self._move_ioncube_ext(contents) self._write_to_file( user, '\n'.join(contents)+'\n', user_ini_path) self._backup_settings(user) self._reload_processes(user) def bulk_replace_extensions(self, user, version, ext_list, website=None): return self.bulk_handle_extensions(user, self.replace_extensions_with_dependenses, version, ext_list, website=website) def replace_extensions(self, user: str, version: str, ext_list: list[str], website: Website \| None = None) -> tuple[list[str], list[str], list[str]]: """ (deprecated) Replaces extensions in user php.ini with supplied ones and print information about dependences and conflicts for old php selector """ (extensions_list, dependencies_list, conflict_dependencies) = \ self.bulk_replace_extensions(user, version, ext_list, website=website) self._print_dependencies_info(dependencies_list) self._print_conflicts_info(conflict_dependencies) return extensions_list def replace_extensions_with_dependenses(self, user: str, version: str, ext_list: list[str], website: Website \| None = None) -> tuple[list[str], list[str], list[str]]: """ Replaces extensions in user php.ini with supplied ones """ self._check_user_in_cagefs(user) resolved_dependencies = [] conflict_dependencies = [] dependencies_list = [] user_ini_path = self._compose_user_ini_path(user, version, website=website) alt_path = self._compose_alt_path(version) contents, extensions, extensions_data = self._load_ini_contents(user_ini_path) extensions_data = {} # we REPLACE extensions ext_list_without_conflicts = self._filter_out_conflicts(ext_list) conflict_dependencies = ClExtSelect.get_conflicts_info(ext_list, ext_list_without_conflicts) extensions = [ext for ext in ext_list if ext in ext_list_without_conflicts] for extension in extensions: resolved = [ext for ext in self._include_dependencies([extension], alt_path, extensions_data) if ext not in resolved_dependencies] dependencies_list = dependencies_list + self.get_dependencies_list(extension, resolved, ext_list) resolved_dependencies.extend(resolved) for ext in resolved_dependencies: contents.extend( self._smooth_data(extensions_data[ext])) contents = self._move_ioncube_ext(contents) self._write_to_file( user, '\n'.join(contents)+'\n', user_ini_path) self._backup_settings(user) self._reload_processes(user) return list(extensions_data.keys()), dependencies_list, conflict_dependencies def bulk_disable_extensions(self, user, version, ext_list, website=None): return self.bulk_handle_extensions(user, self.disable_extensions, version, ext_list, website=website) def disable_extensions(self, user: str, version: str, ext_list: list[str], website: Website \| None = None) -> None: """ Removes extensions from user php.ini """ self._check_user_in_cagefs(user) user_ini_path = self._compose_user_ini_path(user, version, website=website) alt_path = self._compose_alt_path(version) (contents, extensions, extensions_data) = self._load_ini_contents(user_ini_path) for item in set(ext_list): if item not in extensions_data: continue rest_of_set = set(extensions_data.keys()).difference([item]) if (self._is_dependency(item, rest_of_set, alt_path) and not rest_of_set.issubset(set(ext_list))): clprint.print_diag( 'text', {'status': 'WARN', 'message': '%s left as dependency' % (item,)}) continue extensions_data.pop(item, None) for ext in extensions: if ext not in extensions_data: continue contents.extend( self._smooth_data(extensions_data[ext])) self._write_to_file( user, '\n'.join(contents)+'\n', user_ini_path) self._backup_settings(user) self._reload_processes(user) def reset_extensions(self, user, version, website=None): """ Replaces extensions in user php.ini with default ones @param user: string @param version: string @param website: Website or None - website info for per-domain configuration @return: list """ if not version: raise ClSelectExcept.EmptyParam('Version') ext_list = ClExtSelect._get_enabled_extensions(self, version) # replace extensions already adapted for multiple users data = self.replace_extensions(user, version, ext_list, website=website) return data def _move_ioncube_ext(contents): """ PHP ioncube extensions must be at the beginning of extensions list @param contents: list @return: list """ found = False fixed_contents = [] stripped_contents = [] for item in contents: if item.startswith(';---ioncube'): found = True fixed_contents.append(item) continue if found: if item.startswith(';---'): found = False; else: fixed_contents.append(item) continue stripped_contents.append(item) fixed_contents.extend(stripped_contents) return fixed_contents _move_ioncube_ext = staticmethod(_move_ioncube_ext) @staticmethod def _load_ini_contents(path): """ Parses user ini file contents @param path: string @return: tuple """ contents = [] extensions = [] extensions_data = {} is_content = False is_extension = False ext_name = None try: ini = open(path, 'r') for line in ini: line = line.rstrip() # Check if it is content block (Custom PHP options) if line.startswith(';>==='): is_extension = False is_content = True contents.append(line) # Check if it is extension block elif line.startswith(';---'): tmp_ext_name = line.strip(';- ') if tmp_ext_name == "": continue ext_name = tmp_ext_name is_extension = True # Create the key-value pair if ext_name not in extensions_data: extensions_data[ext_name] = [line] # Processing content elif is_content: # Skip comments if line.startswith(';') and not line.startswith(';<==='): continue # Append until the end of block ('<===') contents.append(line) if line.startswith(';<==='): is_content = False # Processing extensions elif ext_name and is_extension: # Skip comments and empty lines if line.startswith(';') or line == "": continue # Append extension data if ext_name not in extensions: extensions.append(ext_name) if ext_name in extensions_data: extensions_data[ext_name].append(line) ini.close() except (OSError, IOError): pass return contents, extensions, extensions_data def _load_user_extensions(self, user: str, version: str, website: Website \| None = None) -> None: """ Loads user alternative extensions list for a version """ user_ini_path = self._compose_user_ini_path(user, version, website=website) extensions = self._skim_over_extensions(user_ini_path) self._user_extensions.extend(extensions) def _compose_user_ini_path(self, user: str, version: str, website: Website \| None = None) -> str: """ Composes user ini file path """ if self.without_cagefs: # Per-domain configuration is only supported with CageFS if website: raise ClSelectExcept.MissingCagefsPackage() homedir = self._clpwd.get_homedir(user) php_dir = 'php%s' % version.replace('.', '') return homedir + '/.cl.selector/alt_' + php_dir + '.ini' uid = str(self._clpwd.get_uid(user)) user_prefix = '/' if clcagefs.in_cagefs() else os.path.join(self.CAGEFS_PATH, uid[-2:], user) if website: # per-website configuration: .../{website_hash}/alt-php{version}/alt_php.ini if get_website_id is None: raise ClSelectExcept.MissingCagefsPackage() website_hash = get_website_id(website.docroot) path = os.path.join(user_prefix, 'etc', 'cl.%s.d' % (self._item,), website_hash, 'alt-%s%s' % (self._item, version.replace('.', '')), self.USER_INI) else: path = os.path.join(user_prefix, 'etc', 'cl.%s.d' % (self._item,), 'alt-%s%s' % (self._item, version.replace('.', '')), self.USER_INI) return path	16.54 KB	2026-02-26 15:33
📄 cluseroptselect.py # -- 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 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 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 = {" ": " ", '"': """, "'": "'", ">": ">", "<": "<", "&": "&"} 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): """ Inserts supplied options into current ones @param optset: string @param decoder: string @param """ 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 ) def insert_json_options(self, user, version, options, append=False, create=True): """ Inserts supplied options into current ones @param user: string @param version: string @param options: object """ self._check_user_in_cagefs(user) 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) 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): """ Handles multiple users with same uids """ 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 ) def delete_options(self, user, version, optset, decoder, quiet=True): """ Deletes supplied options from current ones """ 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 ) def _delete_user(self, user, optset, decoder, version): 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) (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): """ Returns options summary for a user @param user: string @param version: string 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) return self._get_whitelist(version) def reset_options(self, users=None, versions=None): """ Deletes all custom options settings @param users: list @param versions: list """ 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) 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(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_cli=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_cli', '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): """ Parses user ini file and updates values of existing data @param user: string """ self._get_whitelist(version) 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) 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)	23.65 KB	2026-02-26 15:33
📄 cluserselect.py # -- 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 print_function from __future__ import absolute_import from __future__ import division import json import os import sys import uuid import signal import secureio from future.moves import configparser as ConfigParser from stat import S_IRUSR, S_IWUSR, S_IRGRP, S_IROTH from future.utils import iteritems from pathlib import Path import psutil from .clselect import ClSelect from .clselectexcept import ClSelectExcept from clcommon import ClPwd, clcaptain from .clselectprint import clprint from . import utils from clcommon import clcagefs from clcommon.utils import ExternalProgramFailed from clcommon.cpapi import docroot, domain_owner, userdomains from clcommon.cpapi.cpapiexceptions import NoDomain, NoPanelUser try: from clcagefslib.const import BASEDIR from clcagefslib.fs import get_user_prefix from clcagefslib.selector.configure import is_ea4_enabled, read_cpanel_ea4_php_conf, configure_alt_php from clcagefslib.selector.panel.da import da_change_user_php_ini from clcagefslib.selector.panel.isp import ispmanager_create_user_wrapper except ImportError: pass try: from clcagefslib.webisolation.jail_utils import get_website_id from clcagefslib.webisolation.config import load_user_config except ImportError: get_website_id = lambda _: None load_user_config = None class ClUserSelect(ClSelect): CAGEFS_PATH = "/var/cagefs" SELECTOR_PATH = "/usr/selector" NATIVE_PATH = SELECTOR_PATH if clcagefs.in_cagefs() else "/usr/share/cagefs-skeleton/usr/selector" CAGEFS_EXCLUDE = "/etc/cagefs/exclude" SELECTOR2_DIR = ".cl.selector/selector.path" def clean_crui_images(self, users=None): """ Creates flags mod_lsapi_reset_me in users' home directories in order to recreate CRIU images when php version/extensions/options have changed For details see LVEMAN-1210 :param users: list of usernames (strings) """ # There is not reliable way to check if CRIU is enabled inside CageFS # So let's always create the "mod_lsapi_reset_me" flag if not clcagefs.in_cagefs() and not os.path.isfile("/var/run/mod_lsapi/criu.enabled"): return for user in users: pw = self._clpwd.get_pw_by_name(user) path = os.path.join(pw.pw_dir, "mod_lsapi_reset_me") if not os.path.isfile(path): previous_user_data = self._change_uid(user) try: clcaptain.write(path) except (OSError, ExternalProgramFailed) as e: raise ClSelectExcept.UnableToSaveData(path, e) finally: ClUserSelect._restore_uid(previous_user_data) @staticmethod def switch_symlink_for_alt_php(version, pw, document_root=None, exit_on_error=True, configure_multiphp=True): """ Switch symlink for alt php. Create .cagefs directory if not created Rerurn True if error has occured """ if not os.path.isdir(BASEDIR) and not clcagefs.in_cagefs(): print("ERROR: CageFS not installed.") if exit_on_error: sys.exit(1) else: return True if configure_multiphp and is_ea4_enabled(): conf = read_cpanel_ea4_php_conf() if conf: try: # get default system php version selected via MultiPHP Manager in cPanel WHM default_php = conf["default"] # LVEMAN-1170: do not configure PHP Selector when system default version is alt-php if not default_php.startswith("ea-php"): print( "ERROR: system default PHP version is alt-php. " "PHP Selector is disabled. Use cPanel MultiPHP manager instead." ) if exit_on_error: sys.exit(1) else: return True except KeyError: pass # configure alt php - create .cagefs dir and create symlink error = configure_alt_php( pw, version, document_root, write_log=False, drop_perm=(os.geteuid() == 0), configure_multiphp=configure_multiphp ) if error and exit_on_error: sys.exit(1) # When setting global version (document_root=None), sync isolated websites # that don't have per-domain PHP Selector enabled so they follow user-level settings if not document_root and load_user_config is not None: isolated_domains = set(load_user_config(pw.pw_name).enabled_websites) # in some tests (e.g selectorctl.backup.robot) user is created just via useradd -> # plesk cpapi raises NoPanelUser try: target_domains = userdomains(pw.pw_name) except NoPanelUser: target_domains = [] for domain, domain_docroot in target_domains: if domain not in isolated_domains: continue configure_alt_php( pw, version, domain_docroot, write_log=False, drop_perm=(os.geteuid() == 0), configure_multiphp=configure_multiphp ) return error def apply_symlinks_rules(self): if self.without_cagefs: print('ERROR: this option does not work in "single user" mode (when CageFS is disabled)') sys.exit(1) if os.geteuid() != 0: print("ERROR: root privileges required") sys.exit(1) users_vers_dict = self.get_user_version_map() for user, version in iteritems(users_vers_dict): print("Processing user", user) pw = self._clpwd.get_pw_by_name(user) ClUserSelect.switch_symlink_for_alt_php(version, pw, exit_on_error=False, configure_multiphp=False) def __init__(self, item="php", exclude_pid_list=None): ClSelect.__init__(self, item) self._clpwd = ClPwd() self._user_excludes = set() if exclude_pid_list: self.exclude_pid_list = exclude_pid_list else: self.exclude_pid_list = [] def get_version(self, user, domain=None, show_native_version=False): """ Returns alternative version for a user @param user: string @return: string """ self._check_user_in_cagefs(user) alt_path = self._compose_user_alt_path(user) if domain: document_root = docroot(domain)[0] alt_path = self._compose_user_alt_path(user, document_root) if not os.path.isdir(alt_path): alt_path = self._compose_user_alt_path(user) native = self._compose_native_info(show_native_version) if not os.path.isdir(alt_path): return native alternatives = self.get_all_alternatives_data() full_path = os.path.join(alt_path, self._item) if not os.path.islink(full_path): return native link_dst = os.readlink(full_path) if self.without_cagefs: if not self._native_contents: self._load_native_contents(self._item) if link_dst == self._native_contents[self._item]: return native if os.path.dirname(link_dst) == self.SELECTOR_PATH: return native try: version = list( filter((lambda i: alternatives[i]["data"][self._item] == link_dst), list(alternatives.keys())) )[0] return (version, alternatives[version]["version"], alternatives[version]["data"][self._item]) except (IndexError, KeyError): return native def create_dir(self, path, user): if not os.path.isdir(path): previous_user_data = self._change_uid(user) try: clcaptain.mkdir(path) except (OSError, ExternalProgramFailed) as e: raise ClSelectExcept.UnableToSaveData(path, e) finally: ClUserSelect._restore_uid(previous_user_data) def create_selector_symlinks(self, user): """ Creates additional directory and symlinks for use in "without CageFS" mode """ homedir = self._clpwd.get_homedir(user) path_in_home = os.path.join(homedir, self.SELECTOR2_DIR) cur_user = self._change_uid(user) self.create_dir(path_in_home, user) self._create_symlink("../php-cli", path_in_home + "/php", check_existence=True) self._create_symlink("../php", path_in_home + "/php-cgi", check_existence=True) self._restore_uid(cur_user) def get_default_version(self): if os.path.isfile(ClSelect.DEFAULTS_PATH): try: return self._dh.get("versions", self._item) except (ConfigParser.Error, IOError, KeyError): return "native" return "native" def set_version_from_backup(self, user): user_backup_path = os.path.join(self._clpwd.get_homedir(user), ".cl.selector", "defaults.cfg") if not os.path.isfile(user_backup_path): self.set_version(user, self.get_default_version()) else: try: dh = self._get_default_config_handler(user_backup_path) self.set_version(user, dh.get("versions", self._item)) except (ConfigParser.Error, IOError, KeyError) as e: print("Error while restoring settings from backup", str(e)) sys.exit(1) def _version_enabled(self, version): """" Checks if version is enabled, a version is considered "enabled" if there is no "state" option in the config. The presence of the "state" option always means it is disabled, regardless of its value. @param version: string @return: bool - True if version is enabled, False otherwise """ if not self._dh.has_option("%s%s" % (self._item, version), "state"): return True else: return False def set_version(self, user, version, domain=None, return_summary=False, show_native_version=False, exit_on_error=True): """ Sets alternative version for a users with the same uid @param user: string @return: None """ if os.geteuid() != 0 and (message := self.get_version_selection_disabled_msg(user)): raise ClSelectExcept.VersionModificationBlocked(message) # This check is needed to prevent setting a disabled version as selected version for a user if not self._version_enabled(version): raise ClSelectExcept.VersionModificationBlocked('Version %s is disabled by administrator' % version) if domain: try: owner = domain_owner(domain) except NoDomain: owner = None if owner != user: raise ClSelectExcept.SelectorException("Domain %s does not belong to user %s" % (domain, user)) data = utils.apply_for_at_least_one_user( self._set_version, self._clpwd.get_names(self._clpwd.get_uid(user)), ClSelectExcept.NoUserSelector, version, domain, return_summary, show_native_version, exit_on_error, ) if return_summary: return data def _set_version(self, user, version, domain=None, return_summary=False, show_native_version=False, exit_on_error=True): """ Sets alternative version for a user @param user: string @return: None """ if self.without_cagefs: previous_user_data = self._change_uid(user) self._check_user_in_cagefs(user) document_root = "" if domain: document_root = docroot(domain)[0] alt_path = self._compose_user_alt_path(user, document_root) if not os.path.isdir(alt_path): if self.without_cagefs or domain: self.create_dir(alt_path, user) else: raise ClSelectExcept.NoUserSelector(user) alternatives = self.get_all_alternatives_data() if version not in alternatives and version != "native": raise ClSelectExcept.NoSuchAlternativeVersion(version) self._remove_alternatives_links(alt_path) pw = self._clpwd.get_pw_by_name(user) if version == "native": if self.without_cagefs: if not self._native_contents: self._load_native_contents(self._item) for item, native_path in iteritems(self._native_contents): self._create_symlink(native_path, alt_path + "/" + item, user, version) else: ini = "php.ini" new_ini_created = False new_ini_path = os.path.join("%s.etc" % (self.NATIVE_PATH,), ini) if os.path.exists(new_ini_path): src = os.path.join("%s.etc" % self.SELECTOR_PATH, ini) dst = os.path.join(alt_path, ini) self._create_symlink(src, dst, user, version) new_ini_created = True for filename in os.listdir(self.NATIVE_PATH): if self._item not in filename: continue if filename.endswith(".ini") and new_ini_created: continue dst = os.path.join(alt_path, filename) src = os.path.join(self.SELECTOR_PATH, filename) self._create_symlink(src, dst, user, version) else: for item, path in iteritems(alternatives[version]["data"]): self._create_symlink(path, os.path.join(alt_path, item), user, version) if self.without_cagefs: ClUserSelect._restore_uid(previous_user_data) else: ClUserSelect.switch_symlink_for_alt_php(version, pw, document_root=document_root, exit_on_error=exit_on_error) self._switch_php_da_isp(user, version) self._reload_processes(user) self._backup_settings(user) if return_summary: return self.get_summary(user, show_native_version) def get_version_selection_disabled_msg(self, username: str) -> str: """ Returns a message indicating that the selection of the PHP version is disabled for the user, based on the configuration file. Args: username (str): The username for which to check the configuration. Returns: str: The message indicating that version selection is disabled, or an empty string if the configuration file does not exist or does not contain the message. """ uid = self._clpwd.get_uid(username) config_file = Path(f"/var/cloudlinux/cl.selector/uids/{uid}/version_selection_conf.json") if not config_file.exists(): return "" try: with config_file.open(encoding="utf-8") as f: config_data = json.load(f) return config_data.get("version_selection_disabled_msg", "") except (OSError, ValueError): return "" def get_summary(self, user, show_native_version=False): """ Returns a summary of available alternative versions for the given user. Only versions that are enabled, default, or selected are included. The label for the "native" version is replaced if show_native_version is True. @param user: str - Username inside CageFS. @param show_native_version: bool - Whether to replace the 'native' version label. @return: tuple of (str, (bool, bool, bool)) - Each entry contains the version name and a tuple with flags: (enabled, default, selected). """ self._check_user_in_cagefs(user) # Retrieve all alternatives data and "native" alternatives = self.get_all_alternatives_data() native_info = self._compose_native_info(show_native_version) summary = {"native": {"enabled": True, "default": False, "selected": False}} alt_versions = sorted(alternatives.keys()) alt_versions.append("native") selected_version = self.get_version(user)[0] # Populate summary with alternative versions for version in alt_versions: if version not in summary: summary[version] = {} summary[version]["enabled"] = self._version_enabled(version) summary[version]["default"] = False summary[version]["selected"] = False try: default_version = self._dh.get("versions", self._item) except (ConfigParser.NoSectionError, ConfigParser.NoOptionError): default_version = "native" try: summary[default_version]["default"] = True summary[selected_version]["selected"] = True except KeyError: raise ClSelectExcept.NoSuchAlternativeVersion(default_version) summary[native_info[0]] = summary.pop("native") alt_versions.remove("native") alt_versions.append(native_info[0]) result = [] # Final list for summary output # Include only versions where at least one flag is True for idx, v in enumerate(alt_versions): if not any((summary[v]["enabled"], summary[v]["default"], summary[v]["selected"])): continue result.append((v, (summary[v]["enabled"], summary[v]["default"], summary[v]["selected"]))) return tuple(result) def change_to_version(self, new_version, current_version): """ Changes users of a supplied version to specified_version @param version: string @param current_version: string """ users = self.list_users(current_version) for user in users: try: self.set_version(user, new_version, exit_on_error=False) except Exception as e: # catch every errors, print it and go to the next user clprint.print_diag("text", {"status": "ERROR", "message": str(e)}) pass self.clean_crui_images(users) def list_users(self, version): """ Returns users of a certain alternative """ data = self.get_version_user_map() if version in data: return data[version] return [] def list_all_users(self): """ Returns all valid system users @return: list """ if self.without_cagefs: from .clselectctlphp import get_cpanel_user return [get_cpanel_user()] return list(self._get_system_users().difference(self._get_user_excludes())) def cagefs_copy_etc(self, user): config = dict() config["init"] = 0 config["reinit"] = 0 config["verbose"] = 0 LIBDIR = "/usr/share/cagefs" sys.path.append(LIBDIR) try: import cagefsctl except ImportError: print("ERROR: CageFS not installed.") sys.exit(1) cagefs_etc_path = os.path.join(BASEDIR, get_user_prefix(user), user, "etc") if not os.path.exists(cagefs_etc_path + "/cl.selector") or not os.path.exists(cagefs_etc_path + "/cl.php.d"): cagefsctl.cpetc_for_user(user, config) def get_user_version_map(self): """ Returns user version map as dict @return: dict """ actual_users = self.list_all_users() data = {} for user in actual_users: try: data[user] = self.get_version(user, False)[0] except ClSelectExcept.NotCageFSUser: continue return data def get_version_user_map(self, user_names=None, domain=None): """ Returns users grouped by version @return: dict """ actual_users = user_names or self.list_all_users() data = {} for user in actual_users: try: version = self.get_version(user, domain, False)[0] if not version in data: data[version] = [] data[version].append(user) except ClSelectExcept.NotCageFSUser: continue return data def _create_symlink(src, dst, user=None, version=None, check_existence=False): """ Creates symlink from src to dst @param src: string @param dst: string @param user: string @param version: string @param check_existence: bool @return: None """ try: if check_existence: if os.path.islink(dst): if os.readlink(dst) != src: os.unlink(dst) else: return else: utils.remove_file_or_dir(dst) clcaptain.symlink(src, dst) except Exception as e: if user is not None and version is not None: raise ClSelectExcept.UnableToSetAlternative(user, version, e) raise ClSelectExcept.SelectorException("Cannot create symlink from %s to %s (%s)" % (src, dst, e)) _create_symlink = staticmethod(_create_symlink) def _get_user_excludes(self): """ Returns list of user excludes @return: list """ if self._user_excludes: return self._user_excludes if not os.path.isdir(self.CAGEFS_EXCLUDE): return set() for item in os.listdir(self.CAGEFS_EXCLUDE): full_item_path = os.path.join(self.CAGEFS_EXCLUDE, item) self._user_excludes.update( set(map((lambda x: x.strip()), utils.read_file_as_string(full_item_path).splitlines())) ) return self._user_excludes def _check_user_in_cagefs(self, user): """ Check that cagefs enabled for user """ if self.without_cagefs: return if clcagefs.in_cagefs(): return LIBDIR = "/usr/share/cagefs" sys.path.append(LIBDIR) try: import cagefsctl except ImportError: print("ERROR: CageFS not installed.") sys.exit(1) try: if not cagefsctl.is_user_enabled(user): raise ClSelectExcept.NotCageFSUser(user) except AttributeError: print("ERROR: CageFS version is unsupported. Please update CageFS.") sys.exit(1) def _remove_alternatives_links(self, path): """ Removes all symlinks from directory @param path: string @return: None """ for filename in os.listdir(path): if self._item not in filename: continue full_path = os.path.join(path, filename) if not os.path.islink(full_path): continue os.unlink(full_path) def _compose_user_alt_path(self, user, document_root=None): """ Composes and returns user alternative directory path @param user: string @return: string """ if self.without_cagefs: homedir = self._clpwd.get_homedir(user) return homedir + "/.cl.selector" uid = str(self._clpwd.get_uid(user)) document_root_id = "" if document_root: document_root_id = get_website_id(document_root) if document_root else "" base = ( os.path.join(self.CAGEFS_PATH, uid[-2:], user, "etc", "cl.selector") if not clcagefs.in_cagefs() else "/etc/cl.selector" ) return os.path.join(base, document_root_id) if document_root_id else base def _get_system_users(self): """ Returns set of system users @return: set """ users_dict = self._clpwd.get_user_dict() return set(users_dict.keys()) def _delete_if_symlink(file_path): """ Deletes file to be written if it is a symlink """ if os.path.islink(file_path): try: os.unlink(file_path) except OSError: raise ClSelectExcept.UnableToSaveData(file_path, "Cannot delete symlink while saving data") _delete_if_symlink = staticmethod(_delete_if_symlink) def _change_uid(self, user): """ Changes to another uid and returns tuple of previous euid and egid @param user: string @return: tuple """ entry = self._clpwd.get_pw_by_name(user) new_uid = entry.pw_uid new_gid = entry.pw_gid cur_euid = os.geteuid() cur_egid = os.getegid() if cur_euid == new_uid: return cur_euid, cur_egid try: os.setegid(new_gid) os.seteuid(new_uid) secureio.set_capability() return cur_euid, cur_egid except OSError as e: raise ClSelectExcept.UnableToChangeToAnotherUser(user, e) def _restore_uid(uid_and_gid): """ Restores changed uid and gid to original ones @param uid_and_gid: tuple @return: None """ if uid_and_gid[0] != os.geteuid(): secureio.set_capability(clear=True) try: os.setegid(uid_and_gid[1]) os.seteuid(uid_and_gid[0]) except OSError as e: raise ClSelectExcept.UnableToChangeToAnotherUser(str(uid_and_gid[0]), e) _restore_uid = staticmethod(_restore_uid) def _write_to_file(self, user, file_contents, file_path, create=True): """ Saves data to file @param user: string @param file_contents: string @param file_path: string @return: None """ if not create and not os.path.exists(file_path): return self._delete_if_symlink(file_path) previous_user_data = self._change_uid(user) file_directory = os.path.dirname(file_path) try: # Replace tempfile.mkstemp with str(uuid.uuid4()) dirname = "clseltmp_%s" % str(uuid.uuid4()) temp_path = os.path.join(file_directory, dirname) clcaptain.write(temp_path, "%s\n" % (file_contents,)) except (IOError, OSError, ExternalProgramFailed) as e: try: if os.path.exists(temp_path): os.unlink(temp_path) except: pass ClUserSelect._restore_uid(previous_user_data) raise ClSelectExcept.UnableToSaveData(file_path, e) else: try: mask = S_IRUSR \| S_IWUSR \| S_IRGRP \| S_IROTH os.rename(temp_path, file_path) os.chmod(file_path, mask) except OSError: pass ClUserSelect._restore_uid(previous_user_data) def _reload_processes(self, user): """ Reloads user process """ try: next_parent = psutil.Process() for i in range(2): next_parent = next_parent.parent() if next_parent is not None: self.exclude_pid_list.append(next_parent.pid) else: break except psutil.NoSuchProcess: pass try: uid = ClPwd().get_uid(user) except (ClPwd.NoSuchUserException,): # no such user return try: for proc in psutil.process_iter(): try: if uid not in [proc.uids().real, proc.uids().effective] or proc.name().find(self._item) == -1: continue pid = proc.pid except psutil.NoSuchProcess: continue try: if pid not in self.exclude_pid_list: os.kill(pid, signal.SIGHUP) except (OSError,): continue except (OSError, IOError): # psutil reads /proc FS as usual FS, skip read errors pass def _skim_over_extensions(path): """ Get extension names from user extensions file comments """ extensions = [] try: ini = open(path) for line in ini: if line.startswith(";---"): ext = line[4 : line.rfind("---")] extensions.append(ext) ini.close() except (OSError, IOError): pass return extensions _skim_over_extensions = staticmethod(_skim_over_extensions) def _backup_settings(self, user): """ Scans all user settings and backups'em in homedir as INI file @param user: string """ self._check_user_in_cagefs(user) backup_contents = [] user_alt_path = self._compose_user_alt_path(user) user_ext_path = os.path.join(os.path.dirname(user_alt_path), "cl.php.d") alternatives = self.get_all_alternatives_data() user_backup_path = os.path.join(self._clpwd.get_homedir(user), ".cl.selector") if not os.path.isdir(user_backup_path): previous_user_data = self._change_uid(user) try: clcaptain.mkdir(user_backup_path) except (OSError, ExternalProgramFailed) as e: ClUserSelect._restore_uid(previous_user_data) raise ClSelectExcept.UnableToSaveData(user_backup_path, e) ClUserSelect._restore_uid(previous_user_data) user_backup_file = os.path.join(user_backup_path, "defaults.cfg") if os.path.isdir(user_alt_path): version = "[versions]\n%s = %s\n" % (self._item, self.get_version(user)[0]) else: version = "[versions]\n%s = native\n" % (self._item,) backup_contents.append(version) for alt in sorted(alternatives.keys()): if self.without_cagefs: curr_ext_path = user_alt_path + "/alt_php" + alt.replace(".", "") + ".ini" else: curr_ext_path = os.path.join(user_ext_path, "alt-php%s" % ((alt.replace(".", ""),)), "alt_php.ini") extensions = self._skim_over_extensions(curr_ext_path) backup_contents.append("[%s%s]\nmodules = %s\n" % (self._item, alt, ",".join(sorted(extensions)))) self._write_to_file(user, "\n".join(backup_contents), user_backup_file) def _switch_php_da_isp(self, user, version): if self.without_cagefs: return da_change_user_php_ini(user, version) ispmanager_create_user_wrapper(user, version)	30.7 KB	2026-02-26 15:33
📄 locked_extensions.ini ; module version can consist of few versions (separated by comma) ; example: module_name = version_1, version_2, etc... ; --- ; module_name without version, means all module version is locked ; example: module_name = "" or module_name = [python2.6] pip = setuptools = wsgiref = [python2.7] pip = setuptools = wsgiref = [python3.2] pip = setuptools = wsgiref = [python3.3] pip = setuptools = wsgiref = [python3.4] pip = setuptools = wsgiref = [python3.5] pip = setuptools = wsgiref = [ruby1.8] rack = 1.5.2 [ruby1.9] bigdecimal = 1.1.0 io-console = 0.3 json = 1.5.5 minitest = 2.5.1 rake = 0.9.2.2 rack = rdoc = 3.9.5 [ruby2.0] activesupport = 4.1.8 bigdecimal = 1.2.0 i18n = 0.6.11 io-console = 0.4.2 json = 1.7.7 minitest = 5.4.3, 4.3.2 psych = 2.0.0 rake = 0.9.6 rack = rdoc = 4.0.0 s2s-auth = 0.0.3 test-unit = 2.0.0.0 thread_safe = 0.3.4 tzinfo = 1.2.2 [ruby2.1] bigdecimal = 1.2.4 io-console = 0.4.2 json = 1.8.1 minitest = 4.7.5 psych = 2.0.5 rake = 10.1.0 rack = rdoc = 4.1.0 test-unit = 2.1.2.0, 2.1.4.0 [ruby2.3] bigdecimal = 1.2.8 did_you_mean = 1.0.0 io-console = 0.4.5 json = 1.8.3 minitest = 5.8.3 net-telnet = 5.8.3 power_assert = 0.2.6 psych = 2.0.17 rake = 10.4.2 rdoc = 4.2.1 test-unit = 3.1.5 rack =	1.2 KB	2026-02-26 15:33
📄 models.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2025 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT class Website(object): """ Represents a website with its domain and document root. Used for per-domain configuration in PHP Selector. """ def __init__(self, domain: str, docroot: str): """ Initialize Website instance """ self.domain = domain self.docroot = docroot def __repr__(self): return "Website(domain='%s', docroot='%s')" % (self.domain, self.docroot) def __eq__(self, other): if not isinstance(other, Website): return False return self.domain == other.domain and self.docroot == other.docroot	815 B	2026-02-26 15:33
📄 utils.py # 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 print_function from __future__ import absolute_import from __future__ import division import errno import os import pwd import re import subprocess import sys import json import shutil from time import time from typing import AnyStr, List, Dict, Optional # NOQA from pathlib import Path from urllib.parse import urlparse, urlunparse from clcommon.clpwd import resolve_username_and_doc_root from clcommon.cpapi import CP_NAME from clcommon.cpapi.cpapiexceptions import NoDomain, NotSupported, IncorrectData from clcommon.utils import mod_makedirs from clsentry import init_sentry_client from clsentry.utils import get_pkg_version from .clselectprint import clprint from .clselectexcept import ClSelectExcept CAGEFS_ENTER_USER_BIN = "/sbin/cagefs_enter_user" LVEMANDSN = "https://9713d1296f804031b058b8f2d789d7ac:8ddacae32d8246cf8b25cf826bf3fc0a@cl.sentry.cloudlinux.com/12" def run_command(cmd, env_data=None): """ Runs external process and returns output :param cmd: command and arguments as a list :param env_data :return string """ try: output = subprocess.Popen( cmd, stdin=open("/dev/null"), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, close_fds=True, env=env_data, text=True, ) std_out, std_err = output.communicate() except (OSError, IOError) as e: raise ClSelectExcept.FileProcessError(cmd[0], str(e)) if output.returncode != 0: raise ClSelectExcept.ExternalProgramFailed( std_err or "output of the command: %s\n%s" % (" ".join(cmd), std_out) ) return std_out def run_command_full(cmd, env_data=None, preexec_fn=None, cwd=None): """ Runs external process and returns output. Differs from subprocess.check_output, run_command above, and check_output below in that it does not throw an exception if process's return code != 0 :param cmd: command and arguments as a list :param env_data :param preexec_fn: Pre-exec function. None if don't need :param cwd: Directory name to set as current :return Cortege: (ret_code, stdout, stderr) """ try: process = subprocess.Popen( cmd, stdin=open("/dev/null"), stdout=subprocess.PIPE, stderr=subprocess.PIPE, preexec_fn=preexec_fn, close_fds=True, cwd=cwd, env=env_data, ) std_out, std_err = process.communicate() except (OSError, IOError) as e: raise ClSelectExcept.FileProcessError(cmd[0], str(e)) return process.returncode, std_out, std_err def read_file_as_string(filename): """ Reads file contents and returns it as a string """ defaults_contents = "" try: f = open(filename) defaults_contents = f.read() f.close() except (OSError, IOError): pass return defaults_contents def check_call(args, kwargs): check_output(args, *kwargs) def check_output(args, *kwargs): """ check_output(val1, val2, val3, arg4=val4) equivalent check_output(args=(val1, val2, val3), arg4=val4) or equivalent check_output(val1, args=(val2, val3), arg4=val4) DON'T USE check_output((val1, val2, val3), arg4=val4) :param tuple args: arguments for command line :param dict kwargs: parameters for subprocess.Popen :return str: """ error = kwargs.pop("error", None) args_from_kwargs = kwargs.get("args") output = kwargs.pop("output", None) wait = kwargs.pop("wait", None) if args_from_kwargs: args += args_from_kwargs del kwargs["args"] try: p = subprocess.Popen(args=args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, kwargs) except (OSError, IOError) as e: raise ClSelectExcept.FileProcessError(args[0], str(e)) stdout, stderr = p.communicate() if output: output.write(stdout) output.close() if wait is not None: p.wait() if p.returncode: raise ClSelectExcept.ExternalProgramFailed(error or stderr or stdout) return stdout def list_dirs(path): return [d for d in os.listdir(path) if os.path.isdir(os.path.join(path, d))] def versioning(path, versions): yield path for major, minors in versions: yield path + str(major) for minor in minors: yield path + ".".join(map(str, (major, minor))) def mkdir_p(path): try: os.makedirs(path) except OSError as exc: if exc.errno == errno.EEXIST and os.path.isdir(path): pass else: raise def get_abs_rel(user, directory): """ Returns absolute and relative to user's home directories paths, after eliminating all symlinks :param user: Unix user whose home contains the directory :param directory: The directory to obtain absolute and relative paths for """ user_dir = os.path.realpath(pwd.getpwnam(user).pw_dir) abs_dir = os.path.realpath(os.path.join(user_dir, directory)) rel_dir = s_partition(abs_dir, user_dir)[2].lstrip(os.sep) if not all((abs_dir, rel_dir)): raise ClSelectExcept.WrongData("Directory '%s' not in user home ('%s')" % (directory, user_dir)) return abs_dir, rel_dir # TODO: need move to clcommon # safely open/close files in python2.4 # http://stackoverflow.com/questions/3770348/how-to-safely-open-close-files-in-python-2-4 # TODO: consider to remove this method and use builtin def file_read(path, mode="r", errors=None): """ read file and close :param path: file path :param mode: reading mode :return str: """ data = None stream = open(path, mode=mode, errors=errors) try: data = stream.read() finally: stream.close() return data def file_readlines(path, errors=None): """ read litle file and close :param path: :return str: """ data = [] stream = open(path, errors=errors) try: data = stream.readlines() finally: stream.close() return data def file_write(path, line, mode="w", errors=None): """ write litle data to file and close :param path: :return str: """ stream = open(path, mode, errors=errors) try: stream.write(line) finally: stream.close() def file_writelines(path, lines, mode="w", errors=None): """ write litle data to file and close :param path: :return str: """ stream = open(path, mode, errors=errors) try: stream.writelines(lines) finally: stream.close() def s_partition(s, sep): """ str.partition for all python versions :param s: string to parse :param sep: separator to split by :return: cortege - see str.partition function for python 2.5+ """ if sys.version_info[0] >= 2 and sys.version_info[1] >= 5: return s.partition(sep) # Python v2.4 and lower does not support str.partition, so emulating it s_parts = s.split(sep, 1) if len(s_parts) == 2: ret_cortege = (s_parts[0], sep, s_parts[1]) else: ret_cortege = (s_parts[0], "", "") return ret_cortege def grep(pattern, path): """ find regex count in file """ try: f = open(path, "r") except IOError: return None counter = 0 regex = re.compile(pattern) for line in f: if regex.search(line): counter += 1 f.close() return counter def user_should_be_skipped(username): """ Determines is user should be skipped in selectorctl work :param username: user name to check :return: True - user should be skipped, False - not """ from clcommon.cpapi import is_admin # user is DA admin, pass it return is_admin(username) def make_dir(dest_dir): if not os.path.isdir(dest_dir): try: mod_makedirs(dest_dir, 0o755) except (IOError, OSError) as e: print("Error: failed to create", dest_dir, ":", str(e)) sys.exit(1) def remove_file_or_dir(path): try: os.unlink(path) except OSError: pass if os.path.isdir(path): shutil.rmtree(path, True) def make_symlink(src, dst): try: if os.path.islink(dst): if os.readlink(dst) != src: os.unlink(dst) os.symlink(src, dst) else: remove_file_or_dir(dst) os.symlink(src, dst) except OSError as e: print("Error: failed to create symlink", dst, "->", src, ":", str(e)) sys.exit(1) def pretty_json(data): return json.dumps(data, indent=4, separators=(",", ": "), sort_keys=True) # TODO: we have the same safely_resolve_username_and_doc_root in selectorlib, need some refactoring def safely_resolve_username_and_doc_root(user=None, domain=None, msg_format="text", pass_not_supported_panel=False): """ Safely resolve username and doc_root by domain, or resolve document root by username, or resolve document root and username by effective uid :param pass_not_supported_panel: pass not supported panel for php selector :param msg_format: error messages' format (json or text) :param user: str -> name of unix user :param domain: str -> domain of panel user :return: tuple -> user, doc_root """ result_user = user result_doc_root = None try: result_user, result_doc_root = resolve_username_and_doc_root( user=user, domain=domain, ) except NoDomain: clprint.print_diag( msg_format, {"status": "ERROR", "message": "No domain found for user: {}, domain: {}".format(user, domain)} ) sys.exit(1) except NotSupported: if pass_not_supported_panel: pass else: clprint.print_diag( msg_format, {"status": "ERROR", "message": "NodeJs/Ruby/Python selector not supported for %s panel" % CP_NAME}, ) sys.exit(1) except IncorrectData: clprint.print_diag( msg_format, {"status": "ERROR", "message": "Domain %s is not owned by the user %s" % (domain, user)} ) sys.exit(1) return result_user, result_doc_root def get_using_realpath_keys(user, directory, config): """ Transform all directories to realpath, compare and return config value :param user: str -> unix user - owner of the config :param directory: str -> wanted directory key :param config: dict -> config to get value from, should have directories as keys """ abs_directory = get_abs_rel(user, directory)[0] for conf_dir in config: abs_conf_dir = get_abs_rel(user, conf_dir)[0] if abs_directory == abs_conf_dir: return config[conf_dir] raise KeyError("Config does not contain directory: {}".format(directory)) def delete_using_realpath_keys(user, directory, config): """ Transform all directories to realpath, compare and remove directory key :param user: str -> unix user - owner of the config :param directory: str -> wanted directory key :param config: dict -> config to remove key from, should have directories as keys """ abs_directory = get_abs_rel(user, directory)[0] removed_at_least_one = False config_keys = list(config.keys()) for conf_dir in config_keys: abs_conf_dir = get_abs_rel(user, conf_dir)[0] if abs_directory == abs_conf_dir: del config[conf_dir] removed_at_least_one = True if not removed_at_least_one: raise KeyError("Config does not contain directory: {}".format(directory)) def realpaths_are_equal(user, path1, path2): """ Checks that two paths in user home directory are the same after stripping symlinks :param user: Unix user whose home directory contains both paths :type path1: str :type path2: str :return: True if paths are canonically the same otherwise False :rtype: bool """ return get_abs_rel(user, path1)[0] == get_abs_rel(user, path2)[0] def run_process_in_cagefs(user, path_to_utility, args_for_utility, env_vars=None): # type: (AnyStr, AnyStr, List[AnyStr], Optional[Dict[AnyStr, AnyStr]]) -> Dict[AnyStr, AnyStr, int] """ Run any process using the utility cagefs_enter_user """ result = {} # there are cases when we need to store some env vars # (e.g HOME when calling pip during import) if not env_vars: env_vars = {} cmd = [CAGEFS_ENTER_USER_BIN, user, path_to_utility] + args_for_utility process = subprocess.Popen(cmd, env=env_vars, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) stdout, _ = process.communicate() retcode = process.returncode # contition `retcode > 128` because in lve-wrappers: # if (WIFSIGNALED(status)) { # exit(WTERMSIG(status) + 128); # } if not stdout and retcode > 128: stdout = { "details": "User's process \"{}\" failed and didn't return anything. It could mean that user " "has too small limit of PMEM for running of cloudlinux-selector. Please, " "increase the limit to user and try run process again".format(path_to_utility), "result": "error", "timestamp": time(), } else: stdout = stdout.strip() result["output"] = stdout result["returncode"] = retcode result["failed"] = True if retcode else False return result def is_imunify_using_python(version: str) -> bool: """ Check if python version is used by imunify packages The algorithm is as follows: - Verify if imunify services are installed - Check if shebang contains the path to the specified Python version being checked - If shebang contains the path to the imunify virtual environment, determine the Python version it is based on by reading a symlink """ # Init sentry for this function sentry_client = init_sentry_client("lvemanager", release=get_pkg_version("lvemanager"), dsn=LVEMANDSN, handle=False) alt_path = f"/opt/alt/python{version.replace('.', '')}" imunify_venv_path = "/opt/imunify360/venv" imunify_service = Path("/usr/bin/imunify-service") if not imunify_service.is_file(): return False try: with open(imunify_service) as f: first_line = f.readline() if not first_line.startswith("#!"): raise Exception(f'ERROR: "{imunify_service}" doesn\'t have a shebang') if alt_path in first_line: return True if imunify_venv_path in first_line: venv_python_bin = Path(first_line.lstrip("#!").strip()) python_bin = venv_python_bin.readlink() if alt_path in str(python_bin): return True except: sentry_client.captureException() return False def demote(user: str): """ Drops to user """ user_pwd = pwd.getpwnam(user) def func(): os.setgid(user_pwd.pw_gid) os.setuid(user_pwd.pw_uid) os.environ["USER"] = user os.environ["HOME"] = user_pwd.pw_dir return func def apply_for_at_least_one_user(func, users, exception_type, args, kwargs): caught_exception = None is_handled = False for user_alias in users: try: result = func(user_alias, args, *kwargs) except exception_type as e: caught_exception = e else: is_handled \|= True if not is_handled: raise caught_exception return result def transform_wildcard_url(url): """ Transforms a URL containing a wildcard ('') in the netloc. If the netloc of the URL contains a '', it replaces the '' with 'any'. This is necessary because URLs with wildcards are not valid for HTTP requests. Replacing '' with 'any' ensures compatibility with systems or libraries that do not support wildcards in URLs. :param url: str - The URL to transform. :return: str - The transformed URL with the wildcard replaced. """ parsed = urlparse(url) if "" in parsed.netloc: netloc = parsed.netloc.replace("*", "any") parsed = parsed._replace(netloc=netloc) return urlunparse(parsed)	16.22 KB	2026-02-26 15:33

Edit File:

Terminal