FM / clselector

/opt/cloudlinux/venv/lib/python3.11/site-packages/clselector/

/opt/cloudlinux/venv/lib/python3.11/site-packages/clselector

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
clpassenger_detectlib.py	1283	0644	edit dl rm
cl_selector.py	45766	0644	edit dl rm
cl_selector_arg_parse.py	15711	0644	edit dl rm
selectorlib.py	87955	0644	edit dl rm
__init__.py	408	0644	edit dl rm

Edit: /opt/cloudlinux/venv/lib/python3.11/site-packages/clselector/cl_selector.py (45766B)

# 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 re
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

# scanner-triage: this open() runs as the hosting user, not root. CloudlinuxSelector.run
        # reroutes --user/--domain python/nodejs invocations through run_process_in_cagefs
        # (cagefs_enter_user) before this method runs, and is_app_lock_needed excludes PHP/
        # Ruby, so a user-planted .lock symlink cannot cross a privilege boundary.
        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 != '']
                self._selector_lib.validate_config_file_names(config_files)
            # 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

# Allowed shape for an interpreter version token. The string is
    # concatenated into /opt/alt/<...>/<version>/.lock with no
    # canonicalisation, so any `/`, `..`, or NUL would let a less-privileged
    # caller steer the root-side open() outside /opt/alt/.
    _VERSION_TOKEN_RE = re.compile(r'^[A-Za-z0-9][A-Za-z0-9._-]*$')

@classmethod
    def _is_safe_version_token(cls, version):
        return isinstance(version, str) and bool(cls._VERSION_TOKEN_RE.match(version))

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():
            new_version = self._opts['--new-version']
            # Shape-check before the lock path is built and opened as root.
            # acquire_interpreter_lock() runs *before* run_process_in_cagefs
            # drops privileges, so a traversal token would land at an
            # attacker-chosen path under root.
            if not self._is_safe_version_token(new_version):
                self._error_and_exit({
                    'result': 'Invalid value for --new-version: %(version)s',
                    'context': {'version': new_version},
                })
            # 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(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)
        # CLOS-5541 (F-27): defence-in-depth cast at the sink. Upstream
        # get_major_version_from_short() coerces via
        # `str(int(v.split('.')[0]))` today, so these values are already
        # guaranteed numeric-looking strings — but the local `str(int(...))`
        # round-trip makes that guarantee explicit at the boundary where
        # the values start flowing into subprocess argv. If a future
        # upstream regression ever returned a raw un-coerced string,
        # ValueError fires here instead of a possibly-shell-interpretable
        # value reaching a subprocess call. Non-behavioural for correct
        # inputs (int repr of an already-numeric string is the same
        # digits). Keep the outer str() so callers that compare against
        # string version keys (e.g. `to_version not in installed_nj_versions`
        # or `app_info['nodejs_version'] == from_version`) continue to work.
        from_version = str(int(from_version))
        to_version = str(int(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
            # DiD (F-27, CLOS-5541, 2026-07-21): from_version/to_version are
            # already forced through get_major_version_from_short() (a
            # `str(int(...))` reducer) at L750-753 before we get here, so the
            # historic shell=True+bash form was safe. Convert to an argv list
            # + detached Popen anyway so the sink itself no longer relies on
            # the upstream sanitizer for injection safety, and future readers
            # (and scanners) don't have to trace the reducer to convince
            # themselves.
            cmd = [
                self._selector_lib.CLOUDLINUX_SELECTOR_UTILITY,
                'change-version-multiple', '--json',
                '--interpreter=nodejs',
                '--from-version=%s' % from_version,
                '--new-version=%s' % to_version,
                self._bkg_option,
            ]
            subprocess.Popen(
                cmd,
                stdin=subprocess.DEVNULL,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                close_fds=True,
                start_new_session=True,
            )
            # 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

cmd: