[EDIT] FILE: 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 import re 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,)) # Extension tokens are concatenated into a filesystem path under the # alt-php ini dir. Reject anything that isn't a bare extension name so # path-traversal segments (`/`, `\`, `..`, NUL) can't smuggle a # user-controlled .ini into the dependency-resolution queue. _EXT_TOKEN_RE = re.compile(r'^[A-Za-z0-9_][A-Za-z0-9_.+-]*$') @classmethod def _is_safe_ext_token(cls, ext): return isinstance(ext, str) and bool(cls._EXT_TOKEN_RE.match(ext)) 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 for ext in ext_list if cls._is_safe_ext_token(ext)) 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: if cls._is_safe_ext_token(ext_name): 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() if not cls._is_safe_ext_token(ext): return dependencies 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,)})
SAVE CANCEL