Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified
📁 ..	-	2026-03-27 02:38
📄 __init__.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # # fix for PTCLLIB-97 import os # NOQA from clcommon.clpwd import ClPwd from clcommon.clsec import ClSec from clcommon.cllog import ClAuditLog from clcommon.clfunc import ( get_lve_version, validate_cpu, validate_int, page_to_memory, memory_to_page, reload_processes, login_defs, uid_max, exit_with_error ) from clcommon.clhook import (get_config, write_config, get_hook_list_from_file, register_hook, unregister_hook, call_hook_list) from clcommon.clconfpars import load, load_once, change_settings, apache_conf_parser from clcommon.cldebug import listen from clcommon.clexception import FormattedException listen() __all__ = [ "ClPwd", "ClSec", "ClAuditLog", "get_lve_version", "validate_cpu", "validate_int", "page_to_memory", "memory_to_page", "reload_processes", "login_defs", "uid_max", "exit_with_error", "get_config", "write_config", "get_hook_list_from_file", "register_hook", "unregister_hook", "call_hook_list", "load", "load_once", "change_settings", "apache_conf_parser", "listen", "FormattedException" ] __tags__ = ['LU-365'] # tags for external usage	1.37 KB	2026-02-10 15:10
📁 __pycache__	-	2026-03-19 15:49
📄 clcagefs.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import logging import os import re import subprocess import sys CAGEFS_MP_FILENAME = "/etc/cagefs/cagefs.mp" CAGEFSCTL_TOOL = "/usr/sbin/cagefsctl" CAGEFS_SKELETON_PATH = "/usr/share/cagefs-skeleton" CAGEFS_DIR_PATH = "/var/.cagefs" logger = logging.getLogger(__name__) class CagefsMpConflict(Exception): def __init__(self, new_item, existing_item): self._msg = ( f"Conflict in adding '{new_item}' to {CAGEFS_MP_FILENAME} " f"because of pre-existing alternative specification: '{existing_item}'" ) def __str__(self): return self._msg class CagefsNotSupportedError(Exception): """Cagefs Not Supported Exception""" def __init__(self, message): Exception.__init__(self, message) class CagefsMpItem: PREFIX_LIST = "@!%*" _PREFIX_MOUNT_RW = "" _PREFIX_MOUNT_RO = "!" def __init__(self, arg): """Constructor :param arg: Is either path to add to cagefs.mp or a raw line is read from cagefs.mp :param prefix: The same as adding prefix '!' to arg before passing it to ctor""" if arg[:1] == "#": # is a comment? then init as dummy self._path_spec = None elif arg.strip() == "": # init as dummy for empty lines self._path_spec = None else: self._path_spec = arg def mode(self, mode): """Specify mode as in fluent constructor""" if self.prefix() == "@" and mode is not None: self._path_spec = f"{self._path_spec},{mode:03o}" return self def __str__(self): return self._path_spec @staticmethod def _add_slash(path): if path == "": return "/" if path[-1] != "/": return path + "/" return path def pre_exist_in(self, another): adopted = CagefsMpItem._adopt(another) # overkill: just to keep strictly to comparing NULL objects principle if self.is_dummy() or adopted.is_dummy(): return False this_path = CagefsMpItem._add_slash(self.path()) test_preexist_in_path = CagefsMpItem._add_slash(adopted.path()) return this_path.startswith(test_preexist_in_path) def is_compatible_by_prefix_with(self, existing): adopted = CagefsMpItem._adopt(existing) # overkill: just to keep strictly to comparing NULL objects principle if self.is_dummy() or adopted.is_dummy(): return False if self.prefix() == adopted.prefix(): return True prefix_compatibility_map = {CagefsMpItem._PREFIX_MOUNT_RW: [CagefsMpItem._PREFIX_MOUNT_RO]} null_options = [] return self.prefix() in prefix_compatibility_map.get(adopted.prefix(), null_options) def is_dummy(self): return self._path_spec is None @staticmethod def _adopt(x): if isinstance(x, CagefsMpItem): return x else: return CagefsMpItem(x) @staticmethod def _cut_off_mode(path_spec): """Cut off mode from path spec like @/var/run/screen,777 Only one comma per path spec is allowed ;-)""" return path_spec.split(",")[0] @staticmethod def _cut_off_prefix(path_spec): return path_spec.lstrip(CagefsMpItem.PREFIX_LIST) def path(self): return CagefsMpItem._cut_off_prefix(CagefsMpItem._cut_off_mode(self._path_spec)) def prefix(self): if self._path_spec != self.path(): return self._path_spec[0] else: return "" def is_cagefs_present(): return os.path.exists(CAGEFSCTL_TOOL) def _mk_mount_dir_setup_perm(path, mode=0o755, owner_id=None, group_id=None): # -1 means 'unchanged' if group_id is None: group_id = -1 if owner_id is None: owner_id = -1 if not os.path.isdir(path): os.mkdir(path) if mode is not None: os.chmod(path, mode) os.chown(path, owner_id, group_id) def _remount_cagefs(user=None, remount_in_background=False): if not is_cagefs_present(): return if user is None: command = [CAGEFSCTL_TOOL, "--wait-lock", "--remount-all"] else: command = [CAGEFSCTL_TOOL, "--remount", user] if remount_in_background: subprocess.Popen( # pylint: disable=consider-using-with command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) else: subprocess.run( command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, ) def setup_mount_dir_cagefs( path, added_by=None, mode=0o755, owner_id=None, group_id=None, prefix="", remount_cagefs=False, remount_in_background=False, ): """ Add mount point to /etc/cagefs/cagefs.mp :param path: Directory path to be added in cagefs.mp and mounted from within setup_mount_dir_cagefs(). If this directory does not exist, then it is created. :param added_by: package or component, mount dir relates to, or whatever will stay in cagefs.mp with "# added by..." comment :param mode: If is not None: Regardless of whether directory exists or not prior this call, it's permissions will be set to mode. :param owner_id: Regardless of whether directory exists or not prior this call, it's owner id will be set to. If None, the owner won't be changed. :param group_id: Regardless of whether directory exists or not prior this call, it's group id will be set to. If None, the group won't be changed. :param prefix: Mount point prefix. Default is mount as RW. Pass '!' to add read-only mount point. Refer CageFS section at http://docs.cloudlinux.com/ for more options. :param remount_cagefs: If True, cagefs skeleton will be automatically remounted to apply changes. :param remount_in_background: If True, cagefs remount will be done in separate background process, without waiting for completion :returns: None Propagates native EnvironmentError if no CageFS installed or something else goes wrong. Raises CagefsMpConflict if path is already specified in cagefs.mp, but in a way which is opposite to mount_as_readonly param. """ _mk_mount_dir_setup_perm(path, mode, owner_id, group_id) if not is_cagefs_present(): return # Create cagefs.mp if absent. It will be merged when cagefsctl --init. if not os.path.exists(CAGEFS_MP_FILENAME): subprocess.run( [CAGEFSCTL_TOOL, "--create-mp"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, ) add_new_line_to_cagefs_mp() # ^^ # Hereafter we will not care if there was # 'no newline at the end of file' with open(CAGEFS_MP_FILENAME, "r+", encoding="utf-8") as f: new_item = CagefsMpItem(prefix + path).mode(mode) trim_nl_iter = (file_line.rstrip() for file_line in f) pre_exist_option = [x for x in trim_nl_iter if new_item.pre_exist_in(x)] if not pre_exist_option: f.seek(0, 2) # 2: seek to the end of file if added_by is not None: # no newline is allowed added_by = added_by.strip() print("# next line is added by ", added_by, file=f) print(new_item, file=f) if remount_cagefs: _remount_cagefs(remount_in_background=remount_in_background) elif not new_item.is_compatible_by_prefix_with(pre_exist_option[-1]): raise CagefsMpConflict(new_item, pre_exist_option[-1]) def _get_cagefs_mp_lines(): with open(CAGEFS_MP_FILENAME, "r", encoding="utf-8") as f: return f.readlines() def _write_cagefs_mp_lines(lines): with open(CAGEFS_MP_FILENAME, "w", encoding="utf-8") as f: return f.writelines(lines) # next line is added by def add_new_line_to_cagefs_mp(): """ Add new line to the end of /etc/cagefs/cagefs.mp file if it is not there """ lines = _get_cagefs_mp_lines() # file is not empty and last line in the file does not end with new line symbol ? if lines and lines[0] != "" and lines[-1][-1] != "\n": lines[-1] += "\n" _write_cagefs_mp_lines(lines) def remove_mount_dir_cagefs(path, remount_cagefs=False, remount_in_background=False): """ Remove mount points matching given path from cagefs.mp file :param str path: Path that should be removed from file. :param bool remount_cagefs: Remount cagefs skeleton or not :param remount_in_background: If True, cagefs remount will be done in separate background process, without waiting for completion :return: Nothing """ lines = _get_cagefs_mp_lines() r = re.compile(rf"^[{CagefsMpItem.PREFIX_LIST}]?{re.escape(path)}(,\d+)?$") lines_with_excluded_path = [line for line in lines if not r.match(line)] # nothing to do if nothing found by pattern if len(lines) == len(lines_with_excluded_path): return _write_cagefs_mp_lines(lines_with_excluded_path) if remount_cagefs: _remount_cagefs(remount_in_background=remount_in_background) def _is_directory_mounted(path: str) -> bool: """Check the fact that directory is mounted i.e. listed in mountinfo""" with open("/proc/self/mountinfo", encoding="utf-8") as mountinfo_file: for line in mountinfo_file: mountinfo_entry_parts = line.split(maxsplit=5) # See the following link about /proc/<pid>/mountinfo format: # https://docs.kernel.org/filesystems/proc.html#proc-pid-mountinfo-information-about-mounts mount_point = mountinfo_entry_parts[4] if mount_point == path: return True return False def in_cagefs() -> bool: """Recommended way to check whether process is running inside CageFS Please avoid testing for /var/.cagefs existence. Replace with this function call when possible. """ in_cagefs_ = _is_directory_mounted(CAGEFS_DIR_PATH) if not in_cagefs_ and os.path.isdir(CAGEFS_DIR_PATH): logger.warning("%s should not exist in the real file system", CAGEFS_DIR_PATH) return in_cagefs_ def _is_cagefs_enabled(user): """ Check that cagefs enabled for user """ try: cagefs_lib_dir = "/usr/share/cagefs/" if cagefs_lib_dir not in sys.path: sys.path.append(cagefs_lib_dir) import cagefsctl # pylint: disable=import-outside-toplevel except ImportError: return False try: if not cagefsctl.is_user_enabled(user): return False except AttributeError as e: raise CagefsNotSupportedError("ERROR: CageFS version is unsupported. Please update CageFS.") from e return True	10.99 KB	2026-02-10 15:10
📄 clcaptain.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import os import errno from subprocess import Popen, PIPE from clcommon.utils import ExternalProgramFailed CAPTAIN_PATH = '/usr/share/python-cllib/cl.captain' def captain(do_try, cmd, stdin=None): try: do_try() except (OSError, IOError) as e: if e.errno == errno.EDQUOT and os.geteuid() > 0: command = [CAPTAIN_PATH] + [str(s) for s in cmd] with Popen( command, stdin=None if stdin is None else PIPE, stdout=PIPE, stderr=PIPE, text=True, ) as proc: if stdin is None: std_out, std_err = proc.communicate() else: std_out, std_err = proc.communicate(stdin) if proc.returncode != 0: raise ExternalProgramFailed( std_err or f'output of "{command}": {std_out}' ) from e else: raise def mkdir(path, mode=0o755, recursive=False): umask = os.umask(0) if recursive: collector = "/" for subdir in path.split("/"): collector = os.path.join(collector, subdir) if os.path.exists(collector): continue captain(lambda: os.mkdir(collector, mode), ['-d', collector, '-m', f'{mode:o}']) else: captain(lambda: os.mkdir(path, mode), ['-d', path, '-m', f'{mode:o}']) os.umask(umask) def write(path, content=''): def do_try(): with open(path, 'w', encoding='utf-8') as fp: fp.write(content) fp.flush() os.fsync(fp.fileno()) captain(do_try, ['-f', path], content) def symlink(src, dst): captain(lambda: os.symlink(src, dst), ['-l', dst, '-s', src])	1.96 KB	2026-02-10 15:10
📄 clconfig.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import os CPANEL_CONFIG_PATH = "/var/cpanel/cpanel.config" def get_param(path, param): """ Get param from config file :param path: Path to config file :type path: str :param param: Key for param in config file :type param: str """ try: with open(path, encoding="utf-8") as f: data = f.readlines() except (OSError, IOError): return False for line in data: if line.startswith(f"{param}="): return (line.strip().split("="))[1] return None def replace_param(path, param, value): """ Replaces config file with path to config file with line param=value :param path: Path to config file :type path: str :param param: Key for param in config file :type key: str :param value: Value for param in config file :type value: str """ new_line = f'{param}={value}\n' if not os.path.exists(path): # config file is absent with open(path, 'w', encoding='utf-8') as f: f.write(new_line) return # config file present with open(path, encoding='utf-8') as f: data = f.readlines() for i, line in enumerate(data): if line.startswith(f'{param}='): data[i] = new_line with open(path, 'w', encoding='utf-8') as f: f.write(''.join(data)) return # add new line with given param if it wasn't in original config with open(path, 'a', encoding='utf-8') as f: f.write(new_line)	1.68 KB	2026-02-10 15:10
📄 clconfpars.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import configparser import locale import os import re import syslog from collections import namedtuple class WebConfigParsingError(Exception): def __init__(self, message): self.message = message class WebConfigMissing(Exception): def __init__(self, message): self.message = message SECHEAD = 'asection' def load(path, case_sensitive=False, ignore_bad_encoding=False): config = configparser.ConfigParser(allow_no_value=True, interpolation=None, strict=False) if case_sensitive: config.optionxform = str if ignore_bad_encoding: with open(path, 'rb') as f: raw = f.read().decode(locale.getpreferredencoding(), 'replace') else: with open(path, 'r', encoding='utf-8') as f: raw = f.read() config.read_string(f'[{SECHEAD}]\n' + raw, source=path) return dict(config.items(section=SECHEAD)) _QUOTES = "'", '"' def _strip_escape_quotes_of_config_value(val: str) -> str: """ Strips single or double quote char only if the quote present from both sides. """ if val.startswith(_QUOTES) and val.endswith(_QUOTES): return val[1:-1] return val def load_fast(path, delimiter="=", strip_quotes=False): data = {} with open(path, "r", encoding="utf-8", errors="surrogateescape") as f: for line in f.readlines(): parts = line.split(delimiter, 1) try: key, value = parts except ValueError: # Skip broken lines continue value = value.strip() value = ( _strip_escape_quotes_of_config_value(value) if strip_quotes else value ) data[key.strip()] = value return data cache = {} def load_once(path, ignore_errors=False): """ Read ini file once (cached) and return its content as dict """ try: res = cache[path] except KeyError: try: res = cache[path] = load(path) except (IOError, configparser.Error): if not ignore_errors: raise res = cache[path] = {} return res def change_settings(settings_dict, path, tmp_path=None): if not tmp_path: tmp_path = path + ".tmp" used_keys = [] with (open(path, 'r', encoding='utf-8') as fin, open(tmp_path, 'w', encoding='utf-8') as fout): for line in fin: stripped_line = line.strip() if stripped_line and not stripped_line.startswith('#'): key, _ = stripped_line.split('=', 1) key = key.strip() if key in settings_dict: fout.write(f'{key}={settings_dict[key]}\n') used_keys.append(key) continue fout.write(line) with open(tmp_path, 'a', encoding='utf-8') as fout: for key in settings_dict: if key not in used_keys: fout.write(f'{key}={settings_dict[key]}\n') os.rename(tmp_path, path) _NGINX_TOKENS_RE = re.compile( r""" ( # Comments (:? \# .* $ ) # Single-, double-quoted strings and bare strings without whitespaces \| (:? "[^"\n]?" ) \| (:? '[^'\n]?' ) \| (:? [^"';\s\{\}]+ ) # Structural characters \| ; \| \{ \| \} \| \n ) """, re.IGNORECASE \| re.MULTILINE \| re.VERBOSE, ) def _ngx_tokenize(data): tokens = ( match.group(0) for match in _NGINX_TOKENS_RE.finditer(data) if match and match.group(0) ) # Explicitly ignore comments return (tok for tok in tokens if not tok.startswith('#')) def _ngx_take_until(it, val): for tok in it: if tok in val: return yield tok def _ngx_take_until_block_end(it): lvl = 1 for t in it: if t == "{": lvl += 1 elif t == "}": lvl -= 1 if lvl < 1: return yield t def _ngx_scan_block_info(block_tokens, need_fields): """Scan a block for required fields, skips nested blocks""" info = {} for tok in block_tokens: # We need to skip until the end of inner block if it occurs if tok == "{": for _ in _ngx_take_until_block_end(block_tokens): pass # Now gather the value, the last occurrence is in priority if tok in need_fields: value_tokens = _ngx_take_until(block_tokens, ";\n") info[tok] = list(value_tokens) return info def nginx_conf_loose_parser(data): """ Parse content of NGINX configuration in a manner tolerant to minor mistakes and extract relevant fields from all `server` directives. Relevant fields are: - `server_name` - `root` - returned as `document_root` - `ssl` - if `listen` field contains "ssl" word Doesn't handle interpolated values (ex. `${val}`) outside of quoted strings """ tokens = _ngx_tokenize(data) for tok in tokens: if tok != "server": continue # Nothing seems to be allowed between "server" directive and # the opening of his block, so we just discard everything # until first block opening seen for _ in _ngx_take_until(tokens, "{"): pass # Limit further scan by the inside of block block_tokens = _ngx_take_until_block_end(tokens) # By using only `block_tokens` we ensure all blocks are properly delimited info = _ngx_scan_block_info(block_tokens, ("server_name", "root", "listen")) try: server_name = info["server_name"] root = info["root"] except KeyError: continue if not server_name and not root: continue yield { "server_name": _strip_escape_quotes_of_config_value(server_name[0]), "document_root": _strip_escape_quotes_of_config_value(root[0]), "ssl": "ssl" in info.get("listen", []), } def nginx_conf_parser(conf_file): """Parse NGINX config file, see `nginx_conf_loose_parser` for more details""" if not os.path.isfile(conf_file): raise WebConfigMissing(f'File does not exists {conf_file}') dirty_data = read_unicode_file_with_decode_fallback(conf_file) return list(nginx_conf_loose_parser(dirty_data)) def apache_conf_parser(conf_file): if not os.path.isfile(conf_file): raise WebConfigMissing(f'File does not exists {conf_file}') conf_data = [] data_all = read_unicode_file_with_decode_fallback(conf_file).splitlines() data = [i for i in data_all if re.search('^((?!#).)*$', i)] ID = 0 enable = False result = {} vhost = [] while len(data) > 0: out = data.pop(0) if "<VirtualHost" in out: ip_port = out.split()[1] port = '0' try: ip, port = ip_port.split(':') port = port.replace('>', '') except ValueError: ip = ip_port vhost.append(ip) vhost.append(port) enable = True continue if "</VirtualHost>" in out: result[ID] = vhost ID+=1 enable = False vhost = [] continue if enable: vhost.append(out) continue for value in result.values(): # result[i][0] is an IP # result[i][1] is a port data = { 'user' : None, 'server_name' : '', 'document_root' : '', 'server_alias' : None, 'port' : int(value[1]), 'ssl' : False, } for line in value: if "ServerName" in line: data['server_name'] = line.split()[1].strip().replace('www.', '') continue if "DocumentRoot" in line: # remove all whitespaces (first strip) and also quotes (second one) data['document_root'] = line.split()[1].strip().strip('"') continue if "ServerAlias" in line: data['server_alias'] = ','.join(str(n) for n in line.split()[1:]) continue if "SuexecUserGroup" in line: data['user'] = line.split()[1].strip() if "SSLEngine" in line: data['ssl'] = line.split()[1].strip().lower() == 'on' conf_data.append(data) return conf_data def openlitespeed_conf_parser(conf_file): """ Parsing Open Litespeed config Retrieves data from user's openlitespeed.conf :param: full httpd.conf path (/usr/local/directadmin/data/users/<username>/openlitespeed.conf) :return: list of virtual host configurations. """ if not os.path.isfile(conf_file): raise WebConfigMissing(f'File does not exist {conf_file}') conf_data = [] data_all = read_unicode_file_with_decode_fallback(conf_file).splitlines() vhost_dict = None for line in data_all: line = line.strip() if line.startswith('virtualHost'): # Add the previous virtual host to the list if vhost_dict: conf_data.append(vhost_dict) # Start a new virtual host vhost_dict = {'user': None, 'server_name': None, 'document_root': None, 'server_alias': None, 'port': None, 'ssl': False} # Extract port information match_port = re.search(r'-(\d+)', line) vhost_dict['port'] = int(match_port.group(1)) if match_port else None if vhost_dict is None: continue if line.startswith('user'): parts = line.split() if len(parts) == 2: vhost_dict['user'] = parts[1] elif line.startswith('docRoot'): parts = line.split() if len(parts) == 2: vhost_dict['document_root'] = parts[1] elif line.startswith('vhDomain'): parts = line.split() if len(parts) == 2: vhost_dict['server_name'] = parts[1] elif line.startswith('vhAliases'): parts = line.split() if len(parts) == 2: vhost_dict['server_alias'] = parts[1] elif line.startswith('vhssl'): vhost_dict['ssl'] = True # Add the last virtual host to the list if vhost_dict: conf_data.append(vhost_dict) return conf_data PamLVECfg = namedtuple('PamLVECfg', ['min_uid', 'cagefs_enabled', 'groups']) def parse_pam_lve_config(configfile): """ Parse string like: "session required pam_lve.so 500 1 group1,group2" :param configfile: path to config file to parse :type configfile: str :return: PamLVECfg instance when pam_lve configuratiom is found, None otherwise :rtype: namedtuple :raises: IOError, ValueError """ with open(configfile, 'r', encoding='utf-8') as f: for line in f: if line.startswith('#'): continue s = line.split() if len(s) >= 3 and s[2] == 'pam_lve.so': # parse config string taking pam_lve defaults into account min_uid = int(s[3]) if len(s) >= 4 else 500 cagefs_enabled = bool(int(s[4])) if len(s) >= 5 else False groups = s[5].split(',') if len(s) >= 6 else ['wheel'] return PamLVECfg(min_uid, cagefs_enabled, groups) # pam_lve line is not found in config file return None def read_unicode_file_with_decode_fallback(file_path: str) -> str: with open(file_path, 'rb') as f: raw_data = f.read() try: return raw_data.decode() except UnicodeDecodeError: syslog.syslog( syslog.LOG_WARNING, f'Failed to decode "{file_path}" content as utf-8 - loading with placeholders for invalid unicode sequences' ) return raw_data.decode(errors='replace')	12.09 KB	2026-02-10 15:10
📄 clcustomscript.py #!/usr/bin/python # coding=utf-8 import os __author__ = "Aleksandr Shyshatsky" def lvectl_custompanel_script(): """ Retrives custom panel script for lvectl from CL config file :return: Script full path or None if script filename not find in config """ config_param_name = 'CUSTOM_GETPACKAGE_SCRIPT' try: # Try to determine custom script name if os.path.exists(CLSYSCONFIG): with open(CLSYSCONFIG, 'r', encoding='utf-8') as f: file_lines = f.readlines() for line in file_lines: line = line.strip() if line.startswith(config_param_name): line_parts = line.split('=') if len(line_parts) == 2 and line_parts[0].strip() == config_param_name: script_name = line_parts[1].strip() if os.path.exists(script_name): return script_name except (OSError, IOError, IndexError): # We are ignoring all errors, but script name not found in this case pass # Script name not found or error return None CLSYSCONFIG = '/etc/sysconfig/cloudlinux'	1.16 KB	2026-02-10 15:10
📄 cldebug.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # # source code # http://stackoverflow.com/questions/132058/showing-the-stack-trace-from-a-running-python-application import code import traceback import signal def debug(sig, frame): """ Interrupt running process, and provide a python prompt for interactive debugging. """ d = {'_frame': frame} # Allow access to frame object d.update(frame.f_globals) # Unless shadowed by global d.update(frame.f_locals) i = code.InteractiveConsole(d) message = "Signal recieved : entering python shell.\nTraceback:\n" message += ''.join(traceback.format_stack(frame)) i.interact(message) def listen(): signal.signal(signal.SIGUSR1, debug) # Register handler	905 B	2026-02-10 15:10
📄 clemail.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import os import jinja2 # Subject line prefix in template file SUBJECT_LINE_PREFIX = 'Subject:' # TODO: looks like this class is used only in lve-stats # and we should move it there class ClEmail: """ class for generation email messages """ def __init__(self): pass @staticmethod def generate_mail_jinja2(template_path, templ_data, subject=None): """ Generates email message using jinja2 template engine :param template_path: path to jinja2 template :param templ_data: Data to fill template using jinja2 :param subject: Email subject to use if it not found in template :return: Cortege (email_subject, email_body) """ # load and render template templ_loader = jinja2.FileSystemLoader(os.path.dirname(template_path)) templ_envir = jinja2.Environment(loader=templ_loader) message_lines = templ_envir.get_template( os.path.basename(template_path) ).render( templ_data ).split('\n') # Search subject string in template content if len(message_lines) > 2 and message_lines[0].startswith(SUBJECT_LINE_PREFIX) and message_lines[1] == '': subject = message_lines[0].replace(SUBJECT_LINE_PREFIX, '').strip() # Remove Subject line and separator (empty line) from message body message_lines.pop(1) message_lines.pop(0) return subject, '\n'.join(message_lines)	1.65 KB	2026-02-10 15:10
📄 clexception.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # from .clfunc import escape_formatting_chars class FormattedException(Exception): def __init__(self, exc_message): if isinstance(exc_message, dict): self.message = escape_formatting_chars(exc_message["message"]) self.context = exc_message.get("context", {}) self.details = exc_message.get("details") message = self.message % self.context elif isinstance(exc_message, FormattedException): self.message = escape_formatting_chars(exc_message.message) self.context = exc_message.context self.details = exc_message.details message = self.message % self.context else: self.message = escape_formatting_chars(exc_message) self.context = {} self.details = None message = exc_message Exception.__init__(self, message) def __str__(self): return self.message % self.context	1.14 KB	2026-02-10 15:10
📄 clfunc.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import os import re import signal import subprocess import sys LVE_FILE = '/proc/lve/list' # GET VERSION from /proc/lve/list def get_lve_version(): """ Obtain lve process filesystem version """ try: with open(LVE_FILE, encoding='utf-8') as f: line = f.read(3) # we need only first three symbols for parse version lve_procfs_version = [int(line.rsplit(':', 1)[0]), 'OK'] except IOError: lve_procfs_version = [None, f'clcommon: get_lve_version: Can`t open file {LVE_FILE}'] except IndexError: lve_procfs_version = [None, f'clcommon: get_lve_version: Can`t get data from {LVE_FILE}'] except ValueError: lve_procfs_version = [None, f'clcommon: get_lve_version: Can`t parse {LVE_FILE}'] return lve_procfs_version BYTES_CONVERSION_TABLE = { 'K': 1, 'M': 1024, 'G': 1024 * 1024, 'T': 1024 * 1024 * 1024 } def validate_cpu(val: str \| int) -> str \| int \| None: """ Check that the value is a valid CPU limit (0-100 int or speed (% or MHz/GHz)). Return the value if valid, otherwise None. If the value contains decimal part, return only integer part. """ data = str(val) # Validate integer CPU limits regexp_int = re.compile(r'^([1-9][0-9]?\|100)(?:\.(\d+))?$') regex_match = regexp_int.match(data) if regex_match: if isinstance(val, (float, int)): return int(val) return regex_match.group(1) # return only integer part # Validate percentage speeds regexp_speedp = re.compile(r'^([1-9]\|[1-9][0-9])(?:\.(\d+))?%$') regex_match = regexp_speedp.match(data) if regex_match: return f'{regex_match.group(1)}%' # Validate frequency speeds regexp_speedf = re.compile(r'^([1-9]\|[1-9][0-9])(?:\.(\d+))?(mhz\|ghz)$', re.IGNORECASE) regex_match = regexp_speedf.match(data) if regex_match: integer_part, _, unit = regex_match.groups() return f'{integer_part}{unit}' return None def validate_int(val, min_val=0, max_val=sys.maxsize): """ Check that val - is a string number return val as a string """ try: dig_val = int(val) except ValueError: return None if max_val >= dig_val >= min_val: return val def memory_to_page(val, min_val=0, max_val=sys.maxsize): try: suffix = val[-1] if suffix.isdigit(): suffix = 'K' val = val + suffix result = int(float(val[:-1]) * BYTES_CONVERSION_TABLE[suffix.upper()] / 4) if max_val >= result >= min_val: return result except (IndexError, ValueError, KeyError): pass return None def page_to_memory(pages): if pages < 256: # KB return str(pages * 4) + 'K' if pages < 262144: # MB return str(round(float(pages) * 4 / 1024, 2)) + 'M' return str(round(float(pages) * 4 / (1024 * 1024), 2)) + 'G' def reload_processes(item, username): with subprocess.Popen( ['/bin/ps', '-U', username, '-u', username], stdout=subprocess.PIPE, ) as proc: lines = proc.communicate()[0].split(b"\n") for row in lines: parts = row.rstrip().split() try: parts[-1].index(item.encode()) os.kill(int(parts[0]), signal.SIGHUP) except (IndexError, ValueError, OSError): continue def login_defs(key, default=None, _path='/etc/login.defs'): with open(_path, encoding='utf-8') as login_defs: for raw_line in login_defs: if raw_line.startswith('#'): continue # ignore commented lines line = raw_line.split('#', 1)[0] # clear line of comments line_splited = line.split() if len(line_splited) >= 2 and line_splited[0] == key: return line_splited[1] return default def uid_max(default=60000): try: uid_max_ = int(login_defs('UID_MAX')) except (IOError, ValueError, TypeError): uid_max_ = default return uid_max_ def exit_with_error(message, status=1): sys.stderr.write(f"{message}\n") sys.exit(status) def safe_escaped_unicode_to_utf(s): # transforms unicode-escaped string into utf-8 encoded str if '\\u' in s: # str should have escaped unicode symbols try: s = s.decode('unicode_escape').encode('utf-8') except UnicodeDecodeError: pass return s # Copy/paste helper to convert unicode results of json.load back to str # https://stackoverflow.com/questions/956867/how-to-get-string-objects-instead-of-unicode-from-json/13105359#13105359 def byteify(data): if isinstance(data, dict): return {byteify(key): byteify(value) for key, value in data.items()} elif isinstance(data, list): return [byteify(element) for element in data] elif isinstance(data, tuple): return tuple(byteify(element) for element in data) elif isinstance(data, str): return data.encode() else: return data def unicodeify(data): if isinstance(data, dict): return {unicodeify(key): unicodeify(value) for key, value in data.items()} elif isinstance(data, list): return [unicodeify(element) for element in data] elif isinstance(data, tuple): return tuple(unicodeify(element) for element in data) elif isinstance(data, bytes): return data.decode() else: return data def is_ascii_string(s): """ Check is string contains only ASCII characters :param s: string to check :return: True - string contains only ASCII characters """ try: s.encode(encoding='utf-8').decode('ascii') except UnicodeDecodeError: return False else: return True def escape_formatting_chars(text: str) -> str: """ Escape '%' characters inside text, except '%' followed by '(' """ def replace(match_obj): """ Generate string to replace from matched string '% ' -> '%% ' '%%c' -> '%%%%c' """ return match_obj.group()[:-1] * 2 + match_obj.group()[-1] # twice all '%' inside text except '%' followed by '(' and '%' at the end # '% hello' -> '%% hello' # '%% hello' -> '%%%% hello' text = re.sub(r"%+([^(])", replace, text) # replace '%' at the end text = re.sub(r"%$", r"%%", text) return text	6.47 KB	2026-02-10 15:10
📄 clhook.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import os import shutil import subprocess import simplejson as json import configparser CL_CONFIG = '/etc/sysconfig/cloudlinux' HOOK_SECTION = 'hooks' HOOK_SEPARATOR = ';' DEBUG_STRING = 'python-cllib:clhook:' def get_config(file_name): """ read config from file_name return config """ config = configparser.ConfigParser(interpolation=None, strict=False) config.optionxform = str # make config case sensitive config.read(file_name) return config def write_config(config, file_name, debug_local_name): """ write config to file_name return [BOOLEAN_STATE, MESSAGE] """ try: tmp_cfg_path = f'{file_name}.{os.getpid()}' with open(tmp_cfg_path, 'w', encoding='utf-8') as tmpconfig: config.write(tmpconfig) shutil.move(tmp_cfg_path, file_name) except IOError as e: return [False, f'{debug_local_name}{file_name} write error\n{str(e)}'] return [True, 'OK\n'] def get_hook_list_from_file(hook_prefix, config_file = CL_CONFIG): """ return [BOOLEAN_STATE, hook_list] """ hook_list = [] config = get_config(config_file) try: hook_list = config.get(HOOK_SECTION, hook_prefix).split(HOOK_SEPARATOR) except (configparser.NoOptionError, configparser.NoSectionError): return [False, hook_list] return [True, hook_list] def register_hook(hook_path, hook_prefix, config_file = CL_CONFIG): """ add hook to config file return [BOOLEAN_STATE, MESSAGE] """ debug_local_name = DEBUG_STRING + 'register_hook:' config = get_config(config_file) try: hook_list = config.get(HOOK_SECTION, hook_prefix).split(HOOK_SEPARATOR) if hook_path in hook_list: return [True, debug_local_name + 'Hook ' + str(hook_path) + ' already registred\n' ] hook_list.append(hook_path) config.set(HOOK_SECTION, hook_prefix, HOOK_SEPARATOR.join(hook_list)) except configparser.NoSectionError: config.add_section(HOOK_SECTION) config.set(HOOK_SECTION, hook_prefix, hook_path) except configparser.NoOptionError: config.set(HOOK_SECTION, hook_prefix, hook_path) return write_config(config, config_file, debug_local_name) def unregister_hook(hook_path, hook_prefix, config_file = CL_CONFIG): """ remove hook from config_file return [BOOLEAN_STATE, MESSAGE] """ debug_local_name = DEBUG_STRING + 'unregister_hook:' config = get_config(config_file) try: hook_list = config.get(HOOK_SECTION, hook_prefix).split(HOOK_SEPARATOR) hook_list.remove(hook_path) config.set(HOOK_SECTION, hook_prefix, HOOK_SEPARATOR.join(hook_list)) return write_config(config, config_file, debug_local_name) except (configparser.NoSectionError, configparser.NoOptionError, ValueError): return [True, 'OK\n'] def call_hook_list(hook_list, data): """ call hooks and send data as json to stdin return [BOOLEAN_STATUS, message] """ debug_local_name = DEBUG_STRING + 'call_hook:' json_encode = json.dumps(data) for hook in hook_list: if os.path.isfile(hook) and os.access(hook, os.X_OK): try: with subprocess.Popen( hook, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) as proc: proc.stdin.write(json_encode) proc.communicate() except OSError: return [False, debug_local_name + 'Hook call error\n'] else: return [False, debug_local_name + str(hook) + ' : isn`t file or isn`t runnable\n'] return [True, 'OK\n']	3.86 KB	2026-02-10 15:10
📄 cllog.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import datetime import fcntl import sys import psutil class ClAuditLog: def __init__(self, log_file_name: str): self._log_file_name = log_file_name def info_log_write(self, msg: str = ""): dt_now = datetime.datetime.now() current_proc = psutil.Process() # Initialize default values parent_proc_name, parent_proc_cmdline = "<Unknown_name>", "<Unknown_cmdline>" parent_proc_pid = current_proc.ppid() # Attempt to get the parent process properties parent_proc = current_proc.parent() if parent_proc is not None: with parent_proc.oneshot(): parent_proc_name = parent_proc.name() parent_proc_cmdline = " ".join(parent_proc.cmdline()) try: with open(self._log_file_name, "a", encoding="utf-8") as f: fcntl.lockf(f.fileno(), fcntl.LOCK_EX \| fcntl.LOCK_NB) f.write( f"{dt_now.strftime('%Y.%m.%d %H:%M:%S')}: " f"Parent process: {parent_proc_name} ({parent_proc_cmdline}) " f"(Parent PID: {parent_proc_pid}): Args: {sys.argv[1:]} {msg}\n" ) except (IOError, OSError): # Ignore all file errors pass	1.45 KB	2026-02-10 15:10
📄 cloutput.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import simplejson def prepare_attention_json(msg): msg = msg.replace(' \n', '; ').replace('\n', '; ') return simplejson.dumps({'success': 0, 'Error': msg}) def prepare_data_json(jsondata): return simplejson.dumps({'success': 1, 'data': jsondata})	471 B	2026-02-10 15:10
📄 clproc.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import os PROC_LVE = '/proc/lve' LIMIT_LVP_ID = 4294967295 class ProcLveError(Exception): pass class NoSuchLvp(ProcLveError): pass class ProcLve: def __init__(self, proc_lve=PROC_LVE): self._lvp = 'lvp' self.list = 'list' self.proc_lve = proc_lve self._version = None def proc_lve_list(self): """ Return path to /proc/lve/list """ return os.path.join(self.proc_lve, self.list) def version(self): """ Return /proc/lve filesystem version """ if self._version is not None: return self._version with open(self.proc_lve_list(), encoding='utf-8') as f: line = f.read(3) # we need only first three symbols for parse version self._version = int(line.rsplit(':', 1)[0]) return self._version def proc_lve_map(self): return os.path.join(self.proc_lve, 'map') def proc_lve_resellers(self): """ Return path to /proc/lve/resellers """ return os.path.join(self.proc_lve, 'resellers') def resellers_supported(self): """ Check present /proc/lve/resellers """ return os.path.exists(self.proc_lve_resellers()) def get_list_path(self, lvp_id=0): """ Generate path to list file """ if lvp_id == 0: return self.proc_lve_list() else: return os.path.join(self.proc_lve_resellers(), self._lvp + str(lvp_id), self.list) def _extract_id(self, line): return int(line.split()[0].split(',')[-1]) def _lines(self, lvp_id=0): path = self.get_list_path(lvp_id=lvp_id) try: with open(path, encoding='utf-8') as list_: lines = list_.read().strip() except IOError as e: if e.errno == 2: # 2 code error - No such file or directory raise NoSuchLvp(f'No such lvp id {lvp_id}') from e else: raise lines = lines.split('\n')[1:] for line in lines: yield self._extract_id(line), line.rstrip(os.linesep) def lines(self, lvp_id=0, without_limits=True): for lve_id, line in self._lines(lvp_id=lvp_id): if lve_id == LIMIT_LVP_ID and without_limits: continue yield line def lve_id_list(self, lvp_id=0): for lve_id, _ in self._lines(lvp_id=lvp_id): if lve_id in (LIMIT_LVP_ID, 0): continue yield lve_id def lvp_id_list(self): """ Obtain from /proc/lve/resellers lvp id list """ for lvp_dir in os.listdir(self.proc_lve_resellers()): if lvp_dir.startswith(self._lvp): yield int(lvp_dir[len(self._lvp):]) def check_inside_list(self, id_, lvp_id=0): for lve_id in self.lve_id_list(lvp_id=lvp_id): if lve_id == id_: return True def exist_lvp(self, lvp_id): """ Check present lve top container """ return os.path.exists(self.get_list_path(lvp_id)) def detect_inside_lvp(self, id_): """ Find in which lve top container """ for lvp_id in self.lvp_id_list(): if self.exist_lvp(lvp_id) and self.check_inside_list(id_, lvp_id): return lvp_id def map(self): """ Obtain map from /proc/lve/map as dict """ map_dict = {} with open(self.proc_lve_map(), encoding='utf-8') as map_: lines = map_.read().strip() lines = lines.split('\n')[1:] for line in lines: lve_id_str, lvp_id_str = line.strip().split() map_dict[int(lve_id_str)] = int(lvp_id_str) return map_dict def map_lve_id_list(self, lvp_id): return [lve_id_ for lve_id_, lvp_id_ in list(self.map().items()) if lvp_id_ == lvp_id]	4.05 KB	2026-02-10 15:10
📄 clpwd.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import os import pwd import grp from contextlib import contextmanager from .clexception import FormattedException class ClPwd: LOGIN_DEF_FILE = '/etc/login.defs' class NoSuchUserException(FormattedException): def __init__(self, user, args, kwargs): super(ClPwd.NoSuchUserException, self).__init__({ 'message': "No such user (%(user)s)", 'context': {'user': user} }, args, **kwargs) def __init__(self, min_uid=None): self._user_key_map = {} self._uid_key_map = {} self._user_full_map = {} self._uid_full_map = {} if min_uid is None: self._min_uid = self.get_sys_min_uid(500) else: self._min_uid = min_uid def get_user_dict(self): self._load_passwd_database() return self._user_key_map def get_uid_dict(self): self._load_passwd_database() return self._uid_key_map def get_user_full_dict(self): self._load_passwd_database() return self._user_full_map def get_uid_full_dict(self): self._load_passwd_database() return self._uid_full_map def get_pw_by_name(self, user): """ Return pw_entry for user """ try: return self.get_user_full_dict()[user] except KeyError as e: raise ClPwd.NoSuchUserException(user) from e def get_pw_by_uid(self, uid): """ Return list of passwd entries for uid """ try: return self.get_uid_full_dict()[uid] except KeyError as e: raise ClPwd.NoSuchUserException(uid) from e def get_uid(self, user): """ Returns uid for user """ try: return self.get_user_full_dict()[user].pw_uid except KeyError as e: raise ClPwd.NoSuchUserException(user) from e def get_homedir(self, user): """ Returns homedir for a user @param user: string @return: string """ try: return self.get_user_full_dict()[user].pw_dir except KeyError as e: raise ClPwd.NoSuchUserException(user) from e def _load_passwd_database(self): """ Loads the passwd database and fills user_to_uid and user_to_homedir maps """ if not self._uid_full_map: for entry in pwd.getpwall(): self._user_full_map[entry.pw_name] = entry if entry.pw_uid not in self._uid_full_map: self._uid_full_map[entry.pw_uid] = [] self._uid_full_map[entry.pw_uid].append(entry) if entry.pw_uid >= self._min_uid: self._user_key_map[entry.pw_name] = entry if entry.pw_uid not in self._uid_key_map: self._uid_key_map[entry.pw_uid] = [] self._uid_key_map[entry.pw_uid].append(entry) def get_names(self, uid): """ Return names of users with uid specified @param uid: int @return: list of strings """ try: entries = self.get_uid_full_dict()[uid] except KeyError as e: raise ClPwd.NoSuchUserException(uid) from e return [entry.pw_name for entry in entries] def get_sys_min_uid(self, def_min_uid=500): """ Return system defined MIN_UID from /etc/login.def or def_min_uid @param def_min_uid: int @return: MIN_UID: int """ if os.path.exists(self.LOGIN_DEF_FILE): with open(self.LOGIN_DEF_FILE, 'r', encoding='utf-8') as f: lines = f.readlines() for line in lines: if line.startswith('UID_MIN'): try: return int(line.split('UID_MIN')[1].strip()) except ValueError: pass return def_min_uid def drop_user_privileges(user=None, effective_or_real=True, set_env=True): """ Drop current root privileges to user :param effective_or_real: if True - drop euid, else - drop ruid :param user: name of unix user :param set_env: bool -> if true set $HOME and $USER env variables :return: None """ current_euid = os.geteuid() if current_euid == 0 and user: user_pwd = pwd.getpwnam(user) # set user's groups user_groups = [group.gr_gid for group in grp.getgrall() if user in group.gr_mem] main_user_group = user_pwd.pw_gid if main_user_group not in user_groups: user_groups.append(main_user_group) os.setgroups(user_groups) # set effective uid and gid if effective_or_real: os.setegid(user_pwd.pw_gid) os.seteuid(user_pwd.pw_uid) else: os.setgid(user_pwd.pw_gid) os.setuid(user_pwd.pw_uid) if set_env: # set user's env vars os.environ['USER'] = user os.environ['HOME'] = user_pwd.pw_dir @contextmanager def drop_privileges(user): """ Temporary drop privileges to some user :type user: str :raises: ClPwd.NoSuchUserException """ old_uid, old_gid, old_groups = os.getuid(), os.getgid(), os.getgroups() try: drop_user_privileges(user, effective_or_real=True, set_env=False) except KeyError as e: raise ClPwd.NoSuchUserException(user) from e try: yield finally: os.seteuid(old_uid) os.setegid(old_gid) os.setgroups(old_groups) def _resolve_doc_root_by_user(user): """ Resolve document root by username :param user: str -> name of unix user :return: str -> document root """ if user is None: raise FormattedException('Cannot resolve docroot without specified user') # to avoid circular imports from .cpapi import userdomains # pylint: disable=cyclic-import,import-outside-toplevel from .cpapi.cpapiexceptions import NoDomain # pylint: disable=cyclic-import,import-outside-toplevel domains_list = userdomains(user) # get document root for main domain try: _, result = domains_list[0] except IndexError as e: raise NoDomain({ 'message': 'No domain for user %(user)s found', 'context': {'user': user}, }) from e return result def resolve_username_and_doc_root(user=None, domain=None): """ Resolve username and doc_root by domain, or resolve document root by username, or resolve document root and username by effective uid :param user: str -> name of unix user :param domain: str -> domain of panel user :return: tuple -> user, doc_root """ # to avoid circular imports from .cpapi import docroot # pylint: disable=cyclic-import,import-outside-toplevel from .cpapi.cpapiexceptions import IncorrectData # pylint: disable=cyclic-import,import-outside-toplevel result_user = user result_doc_root = None current_euid = os.geteuid() if domain is not None: doc_root, domain_user = docroot(domain) if user is None: result_user = domain_user elif user != domain_user: raise IncorrectData("User and domain are in conflict") result_doc_root = doc_root elif user is not None: # we can obtain user name for domain result_doc_root = _resolve_doc_root_by_user(user=user) elif current_euid != 0: # get doc_root and username by current euid result_user = pwd.getpwuid(current_euid).pw_name result_doc_root = _resolve_doc_root_by_user(user=result_user) return result_user, result_doc_root	7.74 KB	2026-02-10 15:10
📄 clquota.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2022 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import re import os from typing import Optional from clcommon.utils import get_mount_point, run_command, ExternalProgramFailed USER_QUOTA_ON_PATTERN = re.compile(r'user quota on .+ $.+$ is on') def check_quota_enabled(path: str = '/home') -> Optional[str]: """ Check if quotas enabled and initialised for specific path :return: string message with giagnostick information if disabled; None if enabled """ if not os.path.isfile('/sbin/quotaon'): return 'Package "quota" not installed' try: mount_point = get_mount_point(path) except (OSError, ExternalProgramFailed) as e: return f'Error during mount point detection:\n{e}' try: quotaon_cmd = ['/sbin/quotaon', '-up', mount_point] quotaon_output = run_command(quotaon_cmd) except ExternalProgramFailed as e: quotaon_output = str(e) quotaon_output = quotaon_output.strip() # detect "user quota on / (/dev/mapper/VolGroup-lv_root) is on" if not USER_QUOTA_ON_PATTERN.search(quotaon_output): return quotaon_output return None	1.27 KB	2026-02-10 15:10
📄 clsec.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import os class ClSec: class NoRootException(Exception): def __init__(self): Exception.__init__(self, 'You need to be root to perform this command.') def is_root(): if os.geteuid() != 0: return False return True is_root = staticmethod(is_root) def check_root(): if not (ClSec.is_root()): raise ClSec.NoRootException() check_root = staticmethod(check_root)	657 B	2026-02-10 15:10
📄 clwpos_lib.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import json import re import os import subprocess import logging from pathlib import Path from typing import Union, Iterable, Optional, Tuple, List from enum import Enum from clcommon.clpwd import ClPwd from clcommon.utils import get_rhn_systemid_value, is_ubuntu VENDOR_UI_CONFIG_PATH = '/opt/cpvendor/config/cl-manager-ui-settings.json' class PluginType(Enum): """ Plugin types that are currently detected """ OBJECT_CACHE = 'object-cache' ADVANCED_CACHE = 'advanced-cache' class WpPlugins(Enum): """ Static WP plugin names, that are not detected dynamically from drop-in files, dir names, etc """ UNKNOWN = 'Unknown' WP_ROCKET = 'WP Rocket' ACCELERATE_WP = 'AccelerateWP' def clean_comment(line: str, is_multiline_comment: bool) -> Tuple[str, bool]: """ Yep, this bicycle is needed to handle different comment types in .php file https://www.php.net/manual/en/language.basic-syntax.comments.php and ensure that needed line is not under comment """ if is_multiline_comment: if '/' not in line: return '', True else: pos = line.find('/') part1, _ = clean_comment(line[:pos], True) part2, is_multiline_comment = clean_comment(line[pos + 2:], False) return part1 + part2, is_multiline_comment if '//' in line: pos = line.find('//') return line[:pos], False if '#' in line: pos = line.find('#') return line[:pos], False if '/' in line: pos = line.find('/') part1, _ = clean_comment(line[:pos], False) part2, is_multiline_comment = clean_comment(line[pos + 2:], True) return part1 + part2, is_multiline_comment return line, False def _is_real_file(file: str) -> bool: realpath_file = os.path.realpath(file) return os.path.isfile(realpath_file) def _check_wp_config_php(abs_path: Union[str, Path]) -> bool: """ WordPress looks for wp-config.php file in the (1) WordPress root and (2) one directory above the root. Check that there is no wp-settings.php file in the second case. This check helps when there is a nested installation, e.g / is WordPress and /wp_path/ is WordPress. """ try: wp_config_php = os.path.join(abs_path, 'wp-config.php') if os.path.exists(wp_config_php) and _is_real_file(wp_config_php): return True except OSError: pass abs_path_level_up = os.path.join(abs_path, os.pardir) wp_config_php_level_up = os.path.join(abs_path_level_up, 'wp-config.php') wp_settings_php = os.path.join(abs_path_level_up, 'wp-settings.php') wp_settings_php_exists = os.path.exists(wp_settings_php) and _is_real_file(wp_settings_php) return os.path.exists(wp_config_php_level_up) and \ not wp_settings_php_exists and \ _is_real_file(wp_config_php_level_up) def _is_real_dir(dir: str) -> bool: realpath_dir = os.path.realpath(dir) return os.path.isdir(realpath_dir) def _check_wp_includes(abs_path: Union[str, Path]) -> bool: """ Check wp-includes exists and is dir. """ wp_includes = os.path.join(abs_path, 'wp-includes') return 'wp-includes' in os.listdir(abs_path) and _is_real_dir(wp_includes) def is_wp_path(abs_path: Union[str, Path]) -> bool: """ Checks whether passed directory is a wordpress directory by checking presence of wp-includes folder and wp-config.php file. """ try: if not os.path.exists(abs_path): return False # skip paths that can't be read (wrong permissions etc) except OSError: return False try: return _check_wp_config_php(abs_path) and _check_wp_includes(abs_path) except OSError: pass return False def find_wp_paths(doc_root: str, excludes: Optional[List[str]] = None) -> Iterable[str]: """ Returns folder with wordpress Empty string is wp is in docroot dir :param doc_root: root path to start search from :param excludes: list of paths that must be excluded from search, e.g. subdomains """ if not os.path.exists(doc_root): return if is_wp_path(doc_root): yield '' for path in Path(doc_root).iterdir(): if not path.is_dir(): continue if excludes and str(path) in excludes: continue if is_wp_path(path): yield path.name def _is_php_define_var_found(var, path): """ Looks for defined php variable with true value """ r = re.compile(fr'^\sdefine\s$\s((\'{var}\')\|(\"{var}\"))\s,\strue\s$\s;') # let`s find needed setting by reading line by line with open(path, encoding='utf-8', errors='ignore') as f: is_multiline_comment = False while True: line = f.readline() if not line: break cleaned_line, is_multiline_comment = clean_comment(line, is_multiline_comment) if r.match(cleaned_line): return True return False def is_advanced_cache_enabled(wordpress_path: Path): """ Detects whether plugin is really enabled, cause not all plugins are enabled 'on load' # https://kevdees.com/what-are-wordpress-drop-in-plugins/ """ wp_config = wordpress_path.joinpath('wp-config.php') # really strange when main wordpress config is absent if not os.path.exists(wp_config): return False return _is_php_define_var_found('WP_CACHE', wp_config) def wp_rocket_plugin(drop_in_path): """ They are advising to check whether WP_ROCKET_ADVANCED_CACHE is defined to ensure plugin is working https://docs.wp-rocket.me/article/134-advanced-cache-error-message """ if accelerate_wp_plugin(drop_in_path) is None and \ _is_php_define_var_found('WP_ROCKET_ADVANCED_CACHE', drop_in_path): return WpPlugins.WP_ROCKET.value return None def accelerate_wp_plugin(drop_in_path): """ Checking if the plugin folder name exists in the drop-in """ with open(drop_in_path, encoding='utf-8', errors='ignore') as f: if '/clsop' in f.read(): return WpPlugins.ACCELERATE_WP.value return None def get_wp_cache_plugin(wordpress_path: Path, plugin_type: str): """ Looking for object-cache.php or advanced-cache.php in wordpress folder If found - tries to find 'plugin-owner' of <-cache>.php by content comparison If cannot be found -> tries to read <-cache>.php headers looking for Plugin name: <Plugin> """ wp_content_dir = wordpress_path.joinpath("wp-content") activated_cache = wp_content_dir.joinpath(f'{plugin_type}.php') if not os.path.exists(activated_cache): return None if plugin_type == PluginType.ADVANCED_CACHE.value and not is_advanced_cache_enabled(wordpress_path): return None plugins_dir = wp_content_dir.joinpath("plugins") plugin_name = get_wp_cache_plugin_by_scanning_dirs(activated_cache, plugins_dir) \ or get_wp_cache_plugin_by_header(activated_cache) \ or accelerate_wp_plugin(activated_cache) \ or wp_rocket_plugin(activated_cache) \ or WpPlugins.UNKNOWN.value return plugin_name def get_wp_cache_plugin_by_scanning_dirs(activated_plugin: Path, plugins_dir: Path) -> Optional[str]: """ Scanning plugins/ dir and looking for similar <object/advanced_cache>.php """ if not activated_plugin.is_file(): return None if not plugins_dir.is_dir(): return None activated_plugin_bytes = activated_plugin.read_bytes() for plugin in plugins_dir.iterdir(): for root, dirs, files in os.walk(plugin): if activated_plugin.name in files: plugin_object_cache_path = Path(root) / activated_plugin.name if plugin_object_cache_path.read_bytes() == activated_plugin_bytes: return plugin.name return None def get_wp_cache_plugin_by_header(activated_plugin: Path) -> Optional[str]: """ Looking for Plugin name: <Some name> in <object/advanced.php> headers """ if not os.path.exists(activated_plugin): return None # must be enough to loop through headers max_top_lines_count = 30 r = re.compile(r'^.plugin name:\s(?P<plugin_name>[\w ]+)\s*$', re.IGNORECASE) with open(activated_plugin, encoding='utf-8', errors='ignore') as f: for _ in range(max_top_lines_count): line = f.readline() match = r.search(line) if match is not None: return match.group('plugin_name') return None def get_wp_paths_with_enabled_module(user, user_wp_paths, plugin='object_cache'): """ Filter user`s wp paths with paths with enabled module """ paths_with_enabled_module = [] try: home = ClPwd().get_homedir(user) except ClPwd.NoSuchUserException: return [] config = os.path.join(home, '.clwpos', 'clwpos_config.json') if not os.path.exists(config): return [] try: with open(config, encoding='utf-8', errors='ignore') as f: conf = f.read() data = json.loads(conf) except Exception: return [] modules_data = data.get('docroots', {}).get('public_html', {}) for wp_path in user_wp_paths: if wp_path in modules_data and plugin in modules_data[wp_path]: paths_with_enabled_module.append(wp_path) return paths_with_enabled_module def install_accelerate_wp(): packages = [] if not os.path.exists('/usr/bin/cloudlinux-awp-admin'): packages.append('accelerate-wp') if not os.path.exists('/usr/sbin/cloudlinux-ssa-manager'): packages.append('alt-php-ssa') if not os.path.exists('/usr/sbin/cloudlinux-xray-agent'): packages.append('alt-php-xray') if packages: env = os.environ.copy() if is_ubuntu(): install_command = ['apt-get', '-y', 'install'] + packages env.update({'DEBIAN_FRONTEND': 'noninteractive'}) else: install_command = ["yum", "install", "-y"] + packages proc = subprocess.run(install_command, capture_output=True, text=True, check=False, env=env) logging.debug('Installing AccelerateWP packages captured out: %s, err: %s', proc.stdout, proc.stderr) if not os.path.exists('/usr/share/clos_ssa/ssa_enabled'): proc = subprocess.run(['/usr/sbin/cloudlinux-ssa-manager', 'enable-ssa'], capture_output=True, text=True, check=False) logging.debug('Activation SSA captured out: %s, err: %s', proc.stdout, proc.stderr) proc = subprocess.run( ['/usr/sbin/cloudlinux-autotracing', 'enable', '--all'], capture_output=True, text=True, check=False, ) logging.debug('Activation autotracing captured out: %s, err: %s', proc.stdout, proc.stderr) system_id = get_rhn_systemid_value('system_id') if system_id: proc = subprocess.run( ['/usr/sbin/cloudlinux-xray-manager', 'enable-user-agent', '--system_id', system_id.replace('ID-', '')], capture_output=True, text=True, check=False, ) logging.debug('Activation xray manager captured out: %s, err: %s', proc.stdout, proc.stderr) def configure_accelerate_wp(async_set_suite: bool = False, source: str = None): """ 1. Installs needed packages 2. Enables autotracing 3. Allows AccelerateWP Free for all """ install_accelerate_wp() if os.path.exists('/var/clwpos/admin/allowed_for_all_site_optimization.flag'): return if os.path.exists(VENDOR_UI_CONFIG_PATH): with open(VENDOR_UI_CONFIG_PATH, 'r', encoding='utf-8') as f: vendor_config_content = json.load(f) if vendor_config_content.get('features_override', {}).get('accelerate_wp') is False: logging.debug('Activation AccelerateWP Free skipped: Vendor UI skip the enabling of Accelerate WP Free') return command = ['/usr/bin/cloudlinux-awp-admin', 'set-suite', '--allowed-for-all', '--suites', 'accelerate_wp'] if source == 'BILLING_OVERRIDE': command.extend(['--source', 'BILLING_OVERRIDE']) # For WHMCS command if async_set_suite: subprocess.Popen( # pylint: disable=consider-using-with command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) else: proc = subprocess.run(command, check=True, capture_output=True, text=True) logging.debug('Activation AccelerateWP Free captured out: %s, err: %s', proc.stdout, proc.stderr) def configure_accelerate_wp_premium(source: str = None): install_accelerate_wp() # just in case server already allowed for all if os.path.exists('/var/clwpos/admin/allowed_for_all_object_cache.flag'): return command = ['/usr/bin/cloudlinux-awp-admin', 'set-suite', '--visible-for-all', '--suites', 'accelerate_wp_premium'] # If this is a request from WHMCS, we must pass --source to correctly save the suite status # and not overwrite the settings of users if source == 'BILLING_OVERRIDE': command.extend(['--source', 'BILLING_OVERRIDE', '--preserve-user-settings']) # set AccelerateWP visible for all users proc = subprocess.run(command, check=True, capture_output=True, text=True) logging.debug('Activation AccelerateWP Premium captured out: %s, err: %s', proc.stdout, proc.stderr) def configure_accelerate_wp_cdn(source: str = None): install_accelerate_wp() # just in case server already allowed for all if os.path.exists('/var/clwpos/admin/allowed_for_all_cdn.flag'): return # set AccelerateWP visible for all users command_cdn = ['/usr/bin/cloudlinux-awp-admin', 'set-suite', '--allowed-for-all', '--suites', 'accelerate_wp_cdn'] # If this is a request from WHMCS, we must pass --source to correctly save the suite status if source == 'BILLING_OVERRIDE': command_cdn.extend(['--source', 'BILLING_OVERRIDE']) proc = subprocess.run(command_cdn, check=True, capture_output=True, text=True) logging.debug('Activation AccelerateWP CDN captured out: %s, err: %s', proc.stdout, proc.stderr) # set AccelerateWP CDN PRO visible for all users command_cdn_pro = ['/usr/bin/cloudlinux-awp-admin', 'set-suite', '--visible-for-all', '--suites', 'accelerate_wp_cdn_pro'] # If this is a request from WHMCS, we must pass --source to correctly save the suite status # and not overwrite the settings of users if source == 'BILLING_OVERRIDE': command_cdn_pro.extend(['--source', 'BILLING_OVERRIDE', '--preserve-user-settings']) proc = subprocess.run(command_cdn_pro, check=True, capture_output=True, text=True) logging.debug('Activation AccelerateWP CDN PRO captured out: %s, err: %s', proc.stdout, proc.stderr) def configure_upgrade_url(upgrade_url): if not upgrade_url: return options_json = subprocess.run(['/usr/bin/cloudlinux-awp-admin', 'get-options'], check=True, capture_output=True, text=True).stdout options = json.loads(options_json) # nothing to do, upgrade url already set to same value if upgrade_url == options.get('upgrade_url'): return proc = subprocess.run(['/usr/bin/cloudlinux-awp-admin', 'set-options', '--upgrade-url', upgrade_url, '--suite', 'accelerate_wp_premium'], check=True, capture_output=True, text=True) logging.debug('Setting AccelerateWP Premium upgrade url captured out: %s, err: %s', proc.stdout, proc.stderr) def get_locale_from_envars(): """ Locale could be set via those envvars, let`s get them in same priority gettext does for envar in ('LANGUAGE', 'LC_ALL', 'LC_MESSAGES', 'LANG'): LANGUAGE = (unset), LC_ALL = (unset), LC_MESSAGES = "UTF-8", LANG = "uk_UA.UTF-8" """ return (os.environ.get('LANGUAGE') or os.environ.get('LC_ALL') or os.environ.get('LC_MESSAGES') or os.environ.get('LANG') or 'en_US')	16.6 KB	2026-02-10 15:10
📄 const.py # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # for backwards compatibility only from .features import Feature __all__ = ( 'Feature', )	277 B	2026-02-10 15:10
📁 cpapi	-	2026-03-19 15:49
📄 evr_utils.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 # # author: Ruslan Pisarev <rpisarev@cloudlinux.com> # created: 25.07.13 12:36 # description: alternatives,rpm_compare: hash-string for coding EVR # Examples of serialization you can look at tests/test_evr_utils.py from typing import Union, List, Tuple, Optional __all__ = [ 'serialize_evr' ] def encode_int_to_real_numbers_segment(intgr: int) -> List[int]: """ Encode int in real-numbers segment. See http://en.wikipedia.org/wiki/Arithmetic_coding. @param intgr: int for coding in Float an segment [seg_begin, seg_end] @return: list encoding segment """ lst = [] number = intgr while number > 0: number, remainder = divmod(number, 256) lst.append(remainder) lst.append(128 + len(lst)) lst.reverse() return lst def evr_to_float(rpm_data: List[str]) -> str: """ Encode List of parts of Version or Epoch or Release in real-numbers segment. See http://en.wikipedia.org/wiki/Arithmetic_coding. @param rpm_data: list to convert in double @return: Converted string """ evr = [] for elem in rpm_data: if isinstance(elem, int): evr.extend(encode_int_to_real_numbers_segment(elem)) elif isinstance(elem, str) and elem.isdigit(): evr.extend(encode_int_to_real_numbers_segment(int(elem))) elif isinstance(elem, str): evr.extend(ord(ch) for ch in elem) else: raise NameError(f'ThisStrange: {elem}') evr.append(0) converted_str = "".join([f"{n:02x}" for n in evr]) return converted_str def serialize_evr(evr: Union[List[str], Tuple[str], str]) -> str: """ Converts epoch, version and release of package to unique string. Ex: ['1', '2.4.4', '34'] -> '810100008102008104008104000081220000' ['0', '2.2.3', '76.el5_9.cloudlinux'] -> 80000081020081020081030000814c00656c00810500810900636c6f75646c696e75780000 @param evr: List from epoch, version and release @return: str for given list """ ret = '' if not isinstance(evr, (list, tuple)): evr = [evr] for i in evr: ret += evr_to_float(split_segments(i)) + '00' return ret def split_segments(string: Optional[str]) -> List[str]: """ Split str of epoch or version or release to numbers and strings. Ex: '76.el5_9.cloudlinux.2' -> ['76', 'el', '5', '9', 'cloudlinux', '2'] @param string: str of epoch or version or release @return: List strings and numbers from EVR """ if string is None: return [] prev_symb = string[0] begin_index = 0 segments = [] for i, el in enumerate(string): # Continue loop if a type of current symbol is equal to a type of previous symbol # Ex: In string 'el7' symbols 'e' and 'l' have the same type (alphabetic) and # we should take the part of string 'el' as whole if str(prev_symb + el).isdigit() or str(prev_symb + el).isalpha(): prev_symb = el continue # skip segement if it is't alpha/numeric if string[begin_index:i].isalnum(): segments.append(string[begin_index:i]) begin_index = i prev_symb = string[begin_index] # skip end of a string if it isn't alpha/numeric if string[begin_index:].isalnum(): segments.append(string[begin_index:]) return segments	3.55 KB	2026-02-10 15:10
📄 features.py # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT import itertools from enum import StrEnum from clcommon.lib.cledition import ( is_cl_admin_edition, is_cl_shared_pro_edition_safely, is_cl_solo_edition, is_container, ) from clcommon.utils import is_ubuntu class Feature(StrEnum): """ Enum of available CloudLinux features that vendors can disable or enable. """ PHP_SELECTOR = "php_selector" RUBY_SELECTOR = "ruby_selector" PYTHON_SELECTOR = "python_selector" NODEJS_SELECTOR = "nodejs_selector" LSAPI = "mod_lsapi" GOVERNOR = "mysql_governor" LVE = "lve" WIZARD = "wizard" CAGEFS = "cagefs" RESELLER_LIMITS = "reseller_limits" XRAY = "xray" WPOS = "accelerate_wp" AUTOTRACING = "autotracing" ALL_CL_FEATURES: list[Feature] = list(Feature) UBUNTU_BASED_CL_FEATURES_DISABLED = [ Feature.RUBY_SELECTOR, Feature.PYTHON_SELECTOR, Feature.NODEJS_SELECTOR, ] def _get_features_disabled_by_system_type() -> list[Feature]: """ Return overrides that are applied to the list of features depending on the current system base os. Right now we support two backends: rhel-like almalinux and ubuntu. """ if is_ubuntu(): return UBUNTU_BASED_CL_FEATURES_DISABLED return [] SOLO_CL_FEATURES_DISABLED = [ Feature.RUBY_SELECTOR, Feature.GOVERNOR, Feature.LVE, Feature.RESELLER_LIMITS, ] ADMIN_CL_FEATURES_DISABLED = [ Feature.RESELLER_LIMITS, Feature.GOVERNOR, ] SHARED_CL_FEATURES_DISABLED = [ Feature.XRAY, Feature.WPOS, Feature.AUTOTRACING, ] SHARED_PRO_CL_FEATURES_DISABLED: list[Feature] = [] def _get_features_disabled_by_license() -> list[Feature]: if is_cl_solo_edition(skip_jwt_check=True): return SOLO_CL_FEATURES_DISABLED elif is_cl_admin_edition(skip_jwt_check=True): return ADMIN_CL_FEATURES_DISABLED elif is_cl_shared_pro_edition_safely(): return SHARED_PRO_CL_FEATURES_DISABLED else: return SHARED_CL_FEATURES_DISABLED CONTAINER_CL_FEATURES_DISABLED = [ Feature.RESELLER_LIMITS, Feature.GOVERNOR, Feature.LVE, Feature.RUBY_SELECTOR, ] def _get_features_disabled_by_environment() -> list[Feature]: """ Return overrides that are applied to the list of features depending on the current system environment. Right now we support two envs: - bare metal or virtualized servers - virtuoso containers """ if is_container(): return CONTAINER_CL_FEATURES_DISABLED return [] def get_control_panel_disabled_features() -> list[Feature]: # not the best way, but we have to keep backwards compatibility in this lib from clcommon import cpapi # pylint: disable=cyclic-import,import-outside-toplevel return cpapi.get_unsupported_features() def get_cl_feature_status_map() -> dict[Feature, bool]: """ Return dict of CloudLinux features that can be used on current system. The key is the feature name and the value is True if feature is allowed to be used and False otherwise. By default, all features are assumed to be supported later we check different conditions and turn off features that are not supported on current CloudLinux intallation. """ panel_supported_features = get_control_panel_disabled_features() features_disabled_by_system_type = _get_features_disabled_by_system_type() features_disabled_by_license = _get_features_disabled_by_license() features_disabled_by_environment = _get_features_disabled_by_environment() unsuppored_features = set( itertools.chain( features_disabled_by_system_type, features_disabled_by_license, features_disabled_by_environment, panel_supported_features, ), ) # keep feature supported only if all three conditions met: # - feature is supported by current system type (rhel/debian) # - feature is supported by current license type (solo/admin/shared) # - feature is supported by current environment (container/vps) # - feature is supported by current control panel (cpanel/plesk/da) return { feature: feature not in unsuppored_features for feature in ALL_CL_FEATURES } def get_cl_enabled_features() -> list[Feature]: """ Return a list of enabled features. """ return [ feature for feature, is_feature_enabled in get_cl_feature_status_map().items() if is_feature_enabled ] MAX_HOSTING_ACCOUNTS_SOLO_EDITION = 1 MAX_HOSTING_ACCOUNTS_ADMIN_EDITION = 5 def get_hosting_accounts_limit() -> int \| None: """ Return the maximum number of hosting accounts that are currently allowed by CloudLinux license. """ if is_cl_solo_edition(skip_jwt_check=True): return MAX_HOSTING_ACCOUNTS_SOLO_EDITION elif is_cl_admin_edition(skip_jwt_check=True): return MAX_HOSTING_ACCOUNTS_ADMIN_EDITION return None	5.01 KB	2026-02-10 15:10
📄 group_info_reader.py # coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # import platform from time import sleep from typing import Dict # NOQA import requests from clcommon.utils import is_testing_enabled_repo, get_cl_version, is_ubuntu, get_os_version from clcommon.clexception import FormattedException class GroupInfoReaderError(FormattedException): def __init__(self, reason, message=None, details=None): if message is None: message = ( "Encountered an error while getting the remote available groups info. " "Try again later. If the same problem occurs again - contact CloudLinux support." ) super().__init__({ "message": message, "context": {"reason": reason}, "details": details, }) class GroupInfoReader: """ The purpose of this class is to get remote yum info about available groups: 1. alt-php 2. alt-nodejs 3. alt-ruby 4. alt-python There we get special url with json which depends on machine features: cl version: 6, 7, 6 hybrid architecture: x84_64, i386 enabled testing repositories status: beta, stable """ # TODO: consider about caching during some period of time GROUP_INFO = None GROUP_URL = None CL_BASE = "https://repo.cloudlinux.com/other/" UBUNTU_BASE = "https://repo.cloudlinux.com/cloudlinux-ubuntu/" @classmethod def group_url(cls): # type: () -> str """Get url with available groups json""" if cls.GROUP_URL is None: cls.GROUP_URL = cls._get_group_url() return cls.GROUP_URL @classmethod def _get_group_url(cls): # type () -> str """ Final url example for CloudLinux: https://repo.cloudlinux.com/other/cl6/package-info.x86_64.stable.json Final url example for Ubuntu: https://repo.cloudlinux.com/cloudlinux-ubuntu/package-info.20_04.json for cl6, arch x86_64 and disabled testing: stable :return: string with the result url or None if the CL version cannot be identified """ if is_ubuntu(): _, ubuntu_version = get_os_version() # 20.04 -> 20_04 ubuntu_version = ubuntu_version.replace(".", "_") suffix = f"package-info.{ubuntu_version}.json" return cls.UBUNTU_BASE + suffix else: arch = platform.machine() arch = "i386" if arch != "x86_64" else arch repo = "beta" if is_testing_enabled_repo() else "stable" cl_version = get_cl_version() if cl_version is None: return None # replace cl7h -> cl7, due to using same repo link for 7/7h cl_version = cl_version.replace("cl7h", "cl7") suffix = ".".join(["/package-info", arch, repo, "json"]) return cls.CL_BASE + cl_version + suffix @classmethod def get_available_groups(cls): # type: () -> Dict """ Sends request to group url, gets json and converts it to dict :return: dict with groups info """ if cls.GROUP_INFO is not None: return cls.GROUP_INFO url = cls.group_url() if url is None: raise GroupInfoReaderError( "Could not identify CloudLinux version", message="Could not identify CloudLinux version (using kernel version). " "Restart your system. If the same problem occurs again" " - contact CloudLinux support.", ) # requests.get() with 3 retries # Increasing sleep time attempts = 3 for i in range(attempts): try: cls.GROUP_INFO = requests.get(url, timeout=20).json() return cls.GROUP_INFO except requests.exceptions.RequestException as ex: if i + 1 >= attempts: err_message = f"Unable to reach {url}. Check your internet connection \ or try again later. If the same problem occurs again \ - contact CloudLinux support." raise GroupInfoReaderError( f"{url} - link unavailable", message=err_message, details=str(ex), ) from ex sleep(i + 1) return {} @classmethod def get_group_info(cls, group): # type: (str) -> Dict """ Filter dict with all available groups by special group name E.g: group = python we will get dict like that: {'alt-python27': {'version': '2.7.15', 'name': 'alt-python27', 'release': '1.el6'}, 'alt-python33': {'version': '3.3.3', 'name': 'alt-python33', 'release': '1.el6'}, 'alt-python34': {'version': '3.4.4', 'name': 'alt-python34', 'release': '1.el6'}...} :param group :rtype: dict with info per group """ group_info = {} available_groups = cls.get_available_groups() for grp in available_groups.get("groups_info", {}): if group in grp: group_info.update({grp: available_groups["groups_info"][grp]}) return group_info	5.29 KB	2026-02-10 15:10
📁 lib	-	2026-03-19 15:49
📄 lock.py #!/usr/bin/python # 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 # pylint: disable=consider-using-with # pylint: disable=unspecified-encoding import contextlib import fcntl import os def setup_lock(lockfile_path): lock_file_dir = os.path.dirname(lockfile_path) if lock_file_dir and not os.path.exists(lock_file_dir): os.makedirs(lock_file_dir) fd = open(lockfile_path, 'a+') # waiting till lock is acquired fcntl.flock(fd, fcntl.LOCK_EX) fd.seek(0, 0) fd.truncate() fd.write(str(os.getpid())) fd.flush() return fd def release_lock(fd): fcntl.flock(fd.fileno(), fcntl.LOCK_UN) fd.close() @contextlib.contextmanager def acquire_lock(lockfile_path): """ Usage: with acquire_lock(): ... do something ... """ fd = setup_lock(lockfile_path) try: yield finally: release_lock(fd)	1.02 KB	2026-02-10 15:10
📄 mail_helper.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2023 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import email.utils import logging import os import smtplib import subprocess from email.mime import base as mime_base, multipart from logging.handlers import SysLogHandler from typing import Optional from _socket import error as socketerror __author__ = "Aleksandr Shyshatsky" class MailSendFailed(Exception): pass class MailHelper: """Common class to send emails via local smtp or sendmail""" def __init__(self): if not logging.root.handlers: if os.path.exists('/dev/log'): SYS_LOG = SysLogHandler(address='/dev/log', facility=SysLogHandler.LOG_DAEMON) logging.root.addHandler(SYS_LOG) formatter_sys = logging.Formatter(fmt='mail_helper: (%(name)s) [%(levelname)s] %(message)s') SYS_LOG.setFormatter(formatter_sys) # write in system log only ERROR-s and FATALLERROR-s SYS_LOG.setLevel(logging.ERROR) else: logging.basicConfig() self._log = logging.getLogger() def _sendmail_console(self, msg: mime_base.MIMEBase, encoding_name: Optional[str] = None): """ Alternative function that should be called when we cannot send email via smtp :type msg: email.mime.base.MIMEBase :type encoding_name: encoding name to encode message text """ try: if encoding_name: s_message = msg.as_string().encode(encoding_name) else: s_message = msg.as_string() cmd = ["/usr/sbin/sendmail", "-t", "-oi"] with subprocess.Popen( cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) as proc: stdout = proc.communicate(s_message)[0] if proc.returncode != 0: raise subprocess.CalledProcessError(proc.returncode, cmd, stdout) except (OSError, subprocess.CalledProcessError) as ex: raise MailSendFailed(str(ex)) from ex def _sendmail_smtp(self, from_addr, to_addrs, msg, encoding_name: Optional[str] = None): """ Send email using smtp protocol; :type from_addr: str :type to_addrs: List[str] :type msg: email.mime.base.MIMEBase :type encoding_name: encoding name to encode message text """ try: smtp_instance = smtplib.SMTP('localhost') if encoding_name: s_message = msg.as_string().encode(encoding_name) else: s_message = msg.as_string() smtp_instance.sendmail(from_addr, to_addrs, s_message) smtp_instance.quit() except (smtplib.SMTPException, socketerror) as ex: raise MailSendFailed(str(ex)) from ex def _add_message_id(self, from_addr: str, msg: multipart.MIMEMultipart): """Add Message-ID to message if it is missing :type from_addr: str :type msg: multipart.MIMEMultipart """ if not msg['Message-ID']: msg['Message-ID'] = email.utils.make_msgid(domain=from_addr.split('@')[-1]) self._log.debug('Added Message-ID: %s', msg['Message-ID']) def sendmail(self, from_addr, to_addrs, msg, encoding_name: Optional[str] = None): """ Try to send message using local smtp; If that can't be done - try to use /usr/sbin/sendmail; :type from_addr: str :type to_addrs: List[str] :type msg: multipart.MIMEMultipart :type encoding_name: Optional[str] encoding name to encode message text """ self._log.debug("Trying to send mail with params: %s, %s, %s", from_addr, to_addrs, msg) self._add_message_id(from_addr, msg) try: self._sendmail_smtp(from_addr, to_addrs, msg, encoding_name) except MailSendFailed as smtp_exception: self._log.debug("Smtp send failed, trying to send via sendmail") try: self._sendmail_console(msg) except MailSendFailed: self._log.exception("Cannot send email neither via smtp nor using sendmail; " "Please, check your mail server configuration") self._log.error(str(smtp_exception))	4.45 KB	2026-02-10 15:10
📄 mysql_lib.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2023 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import os from typing import Iterable, List from collections import deque import pymysql from pymysql.cursors import DictCursor from clcommon.clconfpars import load_fast PLESK_CONFIG_PATH = '/etc/psa/psa.conf' MY_CONFIG_PATH = '/etc/my.cnf' MY_CONFIG_PATH2 = '/etc/mysql/my.cnf' class MySQLError(Exception): """ Exception related to operation with MySQL. """ class MySQLConnector: def __init__(self, as_dict: bool = False, kwargs): self.add_unix_socket_if_localhost(kwargs.get('host', 'localhost'), kwargs) self._connection = None self._cursor_type = DictCursor if as_dict else None self._connect_kwargs = kwargs def __del__(self): self.close() def __enter__(self): return self.connect() def __exit__(self, exc_type, exc_value, traceback): self.close() def connect(self) -> 'MySQLConnector': if self._connection is not None: return self try: self._connection = pymysql.connect(self._connect_kwargs) return self except pymysql.Error as e: raise MySQLError(str(e)) from e def close(self) -> None: if self._connection is not None: self._connection.close() self._connection = None def commit(self) -> None: if self._connection is not None: self._connection.commit() def cursor(self): return self.connection.cursor(self._cursor_type) @property def connection(self): if self._connection is None: self.connect() return self._connection def execute_query( self, sql_query: str, args: Iterable[object] \| None = None ) -> tuple[object, ...] \| dict[str, object]: """ Execute SQL query and return the result. """ with self.connection.cursor(self._cursor_type) as cursor: cursor.execute(sql_query, args=args) return cursor.fetchall() @staticmethod def add_unix_socket_if_localhost(host: str, kwargs: dict) -> None: """ Add 'unix_socket' to kwargs if host is 'localhost'. It seems that when the host is set to 'localhost', establishing a connection through TCP/IP might encounter issues due to some MySQL configuration options. To prioritize a Unix socket connection, we should include the 'unix_socket' query parameter. """ if host == 'localhost' and 'unix_socket' not in kwargs: kwargs['unix_socket'] = get_unix_socket_path() def get_rfc1738_db_uri(url): """ Get a modified RFC 1738 URL for a MySQL database connection. """ MySQLConnector.add_unix_socket_if_localhost(url.host, url.query) return url def get_unix_socket_path( *, plesk_config_path: str = PLESK_CONFIG_PATH, mysql_configs: Iterable[str] = (MY_CONFIG_PATH, MY_CONFIG_PATH2) ) -> str: """ Get the Unix socket path for MySQL connection. Check Plesk and MySQL config files for the socket path. If found, return that path. If not, return default socket path. """ if os.path.isfile(plesk_config_path): psa_conf = load_fast(plesk_config_path, delimiter=' ') psa_conf_socket = psa_conf.get('MYSQL_SOCKET') if psa_conf_socket is not None: return psa_conf_socket mysql_configs_queue = deque(mysql_configs) while len(mysql_configs_queue) > 0: my_conf_path = mysql_configs_queue.popleft() if not os.path.isfile(my_conf_path): continue my_conf = load_fast(my_conf_path, strip_quotes=True) # Use the first found socket path my_conf_socket = my_conf.get('socket') if my_conf_socket is not None: return my_conf_socket # Add included config files to the queue to check them for the socket path too mysql_configs_queue.extend(get_cnf_paths(my_conf_path)) return '/var/lib/mysql/mysql.sock' def get_cnf_paths(path: str) -> list[str]: """ Get all included config files paths in the given config file marked with !include and !includedir. Returns an empty list if there are no paths. """ include_dirs: List[str] = [] list_of_paths: List[str] = [] try: with open(path, "r", encoding="utf-8", errors="surrogateescape") as config_file: for line in config_file: line = line.strip() try: if line.startswith('!includedir'): included_dir_path = line.removeprefix('!includedir').strip() if os.path.isdir(included_dir_path): include_dirs.append(included_dir_path) elif line.startswith('!include '): included_file_path = line.removeprefix('!include').strip() if os.path.isfile(included_file_path): list_of_paths.append(included_file_path) except IndexError: # skip broken lines pass for dir_path in include_dirs: list_of_paths += get_cnf_files_in_dir(dir_path) except (OSError, IOError) as e: raise MySQLError(f"Error reading config file {path}: {e}") from e return list_of_paths def get_cnf_files_in_dir(path: str) -> list[str]: """ Get paths list in given dir Returns an empty list if there is no paths """ list_of_paths = [] if os.path.isdir(path): for item in os.listdir(path): if os.path.isfile(os.path.join(path, item)) and os.path.splitext(item)[1] == ".cnf": list_of_paths.append(os.path.join(path, item)) return list_of_paths	5.84 KB	2026-02-10 15:10
📄 php_conf_reader.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import copy import re class PhpConfBaseException(Exception): """Base class for all php.conf exceptions""" def __init__(self, message): super().__init__(message) class PhpConfReadError(PhpConfBaseException): def __init__(self, message=None): if message is None: message = "File open/read error" super().__init__(message) class PhpConfLoadException(PhpConfBaseException): def __init__(self, line_num, reason): message = f"Error at line {line_num}: {reason}. Please see " \ "http://docs.cloudlinux.com/index.html?custom_php_ini_options.html" super().__init__(message) class PhpConfNoSuchAlternativeException(PhpConfBaseException): def __init__(self, version): self.php_version = version message = f'No such alternative version ({version})' super().__init__(message) # For unit tests def _open(file_name): return open(file_name, mode='rb') class PhpConfReader: """ Class for read and parse /etc/cl.selector/php.conf During read this file, its syntax check is performed Contains methods for get its data for selectoctl and cagefsctl utilities in required formats """ DIRECTIVES_LIST = ['Directive', 'Default', 'Type', 'Comment', 'Range', 'Remark'] TYPES_LIST = ['value', 'list', 'bool'] def __init__(self, phpconf_path): self.phpconf_path = phpconf_path # /etc/cl.selector/php.conf contents as dictionary # Example: # { 'safe_mode': {'Default': 'Off', 'Comment': 'Enables PHP safe mode.', # 'Remark': '<5.4.0', 'Type': 'bool'}, # 'file_uploads': {'Default': 'On', 'Comment': 'Allows uploading files over HTTP.', # 'Type': 'bool'} # } self._php_conf = None def _read_file(self): """ Reads file and returns its content :return: List of filelines """ try: f = _open(self.phpconf_path) conf_lines = f.readlines() f.close() return conf_lines except (OSError, IOError) as e: # Open/read error raise PhpConfReadError(e.strerror) from e def _php_conf_load(self): data = {} conf_lines = self._read_file() try: directive = None current_section = None # Analyze lines for line_num, line in enumerate(conf_lines, start=1): # line example: 'Directive = upload_max_filesize' line = line.strip().decode('utf-8', "ignore") # pass comments if line.startswith('#'): continue # Empty line - separator between directives if len(line) == 0: directive = None current_section = None continue if '=' not in line: # Separator = not found in line - error raise PhpConfLoadException(line_num, "Required separator = not found") line_parts = line.split('=', 1) directive_name = line_parts[0].strip() directive_value = line_parts[1].strip() if directive_name not in self.DIRECTIVES_LIST: # Invalid directive raise PhpConfLoadException(line_num, f"Unknown directive '{directive_name}' found") if not directive and directive_name != 'Directive': # No current directive and some other directive found - error raise PhpConfLoadException(line_num, f"'{directive_name}' found, but no 'Directive' before it") if directive_name == 'Directive': if directive: # 'Directive' duplicate raise PhpConfLoadException(line_num, f"Directive '{directive_name}' duplicate") # Save current directive name and create placeholder for diirective data directive = directive_value current_section = directive_value data[current_section] = {} else: if directive_name in data[current_section]: # Directive already present raise PhpConfLoadException(line_num, f"Directive '{directive_name}' duplicate") if directive_name == 'Type' and directive_value not in self.TYPES_LIST: raise PhpConfLoadException(line_num, f"Directive is Type, but it value '{directive_value}' is invalid") data[current_section][directive_name] = directive_value # Successfull end self._php_conf = data except PhpConfLoadException as e: # Parsing error raise e except Exception as e: raise PhpConfBaseException(e) from e def get_config_for_selectorctl(self, php_version, php_versions_map): """ Retrives php.conf for selectorctl needs :param php_version: PHP version to filter options :param php_versions_map: Short to full PHP version map. Example: {'4.4', '4.4.9'} :return: dict Example: {'file_uploads': {'default': 'On', 'comment': 'Allows uploading files over HTTP.', 'type': 'bool'}, 'max_execution_time': {'default': '30', 'comment': 'The maximum time in seconds a script is allowed to run ' 'before it is terminated.', 'type': 'value'}, } """ # if config not loaded - load it if not self._php_conf: self._php_conf_load() out_dict = {} for directive_name, directive_data in self._php_conf.items(): # directive_data example: # { 'Default': 'Off', 'Comment': 'Enables PHP safe mode.', # 'Remark': '<5.4.0', 'Type': 'bool'} # if directive has Remark attribute, compare version with supplied and pass if need if 'Remark' in directive_data and not self._check_version(directive_data['Remark'], php_version, php_versions_map): continue out_dict[directive_name] = {} if 'Default' in directive_data: out_dict[directive_name]['default'] = directive_data['Default'] if 'Comment' in directive_data: out_dict[directive_name]['comment'] = directive_data['Comment'] if 'Type' in directive_data: out_dict[directive_name]['type'] = directive_data['Type'] if 'Range' in directive_data: out_dict[directive_name]['range'] = directive_data['Range'] return out_dict def get_config_for_cagefsctl(self): """ Retrives php.conf for selectorctl needs :return: dict Example: {'safe_mode': {'Remark': '<5.4.0', 'Type': 'bool'}, 'file_uploads': {'Type': 'bool'} } """ # if config not loaded - load it if not self._php_conf: self._php_conf_load() out_dict = {} for directive_name, directive_data in self._php_conf.items(): out_dict[directive_name] = copy.deepcopy(directive_data) if 'Comment' in out_dict[directive_name]: del out_dict[directive_name]['Comment'] return out_dict @staticmethod def _check_version(test, version, php_versions_map): """ Compares version in use and version required by PHP feature and return true if PHP feature satisfies :param test: Condition to filter from php.conf ('Remark' option value), such as <5.4.0 :param version: PHP Verson to check filter matching, such as 5.3 :param php_versions_map: Short to full PHP version map. Example: {'4.4': '4.4.9'} :return: bool: true - condition true, false - else """ # if test has 2 section, add third if len(test.split('.')) == 2: test += '.0' patt = re.compile(r'([<>=]{1,2})?(\d+\.\d+\.\d+)\.?') m = patt.match(test) if not m: raise PhpConfNoSuchAlternativeException(test) action = m.group(1) test_int = PhpConfReader._full_version_string_to_int(m.group(2), action) version_int = PhpConfReader._full_version_string_to_int(php_versions_map[version], action) if action == r'<' and version_int < test_int: return True if action == r'<=' and version_int <= test_int: return True if action == r'>' and version_int > test_int: return True if action == r'>=' and version_int >= test_int: return True if not action or action == r'=': if version_int == test_int: return True return False @staticmethod def _full_version_string_to_int(s_ver_full, action): """ Convert version string (such as '5.3.29') to int (10653) :param s_ver_full: PHP version string. Only full, 3-section versions are allowed :param action: Compare type. If None or '=' function will not use third section from version :return: integer """ v_array = [int(x) for x in s_ver_full.split('.')] ver_int = v_array[0] << 11 \| v_array[1] << 7 if not action or action == r'=': return ver_int return ver_int \| v_array[2]	9.77 KB	2026-02-10 15:10
📁 public_hooks	-	2026-03-19 15:49
📄 sysctl.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import os from configparser import ConfigParser from clcommon.utils import run_command, get_file_lines, ExternalProgramFailed from typing import AnyStr, List # NOQA SYSCTL_CL_CONF_FILE = '/etc/sysctl.d/90-cloudlinux.conf' SYSCTL_FILE = '/etc/sysctl.conf' class SysCtlConf: """ For reading params from sysctl """ SYSCTL_BIN = '/sbin/sysctl' def __init__(self, config_file=SYSCTL_FILE, mute_errors=True): # type: (AnyStr, bool) -> None """ :param config_file: path to user defined systcl config file :param mute_errors: T/F value to define should we skip errors or not (used in cldiag checker) """ self.config_file = config_file self.config_tmp_file = f'{self.config_file}.tmp' self.mute_errors = mute_errors def _apply_all(self): # type: () -> None """ Apply all params from sysctl.d & sysctl.conf """ cmd = [ self.SYSCTL_BIN, '--system', ] try: # if invalid param setting found, sysctl --system returns non-zero value on cl6 # on cl7 in such case there will be no error run_command(cmd) except ExternalProgramFailed: if not self.mute_errors: raise @classmethod def _read_sysctl_param(cls, name): # type: (AnyStr) -> AnyStr """ Read sysctl param :param name: name of sysctl param """ cmd = [ cls.SYSCTL_BIN, '-b', '-n', name, ] ret_code, std_out, std_in = run_command( cmd=cmd, return_full_output=True, ) value = std_out.strip() return value def _write_params_to_file(self, lines): # type: (List[AnyStr]) -> None """ Write sysctl params to sysctl.conf :param lines: content for writing to sysctl.conf """ with open(self.config_tmp_file, 'w', encoding='utf-8') as sysctl_conf: lines = ''.join(lines) sysctl_conf.write(lines) sysctl_conf.flush() os.fsync(sysctl_conf.fileno()) os.rename(self.config_tmp_file, self.config_file) @staticmethod def _get_param_name_from_line(line): # type: (AnyStr) -> AnyStr return line.split('=')[0].strip() def _read_sysctl_conf(self): # type: () -> List[AnyStr] """ Read content from sysctl.conf :return: lines from sysctl.conf """ result = get_file_lines(self.config_file) return result def has_parameter(self, param_name): # type: (AnyStr) -> bool file_lines = self._read_sysctl_conf() result = any(param_name == self._get_param_name_from_line(line) for line in file_lines) return result def get(self, name): # type: (AnyStr) -> AnyStr """ Get sysctl param by name :param name: name of sysctl param :return: value of sysctl param """ self._apply_all() value = self._read_sysctl_param(name) return value def set(self, name, value, overwrite=True): # type: (AnyStr, AnyStr, bool) -> None """ Set sysctl param by name :param overwrite: overwrite value of existed parameter :param name: name of sysctl param :param value: value of sysctl param """ param = f'{name} = {value}\n' sysctl_conf_output = list(self._read_sysctl_conf()) idx_param = -1 for i, line in enumerate(sysctl_conf_output): # skip commented strings if line.startswith('#'): continue key = self._get_param_name_from_line(line) if name == key: idx_param = i if idx_param == -1: sysctl_conf_output.append(param) elif overwrite: sysctl_conf_output[idx_param] = param self._write_params_to_file(sysctl_conf_output) self._apply_all() def remove(self, name): # type: (AnyStr) -> None """ Remove systcl param from config :param name: name of sysctl param """ self._apply_all() sysctl_conf_output = list(self._read_sysctl_conf()) idx_list = [] for i, line in enumerate(sysctl_conf_output): key = self._get_param_name_from_line(line) if name == key: idx_list.insert(0, i) for i in idx_list: del sysctl_conf_output[i] self._write_params_to_file(sysctl_conf_output) class SysCtlMigrate: """ Class for migrating of sysctl parameter from /etc/sysctl.conf to /etc/sysctl.conf.d/90-cloudlinux.conf """ MIGRATE_CONFIG_PATH = '/var/lve/cl-sysctl.migrate' MIGRATE_CONFIG_TMP_PATH = f'{MIGRATE_CONFIG_PATH}.tmp' MAIN_SECTION = 'main' def __init__(self): self._src_conf = SysCtlConf(config_file=SYSCTL_FILE) self._dst_conf = SysCtlConf(config_file=SYSCTL_CL_CONF_FILE) # migrate config self._migrate_config = ConfigParser(interpolation=None, strict=False) self._migrate_config.read(self.MIGRATE_CONFIG_PATH) def _is_migration_done(self, param_name): # type: (AnyStr) -> bool result = False if self._migrate_config.has_section(self.MAIN_SECTION) and \ self._migrate_config.has_option(self.MAIN_SECTION, param_name): result = self._migrate_config.getboolean(self.MAIN_SECTION, param_name) return result def _set_migration_state_to_done(self, param_name): # type: (AnyStr) -> None if not self._migrate_config.has_section(self.MAIN_SECTION): self._migrate_config.add_section(self.MAIN_SECTION) self._migrate_config.set(self.MAIN_SECTION, param_name, 'true') with open(self.MIGRATE_CONFIG_TMP_PATH, 'w', encoding='utf-8') as config_tmp: self._migrate_config.write(config_tmp) config_tmp.flush() os.fsync(config_tmp.fileno()) os.rename(self.MIGRATE_CONFIG_TMP_PATH, self.MIGRATE_CONFIG_PATH) def migrate(self, param_name, default_value): # type: (AnyStr, AnyStr) -> None """ Migrate sysctl parameter from one config to another in conformity with presence of parameter in source config and default value. All cases of using you can see in doc: https://docs.google.com/spreadsheets/d/1H_q3TA3CMFCj1YwAOn7G1LZgxcS1F4h90B6jCgiTUpo """ # We won't do anything if paramater already was migrated. if self._is_migration_done(param_name=param_name): return # We use value from src file if parameter is present in it if self._src_conf.has_parameter(param_name): value = self._src_conf.get(param_name) # otherwise, we use default value. else: value = default_value # Remove parameter from src file. self._src_conf.remove(param_name) if value is not None: # Write the migrated parameter to dst file # We shouldn't overwrite existing value in 90-cloudlinux.conf by default value, # but we should migrate value from src cfg to dst cfg overwrite = True if default_value is None else False self._dst_conf.set(param_name, value, overwrite=overwrite) # Set the migrate state to True. self._set_migration_state_to_done(param_name=param_name)	7.61 KB	2026-02-10 15:10
📄 ui_config.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 # import json import copy from collections.abc import Mapping UI_CONFIG_FILE = "/usr/share/l.v.e-manager/lvemanager-config.json" DEFAULT_UI_CONFIG = {'inodeLimits': {'showUserInodesUsage': False}, 'uiSettings': {'hideRubyApp': False, 'hideLVEUserStat': False, 'hidePythonApp': False, 'hideNodeJsApp': False, 'hidePHPextensions': False, 'hideDomainsTab': False, 'hidePhpApp': False, 'hideXrayApp': True, 'hideAccelerateWPApp': True, } } class UIConfig: def __init__(self, ui_config_path=UI_CONFIG_FILE): self.ui_config_path = ui_config_path self.file_content = {} self.config = {} try: with open(ui_config_path, encoding='utf-8') as json_file: self.file_content = json.load(json_file) self.config = self.file_content['ui_config'] except IOError: self.config = DEFAULT_UI_CONFIG except (ValueError, KeyError): self.file_content['ui_config'] = DEFAULT_UI_CONFIG self.config = DEFAULT_UI_CONFIG self._update_file(ui_config_path) def _update_file(self, file_path): with open(file_path, 'w', encoding='utf-8') as update_file: json.dump(self.file_content, update_file) def get_config(self): initial_config = copy.deepcopy(DEFAULT_UI_CONFIG) self._recursive_assign(initial_config, self.config) return initial_config def set_config(self, new_config): initial_config = copy.deepcopy(self.get_config()) self._recursive_assign(initial_config, new_config) with open(self.ui_config_path, 'w', encoding='utf-8') as json_file: json.dump({'ui_config': initial_config}, json_file) def get_param(self, param, parent_param=None): config = self.get_config() if parent_param is None: return config.get(param) elif config.get(parent_param) is not None: return config.get(parent_param).get(param) return None def _recursive_assign(self, initial_config, new_config): """ Function that recursively assignees values received from the UI :param initial_config: Config to which values are assigned :param new_config: New config file received from UI :return: """ for key, value in new_config.items(): if isinstance(value, Mapping): initial_config[key] = self._recursive_assign(initial_config.get(key, {}), value) elif value is not None and initial_config.get(key) is not None: initial_config[key] = value return initial_config	3.12 KB	2026-02-10 15:10
📄 utils.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2022 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import configparser import datetime import grp import io import os import platform import pwd import re import stat import subprocess import sys from configparser import ConfigParser, Error from functools import cache from typing import Any, AnyStr, Dict, Optional, Tuple, Union # NOQA import psutil import secureio from lxml import etree # These commands used to be in this file, but is_ubuntu cannot be imported if they're here # due to a circular import with clcommon.lib.__init__ -> clcommon.lib.mysql_governor_lib. # To avoid that, we move them to a separate file and import into this namespace # so that things like `from clcommon.utils import run_command` still work. from clcommon.utils_cmd import ( # NOQA ExternalProgramFailed, check_command, exec_utility, run_command, ) # Not all imports are used in this file - they are made so that other # modules can import them from utils, not utils_cmd. Therefore, NOQA. RHN_SYSTEMID_FILE = "/etc/sysconfig/rhn/systemid" WEEK_DAYS = ("Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun") LITESPEED_STATUS_FOR_USERS = "/opt/cloudlinux/litespeed_status" NGINX_STATUS_FOR_USERS = "/opt/cloudlinux/nginx_status" APACHE2NGINX_STATE = "/opt/apache2nginx/state" def create_symlink(link_value, link_path): """ Create symlink link_path -> link_value if it does not exist or points to different location :param link_value: path that symlink should point to (symlink value) :type link_value: str :param link_path: path where to create symlink :type link_path: str """ link_to = None if os.path.islink(link_path): try: link_to = os.readlink(link_path) except OSError: pass if link_value != link_to: try: os.unlink(link_path) except OSError: pass os.symlink(link_value, link_path) def get_file_lines(path, unicode_errors_handle=None): """ Read file and return file's lines errors param may be passed to define how handle unicode errors, errors=None is default value of open() :param path: path to file :param unicode_errors_handle: how to handle unicode errors :return: list of file's lines """ content = [] if os.path.isfile(path): with open(path, encoding="utf-8", errors=unicode_errors_handle) as f: content = f.readlines() return content def write_file_lines(path, content, mode): """ Write lines to file :param content: list of lines for writing to file :param path: path to file :param mode: open mode :return: None """ with open(path, mode, encoding="utf-8") as f: f.writelines(content) def delete_line_from_file(path, line): r""" Delete line from file. Return True when line(s) have been deleted, False otherwise (specified line is not found) :param path: path to file :type path: string :param line: line to delete without EOL ('\n') :type line: string :rtype bool """ file_lines = get_file_lines(path) out_file_lines = [item for item in file_lines if line != item.rstrip("\n")] found = len(file_lines) != len(out_file_lines) write_file_lines(path, out_file_lines, "w+") return found def is_root_or_exit(): """ Check whether current user is effectively root and exit if not """ euid = os.geteuid() if euid != 0: try: # Additional admins placed in this special group # by lvemanager hooks to add root-like privileges to them gr_gid = grp.getgrnam("clsupergid").gr_gid if gr_gid in os.getgroups() or os.getegid() == gr_gid: return except KeyError: pass # No group - no privileges print("Error: root privileges required. Abort.", file=sys.stderr) sys.exit(-1) def is_ea4(): """ Detect whether EA4 is installed. :return: True - EA4 present; False - EA4 absent """ return os.path.isfile("/etc/cpanel/ea4/is_ea4") def grep( pattern, path=None, fixed_string=False, match_any_position=True, multiple_search=False, data_from_file=None, ): """ Grep pattern in file :param multiple_search: if True - search all match, False - search first match :param pattern: pattern for search :param path: path to file :param data_from_file: read data from file for parsing :param fixed_string: if True - search only fixed string, False - search by regexp :param match_any_position: if True - search any match position, False - search only from string begin :return: Generator with matched strings """ if data_from_file is None: data_from_file = get_file_lines(path) result = None if not fixed_string: # It's append the symbol ^ to the regexp # if we are searching from the begin of a string and by the regexp if not pattern.startswith("^") and not match_any_position: pattern = f"^{pattern}" pattern_comp = re.compile(pattern) else: pattern_comp = None for line in data_from_file: if fixed_string: if (match_any_position and line.find(pattern) != -1) or line.startswith(pattern): result = line elif pattern_comp.search(line): result = line if multiple_search and result is not None: yield result elif result is not None: break result = None if result is not None: yield result def _parse_systemid_file(): """ :rtype: lxml.etree._ElementTree obj """ return etree.parse(RHN_SYSTEMID_FILE) # pylint: disable=c-extension-no-member def get_rhn_systemid_value(name) -> Optional[str]: """ find a member in xml by name and return value :type name: str :rtype: str\|None """ try: rhn_systemid_xml = _parse_systemid_file() for member in rhn_systemid_xml.iter("member"): if member.find("name").text == name: return member.find("value")[0].text except (OSError, IndexError, KeyError, etree.ParseError): # pylint: disable=c-extension-no-member return None return None def get_file_system_in_which_file_is_stored_on(file_path): # type: (str) -> Dict[str, Any] """ Detect the file system on which a given file is stored. E.g., the file can be stored in NFS and this can affect the normal operation of the file. We want to receive information about FS in emergency situations during reading or writing :param file_path: path to file, for which we want to detect file system :return: dict, which contains two keys: key 'success' can be equals to False if we got error or True if we got normal result key 'details' can contais error string if key 'success' is False or result if key 'success' is True """ result = { "success": False, "details": f'File "{file_path}" isn\'t exists', } if not os.path.exists(file_path): return result # Command: mount \| grep "on $(df <file_name> \| tail -n 1 \| awk '{print $NF}') type" # Result: /usr/tmpDSK on /var/tmp type ext3 (rw,nosuid,noexec,relatime,data=ordered) try: mount_point = subprocess.check_output( [f"df {file_path} \| tail -n 1 \| awk '{{print $NF}}'"], shell=True, executable="/bin/bash", text=True, ).strip() data = subprocess.check_output( [f'mount \| grep "on {mount_point} type"'], shell=True, executable="/bin/bash", text=True, ).strip() result["success"] = True result["details"] = data except (subprocess.CalledProcessError, OSError) as err: result["details"] = f'We can\'t get file system for file "{file_path}". Exception "{err}"' return result def is_testing_enabled_repo() -> bool: """ Check if testing is enabled in /etc/yum.repos.d/cloudlinux.repo config :return: bool value if testing enabled or not """ parser = ConfigParser(interpolation=None, strict=False) try: parser.read("/etc/yum.repos.d/cloudlinux.repo") res = parser.getboolean("cloudlinux-updates-testing", "enabled") # use base exception for config parser class except Error: res = False return res def get_cl_version_ubuntu() -> str \| None: """ Return the CloudLinux OS version. Ubuntu + CL Extension is a separate case from normal CL variants. The string returned is based on /etc/os-release and contains the major version of the Ubuntu system. Previously Ubuntu 20.04 identified itself as "cl8" in this scenario. :return: Appropriate version string or None if the version is not recognized. """ ubuntu_map = { "jammy": "ubuntu22", "focal": "ubuntu20", } # On Ubuntu, this is a codename like "focal" or "jammy" release = get_rhn_systemid_value("os_release") if release and release in ubuntu_map: return ubuntu_map[release] # Gets a version string like "20.04" or "22.04" version_id = get_os_version()[1] if version_id: version_id = version_id.split(".")[0] return f"ubuntu{version_id}" return None def get_cl_version() -> str \| None: """ Return the CloudLinux OS version taking into account the release version from `platform.release()` or the RHN systemid file. E.g: release = 2.6.32-896.16.1.lve1.4.54.el6.x86_64 5.14.0-162.6.1.el9_1.x86_64 The release string is converted to return the appropriate version string: el6 = cl6 el8 = cl8 """ release: str \| None if is_ubuntu(): return get_cl_version_ubuntu() check_vals_decoder = { "el6.": "cl6", "el6h.": "cl6h", "el7.": "cl7", "el7h.": "cl7h", "el8.": "cl8", ".el9": "cl9", # note the dot at the beginning because of possible `el9_1` and so on ".el10": "cl10", } release = platform.release() for check_val, cl_version in check_vals_decoder.items(): if release and check_val in release: return cl_version release = get_rhn_systemid_value("os_release") if release is None: return None accepted_cl_versions = range(6, 10 + 1) # CL6 to CL10 for ver in accepted_cl_versions: if str(ver) in release: return f"cl{ver}" return None @cache def get_virt_type() -> Optional[str]: """ Return virtualization type on current system. It is reachable via virt-what utility. E.g.: 'kvm', 'bhyve', 'openvz', 'qemu' All acceptable outputs are listed here: https://people.redhat.com/~rjones/virt-what/virt-what.txt Output will be returned with at least two rows Sample: > kvm > Furthermore, there is a possibility for multiple text rows Sample: > xen > xen-domU That's why, the result will be taken from a first row. If the output is empty, and there were no errors, the machine is either non-virtual, or virt-what tool isn't familiar with it's hypervisor. But the list of supported hypervisors and containers covers all popular VMs types. :return: virt_type - Optional[AnyStr] - appropriate virtualization type string, - 'physical' if there is no virtualization, - None if there was an error """ try: virt_what_output = run_command(["/usr/sbin/virt-what"]).strip() except (subprocess.CalledProcessError, FileNotFoundError): return None # Check for the non-empty output - virtualization exists if virt_what_output != "": return virt_what_output.split("\n", maxsplit=1)[0] return "physical" def check_pid(pid: int): """ Check for a process existence by os.kill command If os.kill will be used as os.kill(pid, 0), it will just check for a presence of such PID And if such pid can't be reached with kill method, there will be raised OSError """ try: os.kill(pid, 0) except OSError: return False else: return True def is_process_running(pid_file_path: str) -> bool: """ Check if process running using pid file Arguments: pid_file_path: path to the pid file of service Returns: bool: True or False """ if os.path.isfile(pid_file_path): with open(pid_file_path, encoding="utf-8") as f: try: return check_pid(int(f.read().strip())) except ValueError: pass return False def is_litespeed_running() -> bool: """ Detect that server works under Litespeed. Note: be careful when modifying this method. It is used in X-Ray, ask @dkavchuk or someone else from C-Projects team for details. return: True - LS working; False - LS not running (stopped or absent) """ if os.geteuid() == 0: litespeed_pid_path = "/tmp/lshttpd/lshttpd.pid" openlitespeed_pid_path = "/var/run/openlitespeed.pid" return is_process_running(litespeed_pid_path) or is_process_running(openlitespeed_pid_path) if not os.path.exists(LITESPEED_STATUS_FOR_USERS): return False with open(LITESPEED_STATUS_FOR_USERS, encoding="utf-8") as f: status = f.read().strip() return status == "0" def is_nginx_running() -> bool: """ Detect that server works under nginx. Note: be careful when modifying this method. It is used in X-Ray, ask @dkavchuk or someone else from C-Projects team for details. return: True - nginx working; False - nginx not running (stopped or absent) """ if os.geteuid() == 0: pid_file_path = "/run/nginx.pid" return is_process_running(pid_file_path) if not os.path.exists(NGINX_STATUS_FOR_USERS): return False with open(NGINX_STATUS_FOR_USERS, encoding="utf-8") as f: status = f.read().strip() return status == "0" def is_apache2nginx_running() -> bool: """ Detect that server works under apache2nginx. Note: be careful when modifying this method. It is used in X-Ray, ask @dkavchuk or someone else from C-Projects team for details. return: True - apache2nginx working; False - apache2nginx not running (stopped or absent) """ if not os.path.exists(APACHE2NGINX_STATE): return False with open(APACHE2NGINX_STATE, encoding="utf-8") as f: status = f.read().strip() return status == "on" def get_passenger_package_name(): """ Return proper passenger package according to apache version :rtype: str """ if is_ea4(): # Package that brings passenger on cPanel has different names on different systems: # CL6 - ea-ruby24-mod_passenger # CL7/8 - ea-ruby27-mod_passenger # CL9 - ea-apache24-mod-passenger # But they all provide "apache24-passenger" package return "apache24-passenger" return "alt-mod-passenger" def is_package_installed(package_name: str) -> bool: """ Check that the given package is installed on the server. :param package_name: Package name to check. :return: True if package is installed, False otherwise. """ try: if is_ubuntu(): # We can't depend on dpkg -s because it can return 0 even if package is not installed # For example, if a package was removed, but its configuration files are still present pkg_status = run_command(["dpkg-query", "-W", "-f='${Status}\n'", package_name]) # Normal output for installed packages is "install ok installed" # An uninstalled package might return "deinstall ok config-files" - but would still yield 0 in `dpkg -s` if "install ok installed" not in pkg_status: return False else: run_command(["rpm", "-q", package_name]) except ExternalProgramFailed: return False return True def get_package_db_errors(): """ Pick the package manager to check depending on the OS. If we're not on Ubuntu, assume a CL variant with RPM. :return: Error string, if any, None otherwise. """ if is_ubuntu(): return get_apt_db_errors() return get_rpm_db_errors() def get_apt_db_errors(): """ Check the dpkg DB as described in https://man7.org/linux/man-pages/man1/dpkg.1.html See `--audit`. :return: Error string, if any, None otherwise. """ try: with subprocess.Popen( ["dpkg", "--audit"], shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) as proc: std_out, std_err = proc.communicate() # Should return 1 if the audit fails. if proc.returncode != 0: # Check error return f"dpkg audit error: {std_out}\n{std_err}." except OSError as err: return str(err) # There is no dpkg DB errors return None def get_rpm_db_errors(): """ Check RPM DB consistency. :return: None - No RPM DB errors string_message - Error description """ doc_link = ( "https://cloudlinux.zendesk.com/hc/en-us/articles/115004075294-Fix-rpmdb-Thread-died-in-Berkeley-DB-library" ) try: with subprocess.Popen( ["/bin/rpm", "--dbpath", "/var/lib/rpm", "--verifydb"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) as proc: std_out, std_err = proc.communicate() if proc.returncode != 0: # Check error return f"RPM DB check error: {std_out}\n{std_err}.\nSee doc: {doc_link}" except OSError as err: return str(err) # There is no RPM DB errors return None def silence_stdout_until_process_exit(): """ Upon process exit, Sentry sometimes prints: Sentry is attempting to send 1 pending error messages Waiting up to 10 seconds Press Ctrl-C to quit This causes broken JSON in output. See also this issue: https://github.com/getsentry/raven-python/issues/904 """ sys.stdout = io.StringIO() sys.stderr = io.StringIO() def mod_makedirs(path, mod): """ Create directories with desired permissions Changed in version 3.7: The mode argument no longer affects the file permission bits of newly-created intermediate-level directories. Because it we use umask while creating dirs :param mod: desired permissions """ inverted_mod = 0o777 - (mod & 0o777) with secureio.set_umask(inverted_mod): os.makedirs(path, mod) def is_user_present(username: str) -> bool: """ Check user existence in the system """ try: pwd.getpwnam(username) except KeyError: return False return True def is_uid_present(uid: int) -> bool: """ Check uid existence in the system """ try: pwd.getpwuid(uid) except KeyError: return False return True def is_socket_file(path: AnyStr) -> Optional[bool]: """ Check that file by path is socket """ try: mode = os.lstat(path).st_mode except (FileNotFoundError, OSError): return None is_socket = stat.S_ISSOCK(mode) return is_socket def get_system_runlevel() -> int: """ Get number of system run level by command `runlevel`. """ output = subprocess.check_output( "/sbin/runlevel", shell=True, executable="/bin/bash", text=True, ) # output: N 5 # there is N - previous value of runlevel, 5 is current runlevel result = output.strip().split()[1] # `S` means single-mode. Equals to level `1` level = 1 if result == "S" else int(result) return level def _get_service_state_on_init_d_system(service_name: str) -> Tuple[bool, bool]: """ Return state of a service (present and enabled) for init.d system. Returns False, False if a service doesn't exist Returns True, False if a service exists and it's not enabled Returns True, True if a service exists and it's enabled """ runlevel = get_system_runlevel() try: # LANG=C parameter allows to use C programming language (en-US) # locale instead of current, since there can be non-English # results in the chkconfig output, while we search for the `on` output = subprocess.check_output( # the command return non-zero code if a service doesn't exist f"LANG=C /sbin/chkconfig --list {service_name}", shell=True, executable="/bin/bash", text=True, ) except (subprocess.CalledProcessError, FileNotFoundError): return False, False # split output: # `cl_plus_sender 0:off 1:off 2:on 3:on 4:on 5:on 6:off` output_list = output.strip().split() for state_info in output_list[1:]: state_list = state_info.strip().split(":") is_active = state_list[1] == "on" state_runlevel = int(state_list[0]) if runlevel == state_runlevel: return True, is_active return True, False def _get_service_state_on_systemd_system(service_name: str) -> Tuple[bool, bool]: """ Return state of service (present and enabled) for systemd system Returns False, False if a service doesn't exist Returns True, False if a service exists and it's not enabled Returns True, True if a service exists and it's enabled """ try: subprocess.check_call( # the command return non-zero code if a service isn't enabled or # it's not present f"/usr/bin/systemctl is-enabled {service_name} &> /dev/null", shell=True, executable="/bin/bash", ) return True, True except (subprocess.CalledProcessError, FileNotFoundError): try: with subprocess.Popen( [ "/usr/bin/systemctl", "status", service_name, ], stderr=subprocess.DEVNULL, stdout=subprocess.DEVNULL, ) as proc: proc.communicate() # 0 - service is run # 3 - service is stopped # 4 - service doesn't exist if proc.returncode in (0, 3): return True, False return False, False except (subprocess.CalledProcessError, FileNotFoundError): return False, False def service_is_enabled_and_present(service_name: str) -> Tuple[bool, bool]: """ Return state of service (present and enabled) :param service_name: name of a service """ cl_ver = get_cl_version() if cl_ver and "cl6" in cl_ver: is_present, is_enabled = _get_service_state_on_init_d_system( service_name=service_name, ) else: is_present, is_enabled = _get_service_state_on_systemd_system( service_name=service_name, ) return is_present, is_enabled def process_is_running(process_file_path: str, strict_match: bool) -> bool: """ Check that a file in path is running. You can get false-postive if parameter `strict_match` == False, process is not running, but someone on server open file by path `process_file_path` in an editor :param process_file_path: path to a file which is run :param strict_match: we use parameter `process_file_path` as full cmd line with args for comparing if `strict_match` == True. :return: True if it's running, False - is not, """ if not os.path.exists(process_file_path): raise FileNotFoundError(f'Process file in path "{process_file_path}" does not exist') for process in psutil.process_iter(["cmdline"]): try: if (not strict_match and process_file_path in process.cmdline()) or ( strict_match and process_file_path == " ".join(process.cmdline()) ): return True except psutil.NoSuchProcess: continue return False def get_weekday(dt: Union[datetime.datetime, datetime.date]) -> str: """ Get string representation of weekday from datetime.datetime or datetime.date. Returns shortened version of weekday from WEEK_DAYS. """ if not (isinstance(dt, (datetime.datetime, datetime.date))): raise TypeError(f"Require object of type datetime.datetime or datetime.date, but passed {type(dt)}") return WEEK_DAYS[dt.weekday()] def find_module_param_in_config( config_path: str, apache_module_name: str, param_name: str, default: int = None, ) -> Tuple[int, str]: """ Parse httpd config for details about the mpm module used. :param config_path: path for configuration file with modules :param apache_module_name: expected mpm module. Can be `event`, `worker`, `prefork` :param param_name: name of parameter to find :param default: default value for parameter, if there won't be record :return: tuple with param value and text result of operation Example of config file content: <IfModule mpm_prefork_module> ................. MaxRequestWorkers 450 </IfModule> -- <IfModule mpm_worker_module> ................. MaxRequestWorkers 300 </IfModule> -- <IfModule mpm_event_module> ................. MaxRequestWorkers 2048 </IfModule> """ if_module_line = f"<IfModule mpm_{apache_module_name}_module>" section_lines = [] mpm_lines = get_file_lines(config_path) is_section_found = False for line in mpm_lines: line = line.strip() if line == if_module_line: is_section_found = True continue if is_section_found and line == "</IfModule>": # End of section break if is_section_found: section_lines.append(line) # 2. Find directive in found section grep_result_list = list( grep( param_name, multiple_search=True, fixed_string=False, data_from_file=section_lines, ) ) mrw_list = [directive.strip() for directive in grep_result_list if directive.strip().startswith(param_name)] # 3. Parse all lines with directive and find correct # There is no custom setting for parameter if not mrw_list and default is not None: # Plesk case, when we can use defaults return default, "OK" if not mrw_list and default is None: # DA case, when we don't know about defaults return ( 0, f"MaxRequestWorkers directive not found for mpm_{apache_module_name}_module module in {config_path}", ) # There can be few records with MaxRequestWorkers, so we need # to take the last one parts = mrw_list[-1].split(" ") max_request_workers = int(parts[-1]) return max_request_workers, "OK" def get_kmodlve_module_version() -> str: """ Return kmodlve module's version. Content of '/sys/module/kmodlve/version' looks like '2.0-30.el8'. """ kmodlve_module_file_path = "/sys/module/kmodlve/version" if not os.path.isfile(kmodlve_module_file_path): return "" try: with open(kmodlve_module_file_path, encoding="utf-8") as f: module = f.read().strip() module_version, _ = module.rsplit(".", 1) except (OSError, ValueError): return "" return module_version def proc_can_see_other_uid_and_hidepid_synced() -> bool: """ Find out if system has version of the kernel (according to kmodlve module's version) where fs.proc_can_see_other_uid and hidepid options are synchronized. They are only synchronized if kmodlve module's version is equal to version in synced_kmodlve_versions. """ synced_kmodlve_versions = {"cl8": "2.0-30"} cl_version = get_cl_version() if cl_version not in synced_kmodlve_versions: return False synced_version = synced_kmodlve_versions[cl_version] current_version = get_kmodlve_module_version() if current_version == synced_version: return True return False def get_process_pid(pid_filename: str) -> Optional[int]: """ Detrmines working daemon process pid :param pid_filename: PID filename :return: PID from file or None if error (file not found, etc) """ try: with open(pid_filename, encoding="utf-8") as pf: pid = int(pf.read().strip()) # check that pid is still running os.kill(pid, 0) except (OSError, ValueError): pid = None return pid def write_pid_file(pid_filename: str): """ Writes pid file """ with open(pid_filename, "w+", encoding="utf-8") as pf: pf.write(f"{os.getpid()}") def remove_pid_file(pid_filename: str): """ Remove PID file """ try: os.remove(pid_filename) except OSError: pass def demote(uid, gid): """ Set user's real uid and gid to specified ones. Checking equality of real and effective uids is needed because this function may be used by root with effective uid dropped to user's uid. In that case it is needed to set effective uid back to 0 first. """ def func(): real_uid = os.getuid() eff_uid = os.geteuid() if real_uid != eff_uid: os.seteuid(real_uid) os.setgid(gid) os.setuid(uid) return func def get_mount_point(dirpath: str) -> str: """ Get mount point for dirpath directory from output of `df` utility """ if not os.path.isdir(dirpath): raise OSError(f'Directory "{dirpath}" does not exist') fs_info_cmd = ["/bin/df", "-h", dirpath] fs_info = run_command(fs_info_cmd) # example stdout: # [root@localhost ~] # df -h /home # Filesystem Size Used Avail Use% Mounted on # /dev/vda1 10G 3.1G 7.0G 31% / try: mountpoint_info = fs_info.split("\n")[1] mounted_on = mountpoint_info.split(" ")[-1] except IndexError as e: raise OSError(f'Utility "df" returned unexpected output:\n{fs_info}') from e return mounted_on def is_may_detach_mounts_enabled() -> bool: """ Detect if 'may_detach_mounts' kernel option is enabled. More info on the option: https://cloudlinux.atlassian.net/browse/KMODLVE-512 """ may_detach_mounts_file = "/proc/sys/fs/may_detach_mounts" if not os.path.isfile(may_detach_mounts_file): # Missing option means its enabled for all OS except CL6 return get_cl_version() != "cl6" with open(may_detach_mounts_file, encoding="utf-8") as f: val = int(f.read()) return val == 1 def get_filesystem_type(device: str) -> str: """ Return the file system type for the given device. """ partitions = psutil.disk_partitions() for partition in partitions: if partition.device == device: return partition.fstype return "" def get_os_version() -> tuple[str \| None, str \| None]: """ Detect system name and version. :return: tuple (os_name, os_ver) """ # # cat /etc/os-release \| grep -E '^NAME\|^VERSION_ID' # NAME="Ubuntu" # VERSION_ID="20.04" # # cat /etc/os-release \| grep -E '^NAME\|^VERSION_ID' # NAME="CloudLinux" # VERSION_ID="7.9" try: os_release_filename = "/etc/os-release" section_name = "top" config = configparser.ConfigParser() # config.read('/etc/os-release') with open(os_release_filename, encoding="utf-8") as stream: config.read_string(f"[{section_name}]\n" + stream.read()) os_name = config.get(section_name, "NAME").strip('"') os_ver = config.get(section_name, "VERSION_ID").strip('"') except (OSError, configparser.Error): return None, None # CloudLinux 10 doesn't override /etc/os-release, and therefore still identifies # itself as AlmaLinux. We still return "CloudLinux" for our own purposes. if os_name == "AlmaLinux" and os_ver.startswith("10"): os_name = "CloudLinux" return os_name, os_ver def is_ubuntu() -> bool: """ Determine whether this system is Ubuntu. Reads /etc/os-release file to do so. :return: bool flag is_ubuntu """ os_name, _ = get_os_version() return os_name == "Ubuntu" def is_secureboot_enabled() -> bool: """ Determine if secure boot is turned on :return: bool flag is_secureboot_enabled """ enabled = False if os.path.exists("/sys/firmware/efi"): enabled = ( subprocess.call( "mokutil --sb-state \| grep enabled", shell=True, executable="/bin/bash", stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) == 0 ) return enabled def get_username(): """ Get username of current user. """ try: return pwd.getpwuid(os.getuid())[0] except Exception: return None def get_modules_dir_for_alt_php(): """ It differs on Ubuntu and CloudLinux """ if is_ubuntu(): return "usr/lib/x86_64-linux-gnu/php/modules/" return "usr/lib64/php/modules/"	33.03 KB	2026-02-10 15:10
📄 utils_cmd.py # -- coding: utf-8 -- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import os import sys import subprocess class ExternalProgramFailed(Exception): """Exception class for errors related to externally executed commands.""" def __init__(self, message: str = ""): Exception.__init__(self, message) def check_command(cmdname): """ Checks if command is present and exits if no """ if not os.path.exists(cmdname): print(f"Command not found: ({cmdname})") sys.exit(1) def run_command(cmd, env_data=None, return_full_output=False, std_in=None, convert_to_str=True): """ Runs external process and returns output :param cmd: command and arguments as a list :param env_data: environment data for process :param return_full_output: if true, returns (ret_code, std_out, std_err) @return: process stdout if is_full_output==False else - cortege (ret_code, std_out, std_err) without any checking """ cmd_line = " ".join(cmd) try: std_err_obj = subprocess.PIPE if return_full_output else subprocess.STDOUT stdin_arg = subprocess.PIPE if std_in else subprocess.DEVNULL with subprocess.Popen( cmd, stdin=stdin_arg, stdout=subprocess.PIPE, stderr=std_err_obj, close_fds=True, env=env_data, text=convert_to_str, ) as proc: if not std_in: std_out, std_err = proc.communicate() else: std_out, std_err = proc.communicate(std_in) returncode = proc.returncode except OSError as oserr: raise ExternalProgramFailed( f"{cmd_line}. Cannot run command: {str(oserr)}" ) from oserr if return_full_output: return returncode, std_out, std_err if returncode != 0: if not convert_to_str: raise ExternalProgramFailed(f"Error during command execution: {cmd_line}") else: raise ExternalProgramFailed( std_err or f"output of the command: {cmd_line}\n{std_out}" ) return std_out def exec_utility(util_path, params): """ Executes supplied utility with supplied parameters :param util_path: Executable file to run path :param params: utility parameters :return: Cortege (ret_code, utility_stdout) """ args = [] args.append(util_path) args.extend(params) with subprocess.Popen(args, stdout=subprocess.PIPE, text=True) as proc: stdout, _ = proc.communicate() retcode = proc.returncode return retcode, stdout.strip()	2.71 KB	2026-02-10 15:10

Edit File:

Terminal