Wala shell

Advanced PHP File Manager

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

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-27 02:38
📄 __init__.py # coding=utf-8 from __future__ import print_function from __future__ import division from __future__ import absolute_import from clcommon import FormattedException # Possible interpreter statuses ENABLED_STATUS = 'enabled' DISABLED_STATUS = 'disabled' NOT_INSTALLED_STATUS = 'not_installed' INSTALLING_STATUS = 'installing' REMOVING_STATUS = 'removing' # Means that cache update process is currently running CACHE_UPDATING_YUM_STATUS = "cache_updating" # possible application states APP_STARTED_CONST = 'started' APP_STOPPED_CONST = 'stopped' class BaseSelectorError(FormattedException): """Top level error class for admin's part of selector""" pass class AcquireInterpreterLockError(BaseSelectorError): """Error raised when we are unable to lock interpreter version""" def __init__(self, version): super(AcquireInterpreterLockError, self).__init__({ "message": "Unable to lock specified interpreter version %(version)s. Probably it's not " "installed at this moment", "context": { 'version': version }, }) class AcquireApplicationLockError(BaseSelectorError): """Error raised when we are unable to lock application""" def __init__(self, app_root, reason=None): message = "Can't acquire lock for app: %(app)s" if reason: message += " Reason is: {}".format(reason) super(AcquireApplicationLockError, self).__init__({ 'message': message, 'context': { 'app': app_root, }, }) class MissingVirtualenvError(BaseSelectorError): """Error raised when virtualenv directory is absent""" def __init__(self, app_venv): super(MissingVirtualenvError, self).__init__({ 'message': "No such application or it's broken. " "Unable to find app venv folder by this path: '%(app_venv)s'", 'context': { 'app_venv': app_venv, }, }) class MissingAppRootError(BaseSelectorError): """Error raised when application root directory is absent""" def __init__(self, app_root): super(MissingAppRootError, self).__init__({ 'message': "No such application or it's broken. " "Unable to find app-root folder by this path: '%(app_root)s'", 'context': { 'app_root': app_root, }, }) class AbsentFileError(BaseSelectorError): """Error raised when some file does not exist""" def __init__(self, filepath): super(AbsentFileError, self).__init__( {'message': "File %(file)s is absent.", 'context': {'file': filepath}} ) class UnsupportedSelectorError(BaseSelectorError): def __init__(self, interpreter): super(UnsupportedSelectorError, self).__init__( {'message': 'Unsupported interpreter was passed: "%(interpreter)s"', 'context': {'interpreter': interpreter}} )	3 KB	2026-02-26 15:33
📁 __pycache__	-	2026-03-21 15:48
📄 apps_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 absolute_import from __future__ import print_function from __future__ import division import json import os import re from abc import ABCMeta, abstractmethod from datetime import datetime import clselect.clpassenger as clpassenger import secureio from future.utils import iteritems from past.builtins import unicode from clcommon import ClPwd from clcommon.utils import get_file_lines, write_file_lines from clselect.utils import pretty_json, delete_using_realpath_keys, get_abs_rel from clselect import ClSelectExcept from clselect.baseclselect import ( BaseSelectorError, AbsentFileError, MissingVirtualenvError, MissingAppRootError ) from .selector_manager import BaseSelectorManager # NOQA from clselect.utils import file_readlines, file_writelines from clselect.utils import get_using_realpath_keys from future.utils import with_metaclass class BaseApplicationsManager(with_metaclass(ABCMeta, object)): """ Base class that responsible for gathering and set information about applications. """ # needs to be overridden in child class _USER_CONFIG = None _LOG_FILE_NAME = '/var/log/selectorctl.log' INTERPRETER = None VENV_DIR = None BINARY_NAME = None def __init__(self, manager): self._manager = manager # type: BaseSelectorManager self._pwd = ClPwd() @classmethod def write_string_to_log(cls, log_str): """ Writes string to log file :param log_str: String to write :return: None """ try: dt_string = datetime.now().strftime("%Y-%m-%d %H:%M:%S") write_file_lines(cls._LOG_FILE_NAME, ['%s: %s\n' % (dt_string, log_str)], 'a') except (IOError, OSError): pass def add_app_to_config(self, user, app_directory, app_data): """ Add data to user's config (update info for an app with app_directory specified) :param str user: name of unix user :param str app_directory: Application root directory :param dict app_data: Application data :return: None """ current_dict = self.get_user_config_data(user) current_dict[app_directory] = app_data # Write new config to file self.write_full_user_config_data(user, current_dict) def add_env_vars_for_htaccess(self, user_name, app_directory, env_vars, doc_root): """ Add enviroment variables to .htaccess file for LVEMAN-1623 :param str user_name: Name of unix user :param str app_directory: Application root directory :param dict env_vars: Dict of enviroment variables :param str doc_root: doc root of application :return: None """ env_section_begin = "# DO NOT REMOVE OR MODIFY. CLOUDLINUX ENV VARS CONFIGURATION BEGIN" env_section_end = "# DO NOT REMOVE OR MODIFY. CLOUDLINUX ENV VARS CONFIGURATION END" htaccess_file = self.get_htaccess_by_appdir(user_name, app_directory, doc_root, None) if htaccess_file is None: return with open(htaccess_file, "r", errors='surrogateescape') as htaccess: lines = [line for line in htaccess.read().split('\n') if line] if env_section_begin in lines and env_section_end in lines: start = lines.index(env_section_begin) end = lines.index(env_section_end) del lines[start: end + 1] if env_vars is not None: lines.append(env_section_begin) lines.append("<IfModule Litespeed>") for env_var, value in env_vars.items(): lines.append("SetEnv {0} {1}".format(env_var, value)) lines.append("</IfModule>") lines.append(env_section_end) with open(htaccess_file, "w", errors='surrogateescape') as htaccess: htaccess.write('\n'.join(lines)) def remove_app_from_config(self, user, app_dir): """ Removes application from config :param user: User name :param app_dir: Application directory :return: True if app is exists in config, False - if app is not exists in config """ config_data = self.get_user_config_data(user) try: delete_using_realpath_keys(user, app_dir, config_data) # write new config self.write_full_user_config_data(user, config_data) except KeyError: return False else: return True def replace_domain_in_configs(self, username, domain, new_domain, include_subdomains=False): """ Replace domain in config files when it is renamed. :param username: domain owner :param domain: previous name :param new_domain: name after rename :param include_subdomains: whether we should also rename subdomains :return: """ full_config = self.get_user_config_data(username) if not full_config: return # find application with domain for app, config in iteritems(full_config): if include_subdomains: match = re.search(r'(\.\|^)%s$' % domain, config['domain']) else: match = re.search(r'^%s$' % domain, config['domain']) if match is not None: config['domain'] = unicode('{}{}'.format( # Cut out old_domain config['domain'][:-len(domain)], new_domain)) self.write_full_user_config_data(username, full_config) @staticmethod def _find_config_files(user_name, app_directory, patterns=None): """ Return list of detected config files """ abs_path, rel_path = get_abs_rel(user_name, app_directory) return [config for config in patterns if os.path.exists(os.path.join(abs_path, config))] def update_htaccess_file(self, user_name, app_root, doc_root): """ Creates .htaccess file for application based on application config data :param str user_name: User's name :param str app_root: Application root :param doc_root: Document root for the domain :return: None """ app_config = self.get_app_config(user_name, app_root) user_dir = self._pwd.get_homedir(user_name) htaccess_file = self.get_htaccess_by_appdir(user_name, app_root, doc_root, app_config) new_lines = [ '{}\n'.format(clpassenger.HTACCESS_BEGIN), 'PassengerAppRoot "{}"\n'.format(os.path.join(user_dir, app_root)), 'PassengerBaseURI "/{}"\n'.format(app_config['app_uri']), ] new_lines.extend(self.get_interpreter_specific_passenger_lines( self.get_binary_path(user_name, app_root, user_dir), app_config)) passenger_log_file = app_config.get('passenger_log_file', None) if passenger_log_file: new_lines.append('PassengerAppLogFile "%s"\n' % passenger_log_file) new_lines.append(clpassenger.HTACCESS_END + '\n') # Append all existing lines new_lines.extend(file_readlines(htaccess_file, errors='surrogateescape')) new_lines = clpassenger.rm_double_empty_lines(new_lines) # write new .htaccess file_writelines(htaccess_file, new_lines, 'w', errors='surrogateescape') @staticmethod @abstractmethod def get_interpreter_specific_passenger_lines(binary_path, app_config): """ Return list of lines that needs to be added to htaccess and are specific to the interpreter """ raise NotImplementedError def get_binary_path(self, user, app_root, user_dir, binary_name=None): """ Return a path to the environment's interpreter binary Get interpreter path for application :param user: owner of the application :param app_root: app path relative to user home (app-root) :param user_dir: User's home directory :param binary_name: name of binary in virtual environemnt (python, npm, node) :return: path to interpreter binary in virtual environment """ version = self.get_interpreter_version_for_app(user, app_root) if binary_name is None: binary_name = self.BINARY_NAME return os.path.join(user_dir, self.VENV_DIR, app_root, version, 'bin', binary_name) def get_users_dict(self, username=None): """ Retrives info about user(s). :param str \| None username: Username to retrive information. :return: Dictionary with user info. Example: {'user1': pw_struct} """ if username is None: return self._pwd.get_user_dict() return {username: self._pwd.get_pw_by_name(username)} def get_user_config_data(self, user): """ Get all data from user's config :param user: name of unix user :return: json data from user's config as dictionary """ _user_config_data = {} user_config = self._get_path_to_user_config(user) if os.path.isfile(user_config): data = get_file_lines(user_config) joined_data = ''.join(data) try: _user_config_data = json.loads(joined_data) except (ValueError, TypeError): raise ClSelectExcept.WrongData('User config "{}" is broken'.format(self._USER_CONFIG)) return _user_config_data def get_app_config(self, user, app_dir): """ Retrieves full application config :param user: User name :param app_dir: Application directory :return: Application data as dictionary If None - No application data found in config """ user_config_data = self.get_user_config_data(user) try: return get_using_realpath_keys(user, app_dir, user_config_data) except KeyError: return None def get_app_domain(self, username, app_directory): """ Retrieves domain for provided user's application :param username: user name :param app_directory: application root directory :return str: application domain """ app_config = self.get_app_config(username, app_directory) return app_config['domain'] def get_app_uri(self, username, app_directory): """ Retrieves uri for provided user's application :param username: user name :param app_directory: application root directory :return str: application uri """ app_data = self.get_app_config(username, app_directory) return app_data['app_uri'] def get_app_startup_file(self, username, app_directory): """ Retrieves name of startup file for provided user's application :param username: user name :param app_directory: application root directory :return str: name of startup file of application """ app_data = self.get_app_config(username, app_directory) return app_data['startup_file'] def get_app_status(self, username, app_directory): """ Retrieves status for provided user's app_directory :param username: user name :param app_directory: application root directory :return str: status of application """ app_data = self.get_app_config(username, app_directory) return app_data['app_status'] def get_interpreter_version_for_app(self, username, app_directory): """ Retrieves interpreter version for provided user and application :param username: user name :param app_directory: application root directory :return str: major interpreter version """ app_data = self.get_app_config(username, app_directory) if app_data is None: raise ClSelectExcept.ConfigMissingError('Application config is missed ' 'for user: {}'.format(username)) return app_data['%s_version' % self.INTERPRETER] def _get_path_to_user_config(self, username): """ Get full path to user config ~/${_USER_CONFIG} :param username: name of unix user :return: full path to config """ user_home = self._pwd.get_homedir(username) application_config = os.path.join(user_home, self._USER_CONFIG) return application_config def _get_full_version_for_short(self, major_version): """ Retrieves full version for supplied major version. :return: str - Full version or None if full version not found """ full_version = self._manager.pkg.get_full_version(major_version) if full_version != major_version: return full_version return None def _add_single_user_app(self, users_data_dest_dict, user_pw_entry, app_root_dir, app_data): """ Add single application data to user_data_dest_dict :param users_data_dest_dict: Destination dictionary with application data :param user_pw_entry: User's passwd entry :param app_root_dir: Application root directory :param app_data: Application data :return: None """ if user_pw_entry.pw_name in users_data_dest_dict: # User already present in dict users_data_dest_dict[user_pw_entry.pw_name]["applications"][app_root_dir] = app_data else: # No such user in dict - add it user_data = { "homedir": user_pw_entry.pw_dir, "applications": {app_root_dir: app_data} } users_data_dest_dict[user_pw_entry.pw_name] = user_data def _add_detected_config_files_to_application(self, app_data, app_root_dir, user_pw_entry): """Add automatically-detected config files to user-defined list""" config_files_detected = self._find_config_files(user_pw_entry.pw_name, app_root_dir) merged_files = set(app_data['config_files'] + config_files_detected) app_data['config_files'] = list(merged_files) def _add_all_user_apps(self, user_pw_entry, user_data_dest_dict, user_app_data): """ Add all user's apps information to user_data_dest_dict :param user_pw_entry: User's passwd entry :param user_data_dest_dict: Destination dictionary with application data :param user_app_data: User's application data ([node\|python\|ruby]-selector.json file content as dictionary). :return: None """ for app_root_dir, app_data in iteritems(user_app_data): full_interpreter_version = self._get_full_version_for_short(app_data['%s_version' % self.INTERPRETER]) if full_interpreter_version is None or full_interpreter_version not in user_data_dest_dict: # Application's interpreter version is absent in available versions of # interpreters - skip application continue # We added section `users` because version has at least one application if 'users' not in user_data_dest_dict[full_interpreter_version]: user_data_dest_dict[full_interpreter_version]['users'] = {} # Manually add venv path to config in order to display it in lvemanager app_data['virtualenv'] = { 'activate_path': self.get_binary_path( user_pw_entry.pw_name, app_root_dir, user_pw_entry.pw_dir, binary_name='activate') } # we do not need this key (why?) del app_data['%s_version' % self.INTERPRETER] users_dict = user_data_dest_dict[full_interpreter_version]['users'] self._add_detected_config_files_to_application(app_data, app_root_dir, user_pw_entry) self._add_single_user_app(users_dict, user_pw_entry, app_root_dir, app_data) @staticmethod def _add_absent_passenger_log_file(user_config_dict): """ Append absent 'passenger_log_file' key with None value to each application :param user_config_dict: Sourse dictionary to modify :return: Modified dict with 'passenger_log_file' keys """ for app_root, app_data in iteritems(user_config_dict): if 'passenger_log_file' not in app_data: app_data['passenger_log_file'] = None user_config_dict[app_root] = app_data return user_config_dict def read_user_selector_config_json(self, user_homedir, uid, gid, euid=None): """ Read [python\|ruby\|node]-selector.json file from user's directory. :param euid: current effective uid :param user_homedir: user's home directory :param uid: uid for drop rights :param gid: gid for drop rights :return: Cortege contents_dict. None - there is no [python\|ruby\|node]-selector.json file in user's directory :raises BaseSelectorError if error. Exception contents: {'message': "File %(file)s read/parse error: %(error)s", 'context': {'file': node_json_path, 'error': 'some message'}} """ # don't do this like euid=os.geteuid() in method signature! if euid is None: euid = os.geteuid() json_config_path = os.path.join(user_homedir, self._USER_CONFIG) if not os.path.exists(json_config_path): raise AbsentFileError(json_config_path) try: if euid == 0: # reads file with drop rights # to prevent print error messages from secureio.read_file_secure directly to stdout secureio.SILENT_FLAG = True file_lines = secureio.read_file_secure(json_config_path, uid, gid, exit_on_error=False, write_log=False) else: # read file without dropping rights file_lines = get_file_lines(json_config_path) return self._add_absent_passenger_log_file(json.loads(''.join(file_lines))) except (IOError, OSError, TypeError, ValueError) as e: # [python\|node\|ruby]-selector.json is unreadable or have non-json format raise BaseSelectorError({'message': "File %(file)s read/parse error: %(error)s", 'context': {'file': json_config_path, 'error': str(e)}}) def get_htaccess_by_appdir(self, user, app_dir, doc_root, app_config=None): """ Retrieve .htaccess for user and app :param user: Username :param app_dir: App dir :param doc_root: Document root for selected domain :param app_config: Optional app configuration :return: .htaccess full path """ try: if app_config is None: app_config = self.get_app_config(user, app_dir) return os.path.join(doc_root, app_config['app_uri'], '.htaccess') except KeyError: return None def set_app_status(self, user, app_dir, new_status): """ Retrieves application status :param user: User name :param app_dir: Application directory :param new_status: New appication status :type new_status: str :return: None """ user_config_data = self.get_user_config_data(user) try: app_config = get_using_realpath_keys(user, app_dir, user_config_data) app_config['app_status'] = new_status except KeyError: return # Write new config to file self.write_full_user_config_data(user, user_config_data) def write_full_user_config_data(self, user, config_data): """ Write data to user's config :param user: name of unix user :param config_data: data in json format :return: None """ json_config_path = self._get_path_to_user_config(user) if os.geteuid() == 0 or os.getegid() == 0: raise ClSelectExcept.SelectorException( 'root should not write to user config') dumped_data = pretty_json(config_data) try: secureio.write_file_via_tempfile( content=dumped_data, dest_path=json_config_path, perm=0o644, suffix='_tmp', ) except (IOError, OSError, TypeError) as e: raise ClSelectExcept.UnableToSaveData( json_config_path, 'Could not write json user config ({})'.format(e)) def is_version_in_use(self, version): """ Returns True if specified NodeJS version is in use by any app of any user. It will stop search on the first match. :param version: NodeJS interpreter major version :return True \| False """ user_info = self.get_users_dict() for user_name, user_pw_entry in iteritems(user_info): try: # Get user's [python\|ruby\|node]-selector.json file as dictionary user_apps_data = self.read_user_selector_config_json(user_pw_entry.pw_dir, user_pw_entry.pw_uid, user_pw_entry.pw_gid) if user_apps_data is not None: for app in user_apps_data.values(): #pylint: disable=E1101 if app.get('%s_version' % self.INTERPRETER) == version: return True except AbsentFileError: # just skip this error, as it means than config file # is not created yet or it was removed by user pass except BaseSelectorError as e: self.write_string_to_log(e.message % e.context) # pylint: disable=exception-message-attribute return False def get_applications_users_info(self, user=None): """ Retrieves info about all installed NodeJS interpreters and user(s) applictions :param user: User name for read applictions. If None and current euid == 0 - all users will be processed. If current euid != 0 (called under some user), this argument will be ignored and only user with uid == euid will be processed :return: Dictionary with user(s) applications info :raises ClSelectExcept.NoSuchUser or BaseSelectorError """ try: user_info = self.get_users_dict(user) except secureio.clpwd.NoSuchUserException: raise ClSelectExcept.NoSuchUser(user) users_apps_info = self._get_interpreter_info() available_versions = users_apps_info['available_versions'] # Process all needed users for user_name, user_pw_entry in iteritems(user_info): try: # Get user's [python\|ruby\|node]-selector.json file as dictionary user_apps_data = self.read_user_selector_config_json(user_pw_entry.pw_dir, user_pw_entry.pw_uid, user_pw_entry.pw_gid) # User applications data was read successfully - process it self._add_all_user_apps(user_pw_entry, available_versions, user_apps_data) except AbsentFileError: # just skip this error, as it means than config file # is not created yet or it was removed by user pass except BaseSelectorError as e: # Error retrieving data from user's [python\|ruby\|node]-selector.json if os.geteuid() == 0: # we are root - write message to log # TODO: add errors logging of broken configs self.write_string_to_log(e.message % e.context) # pylint: disable=exception-message-attribute users_apps_info['warning'] = 'Some user\'s %s can\'t be read. ' \ 'Some user(s) data absent in output. Please see file %s for details' % \ (self._USER_CONFIG, self._LOG_FILE_NAME) else: # we are user - show exception raise e return users_apps_info def get_app_folders(self, username, app_root, chk_env=True, chk_app_root=True): """ Calculate, check exists and return application folders This method does not check that application exists in config. :raises: NoSuchUserException, MissingVirtualenvError, MissingAppRootError :return: tuple(app_root, app_venv) with absolute paths """ user_home = self._pwd.get_pw_by_name(username).pw_dir app_venv = os.path.join(user_home, self.VENV_DIR, app_root) if chk_env and not os.path.exists(app_venv): raise MissingVirtualenvError(app_venv) app_root = os.path.join(user_home, app_root) if chk_app_root and not os.path.exists(app_root): raise MissingAppRootError(app_root) return app_root, app_venv def _get_interpreter_info(self): """Get initial information about interpreter""" users_apps_info = self._manager.get_summary() return users_apps_info def acquire_interpreter_lock(self, it_version): """ Just a public proxy to internal method that blocks any actions with interpreter """ return self._manager.pkg.acquire_interpreter_lock(it_version)	25.21 KB	2026-02-26 15:33
📄 config.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 json import os from abc import ABCMeta, abstractmethod, abstractproperty from future.utils import iteritems from secureio import write_file_via_tempfile from clselect import utils from .pkgmanager import BasePkgManager # NOQA from . import BaseSelectorError, ENABLED_STATUS, DISABLED_STATUS from future.utils import with_metaclass class BaseSelectorConfig(with_metaclass(ABCMeta, object)): """ Base class that responsible for all interaction with CL selector config files """ def __init__(self, pkg): self.Cfg = self._get_config_object() self.pkg = pkg # type: BasePkgManager self.reload() @abstractproperty def _config_file(self): """Should return path to the config file""" raise NotImplementedError() @abstractmethod def _create_config_dirs(self): """Should create all needed directories for configs""" raise NotImplementedError() @staticmethod def _get_config_object(): """Override this method to change config parameters""" # Useful for IDE-level auto-completion and type checking class Cfg: # Defaults. None values means that it's not specified in config yet # and effective values depends on some logic in class properties default_version = None selector_enabled = None disabled_versions = None return Cfg @property def is_config_exists(self): """Check whether config file exists and is a regular file""" return os.path.isfile(self._config_file) def _dump(self): """ Returns underlying config as a plain dict. It will contain only explicitly configured options (e.g. no elements with None values) """ tmp = {} for k, v in iteritems(self.Cfg.__dict__): if not k.startswith('__') and v is not None: tmp[k] = v return tmp def _reset_cfg(self): """ Reset self.Cfg object to all None values before it will be loaded from file as a part of self.reload() """ for k, v in iteritems(self.Cfg.__dict__): if not k.startswith('__'): setattr(self.Cfg, k, None) def reload(self): data = self._read_file_data() if not data: return # No file or it's empty - nothing to load, use defaults try: tmp = json.loads(data) except (ValueError, TypeError) as e: raise BaseSelectorError('Unable to parse json from {} ; Error: {}' .format(self._config_file, e)) self._reset_cfg() for k, v in iteritems(tmp): setattr(self.Cfg, k, v) def _read_file_data(self): """ Should return: - whole file data for normal case - None if file doesn't exists - '' for empty file """ if not self.is_config_exists: return None try: with open(self._config_file, 'rb') as fd: data = fd.read() except (IOError, OSError) as e: raise BaseSelectorError('Unable to read data from {} ; Error: {}' .format(self._config_file, e)) return data def save(self): if not self.is_config_exists: self._create_config_dirs() data = utils.pretty_json(self._dump()) return self._write_file_data(data) def _write_file_data(self, data): try: write_file_via_tempfile( content=data, dest_path=self._config_file, perm=0o644, suffix='_tmp', ) except (IOError, OSError) as e: raise BaseSelectorError('Could not write system config ({})'.format(e)) def _ensure_version_installed(self, version): if version not in self.pkg.installed_versions: raise BaseSelectorError('Version "{}" is not installed' .format(version)) @property def selector_enabled(self): """Returns effective selector_enabled value""" if self.Cfg.selector_enabled is None: # Selector is disabled by default until explicitly enabled by admin return False return self.Cfg.selector_enabled and bool(self.pkg.installed_versions) @selector_enabled.setter def selector_enabled(self, value): if value and not self.pkg.installed_versions: raise BaseSelectorError( "It's not allowed to enable Selector when " "interpreter is not installed") self.Cfg.selector_enabled = value def get_default_version(self): # If unspecified - we still return None so Frontend can show this # somehow user-friendly return self.Cfg.default_version def set_default_version(self, version): if version is None: # We allow to reset to 'unspecified' state self.Cfg.default_version = None return if version in (self.Cfg.disabled_versions or []): raise BaseSelectorError( "It's not allowed to set disabled version as the default one") self._ensure_version_installed(version) self.Cfg.default_version = version def set_version_status(self, version, new_status): disabled_list = self.Cfg.disabled_versions if new_status == ENABLED_STATUS: if disabled_list is not None and version in disabled_list: disabled_list.remove(version) if len(disabled_list) == 0: self.Cfg.disabled_versions = None elif new_status == DISABLED_STATUS: if version == self.get_default_version(): raise BaseSelectorError("It's not allowed to disable currently " "default version") # We explicitly allow to disable even not installed versions too # for future usage if disabled_list is None: self.Cfg.disabled_versions = [version] else: if version not in disabled_list: disabled_list.append(version) else: raise BaseSelectorError('Unknown version status: "{}"' .format(new_status)) @abstractproperty def available_versions(self): raise NotImplementedError()	6.67 KB	2026-02-26 15:33
📄 pkgmanager.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 fcntl import os import contextlib import psutil import subprocess import simplejson as json # because of unicode handling from abc import ABCMeta, abstractmethod from time import time from . import ( INSTALLING_STATUS, REMOVING_STATUS, AcquireInterpreterLockError, ) from future.utils import with_metaclass from clcommon.utils import is_testing_enabled_repo from clcommon.group_info_reader import GroupInfoReader MAX_CACHE_AGE_SEC = 24 * 3600 class PkgManagerError(Exception): pass class BasePkgManager(with_metaclass(ABCMeta, object)): """ Class responsible for all interactions with Yum, interpreter versions installation/removal and gathering info about already installed versions """ _testing_repo_enabled_cache = None _config_dir = None _versions_info = None _yum_cmd = None _alt_names = None _redirect_log = None _install_cmd = None _remove_cmd = None @classmethod def run_background(cls, command): fnull = open(os.devnull, 'w') return subprocess.Popen( command, stdin=fnull, stdout=fnull, stderr=fnull, shell=True, executable='/bin/bash' ) @property def _testing_enabled(self): if self._testing_repo_enabled_cache is not None: return self._testing_repo_enabled_cache res = is_testing_enabled_repo() self._testing_repo_enabled_cache = res return res @property def _yum_cache_file(self): if self._testing_enabled: return os.path.join(self._config_dir, 'yum_cache.dat.testing_enabled') return os.path.join(self._config_dir, 'yum_cache.dat') def update_yum_cache(self): groups = GroupInfoReader.get_group_info(self._alt_names) groups = list(groups.keys()) with open(self._yum_cache_file, 'w') as f: for group in groups: f.write(f'{group}\n') def _read_yum_cache(self): """Return data from file or None if file is absent or outdated""" try: stat = os.stat(self._yum_cache_file) except OSError: return None if (time() - stat.st_mtime) > MAX_CACHE_AGE_SEC: return None return open(self._yum_cache_file).read() @staticmethod def _remove_silent(f): """ Silently remove file ignoring all errors """ try: os.remove(f) except (OSError, IOError): pass @property def installed_versions(self): """ Returns list of installed interpreter versions by scanning alt_node_dir and cache result. Cache also can be pre-filled at init time for testing/debugging purposes """ if self._versions_info is None: self._versions_info = self._scan_interpreter_versions() return list(self._versions_info.keys()) def get_full_version(self, maj): """ Should return full interpreter version for a particular major version or just fallback to given version if info is not available for any reason. This information is taken from the hash map populated during installed_packages scan. :param maj: Major interpreter version :return: Full interpreter version or Major if info is not available """ if self._versions_info is None: self._versions_info = self._scan_interpreter_versions() try: return self._versions_info[maj]['full_version'] except KeyError: return maj @property def _pid_lock_file(self): return os.path.join(self._config_dir, 'yum.pid.lock') @property def _cache_lock_file(self): return os.path.join(self._config_dir, 'yum_cache.pid.lock') def _write_yum_status(self, pid, version=None, status=None): """ :param pid: pid of Yum process :param version: interpreter version or None for "cache update" case :param status: what yum is currently doing(few predefined statuses) :return: None """ if not os.path.exists(self._config_dir): self._create_config_dirs() json.dump({ 'pid': pid, 'version': str(version), 'status': status, 'time': float(time()), }, open(self._pid_lock_file, 'w')) def _check_yum_in_progress(self): ongoing_yum = self._read_yum_status() if ongoing_yum is not None: return "{} of version '{}' is in progress. " \ "Please, wait till it's done"\ .format(ongoing_yum['status'], ongoing_yum['version']) def _read_yum_status(self): """ Result "None" - means installing/removing of our packages is not currently in progress. However, it doesn't mean that any other yum instance is not running at the same time, but we ok with this because our yum process will start processing automatically once standard /var/run/yum.pid lock is removed by other process :return: None or dict """ if self._pid_lock_file is None: raise NotImplementedError() try: data = json.load(open(self._pid_lock_file)) except Exception: # No file or it's broken: self._remove_silent(self._pid_lock_file) return None if not psutil.pid_exists(data.get('pid')): #pylint: disable=E1101 self._remove_silent(self._pid_lock_file) return None # TODO check timeout and stop it or just run with bash "timeout ..." try: pid, _ = os.waitpid(data['pid'], os.WNOHANG) except OSError: # Case when we exit before completion and yum process is no # longer our child process return data # still working, wait... if pid == 0: # still working, wait... return data self._remove_silent(self._pid_lock_file) return None # It was zombie and has already finished def format_cmd_string_for_installing(self, version): """ Formatting cmd string for installing package :return: formatted cmd string :param version: version of interpreter for installing :rtype: str """ return self._install_cmd.format(version) def format_cmd_string_for_removing(self, version): """ Formatting cmd string for removing package :return: formatted cmd string :param version: version of interpreter for removing :rtype: str """ return self._remove_cmd.format(version) def install_version(self, version): """Return None or Error string""" err = self._verify_action(version) if err: return err if version in self.installed_versions: return 'Version "{}" is already installed'.format(version) available = self.checkout_available() if available is None: return ('Updating available versions cache is currently ' 'in progress. Please, try again in a few minutes') if version not in available: return ('Version "{}" is not available. ' 'Please, make sure you typed it correctly'.format(version)) cmd_string = self.format_cmd_string_for_installing(version) p = self.run_background(cmd_string) self._write_yum_status(p.pid, version, INSTALLING_STATUS) def remove_version(self, version): """Return None or Error string""" err = self._verify_action(version) if err: return err if version not in self.installed_versions: return 'Version "{}" is not installed'.format(version) if self.is_interpreter_locked(version): return "This version is currently in use by another operation. " \ "Please, wait until it's complete and try again" if self._is_version_in_use(version): return "It's not possible to uninstall version which is " \ "currently in use by applications" cmd_string = self.format_cmd_string_for_removing(version) p = self.run_background(cmd_string) self._write_yum_status(p.pid, version, REMOVING_STATUS) def in_progress(self): """ Should return version and it's status for versions that is currently installing\|removing """ ongoing_yum = self._read_yum_status() if ongoing_yum is not None and \ ongoing_yum['status'] in (INSTALLING_STATUS, REMOVING_STATUS,): return { ongoing_yum['version']: { 'status': ongoing_yum['status'], 'base_dir': '', } } return None @contextlib.contextmanager def acquire_interpreter_lock(self, interpreter_version): lock_name = self._get_lock_file_path(interpreter_version) try: lf = open(lock_name, 'w') except IOError: raise AcquireInterpreterLockError(interpreter_version) try: fcntl.flock(lf, fcntl.LOCK_EX \| fcntl.LOCK_NB) except IOError: # TODO: try to use LOCK_SH here # It's ok if it's already locked because we allow multiple # operations for different applications at the same time # with the same "--new-version" pass try: yield finally: # Protection from exception in "context code" lf.close() @abstractmethod def checkout_available(self): raise NotImplementedError() @abstractmethod def _scan_interpreter_versions(self): raise NotImplementedError() @abstractmethod def _create_config_dirs(self): raise NotImplementedError() def is_interpreter_locked(self, interpreter_version): lock_name = self._get_lock_file_path(interpreter_version) if not os.path.isfile(lock_name): return False lf = open(lock_name, 'w') try: fcntl.flock(lf, fcntl.LOCK_EX \| fcntl.LOCK_NB) except IOError: return True finally: lf.close() return False @abstractmethod def _verify_action(self, version): raise NotImplementedError() def _get_lock_file_path(self, version): raise NotImplementedError() @abstractmethod def _is_version_in_use(self, version): raise NotImplementedError()	10.72 KB	2026-02-26 15:33
📄 selector_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 absolute_import from __future__ import print_function from __future__ import division import os from abc import ABCMeta from future.utils import iteritems from clselect.baseclselect import ( ENABLED_STATUS, NOT_INSTALLED_STATUS ) from .config import BaseSelectorConfig # NOQA from .pkgmanager import BasePkgManager # NOQA from future.utils import with_metaclass class BaseSelectorManager(with_metaclass(ABCMeta, object)): """Responsible for actual interpreter selector high-level API""" def __init__(self, cfg, pkg): """ :type cfg: config.BaseSelectorConfig :type pkg: pkgmanager.BasePkgManager """ self.pkg = pkg # type: BasePkgManager self.cfg = cfg # type: BaseSelectorConfig self.is_root_user = os.geteuid() == 0 @property def selector_enabled(self): return self.cfg.selector_enabled @selector_enabled.setter def selector_enabled(self, val): self.cfg.selector_enabled = val self.cfg.save() @property def default_version(self): return self.cfg.get_default_version() def switch_default_version(self, version): self.cfg.set_default_version(version) self.cfg.save() def set_version_status(self, version, status): """Disable/Enable particular interpreter version globally""" self.cfg.set_version_status(version, status) self.cfg.save() def is_version_enabled(self, version): """Check whether particular (installed) version is enabled""" enabled = [k for k, v in iteritems(self.cfg.available_versions) if v['status'] == ENABLED_STATUS] return version in enabled def install_version(self, version): return self.pkg.install_version(version) def uninstall_version(self, version): if version == self.cfg.get_default_version(): return "It's not allowed to uninstall default version" return self.pkg.remove_version(version) def get_summary(self, installed_interpreters_only=False): all_versions = self.cfg.available_versions available_to_install = [] if not installed_interpreters_only and self.is_root_user: # Only root can see "in progress" and "not installed yet" versions # We may change this in future if we have such requirement processed_now = self.pkg.in_progress() if processed_now: all_versions.update(processed_now) available_to_install = self.pkg.checkout_available() if available_to_install: available_to_install = set(available_to_install) installed_and_in_progress = set(all_versions.keys()) diff = { ver: {'base_dir': '', 'status': NOT_INSTALLED_STATUS} for ver in (available_to_install - installed_and_in_progress) } all_versions.update(diff) # Prettify output by converting Major version to Full versions # if possible all_versions = {self.pkg.get_full_version(k): v for k, v in iteritems(all_versions)} if "" in all_versions: all_versions.pop("") return { 'default_version': self.cfg.get_default_version(), 'available_versions': all_versions, # `Updating` means pkg info is not available right now and we # already started background process to rebuild that cache. # Once it finished status becomes `ready` again. # If we run as user it's always `ready` 'cache_status': ('updating' if available_to_install is None else 'ready'), 'selector_enabled': self.selector_enabled, }	3.94 KB	2026-02-26 15:33

Terminal