Server : ApacheSystem : Linux host44.registrar-servers.com 4.18.0-513.18.1.lve.2.el8.x86_64 #1 SMP Sat Mar 30 15:36:11 UTC 2024 x86_64 User : vapecompany ( 2719) PHP Version : 7.4.33 Disable Function : NONE Directory : /opt/cloudlinux/venv/lib/python3.11/site-packages/clselect/ |
Upload File : |
# -*- coding: utf-8 -*-
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
from __future__ import absolute_import
from __future__ import print_function
from __future__ import division
import os
import pwd
import re
from collections import namedtuple
import logging
import traceback
import cldetectlib as detect
from typing import List # NOQA
from future.utils import iteritems
from . import clpassenger
from clcommon.cpapi import cpusers, getCPName
from clcommon.clpwd import drop_privileges, ClPwd
from clcommon.cpapi.plugins.cpanel import is_no_php_binaries_on_cpanel
from .clselectexcept import ClSelectExcept
from .clselectnodejs.node_manager import NodeManager
from .clselectpython.python_manager import PythonManager
from .utils import get_abs_rel
from typing import Optional, List # NOQA
DISABLED_DIRECTORY_NAMES = ['public_html', 'rubyvenv', 'virtualenv', '.cl.selector', '.cagefs', 'www', 'nodevenv']
ApplicationSummary = namedtuple('ApplicationSummary', ('user', 'version', 'doc_root', 'interpreter', 'app_status'))
InterpreterSummary = namedtuple('InterpreterSummary', ('version', 'version_full', 'enabled', 'installed'))
def check_directory(directory):
if ' ' in directory:
raise ValueError('Directory should not contain spaces')
if os.path.isabs(directory):
raise ValueError('Directory should be relative to user\'s home')
if directory[-1] == '/':
raise ValueError('Directory should not contain a trailing slash')
if directory in DISABLED_DIRECTORY_NAMES:
raise ValueError('Directory "%s" not allowed' % directory)
all_valid = re.match(r'[-\.\w\/]+$', directory)
if all_valid is None:
raise ValueError('Directory name contains invalid characters')
def get_alias(alias):
for c in ('#', '?', './'):
if c in alias:
raise ValueError('Alias is not valid')
# root aliases
if alias in ('.', ''):
alias = '/'
return alias.strip('/')
def get_directory(prefix):
return prefix.replace('_', '/').replace('//', '_')
def get_prefix(directory):
return directory.replace('_', '__').replace('/', '_')
def get_user(user):
if not user:
current_euid = os.geteuid()
user = pwd.getpwuid(current_euid).pw_name
if user == 'root':
raise ValueError('User parameter must be specified if current user is root')
return user
def _verify_application(interpreter, app_root, binary_path=None):
"""
Application is valid only if binary and app_root exists
:param binary_path: path to binary in virtual environment
:param app_root: path to root directory of application
:return: result of checking
:rtype: bool
"""
if interpreter in ('nodejs', 'python'):
return os.path.isdir(app_root)
return os.path.isfile(binary_path) and os.path.isdir(app_root)
def server_applications_summary(interpreter):
# type: (str) -> List[ApplicationSummary]
"""Find and return all apps for given interpreter on server"""
domains_data = None
applications = []
# LVEMAN-1408
euid = os.geteuid()
for user in cpusers():
try:
if detect.is_da():
# NB: python will import this only 1 time
from clcommon.cpapi.plugins.directadmin import userdomains
domains_list = userdomains(user, as_root=True), # called as root to save time on right's escalated call
domains_data = list(
filter(
None,
map(
lambda x: x[0] if x else None, # take only domain from (domain, docroot) tuples
domains_list,
)
)
)
# drop permissions in order not to
with drop_privileges(user):
# LVEMAN-1408: Force check user existence after drop user rights
pwd.getpwnam(user)
applications.extend(
_user_applications_short_summary(user, interpreter, domains_data))
except ClPwd.NoSuchUserException:
continue
except KeyError:
# LVEMAN-1408: this may be thrown by pwd.getpwnam in clpassenger.py module.
# It uses python pwd library instead ClPwd
# Send debug message to sentry
logger = logging.getLogger(__name__)
logger.error("DEBUG: user %s present in panel but doesn't exist in system. Process euid=%d. Trace: %s at %s" %
(user, euid, traceback.format_exc(), str(traceback.extract_stack())))
return applications
def _user_applications_short_summary(user, interpreter, domains_docroots_data=None):
"""
Return generator with all applications for given user and interpreter.
To increase performance, only a small part of the
information about the applications is given.
:param user: name of panel user
:param interpreter: name of interpreter (python, ruby, etc)
:param domains_docroots_data: total data about users domains
:return: Generator[ApplicationSummary]
"""
userdomains_data = domains_docroots_data
if interpreter in ('nodejs', 'python'):
try:
if interpreter == 'nodejs':
from .clselectnodejs.apps_manager import ApplicationsManager
else:
from .clselectpython.apps_manager import ApplicationsManager
config_data = ApplicationsManager().get_user_config_data(user)
except ClSelectExcept.WrongData:
return
for app, data in iteritems(config_data):
try:
app_root, _ = get_abs_rel(user, app)
except ClSelectExcept.WrongData:
continue
if _verify_application(interpreter, app_root):
yield ApplicationSummary(
user=user,
version=data['%s_version' % interpreter],
doc_root=data['domain'],
interpreter=interpreter,
app_status=data['app_status']
)
else:
for dummy, data in iteritems(clpassenger.summary(user, userdomains_data=userdomains_data)):
if data['interpreter'] != interpreter:
continue
# return only really existing applications
binary, app_root = data['binary'], data['directory']
if _verify_application(interpreter, app_root, binary):
it_version = os.path.basename(os.path.dirname(os.path.dirname(binary)))
yield ApplicationSummary(user=user, version=it_version, doc_root=data['docroot'],
interpreter=data['interpreter'], app_status='started')
def get_default_version(interpreter):
"""Return default version for given interpreter """
if interpreter == 'nodejs':
return NodeManager().default_version
if interpreter == 'python':
return PythonManager().default_version
return None
def _iter_interpreters(interpreter):
"""Return generator for interpreters all versions"""
if interpreter == 'nodejs':
interpreters = NodeManager().get_summary(installed_interpreters_only=True)
elif interpreter == 'python':
interpreters = PythonManager().get_summary(installed_interpreters_only=True)
else:
raise NotImplementedError()
for version, version_info in iteritems(interpreters['available_versions']):
enabled = version_info['status'] == 'enabled'
if interpreter == 'nodejs':
# for nodejs 6.16.0 -> 6
major_version = str(int(version.split('.')[0]))
else:
# for python 2.7.16 -> 2.7
major_version = '.'.join(version.split('.')[:2])
yield InterpreterSummary(version=major_version, version_full=version, enabled=enabled, installed=True)
def _iter_php_interpreters():
"""
Return generator with all PHP versions on server.
:return: Generator[InterpreterSummary]
"""
from .clselect import ClSelect
php = ClSelect()
versions = php.list_alternatives()
for version, full_version, _ in versions:
yield InterpreterSummary(
version, full_version, installed=True,
enabled=php.is_version_enabled(version))
def interpreter_versions_short_summary(interpreter):
# type: (str) -> List[InterpreterSummary]
"""Find and return all versions for given interpreter on server"""
if interpreter == 'php':
# if there is no installed php - no sense to collect statistics
if getCPName() == 'cPanel' and is_no_php_binaries_on_cpanel():
return []
return list(_iter_php_interpreters())
elif interpreter == 'ruby':
from .clselectruby.interpreters import interpreters
elif interpreter in ('python', 'nodejs'):
return list(_iter_interpreters(interpreter))
else:
raise NotImplementedError
it_list = []
for it in interpreters():
# for now, both python and ruby interpreters cannot be disabled
it_list.append(InterpreterSummary(it.version, it.version_full, installed=True, enabled=True))
return it_list