Edit File: php_manager.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 json import os from pwd import getpwuid from future.utils import iteritems import cldetectlib as detect from clcommon.clconfig import get_param, replace_param from clcommon.ui_config import UIConfig from clselect.clselectctlphp import format_summary, parse_extensions, API_1 from clselect import ClSelect from clselect import ClUserSelect, ClExtSelect, ClUserExtSelect, ClUserOptSelect from clselect.baseclselect import BaseSelectorError from clselect.clselectexcept import ClSelectExcept as ClSelectExceptions from clcommon.cpapi import get_main_username_by_uid class PhpManager(object): """Responsible for actual PhpSelector selector high-level API""" interpreter = 'php' # DirectAdmin parameters DA_PATH = "/usr/local/directadmin/plugins/phpselector/plugin.conf" DA_PARAM = "active" def __init__(self, cfg=None, pkg=None): euid = os.geteuid() self.is_root_user = euid == 0 self.user_name = get_main_username_by_uid(euid) if self.is_root_user: self.cl_select_lib = ClSelect(self.interpreter) else: self.cl_select_lib = ClUserSelect(self.interpreter) @property def selector_enabled(self): """ Get current status for panel :return: bool """ if detect.is_da(): # DirectAdmin return get_param(self.DA_PATH, self.DA_PARAM) == "yes" else: return not UIConfig().get_param('hidePhpApp', 'uiSettings') @selector_enabled.setter def selector_enabled(self, status): """ Enable/disable selector for each panel :param status: bool :return: """ if detect.is_da(): try: replace_param(self.DA_PATH, self.DA_PARAM, "yes" if status else "no") except (OSError, IOError) as e: raise BaseSelectorError("Can not process config file %s with reason: %s" % ( self.DA_PATH, str(e))) UIConfig().set_config({'uiSettings': {'hidePhpApp': not status}}) else: UIConfig().set_config({'uiSettings': {'hidePhpApp': not status}}) @property def extensions_is_hidden(self): """ Extensions was hidden by admin :return: bool """ return UIConfig().get_param('hidePHPextensions', 'uiSettings') is True @property def domains_tab_is_hidden(self): """ Domains tab was hidden by admin :return: bool """ return UIConfig().get_param('hideDomainsTab', 'uiSettings') is True def switch_current_version(self, version): """ Set current version for user :param version: str :return: None (succes) """ if self.is_root_user: raise BaseSelectorError("Not supported as root user") c = ClUserSelect(self.interpreter) c.set_version(self.user_name, str(version)) c.clean_crui_images([self.user_name]) def reset_extensions(self, version): """ Reset extension for selected version :param version: str :return: List of enabled extensions after reset """ if self.is_root_user: raise BaseSelectorError("Not supported as root user") ClSelect.check_multiphp_system_default_version() c = ClUserExtSelect(self.interpreter) extensions = c.reset_extensions(self.user_name, str(version)) c.clean_crui_images([self.user_name]) return {"extensions": extensions} def set_extensions(self, version, extensions): """ Set extensions for php version for admin and user :param version: str :param extensions: list :return: response for user. Error or warning list or no error """ to_enable = [] for k, v in iteritems(extensions): if v == "enabled": to_enable.append(k) elif v != "disabled": return {"status": "ERROR: %s is not a valid state of an extension" % v} if self.is_root_user: ClExtSelect(self.interpreter).replace_extensions(str(version), to_enable) return {} else: if self.extensions_is_hidden: raise BaseSelectorError("Extensions was disabled by admin") c = ClUserExtSelect(self.interpreter) extensions, resolved_dependencies, conflict_dependencies = c\ .bulk_replace_extensions(self.user_name, str(version), to_enable) c.clean_crui_images([self.user_name]) if resolved_dependencies: resolved_dependencies = ['{} enabled as dependency ({})'.format(*ext) for ext in resolved_dependencies] # ToDo: add ability to show several warnings and split messages warnings = (', '.join(resolved_dependencies)) if resolved_dependencies else '' if conflict_dependencies: if warnings: warnings = warnings + '. ' warnings = warnings + '{} skipped as conflicting.'.format(', '.join(conflict_dependencies)) if warnings: return {'warning': warnings} return {} def get_summary(self): """ Return all information of php selector All extensions(user, admin) and options(for user), selected version (for user) :return: dict """ selected_version = None if self.is_root_user: summary_data = self.cl_select_lib.get_summary(False) else: try: summary_data = self.cl_select_lib.get_summary(self.user_name, True) except ClSelectExceptions.NotCageFSUser: raise BaseSelectorError({'message': 'User %(user)s not in CageFs', 'context': {'user': self.user_name}}) # find selected version for version_info in summary_data: (version, (is_enabled, is_default, is_selected)) = version_info if is_selected: selected_version = version break json_data = format_summary(summary_data, format='json', api_version=API_1) json_dict = json.loads(json_data) json_dict['selector_enabled'] = self.selector_enabled json_dict['extensions_is_hidden'] = self.extensions_is_hidden json_dict['domains_tab_is_hidden'] = self.domains_tab_is_hidden if selected_version: # show selected version only for users json_dict['selected_version'] = selected_version try: versions_list = json_dict["available_versions"] except KeyError: if 'message' in json_dict: raise BaseSelectorError(json_dict['message']) else: return {"status": "ERROR", "data": json_dict} for item in versions_list: # get extensions for users only and exclude native version if not self.is_root_user and item['version'].startswith('native'): continue try: extensions_json = self._get_extensions(item['version']) item.update(extensions_json) except ClSelectExceptions.UnableToGetExtensions as e: raise BaseSelectorError(dict(message=str(e), details=e.details)) # get otions for every version for users exclude native if not self.is_root_user and not item['version'].startswith('native'): try: options_json = ClUserOptSelect(self.interpreter).get_options(self.user_name, item['version']) except ClSelectExceptions.UnableToGetExtensions as e: raise BaseSelectorError(dict(message=str(e), details=e.details)) item['options'] = options_json return json_dict def set_options(self, version, options): """ Save php options :param version: version which save options for :type version: str :param options: dict of options for save :type options: dict :return: None (success) """ if self.is_root_user: raise BaseSelectorError("Not supported as root user") ClSelect.check_multiphp_system_default_version() c = ClUserOptSelect(self.interpreter) c.bulk_insert_options( self.user_name, str(version), options) c.clean_crui_images([self.user_name]) def switch_default_version(self, version): ClSelect(self.interpreter).set_version(str(version)) def get_current_version(self, user_names = None): """ Return current versions for list of users or all users :param user_names: array of users :return: information with users and versions """ user_map = ClUserSelect(self.interpreter).get_version_user_map(user_names) result = {} for version, users in user_map.items(): result.update({user: {'version': version} for user in users}) return {'users': result} def _get_extensions(self, version): """ return list of exnensions for selected version (for user and admin) :param version: version get extensions for :type version: str :return: """ if self.is_root_user: # Recreate ClExtSelect for noncache ext_list = ClExtSelect(self.interpreter).list_extensions(version) else: ext_list = ClUserExtSelect(self.interpreter).list_all_extensions(self.user_name, version) return parse_extensions(ext_list, version, fmt='json')