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 -- # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # from __future__ import absolute_import from __future__ import print_function from __future__ import division from .selectorlib import CloudlinuxSelectorLib __all__ = [ "CloudlinuxSelectorLib" ]	408 B	2026-02-26 15:33
📁 __pycache__	-	2026-03-21 15:48
📄 cl_selector.py # coding:utf-8 # license.py - work code for cloudlinux-license utility # # 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 fcntl import sys import time import errno import clcommon.cpapi as cpapi import contextlib import json import os import subprocess import traceback from typing import AnyStr # NOQA from future.utils import iteritems from clcommon import ClPwd from clcommon.clexception import FormattedException from clcommon.mail_helper import MailHelper from clcommon.clfunc import is_ascii_string from cllicense import CloudlinuxLicenseLib from clselect import clselectctl from clselect.utils import get_abs_rel, mkdir_p, run_process_in_cagefs from clselect.baseclselect import BaseSelectorError, AcquireApplicationLockError from cli_utils import print_dictionary, replace_params from clselect.clselectnodejs import CONFIG_DIR from clselect.clselectnodejs.pkgmanager import PkgManager from clselector.clpassenger_detectlib import is_clpassenger_active from collections import defaultdict from email.mime.text import MIMEText from tempfile import mkstemp from .cl_selector_arg_parse import NODEJS, PYTHON, PHP from .cl_selector_arg_parse import parse_cloudlinux_selector_opts from .selectorlib import CloudlinuxSelectorLib, OK_RES_DICT, ClSelectExcept from clselect.models import Website from clselect.clselectexcept import ClSelectExcept as ClSelectExcept_old LOCK = '.lock' # For unit tests def _open(file_name, mode): return open(file_name, mode) class CloudlinuxSelector(object): def __init__(self): self._is_json = False self._opts = {} self._selector_lib = None # For convenient checking during arg parsing and other operations. self._is_root_user = os.geteuid() == 0 self._lock = None self._is_bkg_option_present = False self._bkg_option = '--background' self._nj_ver_move_from = '' self._pid_file_name = os.path.join(CONFIG_DIR, 'cloudlinux-selector_bkg.pid') def is_app_lock_needed(self): """ Check if cloudlinux-selector called with application operations :return: True if lock is need """ # locking is implemented only for python and nodejs if self._opts['--interpreter'] not in [PYTHON, NODEJS]: return False if any([self._opts['change-version-multiple'], self._opts['create']]): return False if any([ self._opts['start'], self._opts['restart'], self._opts['destroy'], self._opts['migrate'], self._opts['stop'], self._opts['install-modules'], self._opts['uninstall-modules'], self._opts['run-script'], self._opts['--app-mode'], self._opts['--env-vars'], self._opts['--new-app-root'], self._opts['--new-domain'], self._opts['--new-app-uri'], self._opts['--new-version'], self._opts['--startup-file']]): return True return False def acquire_app_lock_if_needed( self, ignore_missing_app_root=False, ignore_missing_doc_root=False, ): """ Acquire lock for application if this lock is needed :return: None """ if not self.is_app_lock_needed(): return username, app_root = self._opts['--user'], self._opts['--app-root'] _, app_venv = self._selector_lib.apps_manager.get_app_folders( username, app_root, chk_app_root=not ignore_missing_app_root, chk_env=not ignore_missing_doc_root) if not os.path.exists(app_venv): return lock_file = os.path.join(app_venv, LOCK) try: self._lock = open(lock_file, 'a+') fcntl.flock(self._lock.fileno(), fcntl.LOCK_EX \| fcntl.LOCK_NB) except IOError as e: if e.errno == errno.EDQUOT: reason = 'Disk quota exceeded. Please, free space and try again.' raise AcquireApplicationLockError(app_root, reason=reason) raise AcquireApplicationLockError(app_root) def send_notification_if_needed(self): if self._is_root_user and self._opts['--new-version']: self.send_notification() def send_notification(self): # NOTE(vlebedev): As of now, email notifications about selector changes don't contain enough info to be useful. # Moreover, as of the moment of writing, these messages are plain wrong as they always mention # only NodeJS, not the actual Selector being changed. # An investigation is required to clarify whether this functionality is needed at all # and - if yes - what pieces of information should be supplied in such notifications. # For more info, have a look at Jira: # * https://cloudlinux.atlassian.net/browse/LVEMAN-1904 # * https://cloudlinux.atlassian.net/browse/LVEMAN-1903 return MSG_TEMP = "NodeJS version for your application %s was changed by admin. " \ "Please verify that application functions correctly." msg = MIMEText(MSG_TEMP % self._opts['--app-root']) me = 'CloudlinuxNodejsNotify@noresponse.com' msg['Subject'] = 'NodeJS version for your application %s was changed by admin' % self._opts['--app-root'] msg['From'] = me try: cp_userinfo = cpapi.cpinfo( self._opts['--user'], keyls=('mail', 'dns', 'locale', 'reseller'))[0] user_data_email = cp_userinfo[0] # user's email msg['To'] = user_data_email mailhelper = MailHelper() mailhelper.sendmail(me, [user_data_email], msg) except (IndexError, KeyError, cpapi.cpapiexceptions.NotSupported): # can't get user mail or mail corrupted pass @staticmethod def parse_modules(modules_options): if not modules_options: return () return [module for module in modules_options.strip().split(',') if module] def run(self, argv): """ Run command action """ self._is_json = "--json" in argv # Check background option self._is_bkg_option_present = self._bkg_option in argv if self._is_bkg_option_present: argv.remove(self._bkg_option) try: licence = CloudlinuxLicenseLib() if not licence.get_license_status(): self._is_json = True return self._error_and_exit({"result": "Cloudlinux license isn't valid"}) # get arguments, fill the value of --user argument if only --domain was given self._opts = self._parse_args(argv) self._selector_lib = CloudlinuxSelectorLib(self._opts['--interpreter']) self._selector_lib.check_selector_is_available() if self._selector_lib.should_be_runned_as_user(self._opts): with self._lock_interpreter_if_needed(): result = run_process_in_cagefs( self._opts['--user'], self._selector_lib.CLOUDLINUX_SELECTOR_UTILITY, argv, ) returncode = result['returncode'] self._print_raw_data(result['output']) self.send_notification_if_needed() return returncode elif self._selector_lib.should_run_user_without_cagefs(self._opts): user_run_cmd = ['/usr/bin/sudo', '-u', self._opts['--user'], self._selector_lib.CLOUDLINUX_SELECTOR_UTILITY] + argv with self._lock_interpreter_if_needed(): process = subprocess.Popen(user_run_cmd, env={}) process.communicate() self.send_notification_if_needed() return process.returncode self.acquire_app_lock_if_needed( ignore_missing_app_root=self._opts['destroy'], ignore_missing_doc_root=self._opts['destroy'], ) # ignore app root and doc root for destroy option if self._opts['--passenger-log-file']: # Passenger log filename passed, check it message, log_filename = self._passenger_log_filename_validator(self._opts['--user'], self._opts['--passenger-log-file']) if message == "OK": self._opts['--passenger-log-file'] = log_filename else: self._error_and_exit(dict(result=message)) if self._opts['set']: self.run_set() elif self._opts['migrate']: self.run_migrate_application() elif self._opts['import-applications']: self.run_import_applications() elif self._opts['create']: self.run_create() elif self._opts['destroy']: self.run_destroy() elif self._opts['start']: self.run_start() elif self._opts['restart']: self.run_restart() elif self._opts['stop']: self.run_stop() elif self._opts['read-config']: self.run_read_config() elif self._opts['save-config']: self.run_save_config() elif self._opts['install-modules']: self.run_install_modules() elif self._opts['uninstall-modules']: self.run_uninstall_modules() elif self._opts['install-version'] or self._opts['uninstall-version']: self.run_manage_version() elif self._opts['enable-version'] or self._opts['disable-version']: self.run_disable_or_enable_version() elif self._opts['run-script']: self._print_data( self._selector_lib.run_script( self._opts['--user'], self._opts['--app-root'], self._opts['--script-name'], self._opts['<script_args>'] ) ) elif self._opts['change-version-multiple']: self._start_change_all_apps_versions() elif self._opts['make-defaults-config']: self._selector_lib.replace_mysqli() elif self._opts['setup']: self.run_setup() else: self.run_get() except (ClSelectExcept_old.ConfigNotFound, ClSelectExcept_old.WrongData, ClSelectExcept_old.NoSuchAlternativeVersion) as e: self._error_and_exit(dict(result=str(e))) except (ClSelectExcept_old.NativeNotInstalled, ClSelectExcept_old.MissingCagefsPackage) as e: if not self._opts['make-defaults-config']: # pylint: disable=exception-message-attribute self._error_and_exit(dict(result=e.message, context=e.context)) # hack for alt-php spec that calls this method # just do not print error because it is not needed in rpm log exit(0) except ClSelectExcept_old.FileProcessError as e: self._error_and_exit(dict(result=e)) except FormattedException as e: if e.details: self._error_and_exit(dict(result=e.message, context=e.context, details=e.details)) else: self._error_and_exit(dict(result=e.message, context=e.context)) except Exception as err: msg = traceback.format_exc() list_err_msg = traceback.format_exception_only(type(err), err) if isinstance(list_err_msg, list): err_msg = '\n'.join(list_err_msg) else: err_msg = list_err_msg self._error_and_exit(dict( result=err_msg, details=msg )) finally: if self._is_bkg_option_present: # If we worked in background remove pid file try: os.remove(self._pid_file_name) except: pass return 0 def run_set(self): if self._opts['--default-version'] is not None: self._print_data(self._selector_lib.set_default_version(self._opts['--default-version'])) elif self._opts['--current-version'] is not None: self._print_data( self._selector_lib.set_current_version(self._opts['--current-version'], self._opts['--domain']) ) elif self._opts['--reset-extensions']: self._print_data(self._selector_lib.reset_extensions(self._opts['--version'], self._opts['--website'])) elif self._opts['--selector-status'] is not None: self._print_data(self._selector_lib.set_selector_status(self._opts['--selector-status'])) elif self._opts['--supported-versions'] is not None: self._print_data(self._selector_lib.set_supported_versions(self._opts['--supported-versions'])) elif self._opts['--extensions'] is not None and self._opts['--version'] is not None: self._print_data(self._selector_lib.set_extensions(self._opts['--extensions'], self._opts['--version'], self._opts.get('--website'))) elif self._opts['--options'] is not None and self._opts['--version'] is not None: self._print_data(self._selector_lib.set_options(self._opts['--options'], self._opts['--version'])) elif self._is_nodejs or self._is_python: self.run_change(self._opts['--user'], self._opts['--app-root'], self._opts['--app-mode'], self._opts['--env-vars'], self._opts['--new-app-root'], self._opts['--new-domain'], self._opts['--new-app-uri'], self._opts['--new-version'], self._opts['--startup-file'], self._opts['--skip-web-check'], self._opts['--entry-point'], self._opts['--config-files'], self._opts['--passenger-log-file']) # XXX: should we return some error if no option was selected? def run_setup(self): self._selector_lib.setup_selector() def run_change(self, user, app_root, app_mode, env_vars, new_app_root, new_domain, new_app_uri, new_version, startup_file, skip_web_check, entry_point, config_files, passenger_log_file): """ Call selectorctl to change application parameter :param config_files: names of config files (such as requirements.txt or etc) (only for python) :param entry_point: the specified entrypoint for application (only for python) :param user: application owner :param app_root: application main directory (application name) :param app_mode: application mode :param env_vars: dict with environment variables :param new_app_root: new application main directory (new application name) :param new_domain: new application domain :param new_app_uri: new application uri :param new_version: new version for nodejs interpreter :param startup_file: new startup file for application :param skip_web_check: skip check web application after change it's properties :param passenger_log_file: Passenger log filename :return: None """ if user is None: self._error_and_exit({ 'result': 'ERROR: User is not specified'}) if new_app_root is not None: # Change app-root r = self._selector_lib.relocate(user, app_root, new_app_root) # after relocate we need to change current app_root to new one app_root = new_app_root if r['status'].upper() != 'OK': self._print_data(r) sys.exit(1) if new_app_uri is not None or new_domain is not None: # Change app-uri r = self._selector_lib.transit(user, app_root, new_app_uri, new_domain) if r['status'].upper() != 'OK': self._print_data(r) sys.exit(1) if any((app_mode, env_vars, startup_file, entry_point, config_files is not None, passenger_log_file is not None)): # create list of config files if config_files is not None: config_files = [item for item in config_files.split(',') if item != ''] # Change app-mode, environment variables or startup file r = self._selector_lib.set_variables(user, app_root, app_mode, env_vars, startup_file, entry_point, config_files, passenger_log_file) if r['status'].upper() != 'OK': self._print_data(r) sys.exit(1) if new_version is not None: # Change interpreter version for application r = self._selector_lib.change_version(user, app_root, new_version, skip_web_check) if r['status'].upper() != 'OK': self._print_data(r) sys.exit(1) # print_data create {status:ok, timestamp:} and print it self._print_data({}) def run_import_applications(self): self._print_data(self._selector_lib.run_import_applications()) def run_migrate_application(self): self._print_data(self._selector_lib.run_migrate_application( self._opts['--user'], self._opts['--app-root'])) def run_get(self): if self._opts['--get-default-version']: self._print_data(self._selector_lib.get_default_version()) elif self._opts['--get-selector-status']: self._print_data(self._selector_lib.get_selector_status()) elif self._opts['--get-supported-versions']: self._print_data(self._selector_lib.get_supported_versions()) elif self._opts['--get-current-version']: self._print_data(self._selector_lib.get_current_version(self._opts['--user'], self._opts['--domain'])) elif self._opts['--interpreter'] == PHP: self._print_data(self._selector_lib.get_full()) else: res = {'passenger_active': is_clpassenger_active()} if self._opts['--interpreter'] == NODEJS: res.update(self._selector_lib.get_apps_users_info(self._opts['--user'])) # Applications count from background process remaining_apps_count, total_apps_count = self._get_apps_count_from_pid_file() if remaining_apps_count is not None and total_apps_count is not None: res['remaining_apps_count'] = remaining_apps_count res['total_apps_count'] = total_apps_count elif self._opts['--interpreter'] == PYTHON: res.update(self._selector_lib.get_apps_users_info(self._opts['--user'])) if 'result' in res: self._print_data(res, result=res['result']) else: self._print_data(res) def run_create(self): # Not allow to create application on locked version if self._is_version_locked_by_background_process(self._opts['--version']): self._error_and_exit({ 'result': 'Can\'t create application: Nodejs version %(version)s is locked by background process', 'context': {'version': self._opts['--version']}, }) if not is_clpassenger_active(): # passenger not active, application creation not allowed if self._opts['--interpreter'] == PYTHON: url = 'https://docs.cloudlinux.com/python_selector/#installation' else: url = 'https://docs.cloudlinux.com/index.html?installation.html' self._error_and_exit({ 'result': 'Application creation not allowed, ' 'Phusion Passenger seems absent, please see %(url)s for details', 'context': { 'url': url }, }) self._print_data( self._selector_lib.create_app( self._opts['--app-root'], self._opts['--app-uri'], self._opts['--version'], self._opts['--user'], self._opts['--domain'], self._opts['--app-mode'], self._opts['--startup-file'], self._opts['--env-vars'], self._opts['--entry-point'], self._opts['--passenger-log-file'] )) def run_destroy(self): self._print_data(self._selector_lib.destroy_app(self._opts['--app-root'], self._opts['--user'])) def run_start(self): self._print_data(self._selector_lib.start_app(self._opts['--app-root'], self._opts['--user'])) def run_restart(self): self._print_data(self._selector_lib.restart_app(self._opts['--app-root'], self._opts['--user'])) def run_stop(self): self._print_data(self._selector_lib.stop_app(self._opts['--app-root'], self._opts['--user'])) def run_read_config(self): self._print_data( self._selector_lib.read_app_config( self._opts['--app-root'], self._opts['--config-file'], self._opts['--user'])) def run_save_config(self): self._print_data( self._selector_lib.save_app_config( self._opts['--app-root'], self._opts['--config-file'], self._opts['--content'], self._opts['--user'])) def run_install_modules(self): self._print_data( self._selector_lib.install_modules( self._opts['--app-root'], user=self._opts['--user'], domain=self._opts['--domain'], skip_web_check=self._opts['--skip-web-check'], spec_file=self._opts['--requirements-file'], modules=self.parse_modules(self._opts['--modules']), ) ) def run_uninstall_modules(self): self._print_data( self._selector_lib.uninstall_modules( self._opts['--app-root'], modules=self.parse_modules(self._opts['--modules']), user=self._opts['--user'], domain=self._opts['--domain'], skip_web_check=self._opts['--skip-web-check'], ) ) def run_disable_or_enable_version(self): """ Disable or enable interpreter version :return: None """ version = self._opts['--version'] target_version_status = self._opts['enable-version'] try: self._print_data(self._selector_lib.set_version_status(target_version_status, version)) except BaseSelectorError as e: self._error_and_exit({ 'result': str(e), }) def run_manage_version(self): ver = str(self._opts['--version']) try: if self._opts['install-version']: res = self._selector_lib.selector_manager.install_version(ver) else: res = self._selector_lib.selector_manager.uninstall_version(ver) except Exception as e: res = str(e) if res is None: self._print_data(OK_RES_DICT) elif isinstance(res, dict): self._error_and_exit(res) else: self._error_and_exit({'result': res}) def _parse_args(self, argv): """ Parse CLI arguments """ status, data = parse_cloudlinux_selector_opts( argv, self._is_json, as_from_root=self._is_root_user) if not status: # exit with error if can`t parse CLI arguments self._error_and_exit(replace_params(data)) # For php we check only user exists if data['--interpreter'] == 'php': if data['--user']: try: pwd = ClPwd() pwd.get_pw_by_name(data['--user']) except ClPwd.NoSuchUserException: raise ClSelectExcept( { 'message': 'No such user (%s)', 'context': { 'user': data['--user'] }, } ) # Resolve domain and docroot for PHP extensions configuration website = None if data['--domain']: try: resolved_user, doc_root = CloudlinuxSelectorLib.safely_resolve_username_and_doc_root( data['--user'], data['--domain']) # Update user if it was resolved from domain if resolved_user: data['--user'] = resolved_user website = Website(domain=data['--domain'], docroot=doc_root) except ClSelectExcept: # If domain resolution fails, website will be None (per-user config) pass data['--website'] = website return data # We can't detect CPanel under user in CageFS, so we check CPanel specific directory /usr/local/cpanel # In cageFS this directory present, but we can't read it content # May be this is temporary solution, possibly change after PTCCLIB-170 if not os.path.isdir('/usr/local/cpanel') and (data['import-applications'] or data['migrate']): self._error_and_exit({'result': 'success', 'warning': 'Import/migrate of Python Selector applications is not supported'}) # try to resolve username (e.g. if only domain was specified in cli) # DO NOT RESOLVE DOMAIN HERE! # it leads to confusion between the "user's main domain" # and the "domain where application works" data['--user'], _ = CloudlinuxSelectorLib.safely_resolve_username_and_doc_root( data['--user'], data['--domain']) # validate app_root before passing it to create & transit methods # to make them 'safe' and avoid code duplicates for app_root_arg in ['--app-root', '--new-app-root']: if not data.get(app_root_arg): continue _, directory = get_abs_rel(data['--user'], data[app_root_arg]) try: # directory name must not be one of the reserved names and # should not contain invalid symbols. clselectctl.check_directory(directory) except ValueError as e: self._error_and_exit(dict( result=str(e) )) data[app_root_arg] = directory return data def _error_and_exit(self, message, error_code=1): """ Print error and exit :param dict message: Dictionary with keys "result" as string and optional "context" as dict """ if "status" in message: message["result"] = message["status"] del(message["status"]) if self._is_json: message.update({"timestamp": time.time()}) print_dictionary(message, True) else: try: print(str(message["result"]) % message.get("context", {})) except KeyError: print("Error: %s" % message) sys.exit(error_code) @staticmethod def _print_raw_data(data): # type: (AnyStr) -> None """ Print raw data. Function should be used in case if you want to print a json string as an output from other utilities """ print(data) def _print_data(self, data, force_json=False, result="success"): """ Output data wrapper :param: `dict` data - data for output to stdout :param: `bool` force_json - always output json format """ if isinstance(data, dict): data = data.copy() # data may be Exception object with data and context inside if "data" in data and isinstance(data["data"], dict): data = data["data"] # data may already contain "status", so we wont rewrite it data.setdefault("status", result) # rename "status": "ok" to "result": "success" if data["status"].lower() == "ok": data["result"] = "success" if self._opts['--interpreter'] == PHP and self._selector_lib.check_multiphp_system_default(): data['warning'] = 'MultiPHP system default PHP version is alt-php. ' \ 'PHP Selector does not work and should be disabled!' # do not set result to status, if result was passed elif 'result' not in data and 'status' in data: data["result"] = data["status"] del(data["status"]) # and do update timestamp with current time data.update({"timestamp": time.time()}) print_dictionary(data, self._is_json or force_json) @property def _is_nodejs(self): return self._opts['--interpreter'].lower() == NODEJS @property def _is_python(self): return self._opts['--interpreter'].lower() == PYTHON def _is_interpreter_lock_needed(self): # Only NodeJs & Python has interpreter locking if self._opts['--interpreter'] in [NODEJS, PYTHON]: # We will lock only new version because old is unknown before # we SU to user and read it's app configs. We can implement ugly # workaround later if someone ask it new_version = self._opts['--new-version'] return bool(new_version) return False @contextlib.contextmanager def _lock_interpreter_if_needed(self): """ Wrapper over contextmanager of PkgManager in order not to try acquire lock when it is not needed. """ if self._is_interpreter_lock_needed(): # TODO: we need to simplify access and usage # of apps_manager / pkg_manager methods mgr = self._selector_lib.apps_manager with mgr.acquire_interpreter_lock(self._opts['--new-version']): yield else: yield def _get_nj_versions(self): """ Retrives NodeJS versions from arguments and converts them to major versions :return: Cortege (from_version, to_version) """ from_version = self._opts['--from-version'] to_version = self._opts['--new-version'] from_version = self._selector_lib.get_major_version_from_short(from_version) to_version = self._selector_lib.get_major_version_from_short(to_version) if from_version == to_version: self._error_and_exit({'result': '--from-version and --new-version should be different'}) return from_version, to_version def _check_environment_for_move_apps(self): """ Checks arguments and environment before start group applications move :return: Cortege (from_version, to_version) """ from_version, to_version = self._get_nj_versions() pkg_manager = PkgManager() installed_nj_versions = pkg_manager.installed_versions if to_version not in installed_nj_versions: self._error_and_exit({ 'result': 'Can\'t move NodeJS applications to Nodejs version %(version)s. No such version installed.', 'context': {'version': to_version}, }) # For running process: print error if we trying to start background process and another one already running if not self._is_bkg_option_present and self._is_background_process_already_running(): self._error_and_exit({'result': 'Another background process already started.'}) return from_version, to_version def _start_change_all_apps_versions(self): """ Change all applications all users versions :return: """ from_version, to_version = self._check_environment_for_move_apps() # No background process running if not self._is_bkg_option_present: # Option --background not specified, start background process # For example: # cloudlinux-selector change-version-multiple --json --interpreter=nodejs --from-version=6 --new-version=9 --background command = "%s change-version-multiple --json --interpreter=nodejs --from-version=%s --new-version=%s %s >/dev/null &" %\ (self._selector_lib.CLOUDLINUX_SELECTOR_UTILITY, from_version, to_version, self._bkg_option) subprocess.run(command, shell=True, executable='/bin/bash') # Exit without process end waiting self._print_data(OK_RES_DICT) return # Option --background specified, start move application # Scan all users/apps, build appliction list to move users_apps_list, total_apps_count = self._get_all_apps_by_version(from_version) # Do nothing if application list is empty if not users_apps_list or total_apps_count == 0: return # Create pid file for background process self._write_pid_file(from_version, to_version, total_apps_count) # Move applications self._move_apps_by_list(users_apps_list, to_version, total_apps_count) def _move_apps_by_list(self, apps_dict, to_version, total_apps_count): """ Move applications from list from one NodeJS version to another :type dict :param apps_dict: Application list. List example: {'cltest1': [u'modjsapp_root'], 'cltest2': [u'app2', u'main_app']} :param to_version: Move applications to this version :param total_apps_count: Total applications count for move :return: None """ for user_name, user_app_list in iteritems(apps_dict): for app_root in user_app_list: # cloudlinux-selector set --json --interpreter nodejs --user <str> --app-root <str> --new-version <str> cmd = [ self._selector_lib.CLOUDLINUX_SELECTOR_UTILITY, 'set', '--json', '--interpreter', NODEJS, '--user', user_name, '--app-root', app_root, '--new-version', to_version ] process = subprocess.Popen(cmd) process.communicate() total_apps_count -= 1 # update pid file self._change_pid_file(total_apps_count) time.sleep(30) def _get_all_apps_by_version(self, from_version): """ Retrives list of all NodeJS applications for all users, which uses supplied version of NodeJS :param from_version: Required NodeJS version :return: Cortege: (application_list, application_count). Example: ({'cltest1': [u'modjsapp_root'], 'cltest2': [u'app2', u'main_app']}, 3) """ users_apps_dict = defaultdict(list) # 0 -- we always root here user_info = self._selector_lib.apps_manager.get_users_dict() total_apps_count = 0 for user_name, user_pw_entry in iteritems(user_info): try: user_app_data = self._selector_lib.apps_manager.read_user_selector_config_json( user_pw_entry.pw_dir, user_pw_entry.pw_uid, user_pw_entry.pw_gid, ) # user_app_data example: # {u'modjsapp_root': {u'domain': u'cltest1.com', u'app_uri': u'modjsappuri', u'nodejs_version': u'8', # u'app_status': u'started', u'env_vars': {}, u'app_mode': u'production', # u'config_files': [], u'startup_file': u'app.js'}} for app_root, app_info in iteritems(user_app_data): # if application on from_version - add it to list for move if app_info['nodejs_version'] == from_version: users_apps_dict[user_name].append(app_root) total_apps_count += 1 except (BaseSelectorError, TypeError, KeyError, AttributeError): # Skip user if config is unreadable continue return users_apps_dict, total_apps_count def _is_background_process_already_running(self): """ Determine is background process already working :return: True\|False """ try: data = json.load(_open(self._pid_file_name, 'r')) self._nj_ver_move_from = data['from_version'] return True except: pass # No background process found return False def _is_version_locked_by_background_process(self, nj_version): """ Checks if NodeJS version blocked by background operation :param nj_version: NodeJS version to check :return: True - version is locked, False - not locked """ if self._opts['--interpreter'] == PYTHON: return False # Check version and use default version if need nj_version = self._selector_lib.resolve_version(nj_version) nj_version = self._selector_lib.get_major_version_from_short(nj_version) is_bkg_process_present = self._is_background_process_already_running() if is_bkg_process_present and nj_version == self._nj_ver_move_from: return True return False def _write_pid_file(self, from_version, to_version, total_apps_count): """ Creates pid file for background process move version from version to version :param from_version: Move from NJ version :param to_version: Move to NJ version :param total_apps_count: Total application count to move :return: None """ json.dump({ 'pid': os.getpid(), 'from_version': str(from_version), 'to_version': str(to_version), 'total_apps_count': total_apps_count, 'remaining_apps_count': total_apps_count, 'time': float(time.time()), }, _open(self._pid_file_name, 'w')) # Make file readable by anyone os.chmod(self._pid_file_name, 0o644) def _read_pid_file(self): """ Reads pid file and returns it's content as dictionary :return: Dictionary """ f = _open(self._pid_file_name, 'r') pid_data = json.load(f) f.close() return pid_data def _change_pid_file(self, remaining_apps_count): """ Creates pid file for background process move version from version to version :param remaining_apps_count: Remaining application count to move :return: None """ try: pid_data = self._read_pid_file() pid_data['remaining_apps_count'] = remaining_apps_count _, temp_file_name = mkstemp(dir=CONFIG_DIR) json.dump(pid_data, _open(temp_file_name, 'w')) os.rename(temp_file_name, self._pid_file_name) # Make file readable by anyone os.chmod(self._pid_file_name, 0o644) except (OSError, IOError, KeyError): return def _get_apps_count_from_pid_file(self): """ Retrieves application counts from pid file :return: Cortege (remaining_apps_count, total_apps_count) If no background process started, returns None, None """ try: f = _open(self._pid_file_name, 'r') pid_data = json.load(f) f.close() return pid_data['remaining_apps_count'], pid_data['total_apps_count'] except (OSError, IOError, KeyError): return None, None @staticmethod def _passenger_log_filename_validator(username, log_filename): """ Validates passenger log file name :param username: User's name :param log_filename: passenger log file name to validate :return: tuple: (message, log_filename). message: "OK" - filename is valid, any other string - invalid, error text log_filename: corrected log filename - simlink dereferencing, appends user's homedir for relative paths, etc """ pwd = ClPwd() user_homedir = pwd.get_homedir(username) try: if not is_ascii_string(log_filename): return "ERROR: Passenger log filename should contain only english letters", None if os.path.isdir(log_filename): return "ERROR: Passenger log file should be a filename, not a directory name", None if not log_filename.startswith(os.path.sep): log_filename = os.path.join(user_homedir, log_filename) log_realpath = os.path.realpath(log_filename) if log_realpath.startswith(user_homedir+os.sep): dirname = os.path.dirname(log_realpath) if not os.path.exists(dirname): mkdir_p(dirname) return "OK", log_realpath except (OSError, IOError) as exc: return "%s" % str(exc), None return "ERROR: Passenger log file should be placed in user's home", None	41.31 KB	2026-02-26 15:33
📄 cl_selector_arg_parse.py # -- coding: utf-8 -- # Command line arguments parser for cloudlinux-selector utility # cloudlinux-license Utility to check/set Cloudlinux license # # 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 docopt import docopt from docopt import DocoptExit from schema import Schema, And, Use, Or, SchemaError from .selectorlib import CloudlinuxSelectorLib NODEJS = 'nodejs' PYTHON = 'python' RUBY = 'ruby' PHP = 'php' def _ensure_command_allowed(interpreter, args, as_from_root, selector_status): """ Do some additional checks to restrict commands not available for current user or interpreter or whatever and do this only after args parsing """ if not as_from_root and any([ args["install-version"], args["uninstall-version"], args["enable-version"], args["disable-version"], ]): raise SchemaError( None, 'This command should be run from administrator only') if interpreter != NODEJS and any([ args["change-version-multiple"], ]): raise SchemaError(None, 'This command is supported only for NodeJS') if interpreter != PYTHON and any([ args["import-applications"], args["migrate"], args["uninstall-modules"], ]): raise SchemaError(None, 'This command is supported only for Python') if interpreter not in (NODEJS, PYTHON) and any([ args["enable-version"], args["disable-version"], args["create"], args["read-config"], args["save-config"], args["install-version"], args["uninstall-version"], args["start"], args["restart"], args["stop"], args["destroy"], args["install-modules"], args["run-script"], ]): raise SchemaError(None, 'This command is supported only for NodeJS and Python') if interpreter != PHP and (args["make-defaults-config"] or args['setup']): raise SchemaError(None, 'This command is supported only for %s' % PHP) if not as_from_root and any([ args['make-defaults-config'], args['setup'], args['--selector-status'], args['--default-version'], args['--supported-versions'] ]): raise SchemaError(None, 'Specified option(s) only for root') if interpreter != PYTHON and any([ args["--entry-point"] ]): raise SchemaError(None, 'This options(s) only for Python') if interpreter in (NODEJS, PYTHON) and not args['get'] and not selector_status and not _run_from_admin(): raise SchemaError(None, 'Selector is disabled') def _run_from_admin(): """ Check who is owner of the parent process. if owner is root - return True if parent process can't be found - return True :return: """ try: return os.stat(os.path.join('/proc/', str(os.getppid()))).st_uid == 0 except OSError: return True def parse_cloudlinux_selector_opts(argv, _is_json_need=False, as_from_root=True): """ Parse arguments for cloudlinux-selector command :param argv: sys.argv :param _is_json_need: sys.argv contains --json key :param as_from_root: True if we assume that root has called this util :return cortege: (error_flag, s_message) """ # program name prog_name = "cloudlinux-selector" docstring = """Utility to get/set Cloudlinux selector options Usage: {0} [get] [--json] [--interpreter <str>] [(--get-supported-versions \| --get-default-version \| --get-selector-status)] [(--user <str> \| --domain <str>)] {0} get [--json] [--interpreter <str>] [--get-current-version] [--user <str>] [--domain <str>] {0} set [--json] [--interpreter <str>] (--selector-status <enabled,disabled> \| --default-version <str> \| --supported-versions <str> \| --current-version <str>) [--user <str>] [--domain <str>] {0} set [--json] [--interpreter <str>] --version <str> (--extensions <str> \| --options <str>) [(--user <str> \| --domain <str>)] {0} set [--json] [--interpreter <str>] [(--user <str> \| --domain <str>)] --app-root <str> [--app-mode <str>] [--new-app-root <str>] [--new-domain <str>] [--new-app-uri <str>] [--new-version <str>] [--startup-file <str>] [--env-vars <str>] [--skip-web-check] [--entry-point <str>] [--config-files <str>] [--passenger-log-file <str>] {0} create [--json] [--interpreter <str>] [(--user <str> \| --domain <str>)] --app-root <str> --app-uri <str> [--version <str>] [--app-mode <str>] [--startup-file <str>] [--entry-point <str>] [--env-vars <str>] [--passenger-log-file <str>] {0} set [--json] [--interpreter <str>] --reset-extensions --version <str> [(--user <str> \| --domain <str>)] {0} (enable-version \| disable-version) [--json] [--interpreter <str>] --version <str> {0} install-modules [--json] [--interpreter <str>] [(--user <str> \| --domain <str>)] --app-root <str> [(--requirements-file <str> \| --modules <str>)] [--skip-web-check] {0} uninstall-modules [--json] [--interpreter <str>] [(--user <str> \| --domain <str>)] --app-root <str> --modules <str> {0} install-version [--json] [--interpreter <str>] --version <str> {0} uninstall-version [--json] [--interpreter <str>] --version <str> {0} read-config [--json] [--interpreter <str>] [(--user <str> \| --domain <str>)] --app-root <str> --config-file <str> {0} save-config [--json] [--interpreter <str>] [(--user <str> \| --domain <str>)] --app-root <str> --config-file <str> --content <str> {0} (start \| restart \| stop \| destroy) [--json] [--interpreter <str>] [(--user <str> \| --domain <str>)] --app-root <str> {0} run-script [--json] [--interpreter <str>] [(--user <str> \| --domain <str>)] --app-root <str> --script-name <str> {0} run-script [--json] [--interpreter <str>] [(--user <str> \| --domain <str>)] --app-root <str> --script-name <str> -- <script_args>... {0} change-version-multiple --json --interpreter <str> --from-version <str> --new-version <str> {0} make-defaults-config [--json] [--interpreter <str>] {0} import-applications [--json] --interpreter <str> {0} migrate [--json] --interpreter <str> --user <str> --app-root <str> {0} setup [--json] --interpreter <str> {0} (-h \| --help) Options: --json Return data in JSON format. --interpreter <str> One of php/nodejs/python, default is php --get-supported-versions Return info about supported versions --get-current-version Return current version of interpretator for user --get-default-version Return info about default version only --get-selector-status Return info about selector status --reset-extensions Replace user extensions with version default extensions --supported-versions <str> Set supported versions of interpreter --default-version <str> Set default version of interpreter --current-version <str> Set alternative as user default --selector-status <enabled\|disabled> Set selector status enabled or disabled --version <str> Version of interpreter --extensions <str> JSON dict with extensions and their status --options <str> JSON dict with options and their values --user <str> Username to operate on --app-root <str> Root of an application to be created --domain <str> Domain to work in --app-uri <str> URI path to get the application being created --config-file <str> path to config file to be read or saved --config-files <str> names of config files (such as requirements.txt or etc) (only for python interpreter) --content <str> Base64-encoded config file contents to be saved --app-mode <str> Application mode --startup-file <str> Startup application file --env-vars <str> Environment variable as json string --new-app-root <str> Set new application directory --new-app-uri <str> Set new application uri --new-domain <str> Set new domain for application --new-version <str> Set new nodejs version for application --from-version <str> Old NodeJS version for group change version operations --script-name <str> Command for an npm script to be run --skip-web-check Skip check web application after change it's properties --entry-point <str> Use the specified entrypoint for application (only for python interpeter) --requirements-file <str> Use the specified file for install required modules --modules <str> Install comma-separated list of modules --passenger-log-file <str> Set passenger log full filename -h, --help Show this help message and exit """.format(prog_name) try: args = docopt(docstring, argv) except DocoptExit: s_error_string = 'ERROR: Invalid parameter passed' if not _is_json_need: s_error_string += "\n\n" + docstring return False, s_error_string interpreter = args.get('--interpreter') def _convert_version(x): """For the NodeJS, ignore all the version parts except the major one""" parts = x.split('.') if interpreter == NODEJS: return str(int(parts[0])) if interpreter == PYTHON: ver = x if len(parts) == 3: ver = x.rsplit('.', 1)[0] return ver return float(x) # keep old behavior for PHP _version_validator = Or(None, And(str, lambda x: x == "native"), And(str, Use(_convert_version)), error="Version must be Interpreter version number or native") def _json_string_to_dict(s_json): j_dict = json.loads(s_json) # json.loads in some cases returns a string instead dictionary. # So try to get keys to ensure that j_dict really dictionary list(j_dict.keys()) return j_dict s_json_error = '%s option should contain a valid JSON' s = Schema({ "get": bool, "set": bool, "import-applications": bool, "migrate": bool, "create": bool, "read-config": bool, "enable-version": bool, "disable-version": bool, "save-config": bool, "install-modules": bool, "uninstall-modules": bool, "start": bool, "restart": bool, "stop": bool, "destroy": bool, "install-version": bool, "uninstall-version": bool, "run-script": bool, "change-version-multiple": bool, "make-defaults-config": bool, "setup": bool, "<script_args>": Or(None, list), "--": bool, "--json": bool, "--help": bool, "--interpreter": Or(None, PHP, PYTHON, RUBY, NODEJS), "--get-supported-versions": bool, "--get-current-version": bool, "--get-default-version": bool, "--get-selector-status": bool, "--reset-extensions": bool, "--supported-versions": Or(None, And(str, Use(_json_string_to_dict)), error=s_json_error % "--supported-versions"), "--default-version": _version_validator, "--current-version": _version_validator, "--selector-status": Or(None, And(str, lambda x: x in ["enabled", "disabled"]), error="Selector status must be enabled or disabled"), "--version": _version_validator, "--extensions": Or(None, And(str, Use(_json_string_to_dict)), error=s_json_error % "--extensions"), "--options": Or(None, And(str, Use(_json_string_to_dict)), error=s_json_error % "--options"), "--user": Or(None, str), "--app-root": Or(None, str), "--domain": Or(None, str), "--app-uri": Or(None, str), "--config-file": Or(None, str), "--content": Or(None, str), "--app-mode": Or(None, str), "--startup-file": Or(None, And(str, lambda x: x != "package.json"), error='Cannot set "package.json" as startup file'), "--env-vars": Or(None, str), "--config-files": Or(None, str), "--new-app-root": Or(None, str), "--new-app-uri": Or(None, str), "--new-domain": Or(None, str), "--new-version": Or(None, str), "--from-version": Or(None, str), "--script-name": Or(None, str), "--entry-point": Or(None, str), "--requirements-file": Or(None, str), "--modules": Or(None, And(str, lambda x: bool(x)), error="modules should be a comma-separated list of packages"), "--skip-web-check": bool, "--passenger-log-file": Or(None, str) }) def _check_users_part_cli(_args): """ Check args for existing of mandatory arguments :param _args: parsed arguments from command line :return: True if checking passed, False - not passed """ if interpreter not in (NODEJS, PYTHON): return if as_from_root and any(( _args['create'], _args['read-config'], _args['save-config'], _args['install-modules'], _args['uninstall-modules'], _args['stop'], _args['restart'], _args['start'], _args['destroy'], _args['run-script'], )) and not (_args['--user'] or _args['--domain']): raise SchemaError(None, 'Domain or user argument is mandatory while calling selector under root') if args["--interpreter"] == PYTHON and args['--entry-point'] and not args['--startup-file']: raise SchemaError(None, '--entry-point option is requires --startup-file option for interpreter python') if args["--app-mode"] and args["--interpreter"] != NODEJS: raise SchemaError(None, '--app-mode option is requires only for interpreter nodejs') if args["--entry-point"] and args["--interpreter"] != PYTHON: raise SchemaError(None, '--entry-point option is requires only for interpreter python') try: args = s.validate(args) except SchemaError as e: return False, str(e) if not args['--json'] and not args['import-applications']: return False, "use --json option, other modes currently unsupported" if interpreter is None or interpreter in (PHP, ): args["--interpreter"] = PHP selector_status = False elif interpreter in (RUBY,): msg = 'ruby interpreter currently unsupported. Use selectorctl instead' return False, msg elif interpreter in (PYTHON, NODEJS): lib = CloudlinuxSelectorLib(args["--interpreter"]) selector_status = list(lib.get_selector_status().values())[0] try: _ensure_command_allowed(interpreter, args, as_from_root, selector_status) _check_users_part_cli(args) except SchemaError as e: return False, str(e) return True, args	15.34 KB	2026-02-26 15:33
📄 clpassenger_detectlib.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 glob import os def is_clpassenger_active(): """ Detect if passenger active on server. Detects: - CL passenger ea-apache24-mod-alt-passenger/alt-mod-passenger - CL EA4 passenger ea-ruby24-mod_passenger - CL EA4 passenger ea-ruby27-mod_passenger - alt-ruby27-mod_passenger (replaces ea-apache24-mod-alt-passenger since 6.1.2) - CentOS passenger :return bool: True/False - passenger found/not found """ # CentOS passenger binary orig_passenger_file = "/usr/bin/passenger" # List of CL passenger dirs passenger_dirs = glob.glob('/opt/passenger*') # EA4-ruby24 passenger is_ea4_ruby24_passenger = os.path.isdir('/opt/cpanel/ea-ruby24/root/usr/share/passenger') # EA4-ruby27 passenger is_ea4_ruby27_passenger = os.path.isdir('/opt/cpanel/ea-ruby27/root/usr/share/passenger') # alt-ruby27 passenger is_alt_ruby27_passenger = os.path.isdir('/opt/alt/ruby27/share/passenger') if (os.path.isfile(orig_passenger_file) or passenger_dirs or is_ea4_ruby24_passenger or is_ea4_ruby27_passenger or is_alt_ruby27_passenger): return True return False	1.43 KB	2026-02-26 15:33
📄 selectorlib.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 import base64 import collections import glob import json import os import pathlib import pwd import re import shutil import subprocess import sys import traceback from typing import Set # NOQA from future.utils import iteritems import cldetectlib as detect import clselect.clselectctlnodejsuser import clselect.clselectexcept import clselect.clselectctlpython import clselect.clselectctlruby import clselect.clselectpythonuser import clselect.clselectnodejsuser from clcommon.clexception import FormattedException from clcommon.clpwd import resolve_username_and_doc_root from clcommon.cpapi import CP_NAME, docroot from clcommon.cpapi.cpapiexceptions import NoDomain, NotSupported, IncorrectData from cllimits.lib import exec_utility from clselect import clselectctl from clselect import clpassenger from clselect import ClUserSelect, ClSelect, ClExtSelect from clselect.baseclselect import APP_STARTED_CONST, ENABLED_STATUS, DISABLED_STATUS, BaseSelectorError from clselect.clselectctlnodejsuser import validate_env_vars from clselect.clselectctlphp import format_summary, API_1 from clselect.clselectexcept import ClSelectExcept as ClSelectExceptions from clselect.clselectnodejs import NodeJSConfigError from clselect.models import Website from clselect.clselectnodejs.node_manager import NodeManager from clselect.clselectpython.apps_manager import ( PythonAppFormatVersion, get_venv_rel_path ) from clselect.clselectpython.python_manager import PythonManager from clselect.clselectpythonuser.environments import Environment as PythonEnvironment # NOQA from clselect.clselectnodejsuser.environments import Environment as NodeJsEnvironment # NOQA from clselect.utils import ( mkdir_p, file_read, file_write, get_using_realpath_keys, get_abs_rel, delete_using_realpath_keys, ) from secureio import get_perm from clconfig.ui_config_lib import _set_ui_config, UIConfigException class ClSelectExcept(FormattedException): pass class ClSelectDomainNotFound(ClSelectExcept): """ Custom exception in case if user doesn't have the specific domain """ pass OK_RES_DICT = {"status": "ok"} class CloudlinuxSelectorLib(object): def __init__(self, interpreter): self.interpreter = interpreter self._SELECTORCTL_UTILITY = "/usr/bin/selectorctl" self.DYNAMIC_UI_CTL_CMD = '/usr/share/l.v.e-manager/utils/dynamicui.py' self.CLOUDLINUX_SELECTOR_UTILITY = '/usr/sbin/cloudlinux-selector' self.NODEJS_INTERPRETER = "nodejs" self.PYTHON_INTERPRETER = "python" self.RUBY_INTERPRETER = "ruby" self.PHP_INTERPRETER = "php" # self.selector_manager - responsible for actual selector high-level API # self.apps_manager - responsible for gathering and set information about applications # self.selector_user_lib - set of modules which responsible user's part of work with virtual envs and etc # self.selector_old_lib - old libraries for work with applications self.selector_manager = None self.apps_manager = None self.selector_user_lib = None self.selector_old_lib = None if self.interpreter == self.NODEJS_INTERPRETER: from clselect.clselectnodejs.apps_manager import ApplicationsManager self.apps_manager = ApplicationsManager() self.selector_manager = NodeManager() self.selector_user_lib = clselect.clselectnodejsuser self.selector_old_lib = clselect.clselectctlnodejsuser elif self.interpreter == self.PYTHON_INTERPRETER: from clselect.clselectpython.apps_manager import ApplicationsManager self.apps_manager = ApplicationsManager() self.selector_manager = PythonManager() self.selector_user_lib = clselect.clselectpythonuser self.selector_old_lib = clselect.clselectctlpython elif self.interpreter == self.PHP_INTERPRETER: from clselect.clselectphp.php_manager import PhpManager self.selector_manager = PhpManager() def check_selector_is_available(self): """ Checks that selector is able to work on current os environment :return: """ if self.interpreter != self.PHP_INTERPRETER: return # check that native version is installed for php self.selector_manager.cl_select_lib.check_requirements() def safely_resolve_doc_root_for_app(self, username, app_root): """Get doc_root from application config or raise exception""" domain = self.apps_manager.get_app_domain(username, app_root) _, doc_root = self.safely_resolve_username_and_doc_root(username, domain) return doc_root @staticmethod def safely_resolve_username_and_doc_root(user=None, domain=None): """ Safely resolve username and doc_root by domain, or resolve document root by username, or resolve document root and username by effective uid :param user: str -> name of unix user :param domain: str -> domain of panel user :return: tuple -> user, doc_root """ try: result_user, result_doc_root = resolve_username_and_doc_root( user=user, domain=domain, ) except NoDomain: raise ClSelectDomainNotFound( { 'message': 'No such domain: %(domain)s' if domain is not None else 'No such user: %(user)s', 'context': { 'domain': domain, 'user': user, }, } ) except NotSupported: raise ClSelectExcept( { 'message': 'Nodejs selector not supported for %(panel)s', 'context': { 'panel': CP_NAME }, } ) except IncorrectData: raise ClSelectExcept( { 'message': 'Domain %(domain)s is not owned by the user %(user)s', 'context': { 'domain': domain, 'user': user }, } ) return result_user, result_doc_root @staticmethod def should_be_runned_as_user(opts): """ Check whether selector should be run through "su - user" :param opts: dict of parsed cli params :return: True if should be run through su or False if not """ result = True euid, egid = get_perm() if opts is None: result = False elif not isinstance(opts, dict): result = False elif opts['--user'] is None and opts['--domain'] is None: # if --user and --domain are absent - root result = False elif euid != 0 and egid != 0: result = False elif opts['--interpreter'] == 'php': result = False return result @staticmethod def should_run_user_without_cagefs(opts): return opts['--interpreter'] == 'php' and\ opts['--user'] is not None and\ os.geteuid() == 0 @staticmethod def _get_user_pwd_data(user=None): """ Resolves user eigher with passed username or with getting current user ID :param user: str or None -> username to be resolved :return: obj -> pwd user object """ userdata = pwd.getpwnam(user) return userdata @staticmethod def user_and_domain_checker(user=None, domain=None): """ Check if user and domain are None :param user: name of unix user :param domain: domain of panel user :return: None """ if user is None and domain is None: raise ClSelectExcept( { 'message': 'User or domain parameter must be specified if current user is root', 'context': {} } ) @staticmethod def _return_error(result, *kwargs): err = {"status": "ERROR: %s" % result} if len(kwargs) > 0: err.update(kwargs) return err @staticmethod def _return_with_status_error(result, details=None): """ Construct error dict in one place :param result: error string :return: dict with 'status':'error' and error message """ err = { 'status': 'error', 'result': result } if details: err.update({'details': details}) return err def _change_selector_status(self, status): """ Set CL selector status in it's config :param status: set status of selector :return: error or ok message """ if status not in (ENABLED_STATUS, DISABLED_STATUS,): return self._return_error( 'Unknown selector status provided: "{}". ' "Can be only 'enabled' or 'disabled'".format(status)) try: self.selector_manager.selector_enabled = status == ENABLED_STATUS except BaseSelectorError as e: return self._return_error(e) # Backward compatibility with cloudlinux-config on python if self.interpreter == self.PYTHON_INTERPRETER: # Create hidePythonApp dictionary config_dict = {'uiSettings': {'hidePythonApp': status != ENABLED_STATUS}} try: # _set_ui_config writes changes and updates UI _set_ui_config(config_dict) # We dont need to call self.update_ui after return OK_RES_DICT except UIConfigException: pass # Backward compatibility with cloudlinux-config on nodejs if self.interpreter == self.NODEJS_INTERPRETER: config_dict = {'uiSettings': {'hideNodeJsApp': status != ENABLED_STATUS}} try: _set_ui_config(config_dict) return OK_RES_DICT except UIConfigException: pass self.update_ui() return OK_RES_DICT @staticmethod def get_nodejs_selector_status(): res = {'selector_enabled': False} if NodeManager().selector_enabled: res['selector_enabled'] = True return res def get_php_selector_status(self): return {'PHPSelector': ENABLED_STATUS if self.selector_manager.selector_enabled else DISABLED_STATUS} def get_summary(self): try: res = self.selector_manager.get_summary() res.update(self.get_selector_status()) return res except BaseSelectorError as e: return self._return_error(e) @staticmethod def get_python_selector_status(): res = {'selector_enabled': False} if PythonManager().selector_enabled: res['selector_enabled'] = True return res @staticmethod def get_user_home(user): try: return pwd.getpwnam(user).pw_dir except KeyError: raise ClSelectExcept({ 'message': 'No such user: `%(user)s`' }) def get_nodejs_summary(self): res = NodeManager().get_summary() res.update(self.get_selector_status()) return res def run_script(self, user, app_root, script_name, script_args=None): """ Runs script to execute other script inside user app environment :param user: str -> owner of application :param app_root: str -> application directory :param script_name: str -> name of script :param script_args: list of str -> arguments for the script :return: dict """ user_home = self.get_user_home(user) if self.interpreter == self.NODEJS_INTERPRETER: interpreter_path = self.apps_manager.get_binary_path( user, app_root, user_home, 'npm') elif self.interpreter == self.PYTHON_INTERPRETER: interpreter_path = self.apps_manager.get_binary_path( user, app_root, user_home, 'python') else: raise NotImplementedError() cmd = [interpreter_path] if self.interpreter == self.NODEJS_INTERPRETER: cmd.append('run-script') cmd.append(script_name) if script_args is not None: if self.interpreter == self.NODEJS_INTERPRETER: cmd.append('--') cmd.extend(script_args) result = self._run_interpreter(cmd, user_home, app_root) return result @staticmethod def _run_interpreter(cmd, user_home, app_root): """ Run interpreter in users environment :param cmd: list -> command to execute :param user_home: -> user home directory :param app_root: -> app path :return: dict """ p = subprocess.Popen( args=cmd, cwd=os.path.join(user_home, app_root), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) try: stdout, stderr = p.communicate() except OSError as e: raise ClSelectExcept({ 'message': ('run-script call: `%(args)s` failed ' 'with error: %(err)s'), 'context': {'args': cmd, 'err': e} }) output_string = ( 'returncode: {}\n' 'stdout:\n{}\n' 'stderr:\n{}\n' ).format( p.returncode, stdout.strip(), stderr.strip()) if any('out of memory' in output.lower() for output in (stdout, stderr)): pid = os.getpid() proc_limits_path = f'/proc/{pid}/limits' process_limits = pathlib.Path(proc_limits_path).read_text() output_string += ( '\nOut of memory error may be caused by hitting LVE limits\n' 'or "Max data size", "Max address space" or "Max resident set" process limits\n' 'Please check LVE limits and process limits. Readjust them if necessary\n' 'More info: https://docs.cloudlinux.com/shared/cloudlinux_os_components/#known-restrictions-and-issues' f'\n\nprocess limits "{proc_limits_path}":\n{process_limits}\n' ) result = {'data': base64.b64encode(output_string.encode()).decode()} result.update({'status': 'success'}) if p.returncode != 0: result['warning'] = f'Script exit code: {p.returncode}' return result def get_full(self): if self.interpreter in (self.NODEJS_INTERPRETER, self.PHP_INTERPRETER): return self.selector_manager.get_summary() @staticmethod def _add_statistics_field(versions_list): """ Add selector usage statistics (amount of domains that use some version, etc) Fist parameter is an array with such format: [{'version': '5.6'}, {'version': '7.6'}] Output is an array with such format: [{'version': '5.6', 'users': 10}, ...] :type versions_list: list :rtype: list """ user_version_map = ClUserSelect().get_user_version_map() # count users per php version version_user_map = collections.Counter() for user, version in iteritems(user_version_map): version_user_map[version] += 1 # and add additional field to the output for item in versions_list: item['total_users'] = version_user_map[item['version']] return versions_list def get_supported_versions(self): """ Retrieves supported versions list and default version """ if self.interpreter == self.PHP_INTERPRETER: # ToDo: make get_supported_versions do not request all information (like it return PHP) data = ClSelect(self.interpreter).get_summary(False) json_data = format_summary(data, format='json', api_version=API_1) selectorctl_result = json.loads(json_data) # Our Interpreter-specific selectorctl commands should support API >= 1 elif self.selector_manager is not None: selectorctl_result = self.selector_manager.get_summary() else: selectorctl_result = {} if selectorctl_result.get('status') == 'ERROR': # e.g. {"status": "ERROR", "message": "alt-php packages not found"} result = {"status": selectorctl_result['message']} else: result = selectorctl_result return result def get_current_version(self, user=None, domain=None): """ Retrives current version for user """ user = [user] if user else None if self.interpreter == self.PHP_INTERPRETER: return self.selector_manager.get_current_version(user, domain) else: return self._return_error('Supported only by php selector') def get_default_version(self): json_dict = self.get_supported_versions() try: default_version = json_dict['default_version'] except KeyError: if 'message' in json_dict: return self._return_error(json_dict['message']) else: return {"status": "ERROR", "data": json_dict} return {'default_version': default_version} def get_selector_status(self): if self.interpreter == self.NODEJS_INTERPRETER: return self.get_nodejs_selector_status() elif self.interpreter == self.PYTHON_INTERPRETER: return self.get_python_selector_status() elif self.interpreter == self.PHP_INTERPRETER: return self.get_php_selector_status() def php_selector_is_disabled(self): return not self.selector_manager.selector_enabled def php_selector_is_enabled(self): return self.selector_manager.selector_enabled def check_multiphp_system_default(self): """ Returns True when MultiPHP system default PHP version is alt-php and PHP Selector is NOT disabled For details please see LVEMAN-1170 """ if self.interpreter != self.PHP_INTERPRETER: return False try: from clcagefslib.selector.configure import is_ea4_enabled, read_cpanel_ea4_php_conf except ImportError: return False if is_ea4_enabled() and not self.php_selector_is_disabled(): 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'] if not default_php.startswith('ea-php'): return True except KeyError: pass return False def set_supported_versions(self, versions): if self.interpreter in (self.NODEJS_INTERPRETER, self.PYTHON_INTERPRETER,): for version, enable_disable in iteritems(versions): try: result = self.set_version_status(enable_disable, version) except BaseSelectorError as e: return self._return_error(e) if result != OK_RES_DICT: return result return OK_RES_DICT json_dict = self.get_supported_versions() try: alternatives_list = json_dict["available_versions"] except KeyError: return self._return_error("corrupted answer from %s --json --summary --interpreter %s" % ( self._SELECTORCTL_UTILITY, self.interpreter)) alternatives_versions = {x["version"] for x in alternatives_list} if set(versions.keys()) - alternatives_versions: return self._return_error( "invalid alternative versions (%s), see %s --summary --interpreter %s for valid versions" % ( ', '.join(set(versions.keys()) - alternatives_versions), self._SELECTORCTL_UTILITY, self.interpreter)) # TODO: change error message success = 0 errors_list = [] for version, to_enable in iteritems(versions): if to_enable: ClSelect(self.interpreter).enable_version(str(version)) else: ClSelect(self.interpreter).disable_version(str(version)) success += 1 if success == len(versions): return OK_RES_DICT elif success > 0: return { "status": "WARNING: only {} of {} commands was successful. " "Errors was: {}".format(success, len(versions), errors_list) } else: return self._return_error("All commands were failed" "Errors was: {}".format(errors_list)) def set_default_version(self, version): try: if self.selector_manager is not None: self.selector_manager.switch_default_version(version) except clselect.clselectexcept.ClSelectExcept.NoSuchAlternativeVersion: # No such alt-php version raise ClSelectExcept({'message': "No such php version: %(ver)s", 'context': {'ver': version}}) except (clselect.clselectexcept.BaseClSelectException, NodeJSConfigError) as e: return self._return_with_status_error(str(e)) return OK_RES_DICT def set_version_status(self, target_version_status, version): """ Disable or enable version of selector :param target_version_status: disable or enable version of interpreter :param version: version of interpreter :return: OK_RES_DICT """ if target_version_status: self.selector_manager.set_version_status(version, ENABLED_STATUS) else: self.selector_manager.set_version_status(version, DISABLED_STATUS) return OK_RES_DICT def set_current_version(self, version, domain): try: if self.interpreter == self.PHP_INTERPRETER: self.selector_manager.switch_current_version(version, domain) except clselect.clselectexcept.ClSelectExcept.NoSuchAlternativeVersion: # No such alt-php version raise ClSelectExcept({'message': "No such php version: %(ver)s", 'context': {'ver': version}}) except clselect.clselectexcept.BaseClSelectException as e: return self._return_with_status_error(str(e)) return OK_RES_DICT def reset_extensions(self, version: str, website: Website \| None = None) -> dict: """Reset extensions to default for selected version.""" extensions = {} try: if self.interpreter == self.PHP_INTERPRETER: extensions = self.selector_manager.reset_extensions(version, website=website) except clselect.clselectexcept.ClSelectExcept.NoSuchAlternativeVersion: # No such alt-php version raise ClSelectExcept({'message': "No such php version: %(ver)s", 'context': {'ver': version}}) except clselect.clselectexcept.BaseClSelectException as e: return self._return_with_status_error(str(e)) return extensions def set_selector_status(self, status): if self.interpreter in (self.NODEJS_INTERPRETER, self.PYTHON_INTERPRETER, self.PHP_INTERPRETER): return self._change_selector_status(status) def update_ui(self): if detect.is_cpanel() or detect.is_plesk() or detect.is_da(): retcode, out, err = exec_utility( self.DYNAMIC_UI_CTL_CMD, ['--sync-conf=all'], stderr=True) if retcode != 0: return self._return_error( "Can not sync UI with reason: {} {}".format(out, err)) return None def set_extensions(self, extensions: dict[str, str], version: str, website: Website \| None = None) -> dict: """Set extensions for selected version.""" result = OK_RES_DICT.copy() if self.interpreter == self.PHP_INTERPRETER: try: data = self.selector_manager.set_extensions(version, extensions, website=website) if data: result.update(data) except clselect.clselectexcept.BaseClSelectException as e: return self._return_with_status_error(str(e)) return result def set_options(self, options, version): if self.interpreter == self.PHP_INTERPRETER: try: self.selector_manager.set_options(version, options) except clselect.clselectexcept.BaseClSelectException as e: return self._return_with_status_error(str(e)) return OK_RES_DICT def resolve_version(self, version): """ Attempts to get or verify version to be passed to external program Currently supported version is one digit (6 or 8). If version is None, return a default version :param version: str or None -> version to be verified or found :return: str -> digit as string """ if version is None: default_version = self.get_default_version().get('default_version', '') if default_version is None or default_version == '': # Interpreter default version not defined raise ClSelectExcept("{} default version not defined".format(self.interpreter)) if self.interpreter == self.NODEJS_INTERPRETER: m = re.match(r'(?P<version>\d+)', default_version) if not m: raise ClSelectExcept({'message': "Incorrect selector version: %(ver)s", 'context': {'ver': default_version}}) return m.group('version') else: return default_version if isinstance(version, (int, float)): if self.interpreter == self.NODEJS_INTERPRETER: # For NodeJS use only major version return str(int(version)) # TODO: check among supported versions else: # For Python use full version return str(version) if version == 'native': raise ClSelectExcept({'message': "Unsupported version: %(ver)s", 'context': {'ver': version}}) return version # TODO: do check among supported versions def create_app(self, app_root, app_uri, version, user=None, domain=None, app_mode=None, startup_file=None, env_vars=None, entry_point=None, passenger_log_file=None): """ Creates application for specified user, interpreter and version If user is None we hope that the external application resolves a user Currently NodeJS supported only :param domain: str -> domain of the application :param app_root: str -> app path relative to user home :param app_uri: str -> URI path of the application :param version: str or None -> version of the interpreter :param user: str or None -> username of user who owns the app :param app_mode: str or None -> application mode (development or production) :param startup_file: str or None -> main application file :param env_vars: json_string or None -> enviroment variables for application :param entry_point: Application entry point (used only for python interpreter). :param passenger_log_file: Passenger log filename :return: dict """ self.user_and_domain_checker(user, domain) version = self.resolve_version(version) if env_vars is not None: env_vars = validate_env_vars(json.loads(env_vars)) user, doc_root = self.safely_resolve_username_and_doc_root(user, domain) try: if self.interpreter == self.PYTHON_INTERPRETER: self.selector_old_lib.create(user, app_root, app_uri, version, doc_root=doc_root, env_vars=env_vars, startup_file=startup_file, domain_name=domain, entry_point=entry_point, apps_manager=self.apps_manager, passenger_log_file=passenger_log_file) elif self.interpreter == self.RUBY_INTERPRETER: clselect.clselectctlruby.create(user, app_root, app_uri, version, doc_root=doc_root) # ruby elif self.interpreter == self.NODEJS_INTERPRETER: # args[0] - directory, args[1] - alias (app-uri) # nodejs self.selector_old_lib.create( user, app_root, app_uri, version=version, doc_root=doc_root, app_mode=app_mode, env_vars=env_vars, startup_file=startup_file, domain_name=domain, apps_manager=self.apps_manager, passenger_log_file=passenger_log_file ) except clselect.clselectexcept.BaseClSelectException as e: return self._return_with_status_error(result=str(e)) # TODO: catch ClSelectExcept.BusyApplicationRoot return OK_RES_DICT @staticmethod def _get_application_path(app_root, userdata): """ Resolve app root to absolute path and checks if it exists :param app_root: str -> relative a user homedir app path :param userdata: obj -> pwd user object :return: str -> absolute path to app """ app_path = os.path.join(userdata.pw_dir, app_root) if not os.path.isdir(app_path): raise ClSelectExcept({'message': "No such application: %(app)s", 'context': {'app': app_root}}) return app_path def read_app_config(self, app_root, config_path, user=None): """ Reads file and returns its content as Base64 string :param app_root: str -> path to app relative to user home :param user: str -> username to resolve app path :param config_path: str -> file to be read (relative to app path) :return: dict """ result = OK_RES_DICT.copy() self.user_and_domain_checker(user, None) userdata = self._get_user_pwd_data(user) app_path = self._get_application_path(app_root, userdata) full_config_path = os.path.join(app_path, config_path) if not os.path.exists(full_config_path): raise ClSelectExcept( { 'message': "Configuration file not found: %(path)s", 'context': {'path': full_config_path} } ) try: with open(full_config_path, 'rb') as f: data = f.read() result.update({'data': base64.b64encode(data).decode()}) return result except Exception as e: return self._return_with_status_error(getattr(e, 'strerror', '')) def save_app_config(self, app_root, config_path, content, user=None): """ Saves data passed as Base64 string to specified file :param content: data for saving in app's config :param app_root: str -> path to app relative to user home :param user: str -> username to resolve app path :param config_path: str -> file to be read (relative to app path) :param content: str -> Base64-encoded string :return: dict """ self.user_and_domain_checker(user, None) userdata = self._get_user_pwd_data(user) app_path = self._get_application_path(app_root, userdata) full_config_path = os.path.join(app_path, config_path) try: with open(full_config_path, 'wb') as f: f.write(base64.b64decode(content)) return OK_RES_DICT except Exception as e: return self._return_with_status_error(getattr(e, 'strerror', '')) @staticmethod def _add_user_or_domain(user, domain, args): result_args = list(args) if domain is not None: result_args.extend(['--domain', domain]) elif user is not None: result_args.extend(['--user', user]) else: raise ClSelectExcept('User or domain parameter must be specified if current user is root') return result_args def uninstall_modules(self, app_root, modules, user=None, domain=None, skip_web_check=False): """ Uninstall described modules for user's webapp :param app_root: directory with webapp :param modules: comma-separated list of modules to uninstall :param user: name of unix user :param domain: domain of user :param skip_web_check: skip check web application after changing its properties :return: None """ self.user_and_domain_checker(user, domain) if self.interpreter != self.PYTHON_INTERPRETER: raise ClSelectExcept({ 'message': 'Uninstall command is available only for python interpreter, not %(interp)s', 'context': { 'interp': self.interpreter, }, }) try: for module in modules: self.selector_old_lib.uninstall(user, app_root, module) except clselect.clselectexcept.BaseClSelectException as e: return self._return_with_status_error(str(e)) else: return OK_RES_DICT def install_modules(self, app_root, user=None, domain=None, skip_web_check=False, spec_file=None, modules=()): """ Install described modules for user's webapp :type domain: domain of user :param user: name of unix user :param app_root: directory with webapp :param skip_web_check: skip check web application after change it's properties :param spec_file: file containing modules and their versions to install :param modules: list of installed modules :return: None """ self.user_and_domain_checker(user, domain) if self.interpreter == self.NODEJS_INTERPRETER: try: self.selector_old_lib.install(user, app_root, skip_web_check=skip_web_check, apps_manager=self.apps_manager) except clselect.clselectexcept.BaseClSelectException as e: return self._return_with_status_error(str(e)) else: return OK_RES_DICT elif self.interpreter == self.PYTHON_INTERPRETER: try: if modules: for module in modules: self.selector_old_lib.install(user, app_root, module, None, skip_web_check=skip_web_check, apps_manager=self.apps_manager) elif spec_file: self.selector_old_lib.install(user, app_root, None, spec_file, skip_web_check=skip_web_check, apps_manager=self.apps_manager) else: err = "Please, specify modules or requirements file with modules" return self._return_with_status_error(err) except clselect.clselectexcept.BaseClSelectException as e: return self._return_with_status_error(str(e)) else: return OK_RES_DICT else: raise ClSelectExcept({ 'message': 'Unknown interpreter: %(interp)s', 'context': { 'interp': self.interpreter, }, }) def destroy_app(self, app_root, user): """ Destroy specified application root directory and user name :param app_root: Application directory :param user: name of unix user :return: dict """ self.user_and_domain_checker(user, None) try: if self.interpreter == self.RUBY_INTERPRETER: clselect.clselectctlruby.destroy(user, app_root) else: try: doc_root = self.safely_resolve_doc_root_for_app(user, app_root) except TypeError: raise ClSelectExceptions.WrongData( 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 } ) except ClSelectDomainNotFound: # We still want to clean up an application doc_root = None if self.interpreter in (self.NODEJS_INTERPRETER, self.PYTHON_INTERPRETER): self.selector_old_lib.destroy(user, app_root, doc_root=doc_root, apps_manager=self.apps_manager) else: raise ClSelectExceptions.InterpreterError( message='Unknown interpreter: %(interp)s', context={'interp': self.interpreter}) except clselect.clselectexcept.BaseClSelectException as e: return self._return_with_status_error(str(e)) return OK_RES_DICT def start_app(self, app_root, username): """ Start specified application root directory and user name :param app_root: Application directory :param username: name of unix user :return: dict """ self.user_and_domain_checker(username, None) try: doc_root = self.safely_resolve_doc_root_for_app(username, app_root) if self.interpreter in (self.NODEJS_INTERPRETER, self.PYTHON_INTERPRETER): self.selector_old_lib.start(username, app_root, doc_root, self.apps_manager) else: raise ClSelectExceptions.InterpreterError( message='Unknown interpreter: %(interp)s', context={'interp': self.interpreter}) return OK_RES_DICT except clselect.clselectexcept.BaseClSelectException as e: return self._return_with_status_error(str(e)) def restart_app(self, app_root, username): """ Destroy specified application root directory and user name :param app_root: Application directory :param username: name of unix user :return: dict """ self.user_and_domain_checker(username, None) try: doc_root = self.safely_resolve_doc_root_for_app(username, app_root) if self.interpreter == self.NODEJS_INTERPRETER: self.selector_old_lib.restart(username, app_root, doc_root, self.apps_manager) elif self.interpreter == self.PYTHON_INTERPRETER: self.selector_old_lib.restart(username, app_root, doc_root, self.apps_manager) else: raise ClSelectExceptions.InterpreterError( message='Unknown interpreter: %(interp)s', context={'interp': self.interpreter}) return OK_RES_DICT except clselect.clselectexcept.BaseClSelectException as e: return self._return_with_status_error(str(e)) def stop_app(self, app_root, user): """ Start specified application root directory and user name :param app_root: Application directory :param user: name of unix user :return: dict """ self.user_and_domain_checker(user, None) try: doc_root = self.safely_resolve_doc_root_for_app(user, app_root) if self.interpreter == self.NODEJS_INTERPRETER: self.selector_old_lib.stop(user, app_root, doc_root, self.apps_manager) elif self.interpreter == self.PYTHON_INTERPRETER: self.selector_old_lib.stop(user, app_root, doc_root, self.apps_manager) else: raise ClSelectExceptions.InterpreterError( message='Unknown interpreter: %(interp)s', context={'interp': self.interpreter}) return OK_RES_DICT except clselect.clselectexcept.BaseClSelectException as e: return self._return_with_status_error(str(e)) @staticmethod def _replace_old_env_and_prompt_in_binaries_in_environment(new_env, old_env, new_rel, old_rel): # type: (str, str, str, str) -> None """ Replace old prompt and env_var in binaries in new environment Working with bytes here, because of python binary """ old_prompt = ('(' + old_rel + ':').encode() new_prompt = ('(' + new_rel + ':').encode() for venv_bin_file in glob.glob(os.path.join(new_env, '', 'bin', '*')): if not os.path.isdir(venv_bin_file): try: old_activate = file_read(venv_bin_file, mode='rb') if old_env.encode() in old_activate or old_prompt in old_activate: _new_activate = old_activate.replace(old_env.encode(), new_env.encode()) new_activate = _new_activate.replace(old_prompt, new_prompt) file_write(venv_bin_file, new_activate, 'wb') except IOError: _, _, traceback_ = sys.exc_info() sys.stderr.write(str(traceback.print_tb(traceback_))) def get_app_summary(self, username, app_config_dict_full, app_root): """ Retrieve application info from user's applications config. Analog of function clpassenger.summary :param username: User name :param app_config_dict_full: Full user's application config. :param app_root: Application root :return: Dictionary with application info Example: { 'binary': '/home/cltest1/virtualenv/new_app_root/2.7/bin/python', # + 'domain': 'cltest1.com', # + 'alias': 'app1', # + 'htaccess': '/home/cltest1/public_html/app1/.htaccess', # + 'interpreter': 'python', # + 'directory': '/home/cltest1/new_app_root', # + 'docroot': '/home/cltest1/public_html', # + 'domains': ['cltest1.com'] } """ try: app_config_dict = get_using_realpath_keys(username, app_root, app_config_dict_full) user_home = pwd.getpwnam(username).pw_dir doc_root = docroot(app_config_dict['domain'])[0] app_info_dict = { app_root: { 'interpreter': self.apps_manager.INTERPRETER, 'binary': self.apps_manager.get_binary_path(username, app_root, user_home), 'alias': app_config_dict['app_uri'], 'domain': app_config_dict['domain'], 'docroot': doc_root, 'htaccess': '/'.join([doc_root, app_config_dict['app_uri'], '.htaccess']), 'directory': '/'.join([user_home, app_root]), 'domains': [app_config_dict['domain']] } } return app_info_dict except KeyError: # we return empty dict because app doesn't exist return {} @staticmethod def _move_app_from_old_dir_to_new(old_directory, new_directory): # type: (str, str) -> None """ Move all items from old directory of application to new directory :param old_directory: full real path to old directory of applicaton :param new_directory: full real path to new directory of applicaton """ if not os.path.exists(new_directory): mkdir_p(new_directory) os.rename(old_directory, new_directory) else: # move all items from old directory to new for item in os.listdir(old_directory): shutil.move(os.path.join(old_directory, item), new_directory) os.rmdir(old_directory) def _relocate(self, user, old_directory, new_directory): """ Move user's application from directory to new_directory :param user: application owner. unix like user name :param old_directory: current directory with application :param new_directory: new directory for application :return: None """ full_config = self.apps_manager.get_user_config_data(user) try: app_config = get_using_realpath_keys(user, old_directory, full_config) except KeyError: raise ClSelectExceptions.NoSuchApplication( 'No such application (or application not configured) "%s"' % old_directory) old_abs, old_rel = get_abs_rel(user, old_directory) new_abs, new_rel = get_abs_rel(user, new_directory) try: # Directory name must not be one of the reserved names and # should not contain invalid symbols. clselectctl.check_directory(new_rel) except ValueError as e: raise ClSelectExceptions.WrongData(str(e)) # Get application summary for the application # Application summary example # {'new_app_root': # {'binary': '/home/cltest1/virtualenv/new_app_root/2.7/bin/python', # 'domain': 'cltest1.com', # 'alias': 'app1', # 'htaccess': '/home/cltest1/public_html/app1/.htaccess', # 'interpreter': 'python', # 'directory': '/home/cltest1/new_app_root', # 'docroot': '/home/cltest1/public_html', # 'domains': ['cltest1.com']} # } # TODO: why do we check only for applications of same type and not other (node/ruby/python)? new_user_summary = self.get_app_summary(user, full_config, new_rel) try: get_using_realpath_keys(user, new_rel, new_user_summary) except KeyError: pass else: raise ClSelectExceptions.AppRootBusy(new_abs) old_user_summary = self.get_app_summary(user, full_config, old_rel) try: old_user_app_summary = get_using_realpath_keys(user, old_rel, old_user_summary) except KeyError: raise ClSelectExceptions.WrongData("No such application (or application not configured) \"%s\"" % old_directory) doc_root = old_user_app_summary['docroot'] alias = old_user_app_summary['alias'] env_name = self.selector_old_lib._get_environment( user, old_directory, app_summary=old_user_summary[old_directory]).name if self.interpreter == self.PYTHON_INTERPRETER: ver, rel_venv = get_venv_rel_path(user, old_directory) _old_env, _ = get_abs_rel( user, rel_venv) _new_env, _ = get_abs_rel( user, get_venv_rel_path(user, new_directory, version=ver)[1]) elif self.interpreter == self.NODEJS_INTERPRETER: _old_env, _ = get_abs_rel(user, os.path.join( self.selector_user_lib.environments.DEFAULT_PREFIX, old_rel)) _new_env, _ = get_abs_rel(user, os.path.join( self.selector_user_lib.environments.DEFAULT_PREFIX, new_rel)) else: raise NotImplementedError() old_env = os.path.join(_old_env, '') new_env = os.path.join(_new_env, '') new_env_dir = os.path.dirname(_new_env) # needed to avoid copy in shutil.move if not os.path.exists(new_env_dir): os.makedirs(new_env_dir) shutil.move(old_env, new_env) self._move_app_from_old_dir_to_new(old_abs, new_abs) if self.interpreter == self.PYTHON_INTERPRETER: self._replace_old_env_and_prompt_in_binaries_in_environment(new_env, old_env, new_rel, old_rel) if self.interpreter == self.PYTHON_INTERPRETER: _, prefix = get_venv_rel_path(user, new_directory) elif self.interpreter == self.NODEJS_INTERPRETER: prefix = self.selector_old_lib._get_prefix(user, new_directory) else: raise NotImplementedError() environment = self.selector_user_lib.environments.Environment(env_name, user, prefix) binary = environment.interpreter().binary app_status = get_using_realpath_keys(user, old_directory, full_config)['app_status'] if app_status == APP_STARTED_CONST: # Clear .htaccess from CL's directives clpassenger.unconfigure(user, old_directory) passenger_log_file_to_set = full_config[old_directory].get('passenger_log_file', None) clpassenger.configure(user, new_directory, alias, self.interpreter, binary, doc_root=doc_root, startup_file=app_config['startup_file'], passenger_log_file=passenger_log_file_to_set) clpassenger.restart(user, new_directory) # update config delete_using_realpath_keys(user, old_directory, full_config) full_config[new_directory] = app_config self.apps_manager.write_full_user_config_data(user, full_config) def relocate(self, user, old_app_root, new_app_root): """ Call selectorctl to relocate application from old_app_root to new_app_root :param user: application owner :param old_app_root: current application directory (current application name) :param new_app_root: new application directory (new application name) :return: json """ try: if self.interpreter in (self.PYTHON_INTERPRETER, self.NODEJS_INTERPRETER): self._relocate(user, old_app_root, new_app_root) elif self.interpreter == self.RUBY_INTERPRETER: # last param in relocate not used clselect.clselectctlruby.relocate(user, old_app_root, new_app_root, None) except clselect.clselectexcept.BaseClSelectException as e: return self._return_with_status_error(str(e)) return OK_RES_DICT # TODO: we have something similar in clpassenger.move # one day we should remove one of that methods @staticmethod def _transit_htaccess_file(old_doc_root, old_alias, new_doc_root, new_alias): # type: (str, str, str, str) -> None """ :param old_doc_root: path to old doc root of application :param old_alias: old alias (uri) of application :param new_doc_root: path to new doc root of application :param new_alias: new alias (uri) of application :return: None """ # Copy existing .htaccess to new location htaccess = '.htaccess' # Get path to old .htaccess old_htaccess_file = os.path.join(old_doc_root, old_alias, htaccess) # Create path for new .htaccess new_htaccess_file = os.path.join(new_doc_root, new_alias, htaccess) if os.path.realpath(old_htaccess_file) \ == os.path.realpath(new_htaccess_file): return new_htaccess_path = os.path.dirname(new_htaccess_file) if not os.path.isdir(new_htaccess_path): os.makedirs(new_htaccess_path) shutil.copy(old_htaccess_file, new_htaccess_file) def _transit(self, user, directory, new_doc_root, new_domain, alias=None): """ Change application URI :param user: application owner. unix like user name :param directory: directory with application. (app-root) :param alias: new alias (app-uri) for application or None if change only the domain :param new_doc_root: NEW doc_root to transit application to :param new_domain: NEW domain to transit application to :return: None """ full_config = self.apps_manager.get_user_config_data(user) try: app_config = get_using_realpath_keys(user, directory, full_config) except KeyError: raise ClSelectExceptions.NoSuchApplication( 'No such application (or application not configured) "{}"'.format( directory)) apps_summary = self.get_app_summary(user, full_config, directory) try: old_app_summary = get_using_realpath_keys(user, directory, apps_summary) except KeyError: raise ClSelectExceptions.WrongData( 'No such application ' '(or application not configured) "{}"'.format(directory)) old_alias = old_app_summary['alias'] old_doc_root = old_app_summary['docroot'] new_alias = old_alias if alias is None else clselectctl.get_alias(alias) environment = self.selector_old_lib._get_environment( user, directory, app_summary=apps_summary[directory]) binary = environment.interpreter().binary if full_config[directory]['app_status'] == APP_STARTED_CONST: passenger_log_file_to_set = full_config[directory].get('passenger_log_file', None) clpassenger.configure(user, directory, new_alias, self.interpreter, binary, True, 'transit', doc_root=new_doc_root, startup_file=app_config['startup_file'], passenger_log_file=passenger_log_file_to_set) clpassenger.move(user, directory, old_alias, new_alias, old_doc_root=old_doc_root, new_doc_root=new_doc_root) clpassenger.restart(user, directory) else: # New doc root should be equal to old doc root # if we don't want to change domain for application. if new_doc_root is None: new_doc_root = old_doc_root self._transit_htaccess_file(old_doc_root, old_alias, new_doc_root, new_alias) app_config['app_uri'] = new_alias if new_domain is not None: app_config['domain'] = new_domain self.apps_manager.write_full_user_config_data(user, full_config) def transit(self, user, app_root, new_app_uri=None, new_domain=None): """ Call selectorctl to transit application to new_app_uri :param user: application owner :param app_root: application directory (application name) :param new_app_uri: new uri or None if change only the domain :param new_domain: new domain or None if change only the app_uri :return: json """ try: if new_domain is None: new_doc_root = None else: _, new_doc_root = self.safely_resolve_username_and_doc_root(user, new_domain) if self.interpreter in (self.PYTHON_INTERPRETER, self.NODEJS_INTERPRETER): self._transit(user, app_root, new_doc_root, new_domain, new_app_uri) elif self.interpreter == self.RUBY_INTERPRETER: clselect.clselectctlruby.transit(user, app_root, new_app_uri, doc_root=new_doc_root) else: raise NotImplementedError() except clselect.clselectexcept.BaseClSelectException as e: return self._return_with_status_error(str(e)) return OK_RES_DICT def _relocate_nodejs_extensions(self, user, app_root, new_env, old_extensions): # type: (str, str, NodeJsEnvironment, Set) -> None """ Install nodejs extensions to new nodejs environment and change symlink <user_homedir>/<app-root>/node_modules to new environment. Raise exception `WebAppError` if npm will return non-zero code """ npm_ret_code = 0 if old_extensions: try: npm_ret_code = new_env.extension_install_single_call(old_extensions) except ClSelectExceptions.FileProcessError: pass # Change symlink <user_homedir>/<app-root>/node_modules to new environment self.selector_old_lib._create_symlink_to_node_modules(user, new_env.path, app_root) if npm_ret_code != 0: raise ClSelectExceptions.WebAppError("Module installation has been failed. Please, check npm logs.") @staticmethod def _relocate_python_extensions(new_env, old_extensions): # type: (PythonEnvironment, Set) -> None """ Install python extensions to new python environment. They are equivalent to extensions from old environment. Remove python extensions which not existing in old environment from new environment """ new_extensions = set(new_env.extensions()) for extension in new_extensions - old_extensions: try: new_env.extension_uninstall(extension) except ClSelectExceptions.ExternalProgramFailed: # TODO: logging # https://cloudlinux.atlassian.net/browse/LVEMAN-1465 pass for extension in old_extensions - new_extensions: try: new_env.extension_install(extension) except ClSelectExceptions.ExternalProgramFailed: # TODO: logging # https://cloudlinux.atlassian.net/browse/LVEMAN-1465 pass def _change_version(self, user, directory, version=None, skip_web_check=False): """ Set current interpreter version for the application :param user: application owner. unix like user name :param directory: app_root - main directory with user application :param version: new version of python interpreter or None if we get current :param skip_web_check: skip check web application after change it's properties :return: None """ full_config = self.apps_manager.get_user_config_data(user) try: app_config = get_using_realpath_keys(user, directory, full_config) except KeyError: raise ClSelectExceptions.NoSuchApplication( 'No such application (or application not configured) "%s"' % directory) old_environment = self.selector_old_lib._get_environment(user, directory) # reads .htaccess if not version: return {old_environment.name: old_environment.interpreter().as_dict()} # SET new interpreter: new_environment = self.selector_old_lib._create_environment(user, directory, version) self._ensure_version_enabled(version, user) # Get extensions list for old environment installed_extensions = set(old_environment.extensions()) if self.interpreter == self.PYTHON_INTERPRETER: self._relocate_python_extensions(new_environment, installed_extensions) elif self.interpreter == self.NODEJS_INTERPRETER: self._relocate_nodejs_extensions(user, directory, new_environment, installed_extensions) # Reconfigure clpassenger user_summary_data = clpassenger.summary(user) app_summary = get_using_realpath_keys(user, directory, user_summary_data) doc_root = app_summary['docroot'] binary = new_environment.interpreter().binary alias, app_domain = self.selector_old_lib._get_info_about_webapp(app_summary, user) def action(): # Clear .htaccess clpassenger.unconfigure(user, directory) passenger_log_file_to_set = full_config[directory].get('passenger_log_file', None) clpassenger.configure(user, directory, alias, self.interpreter, binary, doc_root=doc_root, startup_file=app_config['startup_file'], passenger_log_file=passenger_log_file_to_set) # Create restart.txt file in tmp directory clpassenger.restart(user, directory) full_config[directory]['%s_version' % self.interpreter] = version self.apps_manager.write_full_user_config_data(user, full_config) if not skip_web_check: try: self.selector_old_lib.check_response_from_webapp( domain=app_domain, alias=alias, action=action, ) except ClSelectExceptions.WebAppError as err: raise ClSelectExceptions.WebAppError('An error occured during changing version. %s' % err) else: action() def change_version(self, user, app_root, new_version, skip_web_check): """ Call selectorctl to change current interpreter version to new_version for application :param user: application owner :param app_root: application directory (application name) :param new_version: new nodejs interpreter version :param skip_web_check: skip check web application after change it's properties :return: json """ try: if self.interpreter not in (self.NODEJS_INTERPRETER, self.PYTHON_INTERPRETER): raise NotImplementedError # forbid user to change python version unless he migrates application # to new python selector version if self.interpreter == self.PYTHON_INTERPRETER: app_version = self.apps_manager.get_app_config(user, app_root).get( 'app_version', PythonAppFormatVersion.LEGACY) if app_version == PythonAppFormatVersion.LEGACY: return { 'status': 'This application was created by too old version ' 'of python selector and we cannot change version ' 'without migration to the new application format. ' 'To do that you can use `cloudlinux-selector migrate` ' 'command or just click button in web UI.', } if self.apps_manager.get_app_status(user, app_root) == APP_STARTED_CONST: self._change_version(user, app_root, new_version, skip_web_check=skip_web_check) else: # Supplied application is stopper - run special function for change interpreter version for it self._change_version_for_stopped_app(user, app_root, new_version) return OK_RES_DICT except clselect.clselectexcept.BaseClSelectException as e: return self._return_with_status_error(str(e)) def set_variables_for_litespeed(self, user, app_root, env_vars): if self.interpreter == self.PYTHON_INTERPRETER or self.interpreter == self.NODEJS_INTERPRETER: doc_root = self.safely_resolve_doc_root_for_app(user, app_root) # need for add_env_vars_for_htaccess try: self.apps_manager.add_env_vars_for_htaccess(user, app_root, env_vars, doc_root) except Exception as err: raise ClSelectExceptions.WebAppError( "Unable to set environment variables in htaccess file for the application." "Error: {}".format(err) ) def _set_variables(self, user, directory, app_mode, env_vars, startup_file, entry_point, config_files, passenger_log_file): """ Set application mode, environment variables and startup_file for application :param config_files: names of config files (such as requirements.txt or etc) :param entry_point: the specified entrypoint for application :param user: application owner. unix like user name :param directory: directory with application :param app_mode: expected application mode :param env_vars: dict with environment variables for application :param startup_file: main file for application :param passenger_log_file: Passenger log filename :return: None """ full_config = self.apps_manager.get_user_config_data(user) try: app_config_data = get_using_realpath_keys(user, directory, full_config) except KeyError: raise ClSelectExceptions.NoSuchApplication( 'No such application (or application not configured) "%s"' % directory) if self.interpreter == self.PYTHON_INTERPRETER \ and env_vars is not None \ and app_config_data['env_vars'] != env_vars \ and app_config_data['app_version'] == PythonAppFormatVersion.LEGACY: raise ClSelectExceptions.WebAppError( "Unable to set environment variables. " "Application was created too long time ago. " "Please, migrate your application to newer version " "before changing interpreter version" ) # Update user's config if app_mode is not None: app_config_data['app_mode'] = app_mode if env_vars is not None: # Python env vars task # TODO: LVEMAN-1466 app_config_data['env_vars'] = env_vars if entry_point is not None: app_config_data['entry_point'] = entry_point if config_files is not None: app_config_data['config_files'] = config_files if startup_file is not None and startup_file != app_config_data.get('startup_file') or \ passenger_log_file is not None: if startup_file is not None and startup_file != app_config_data.get('startup_file'): # Startup file changing app_config_data['startup_file'] = startup_file if self.interpreter == self.PYTHON_INTERPRETER: startup_file_full_path = self.selector_old_lib._get_full_path_to_startup_file( user, directory, startup_file ) # We are using the main startup file with name `passenger_wsgi.py` in case custom name of # startup file, because passenger doesn't support directive for set entry_point. # In tje file `passenger_wsgi.py` we can set entry_point and custom name of startup file. self.selector_old_lib.setup_wsgi(user, directory, startup_file_full_path, entry_point) if passenger_log_file is not None: # Set/remove PassengerAppLogFile # Remove passenger log from app config if passenger_log_file path is empty app_config_data['passenger_log_file'] = None if passenger_log_file == '' else passenger_log_file env = self.selector_old_lib._get_environment(user, directory) user_summary = clpassenger.summary(user) user_app_summary = get_using_realpath_keys(user, directory, user_summary) alias = user_app_summary['alias'] binary = env.interpreter().binary doc_root = user_app_summary['docroot'] htaccess_path = user_app_summary['htaccess'] clpassenger._unconfigure(htaccess=htaccess_path) # If main file is not configured, use default value if startup_file is None and app_config_data.get('startup_file') is None: clpassenger.configure(user, directory, alias, self.interpreter, binary, doc_root=doc_root, passenger_log_file=passenger_log_file) else: clpassenger.configure(user, directory, alias, self.interpreter, binary, doc_root=doc_root, startup_file=app_config_data.get('startup_file'), passenger_log_file=passenger_log_file) clpassenger.restart(user, directory) self.apps_manager.write_full_user_config_data(user, full_config) self.set_variables_for_litespeed(user, directory, env_vars) def set_variables(self, user, app_root, app_mode, env_vars, startup_file, entry_point, config_files, passenger_log_file): """ Call selectorctl to set variables for application :param config_files: names of config files (such as requirements.txt or etc) (only for python) :param entry_point: the specified entrypoint for application (only for python) :param user: application owner :param app_root: application directory (application name) :param app_mode: application mode :param env_vars: json_string with environment variables for application :param startup_file: main file for application :param passenger_log_file: Passenger log filename :return: json """ if env_vars is not None: try: env_dict = validate_env_vars(json.loads(env_vars)) except (TypeError, ValueError): return self._return_with_status_error('wrong json format for environment variable list') else: env_dict = None try: if self.interpreter in (self.NODEJS_INTERPRETER, self.PYTHON_INTERPRETER) and \ self.apps_manager.get_app_status(user, app_root) == APP_STARTED_CONST: self._set_variables(user, app_root, app_mode, env_dict, startup_file, entry_point, config_files, passenger_log_file) else: # Supplied application is stopped - run special function for change a few variables of application self._set_variables_for_stopped_app(user, app_root, app_mode, env_dict, startup_file, entry_point, config_files, passenger_log_file) return OK_RES_DICT except clselect.clselectexcept.BaseClSelectException as e: return self._return_with_status_error(str(e)) def get_apps_users_info(self, user=None): """ Retrieves info about all installed interpreters and user(s) applictions :param user: User name for read applictions. If None all users will be processed :return: Dict with info """ try: result_dict = self.apps_manager.get_applications_users_info(user) return result_dict except BaseSelectorError as e: return {'result': e.message, 'context': e.context} # pylint: disable=exception-message-attribute def _ensure_version_enabled(self, new_version, username): """ Check whether particular interpreter version is enabled and raises exception if not :param username: user to include in exception :param new_version: new interpreter version """ if not self.selector_manager.is_version_enabled(new_version): raise clselect.ClSelectExcept.UnableToSetAlternative(username, new_version, 'version is not enabled') def _change_version_for_stopped_app(self, username, app_root, new_version): """ Changes version for stopped application :param username: application owner :param app_root: application directory (application name) :param new_version: new nodejs interpreter version :return: None """ self._ensure_version_enabled(new_version, username) # Get extensions list fom old environment old_version = self.apps_manager.get_interpreter_version_for_app(username, app_root) old_environment = self.selector_old_lib._create_environment(username, app_root, old_version, None) old_extensions = set(old_environment.extensions()) # Create new environment new_environment = self.selector_old_lib._create_environment(username, app_root, new_version, None) # install extension to new app for extension in old_extensions: try: new_environment.extension_install(extension) except clselect.clselectexcept.ClSelectExcept.ExternalProgramFailed: pass # Update user's app config app_config = self.apps_manager.get_user_config_data(username) app_config[app_root]['%s_version' % self.interpreter] = new_version self.apps_manager.write_full_user_config_data(username, app_config) def _set_variables_for_stopped_app(self, username, app_root, app_mode, env_vars_dict, startup_file, entry_point, config_files, passenger_log_file): """ Sets new app_mode, environment variables and startup file for stopped NodeJS application :param config_files: names of config files (such as requirements.txt or etc) (only for python) :param entry_point: the specified entrypoint for application (only for python) :param str username: application owner :param str app_root: application directory (application name) :param str app_mode: New application mode, can be None :param dict env_vars_dict: New environment variables, can be None :param startup_file: New startup file, can be None :param passenger_log_file: Passenger log filename :return: None """ # Update user's app config app_config = self.apps_manager.get_user_config_data(username) if app_mode: app_config[app_root]['app_mode'] = app_mode if env_vars_dict: app_config[app_root]['env_vars'] = env_vars_dict if startup_file and startup_file != app_config[app_root].get('startup_file'): app_config[app_root]['startup_file'] = startup_file if entry_point: app_config[app_root]['entry_point'] = entry_point if config_files: app_config[app_root]['config_files'] = config_files if passenger_log_file is not None and passenger_log_file != '': app_config[app_root]['passenger_log_file'] = passenger_log_file else: app_config[app_root]['passenger_log_file'] = None self.apps_manager.write_full_user_config_data(username, app_config) self.set_variables_for_litespeed(username, app_root, env_vars_dict) @staticmethod def get_major_version_from_short(version): """ Retrieves major version from full. If already short, return it with no difference :param version: Full/short :return: Short version as string """ return str(int(version.split('.')[0])) @staticmethod def replace_mysqli(): """ Replace mysqli extension to nd_mysqli for defaults. Warning: only for PHP. See LVEMAN-1399 for details :return: """ # Get available alt-php versions list. # For example: ['5.1', '5.2', '5.3', '5.4', '5.5', '5.6', '4.4', '7.2', '7.0', '7.1'] alt_php_versions_list = list(ClSelect('php').get_all_alternatives_data().keys()) cl_ext_select = ClExtSelect() # php by default for alt_php_ver in alt_php_versions_list: # Replace mysqli -> nd_mysqli for the version for new installations according to LVEMAN-1399 cl_ext_select.list_extensions(alt_php_ver) @classmethod def setup_selector(cls): """ Setup php selector for work (suggested to use after native php is installed) """ subprocess.check_output(['cagefsctl', '--force-update']) subprocess.check_output(['cagefsctl', '--remount-all']) subprocess.check_output(['/usr/sbin/cloudlinux-selector', 'make-defaults-config', '--json', '--interpreter', 'php']) subprocess.check_output(['/usr/share/l.v.e-manager/utils/cache_phpdata.py']) cls.replace_mysqli() def run_import_applications(self): """ Scan users home dirs for .htaccess files and import applications to new config file. """ if self.interpreter != self.PYTHON_INTERPRETER: raise NotImplementedError # We don't need to import apps for DA elif detect.is_da(): return OK_RES_DICT try: self.apps_manager.import_legacy_applications_to_config() return OK_RES_DICT except Exception as e: return self._return_with_status_error(str(e)) def run_migrate_application(self, user, app_root): """ Convert applications created in older selector versions to new format """ if self.interpreter != self.PYTHON_INTERPRETER: raise NotImplementedError("Migration is only available " "for python selector") self.apps_manager.migrate_application(user, app_root) return OK_RES_DICT	75.73 KB	2026-02-26 15:33

Terminal