aDriv4 website!

Linux server64.web-hosting.com 4.18.0-513.18.1.lve.el8.x86_64 #1 SMP Thu Feb 22 12:55:50 UTC 2024 x86_64
LiteSpeed
198.54.116.248

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'],
                                                            self._opts.get('--website')))
        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(self._opts.get('--website')))
        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

Name	Type	Size	Owner/Group	Permission	Last Modified	Actions
.	dir	-	root/root	0755	2026-04-30 02:10:08
..	dir	-	root/root	0755	2026-05-04 07:16:41
__pycache__	dir	-	root/root	0755	2026-04-30 02:10:50
__init__.py	file	408 B	root/root	0644	2026-04-10 12:50:08
cl_selector.py	file	41.43 KB	root/root	0644	2026-04-10 12:50:08
cl_selector_arg_parse.py	file	15.34 KB	root/root	0644	2026-04-10 12:50:08
clpassenger_detectlib.py	file	1.25 KB	root/root	0644	2026-04-10 12:50:08
selectorlib.py	file	75.87 KB	root/root	0644	2026-04-10 12:50:08

For more Private tool come to me!

https://t.me/aDriv4vz