Current File : /opt/cloudlinux/venv/lib/python3.11/site-packages/clselect//clselect.py
# -*- coding: utf-8 -*-
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
from __future__ import absolute_import
from __future__ import print_function
from __future__ import division
import os
import re
import sys
from glob import glob
from builtins import map
from future.moves import configparser as ConfigParser
from .clselectexcept import ClSelectExcept, BaseClSelectException
from .clselectprint import clprint
from clcommon import clcagefs
class ClSelect(object):
BASE_ETC_CL_SELECTOR = '/etc/cl.selector.conf.d' if clcagefs.in_cagefs() else '/etc/cl.selector'
CONFIG_PATH = f'{BASE_ETC_CL_SELECTOR}/selector.conf'
DEFAULTS_PATH = f'{BASE_ETC_CL_SELECTOR}/defaults.cfg'
DEFAULT_PHP_PATH = '/usr/bin/php'
NATIVE_CONF = f'{BASE_ETC_CL_SELECTOR}/native.conf'
USER_CONF = f'{BASE_ETC_CL_SELECTOR}/user.conf'
CONFIGS_DIR = f'{BASE_ETC_CL_SELECTOR}/php.extensions.d/'
_CAGFSCTL = "/usr/sbin/cagefsctl"
# Cache PHP index file
CACHEFILE_DIR = '/var/lve'
CACHEFILE_CAGEFS_DIR = '/var/lve/php.dat.d'
CACHEFILE_PHP_PATTERN = '/php%s.dat'
CACHEFILE_PATTERN = CACHEFILE_DIR + CACHEFILE_PHP_PATTERN
CACHEFILE_CAGEFS_PATTERN = CACHEFILE_CAGEFS_DIR + CACHEFILE_PHP_PATTERN
CACHEFILE_PHP_NATIVE_PATTERN = '/php_native_ver.dat'
CACHEFILE_NATIVE_VER_PATTERN = CACHEFILE_DIR + CACHEFILE_PHP_NATIVE_PATTERN
CACHEFILE_NATIVE_VER_CAGEFS_PATTERN = CACHEFILE_CAGEFS_DIR + CACHEFILE_PHP_NATIVE_PATTERN
@staticmethod
def check_multiphp_system_default_version():
if clcagefs.in_cagefs():
return
try:
from clcagefslib.selector.configure import multiphp_system_default_is_ea_php, selector_modules_must_be_used
except ImportError:
raise BaseClSelectException('CageFS not installed.')
if not multiphp_system_default_is_ea_php() and not selector_modules_must_be_used():
raise BaseClSelectException('system default PHP version is alt-php. PHP Selector is disabled. Use cPanel MultiPHP manager instead.')
@staticmethod
def work_without_cagefs():
return os.path.exists(ClSelect.USER_CONF)
def __init__(self, item='php'):
self._item = item
self._dh = self._get_default_config_handler()
self._selector_contents = {}
self._native_contents = {}
self._hidden_extensions = set()
self._native_version = None
self.without_cagefs = ClSelect.work_without_cagefs()
self._load_config_files()
def check_requirements(self):
# check that selectorctl and cagefsctl are available
utilities = [
(
lambda: os.path.exists(self._get_native_path('cli')),
ClSelectExcept.NativeNotInstalled(self._get_native_path('cli'))
),
(
lambda: clcagefs.in_cagefs() or os.path.exists(self._CAGFSCTL),
ClSelectExcept.MissingCagefsPackage()
),
]
for predicate, error in utilities:
if predicate():
continue
raise error
def _load_config_files(self):
for filename in glob(os.path.join(self.CONFIGS_DIR, '*.cfg')):
self._load_config_file(filename)
def _load_config_file(self, filepath):
dh = ConfigParser.SafeConfigParser(interpolation=None, strict=False)
try:
dh.read(filepath)
except ConfigParser.Error as e:
raise ClSelectExcept.FileProcessError(
filepath, message="Config is malformed, error: %s" % str(e))
try:
self._hidden_extensions.update(
dh.get('extensions', 'hide_extensions').split(','))
except (ConfigParser.NoSectionError, ConfigParser.NoOptionError):
pass
def list_alternatives(self):
"""
Returns alternatives summary as tuple
:rtype: tuple
"""
alternatives = self.get_all_alternatives_data()
list_of_alternatives = []
for alt in sorted(alternatives.keys()):
try:
list_of_alternatives.append(
(alt,
alternatives[alt]['version'],
alternatives[alt]['data'][self._item]))
except KeyError:
continue
return tuple(list_of_alternatives)
def get_all_alternatives_data(self):
"""
Returns dict of all selector config contents.
If no data loads them
:return:
{'4.4': {'version': '4.4.9', 'data': {'lsphp': '/opt/alt/php44/usr/bin/lsphp',
'php.ini': '/opt/alt/php44/etc/php.ini',
'php': '/opt/alt/php44/usr/bin/php-cgi',
'php-cli': '/opt/alt/php44/usr/bin/php'}}}
:rtype: dict
"""
if not self._selector_contents:
try:
self._load_alternatives_config()
except (ClSelectExcept.ConfigNotFound, ClSelectExcept.WrongConfigFormat):
return {}
return self._selector_contents
def get_alternatives_data(self, version):
"""
Returns selector config contents of certain version as dict.
If no data loads them
@param version: string, selector version
@return: dict
"""
if not self._selector_contents:
self._load_alternatives_config()
try:
return {version: self._selector_contents[version]}
except KeyError:
raise ClSelectExcept.NoSuchAlternativeVersion(version)
def get_version(self, show_native_version=False):
"""
Gets default selector version
"""
alternatives = self.get_all_alternatives_data()
try:
version = self._dh.get('versions', self._item)
return (
version,
alternatives[version]['version'],
alternatives[version]['data'][self._item])
except (ConfigParser.NoSectionError, KeyError):
return self._compose_native_info(show_native_version)
def set_version(self, version):
"""
Sets default selector version
"""
alternatives = self.get_all_alternatives_data()
self._check_alternative(version, alternatives)
defaults_contents = self._process_ini_file(
self.DEFAULTS_PATH,
('versions',),
self._add_or_change_option,
(self._item, version))
self._write_to_file(
'\n'.join(defaults_contents), self.DEFAULTS_PATH)
def enable_version(self, version):
"""
Removes disabled state from version
"""
alternatives = self.get_all_alternatives_data()
self._check_alternative(version, alternatives)
defaults_contents = self._process_ini_file(
self.DEFAULTS_PATH,
(self._item, version),
self._remove_option,
'state')
self._write_to_file(
'\n'.join(defaults_contents), self.DEFAULTS_PATH)
def disable_version(self, version):
"""
Marks a vesrion as disabled
"""
alternatives = self.get_all_alternatives_data()
self._check_alternative(version, alternatives)
defaults_contents = self._process_ini_file(
self.DEFAULTS_PATH,
(self._item, version),
self._add_or_change_option,
('state', 'disabled'))
self._write_to_file(
'\n'.join(defaults_contents), self.DEFAULTS_PATH)
def is_version_enabled(self, version):
"""
Method that allows you to check if some version is enabled in config.
E.g. is_version_enabled('5.4') -> True
:rtype: bool
"""
return not self._dh.has_option(
"%s%s" % (self._item, version), 'state')
def get_summary(self, show_native_version=False):
"""
Returns state of alternatives
@return: tuple[version, tuple[isEnabled, isDefault]]
"""
alternatives = self.get_all_alternatives_data()
native_info = self._compose_native_info(show_native_version)
summary = {'native': {'enabled': True, 'default': False}}
alt_versions = sorted(alternatives.keys()) + ['native']
for version in alt_versions:
if version not in summary:
summary[version] = {}
summary[version]['enabled'] = self.is_version_enabled(version)
summary[version]['default'] = False
try:
default_version = self._dh.get('versions', self._item)
except (ConfigParser.NoSectionError, ConfigParser.NoOptionError):
default_version = 'native'
try:
summary[default_version]['default'] = True
except KeyError:
raise ClSelectExcept.NoSuchAlternativeVersion(default_version)
summary[native_info[0]] = summary.pop('native')
alt_versions.remove('native')
alt_versions.append(native_info[0])
for idx in range(len(alt_versions)):
v = alt_versions[idx]
alt_versions[idx] = (
v,
(summary[v]['enabled'],
summary[v]['default']))
return tuple(alt_versions)
def _load_alternatives_config(self):
"""
Parses selector config file and fills an instance config dict. Example:
{'4.4': {'version': '4.4.9', 'data': {'lsphp': '/opt/alt/php44/usr/bin/lsphp',
'php.ini': '/opt/alt/php44/etc/php.ini',
'php': '/opt/alt/php44/usr/bin/php-cgi',
'php-cli': '/opt/alt/php44/usr/bin/php'}}}
:raises ClSelectExcept.ConfigNotFound:
:raises ClSelectExcept.WrongConfigFormat:
"""
try:
f = open(self.CONFIG_PATH)
for line in f:
stripped_line = line.strip()
if stripped_line == "":
continue
(item, short_version, long_version,
item_path) = stripped_line.split()
if self._item not in item:
continue
if short_version not in self._selector_contents:
self._selector_contents[short_version] = {}
self._selector_contents[short_version]['version'] = long_version
if 'data' not in self._selector_contents[short_version]:
self._selector_contents[short_version]['data'] = {}
self._selector_contents[short_version]['data'][item] = item_path
if not self._selector_contents:
raise ClSelectExcept.ConfigNotFound(None,
message = 'alt-php packages not found')
except (OSError, IOError) as e:
raise ClSelectExcept.ConfigNotFound(
'Cannot read %s: %s. Native used' % (self.CONFIG_PATH, e),
message = 'alt-php packages not found')
except ValueError:
raise ClSelectExcept.WrongConfigFormat(self.CONFIG_PATH)
def _get_default_config_handler(self, path=None):
"""
Gets ConfigParser handler for future use
"""
dh = ConfigParser.ConfigParser(interpolation=None, strict=False)
dh.optionxform = str
if path:
dh.read(path)
else:
dh.read(self.DEFAULTS_PATH)
return dh
def _check_alternative(version, alternatives):
if version != 'native' and version not in alternatives:
raise ClSelectExcept.NoSuchAlternativeVersion(version)
_check_alternative = staticmethod(_check_alternative)
def _make_section_header(section_info):
"""
Gets section header data tuple and returns ini section header string
@param section_info: tuple
@return: string
"""
section_fmt = "[%s]" % ''.join(['%s'] * len(section_info))
return section_fmt % section_info
_make_section_header = staticmethod(_make_section_header)
def _smooth_data(data):
"""
Removes empty lines from list and appends newline if missing
"""
data = list(filter((lambda i: i != ''), data))
if not data or data[-1] != '\n':
data.append('\n')
return data
_smooth_data = staticmethod(_smooth_data)
def _process_ini_file(self, path, section_info, function, data, trace=True, action = None):
"""
Parses ini file by sections,
calls supplied callable to modify section is question,
returns file as list of strings
"""
contents = []
no_section_contents = []
section = []
in_section = False
found = False
has_default = False
section_header = self._make_section_header(section_info)
try:
f = open(path)
for line in f:
line = line.strip()
if line.startswith('['):
in_section = True
if '[versions]' in line:
has_default = True
if section_header == line:
found = True
if len(no_section_contents) != 0:
contents.extend(no_section_contents)
no_section_contents = []
contents.extend(function(section_info, section, data, trace))
section = [line]
continue
if in_section:
section.append(line)
else:
no_section_contents.append(line)
contents.extend(function(section_info, section, data, trace))
f.close()
except (OSError, IOError):
pass
if not has_default and '[versions]' not in section_header:
default = ['[versions]', '%s = native' % self._item, '']
default.extend(contents)
contents = default
if not found:
try:
build_in = self._get_builtins('native')
except ClSelectExcept.UnableToGetExtensions:
pass # skip if native version not installed
if action == 'disable_extentions':
contents.extend(function(section_info, [section_header, 'modules = ' + ','.join(build_in)], data, trace))
elif action == 'enable_extentions':
data.extend(build_in)
contents.extend(function(section_info, [section_header], data, trace))
else:
contents.extend(function(section_info, [section_header], data, trace))
return contents
def _get_php_binary_path(self, version):
"""
Retrives path to php binary for supplied version
:param version: php version to retrive path
:return: path to php binary. If alternative version not found native php binary path returned
"""
item = "%s-cli" % self._item
alternatives = self.get_all_alternatives_data()
try:
path = alternatives[version]['data'][item]
except KeyError:
path = self._get_native_path(suffix='cli')
return path
def get_all_php_binaries_paths(self):
"""
Retrives paths to php binary for all versions
:return: Dictionary version -> path. Example:
{ '5.2': '/opt/alt/php52/usr/bin/php',
'5.3': '/opt/alt/php53/usr/bin/php',
'native': '/usr/bin/php'
}
"""
alternatives = self.get_all_alternatives_data()
paths_dict = {'native': self._get_php_binary_path('native')}
for version in alternatives.keys():
paths_dict[version] = self._get_php_binary_path(version)
return paths_dict
def _read_php_cache_file(self, version):
"""
Retrives contents of cache file for supplied php version
:param version: PHP version to read file
:return: file contents
"""
filename = self.CACHEFILE_CAGEFS_PATTERN % version if clcagefs.in_cagefs() else self.CACHEFILE_PATTERN % version
with open(filename, "r") as f:
return f.read()
def _get_builtins(self, version):
"""
Gets php extensions from the /var/lve/phpX.X.dat cache file,
which contains list of modules that are either compiled-in
or enabled in /opt/alt/phpXX/etc/php.ini config file
"""
builtins = []
# Read php output from cache file for approptiate php version
try:
output = self._read_php_cache_file(version)
except (OSError, IOError):
raise ClSelectExcept.UnableToGetExtensions(version)
# Section that contains list of modules usually starts and ends with
# [PHP Modules] and [Zend Modules] headers respectively
start_pattern, end_pattern = "[PHP Modules]", "[Zend"
start_index, end_index = output.find(start_pattern), output.find(end_pattern)
start_index = 0 if start_index == -1 else start_index + len(start_pattern)
modules_list = output[start_index:end_index]
module_pattern = re.compile(r"\w")
for ext in modules_list.split("\n"):
if not module_pattern.match(ext):
continue
module = "_".join(re.split("\s+", ext.lower()))
if module not in self._hidden_extensions:
builtins.append(module)
return builtins
def _remove_option(self, section_info, section, data, trace=True):
"""
Adds 'modules' option to section or extends it
@param section_info: tuple (item and version)
@param section: list
@param data: string
@return: list
"""
section_header = self._make_section_header(section_info)
if len(section) == 0 or section_header != section[0]:
return section
return self._smooth_data(
list(filter((lambda i: not i.startswith(data)), section)))
def _add_or_change_option(self, section_info, section, data, trace=True):
"""
Adds 'modules' option to section or extends it
@param section_info: tuple
@param section: list
@param data: tuple
@return: list
"""
section_header = self._make_section_header(section_info)
if len(section) == 0 or section_header != section[0]:
return section
oidx = None
for idx in range(len(section)):
if section[idx].startswith(data[0]):
oidx = idx
break
option = "%s = %s" % data
if oidx:
section[oidx] = option
else:
section.append(option)
return self._smooth_data(section)
def _write_to_file(self, file_contents, file_path):
"""
Saves data to file
"""
try:
f = open(file_path, 'w')
f.write("%s\n" % file_contents)
f.close()
except (OSError, IOError) as e:
raise ClSelectExcept.UnableToSaveData(file_path, e)
def _get_native_path(self, suffix=None):
"""
Returns path for native interpreter
"""
suffixes = {
'cli': '-cli',
'ini': '.ini',
'fpm': '-fpm'}
item = self._item
if suffix and suffix in suffixes:
item = "%s%s" % (self._item, suffixes[suffix])
if not self._native_contents:
self._load_native_contents(self._item)
if item in self._native_contents:
path = self._native_contents[item]
if os.path.isfile(path) and not os.path.islink(path):
return path
return self._native_contents[self._item]
def _load_native_contents(self, value):
"""
Parses native.conf file and loads it as dict, for example:
{'php-fpm': '/usr/local/sbin/php-fpm', 'php.ini': '/usr/local/lib/php.ini', 'php': '/usr/bin/php', 'php-cli': '/usr/bin/php'}
"""
try:
f = open(self.NATIVE_CONF)
for line in f:
if line.startswith('#'):
continue
if value not in line:
continue
if '=' not in line:
continue
item, path = list(map((lambda x: x.strip()), line.split('=')))
self._native_contents[item] = path
if value not in self._native_contents:
self._native_contents[value] = self.DEFAULT_PHP_PATH
f.close()
except (OSError, IOError):
self._native_contents[value] = self.DEFAULT_PHP_PATH
def _compose_native_info(self, show_version=False):
if not show_version:
return 'native', 'native', self._get_native_path()
native_version = self.get_native_version(verbose=False)
if native_version:
return ('native (%s)' % (native_version[0],),
'native (%s)' % (native_version[1],),
self._get_native_path())
return 'native', 'native', self._get_native_path()
def get_native_version(self, verbose=True):
if self._native_version:
return self._native_version
version_pattern = re.compile(r'PHP\s+(?P<full>(?P<short>\d+\.\d+)\.\d+)')
# Read php output from cache file for approptiate php version
try:
f = open(self.CACHEFILE_NATIVE_VER_PATTERN, "r")
data = f.read()
f.close()
except (OSError, IOError) as e:
if verbose:
clprint.print_diag('text', {'status': 'ERROR', 'message': str(e)})
return None
for line in data.splitlines():
m = version_pattern.match(line)
if m:
short, full = m.group('short'), m.group('full')
self._native_version = (short, full)
return short, full
return None
