Wala shell

Advanced PHP File Manager

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

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-19 15:49
📁 GeneralPanel	-	2026-03-19 15:49
📄 __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 """ CloudLinux control panel API API for Hosting control panel """ import os from functools import lru_cache from .const import CACHE_CPNAME, UNKNOWN_CP_NAME from clcommon.features import ( Feature, get_cl_feature_status_map, get_hosting_accounts_limit ) from .pluginlib import ( get_cp_plugin_module, detect_panel_fast, CPANEL_NAME, DIRECTADMIN_NAME, PLESK_NAME, ISPMANAGER_NAME, INTERWORX_NAME, INTEGRATED_NAME, OFFICIAL_PLUGINS ) # NOQA needed for local disable pyflakes: F401 / 'NotSupported' warning during prospector checks from .cpapiexceptions import NotSupported, NoPanelUser # NOQA from clcommon.lib.cledition import ( is_cl_solo_edition, CLEditionDetectionError, is_container ) # If Hoster has own third-party plugins configured we can't bypass # our cache because we need to copy them to cache dir to be able to # import them and use to detect ControlPanel they provides. # So we do fast detection of our "official" plugins because they used in # most (all?) cases, and if we can't detect any CP then just fallback to # logic with cache _CP_NAME, _CP_PLUGIN_NAME = detect_panel_fast() def is_cl_supported_panel(): """ Is current panel has native CL support :return: True/False """ return _CP_NAME in [ CPANEL_NAME, DIRECTADMIN_NAME, PLESK_NAME, ISPMANAGER_NAME, INTERWORX_NAME ] def admin_packages(raise_exc=False): """ Return list of available admin's packages :param raise_exc: hack for lvemanager :rtype: list :return: List of packages. For example ['BusinessPackage', 'Package2'] """ return _panel_class.admin_packages(raise_exc) def resellers_packages(raise_exc=False): """ Return dictionary, contains available resellers packages, grouped by resellers :rtype: dict :return: Dictionary. Example: {'res1': ['BusinessPackage', 'UltraPackage', 'Package'], 'res2': ['SimplePackage', 'Package'] } """ return _panel_class.resellers_packages() def getCPName(): """ Return panel name :rtype: str :return: Panel name """ # if we are root, get panel name from script if os.getegid() == 0: return _panel_class.getCPName() # if we are user: # - for integrated control panel return real # name from scripts (they should work in cagefs) if OFFICIAL_PLUGINS[INTEGRATED_NAME] == _CP_PLUGIN_NAME: return _panel_class.getCPName() # - check _CP_NAME, probably detect_panel_fast was # able to detect control panel elif _CP_NAME is not None: return _CP_NAME # - last chance - read panel name from cache file controlpanelname = UNKNOWN_CP_NAME if os.path.isfile(CACHE_CPNAME): with open(CACHE_CPNAME, encoding='utf-8') as cache_stream: controlpanelname = cache_stream.readline().strip() return controlpanelname def get_cp_description(): """ Retrieve panel name and it's version :rtype: dict :return: dict: { 'name': 'panel_name', 'version': 'panel_version', 'additional_info': 'add_info'} or None if can't get info """ return _panel_class.get_cp_description() def cpusers(): """ Generates a list of cpusers registered in the control panel :return: list of cpusers registered in the control panel :rtype: tuple """ if is_cl_supported_panel(): # For CL supported panels we use already present plugin functions return _panel_class.cpusers() users_info = _panel_class.cpinfo(keyls=('cplogin', )) _list_users = [user_[0] for user_ in users_info] return tuple(_list_users) def resellers(): """ Generates a list of resellers in the control panel :return: list of cpusers registered in the control panel :rtype: tuple """ return _panel_class.resellers() def admins(): """ Generates a set of admins in the control panel :return: set of cpusers registered in the control panel :rtype: set :raise: NotSupported """ return _panel_class.admins() def db_access(): """ Getting root access to mysql database. For example {'login': 'root', 'db': 'mysql', 'host': 'localhost', 'pass': '9pJUv38sAqqW'} :return: root access to mysql database :rtype: dict :raises: NoDBAccessData """ return _panel_class.db_access() def dblogin_cplogin_pairs(cplogin_lst=None, with_system_users=False): """ Returs a list of pairs, the database user login - user login control panel For example: (('nata2_someuse', 'nata2'), ('testsome_dfrtbus', 'testsome')) :param list\|tuple\|None cplogin_lst: list of control panel users :param bool with_system_users: add system users to dbmapping :return: list of pairs, the database user login - user login control panel :rtype: tuple :raises: NotSupported, NoPackage """ return _panel_class.dblogin_cplogin_pairs(cplogin_lst, with_system_users) def cpinfo(cpuser=None, keyls=('cplogin', 'package', 'mail', 'reseller', 'dns'), search_sys_users=True): """ Retrieves info about panel user(s) :param str\|unicode\|list\|tuple\|None cpuser: user login :param keyls: list of data which is necessary to obtain the user, the valuescan be: cplogin - name/login user control panel mail - Email users reseller - name reseller/owner users locale - localization of the user account package - User name of the package dns - domain of the user :param bool search_sys_users: search for cpuser in sys_users or in control panel users (e.g. for Plesk) :return: returns a tuple of tuples of data in the same sequence as specified keys in keylst :rtype: tuple """ return _panel_class.cpinfo(cpuser, keyls, search_sys_users=search_sys_users) def get_admin_email(): """ Gets admin email Note: be careful when modifying this method. It is used in SSA, ask @dkavchuk or someone else from C-Projects team for details :rtype: string :return: string - admin email """ if is_cl_supported_panel(): return _panel_class.get_admin_email() else: try: return _panel_class.get_admin_emails_list()[0] except IndexError: return '' def get_admin_locale(): DEFAULT_LOCALE = 'en' if is_cl_supported_panel(): try: return _panel_class.get_admin_locale() except NotSupported: return DEFAULT_LOCALE return DEFAULT_LOCALE def docroot(domain): """ Return document root for domain :param str\|unicode domain: :rtype tuple :return Tuple: (document_root, owner) """ return _panel_class.docroot(domain) def useraliases(cpuser, domain): """ Return domain and document root pairs for control panel user first domain is main domain :param str\|unicode cpuser: user login :rtype: list :return list of tuples (domain_name, documen_root) """ return _panel_class.useraliases(cpuser, domain) def userdomains(cpuser): """ Return domain and document root pairs for control panel user first domain is main domain :param str\|unicode cpuser: user login :rtype: list :return list of tuples (domain_name, documen_root) """ return _panel_class.userdomains(cpuser) def useraddondomains(cpuser): """ Return addon domain and mountpoint domain pairs for control panel user :param str\|unicode cpuser: user login :return: mapping of addon domain names to their mountpoint names :rtype: dict """ return _panel_class.useraddondomains(cpuser) def homedirs(): """ Detects and returns list of folders contained the home dirs of users of the cPanel :rtype: list :return: list of folders, which are parent of home dirs of users of the panel """ return _panel_class.homedirs() def domain_owner(domain): """ Return domain's owner :param domain: Domain/sub-domain/add-domain name :rtype: str :return: user name or None if domain not found """ if is_cl_supported_panel(): return _panel_class.domain_owner(domain) _, _domain_owner = docroot(domain) return _domain_owner def reseller_users(resellername): """ Return reseller users :param resellername: reseller name; return empty list if None :rtype list :return list[str]: user names list """ return _panel_class.reseller_users(resellername) def reseller_domains(resellername=None): """ Return reseller users and their main domains :param resellername: reseller name; autodetect name if None :rtype disct :return dict[str, str]: pairs user <==> domain """ return _panel_class.reseller_domains(resellername) def is_reseller(username): """ Check if user is reseller; :type username: str :rtype: bool """ return _panel_class.is_reseller(username) def get_user_login_url(domain): """ Get login url for current panel; :type domain: str :rtype: str """ return _panel_class.get_user_login_url(domain) def is_admin(username): """ Return True if username is in admin names :param str username: user to check :rtype bool :return: True if user is admin """ return _panel_class.is_admin(username) def is_reseller_limits_supported(): """ Return True if panel supports reseller limits (LVE10+) :return: bool - True - yes, False - no :rtype bool """ panel_features = get_cl_feature_status_map() try: return panel_features[Feature.RESELLER_LIMITS] except KeyError: return False # The method is implemented only for Plesk, # because the resellers on Plesk can be without the system users. # For other panels the logic of method distributed by the code of lve-utils package def get_reseller_id_pairs(): """ Get dict reseller => id Optional method for panels without hard link reseller <=> system user :rtype: dict[str,int] - {'res1': id1} """ return _panel_class.get_reseller_id_pairs() # TODO: PTCCLIB-158 # The method should be private and moved to DA plugin def get_encoding_name(): """ Retrive encoding name, used for package/reseller names, from panel :return: """ return _panel_class.get_encoding_name() def is_panel_feature_supported(feature: Feature) -> bool: """ Get if feature is supported by control panel. :param feature: constant from Feature class :return: boolean """ try: features = get_supported_cl_features() return features is None or features.get(feature, False) except CLEditionDetectionError: return False def get_apache_connections_number(): """ Retrieves Apache's connections number (from mod_status) For CM """ return _panel_class.get_apache_connections_number() def get_apache_ports_list(): """ Retrieves active httpd's ports from httpd's config :return: list of apache's ports """ return _panel_class.get_apache_ports_list() def get_apache_max_request_workers(): """ Get current maximum request apache workers from httpd's config :return: tuple (max_req_num, message) max_req_num - Maximum request apache workers number or 0 if error message - OK/Trace """ return _panel_class.get_apache_max_request_workers() def get_main_username_by_uid(uid: int) -> str: """ Get "main" panel username by uid. :param uid: uid :return Username """ return _panel_class.get_main_username_by_uid(uid) def get_user_emails_list(username, domain): return _panel_class.get_user_emails_list(username, domain) def panel_login_link(username): return _panel_class.panel_login_link(username) def panel_awp_link(username): return _panel_class.panel_awp_link(username) def get_unsupported_features() -> list[Feature]: """ Returns features that are not supported on current control panel. """ return _panel_class.get_unsupported_cl_features() or [] # Global caching need to make this method work fast, # especially when we call it on vendors control panel. @lru_cache() def get_supported_cl_features(): """ Left here mostly for backwards compatibility. See get_cl_feature_status_map docstring. """ return get_cl_feature_status_map() def is_throttling_supported(): if is_cl_solo_edition(skip_jwt_check=True) or is_container(): return False return True def get_hosting_accounts_count() -> int: """ Get panel users count, admin accounts are not counted :return: Users count. """ return _panel_class.get_hosting_accounts_count() def is_hitting_max_accounts_limit(): """ Check if customer hitting max accounts count per edition """ max_accounts_limit = get_hosting_accounts_limit() if not max_accounts_limit: return False current_accounts_count = _panel_class.get_hosting_accounts_count() return current_accounts_count > max_accounts_limit def get_max_accounts_limit(): # warn: deprecated, use features.get_hosting_accounts_limit directly return get_hosting_accounts_limit() def count_hosting_accounts(): """ Retrieve users count and admins count, substract the last value from the first one and dump the result on disk """ try: users_list = cpusers() except NotSupported as e: raise NotSupported("cpusers() method not supported") from e # admins returns set or list depending on the control panel in use admins_collection = admins() return sum(1 for user in users_list if user not in admins_collection) def is_wp2_environment() -> bool: """ They have symlink /usr/local/cpanel/server.type -> wp2 """ wp2_marker = '/usr/local/cpanel/server.type' if not os.path.islink(wp2_marker): return False try: return os.readlink('/usr/local/cpanel/server.type') == 'wp2' except Exception: return False # Loading api _apiplugin = get_cp_plugin_module(_CP_PLUGIN_NAME) _panel_class = _apiplugin.PanelPlugin() CP_NAME = getCPName() # TODO: make a proxy that will do this stuff automatically get_domains_php_info = _panel_class.get_domains_php_info get_system_php_info = _panel_class.get_system_php_info list_all = _panel_class.list_all list_users = _panel_class.list_users get_reseller_users = _panel_class.get_reseller_users reseller_package_by_uid = _panel_class.reseller_package_by_uid get_uids_list_by_package = _panel_class.get_uids_list_by_package get_installed_php_versions = _panel_class.get_installed_php_versions get_customer_login = _panel_class.get_customer_login get_domain_login = _panel_class.get_domain_login get_server_ip = _panel_class.get_server_ip invalidate_cpapi_cache = _panel_class.invalidate_cpapi_cache suspended_users_list = _panel_class.suspended_users_list	14.85 KB	2026-02-10 15:10
📁 __pycache__	-	2026-03-19 15:49
📄 apilink.py from .plugins import vendors as api	36 B	2026-03-19 15:49
📁 cache	-	2026-03-19 15:49
📄 clcpapi.py #!/opt/cloudlinux/venv/bin/python3 -bb # -- 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 """ CloudLinux control panel API command line interface """ import argparse from clcommon import cpapi from clcommon.cloutput import prepare_data_json, prepare_attention_json from clcommon.cpapi import cpapiexceptions def cpapi_parser(): parser = argparse.ArgumentParser(description="%(prog)s - CloudLinux control panel API command line interface") parser.add_argument('-j', '--json', action='store_true', default=False, help='Output in json format') subparsers = parser.add_subparsers(dest='command') docroot_parser = subparsers.add_parser('docroot') docroot_parser.add_argument('domain', help='Return document root for domain') userdomains_parser = subparsers.add_parser('userdomains') userdomains_parser.add_argument('userdomains', help='Return domain and document root pairs for control panel user') dblogin_cplogin_pairs_parser = subparsers.add_parser("dblogin_cplogin_pairs") dblogin_cplogin_pairs_parser.add_argument("--cplogin_lst", nargs="*", help="Return mapping for this system users") dblogin_cplogin_pairs_parser.add_argument("--with_system_users", action="store_true", default=False, help="Show system users in output") reseller_users = subparsers.add_parser("reseller-users", help="Show reseller users") reseller_users.add_argument( 'resellername', type=str, nargs='?', help='Reseller name') return parser def flat_plain(data, sep=' '): return '\n'.join([sep.join(tuple_) for tuple_ in data]) def main(): parser = cpapi_parser() namespace = parser.parse_args() output_ = None exit_code = 0 try: if namespace.command == 'docroot': domain, user = cpapi.docroot(namespace.domain) if namespace.json: output_ = prepare_data_json({'domain': domain, 'user': user}) else: output_ = domain + ' ' + user elif namespace.command == 'userdomains': domain_docroot_pairs = cpapi.userdomains(namespace.userdomains) if namespace.json: data = [{'domain': domain_docroot[0], 'docroot': domain_docroot[1]} for domain_docroot in domain_docroot_pairs] output_ = prepare_data_json(data) else: output_ = flat_plain(domain_docroot_pairs) elif namespace.command == 'dblogin_cplogin_pairs': db_mapping = cpapi.dblogin_cplogin_pairs(namespace.cplogin_lst, namespace.with_system_users) if namespace.json: output_ = prepare_data_json(db_mapping) else: output_ = flat_plain(db_mapping) elif namespace.command == 'reseller-users': users = cpapi.reseller_users(namespace.resellername) if namespace.json: output_ = prepare_data_json(users) else: output_ = '\n'.join(users) except cpapiexceptions.CPAPIException as e_: if namespace.json: output_ = prepare_attention_json(str(e_)) else: exit_code = 1 output_ = 'ERROR: ' + str(e_) return output_, exit_code if __name__ == '__main__': import sys output_, exit_code = main() if output_: print(output_) if exit_code: sys.exit(exit_code)	3.64 KB	2026-02-10 15:10
📄 const.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 """ Commonly used constants """ CLCPAPI_NAME = 'CloudLinux control panel API' SUPPORTED_CPINFO = ('cplogin', 'package', 'mail', 'reseller', 'dns', 'locale') CACHE_CPNAME = '/var/cache/controlpanelname' UNKNOWN_CP_NAME = 'Unknown' UNKNOWN_CP_IMPORT = 'from clcommon.cpapi.plugins import nopanel as api'	511 B	2026-02-10 15:10
📄 cpapicustombin.py # -- coding: utf-8 -- """ common helper functions, related to processing api using custom binary currently available for Directadmin and Plesk """ import os import json from typing import List, Tuple, Union, Optional from clcommon.utils import exec_utility from clcommon.clpwd import ClPwd USERDOMAINS = '/usr/share/python-cllib/userdomains' def get_domains_via_custom_binary() -> Tuple[int, Union[str, List[Tuple[str, str]]]]: """ Calls USERDOMAINS(userdomains.c) bin and returns rc and loaded json output This is equals to call `userdomains(pwd.getpwuid(os.getuid()).pw_name, as_root=True)` Returns: return code of USERDOMAINS and json output if rc == 0: `out` is out from `userdomains()` NB: there is no tuples in JSON so da_out's type is List[List[str, str]] instead of List[Tuple[str, str]] if rc != 0: `out` type is str error codes: 1: diradmin/psaadmin user not found 2: not DA/Plesk panel 3: seteuid() syscall failed 4: executed as root 10: exception during `userdomains()` call 11: no such user in panel """ rc, out = exec_utility(USERDOMAINS, []) try: res = json.loads(out) except json.JSONDecodeError: pass else: if rc == 0: return rc, [ (row[0], row[1]) for row in res if len(row) == 2 and all(isinstance(p, str) for p in row) ] return rc, out def _docroot_under_user_via_custom_bin(domain: str) -> Optional[ Tuple[str, str]]: """ New method for getting doc_root for domain under user Method parses /usr/local/directadmin/... user's web server config See: _get_domains_list_as_root() :return: (doc_root, username) tuple """ clpwd = ClPwd() user_pw = clpwd.get_pw_by_uid(os.getuid())[0] # domains type (if rc == 0): List[Tuple[str, str]] # domain, docroot rc, domains = get_domains_via_custom_binary() if rc != 0: return None for dom, _docroot in domains: if dom == domain: return _docroot, user_pw.pw_name return None	2.12 KB	2026-02-10 15:10
📄 cpapiexceptions.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 clcommon.clexception import FormattedException from .const import CLCPAPI_NAME class CPAPIException(FormattedException): def __init__(self, exc_message): super().__init__(exc_message) self.message = f'{CLCPAPI_NAME}: {self.message}' class NotSupported(CPAPIException): pass class NoDBAccessData(CPAPIException): pass class NoPanelUser(CPAPIException): pass class NoPanelPackage(CPAPIException): pass class NoPackage(CPAPIException): pass class PluginImportError(CPAPIException): pass class NoDomain(CPAPIException): pass class DuplicateData(CPAPIException): pass class ReadFileError(CPAPIException): pass class CPAPIExternalProgramFailed(CPAPIException): pass class ParsingError(CPAPIException): pass class IncorrectData(CPAPIException): pass class EncodingError(CPAPIException): def __init__(self, message, **context): exc_message = { 'message': message, 'context': context} super().__init__(exc_message) class CpApiTypeError(CPAPIException): def __init__(self, funcname=None, supportedtypes=None, received_type=None): exc_message = { 'message': "in function %(funcname)s argument must be %(supportedtypes)s, not '%(received_type)s'", 'context': { 'funcname': funcname or '<api function>', 'supportedtypes': supportedtypes or '<supported types>', 'received_type': received_type or '<received type>' }} super().__init__(exc_message)	1.75 KB	2026-02-10 15:10
📄 cpapirebuildcache #!/opt/cloudlinux/venv/bin/python3 -bb # -- 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 """ Check with custom plug-ins directory, and try to use these plugins. Detection control panel and record the panel name to file /var/cache/controlpanelname """ from clcommon.cpapi import pluginlib pluginlib.rebuild_cache()	471 B	2026-02-10 15:10
📄 panel_hooks_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 os import subprocess import sys # Exit code when LVE doesn't support reseller limits TOO_OLD_LVE = 38 EXIT_NO_USER_FOUND = 1 ERR_NO_USER_FOUND = 'Unable to find username in hook cmdline' LVECTL = '/usr/sbin/lvectl' WHMAPI1 = '/usr/sbin/whmapi1' LVE_DIR = '/var/lve' TMP_DIR = os.path.join(LVE_DIR, 'tmp') def syncmap_supported(): test_cmd = rf"{LVECTL} --help \| grep -P '^sync-map\s+' &> /dev/null" return not subprocess.call(test_cmd, shell=True, executable='/bin/bash') # still used in lve-utils several times # move it there and then remove this file completely def call_sync_map(): """ Run lvectl sync-map :return: None """ ret_code = subprocess.call(f'{LVECTL} sync-map &> /dev/null', shell=True, executable='/bin/bash') if ret_code not in (0, TOO_OLD_LVE): # ignore calls with unsupported LVE if syncmap_supported(): # ignore only no such command "sync-map" error print(f'Error during "lvectl sync-map", code: {ret_code}', file=sys.stderr)	1.22 KB	2026-02-10 15:10
📄 pluginlib.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 """ Library to work with plug-ins """ import os import py_compile import subprocess import sys import pwd import glob import shutil import syslog from importlib import import_module from configparser import ConfigParser from clcommon.clcustomscript import lvectl_custompanel_script, CLSYSCONFIG from .const import CACHE_CPNAME, UNKNOWN_CP_NAME, UNKNOWN_CP_IMPORT from .cpapiexceptions import PluginImportError clsysconfig = ConfigParser(interpolation=None, strict=False) clsysconfig.read(CLSYSCONFIG) CONTROLPANELNAME_VAR = '__cpname__' DETECTFUNCNAME_VAR = 'detect' PLUGINS_PKG_PREFIX = 'clcommon.cpapi' PLUGINS_DIR = None if clsysconfig.has_option(section='cpapi', option='plugindir'): PLUGINS_DIR = clsysconfig.get(section='cpapi', option='plugindir') if not os.path.exists(PLUGINS_DIR): print("WARNING: plugindir is configured, but doesn't exists!") API_LINK_PATH = os.path.join(os.path.dirname(__file__), 'apilink.py') CPANEL_NAME = 'cPanel' DIRECTADMIN_NAME = 'DirectAdmin' PLESK_NAME = 'Plesk' ISPMANAGER_NAME = 'ISPManager' INTERWORX_NAME = 'InterWorx' UNKNOWN_NAME = 'Unknown' INTEGRATED_NAME = 'Integrated' OFFICIAL_PLUGINS = { CPANEL_NAME: 'cpanel', DIRECTADMIN_NAME: 'directadmin', PLESK_NAME: 'plesk', ISPMANAGER_NAME: 'ispmanager', INTERWORX_NAME: 'interworx', UNKNOWN_NAME: 'backward_plugin', INTEGRATED_NAME: 'vendors' } def detect_panel_fast(): """ This function will try to detect our officially supported control panels :return: tuple of CP name and CP plugin name to import """ panel_name, official_plugin_name = None, None if os.path.isfile('/opt/cpvendor/etc/integration.ini'): panel_name, official_plugin_name = INTEGRATED_NAME, OFFICIAL_PLUGINS[INTEGRATED_NAME] elif os.path.isfile('/usr/local/cpanel/cpanel'): panel_name, official_plugin_name = CPANEL_NAME, OFFICIAL_PLUGINS[CPANEL_NAME] elif os.path.isfile('/usr/local/directadmin/directadmin') or\ os.path.isfile('/usr/local/directadmin/custombuild/build'): panel_name, official_plugin_name = DIRECTADMIN_NAME, OFFICIAL_PLUGINS[DIRECTADMIN_NAME] elif os.path.isfile('/usr/local/psa/version'): panel_name, official_plugin_name = PLESK_NAME, OFFICIAL_PLUGINS[PLESK_NAME] # ispmanager must have: # v5: /usr/local/mgr5/ directory, # v4: /usr/local/ispmgr/bin/ispmgr file elif os.path.isfile('/usr/local/ispmgr/bin/ispmgr') or os.path.isdir('/usr/local/mgr5'): panel_name, official_plugin_name = ISPMANAGER_NAME, OFFICIAL_PLUGINS[ISPMANAGER_NAME] elif os.path.isdir('/usr/local/interworx'): panel_name, official_plugin_name = INTERWORX_NAME, OFFICIAL_PLUGINS[INTERWORX_NAME] elif lvectl_custompanel_script() is not None: # Check custom panel script to use backward plugin panel_name, official_plugin_name = UNKNOWN_NAME, OFFICIAL_PLUGINS[UNKNOWN_NAME] return panel_name, official_plugin_name def get_cp_plugin_module(plugin_name=None): if plugin_name: # Case when some "official" plugin was already detected return import_module(f'{PLUGINS_PKG_PREFIX}.plugins.{plugin_name}') # Use cached module if detect_panel_fast() can't detect anything # Actual import statement in file is updated on each rebuild_cache() from .apilink import api as _apiplugin # pylint: disable=import-outside-toplevel return _apiplugin def detectplugin(plugin_pack, ignore_errors=False): """ Scan directory for presence of plugins. Plugin is a file that has the extension ".py" and a non-empty variable "__cpname__" (the name of the control panel). The file must contain a function "detect" which returns True in case of presence of the corresponding control panel. >>> detectplugin('plugins') ('from .plugins import cpanel as api', 'cPanel') :param plugin_pack: package name or the name of the plug-ins directory ('cache' - cache plugins users; 'plugins' - officially supported plug-ins) :rtype: tuple :return: a pair of values: (line to import the package, the name of the control panel) """ plugin_dir = os.path.join(os.path.dirname(__file__), plugin_pack) plugin_path_pattern = os.path.join(plugin_dir, '*.py') modules = [ os.path.splitext(os.path.basename(py_full_path))[0] for py_full_path in glob.glob(plugin_path_pattern) ] for mod_name in modules: if mod_name == '__init__': continue absolute_module_name = f'{PLUGINS_PKG_PREFIX}.{plugin_pack}.{mod_name}' import_string = f'from .{plugin_pack} import {mod_name} as api' try: api = import_module(absolute_module_name) except ImportError as e: if ignore_errors: continue raise PluginImportError(f'Can not import {absolute_module_name} plugin') from e try: panel_class = api.PanelPlugin() # Try to detect panel panel_data = panel_class.get_cp_description() return import_string, panel_data['name'] except (AttributeError, TypeError, KeyError) as exc: syslog.syslog( syslog.LOG_WARNING, f'cpapi: Plugin {mod_name} does not satisfy cpapi requirements:\n{exc}', ) return None, None def getuser(): """ Get current user's username based on os.getuid() :rtype: str """ uid = os.getuid() if uid == 0: return 'root' return pwd.getpwuid(uid)[0] def _reinit_module(init_path): """create (or rewrite) an empty __init__.py and it's .pyc/.pyo parts""" with open(init_path, 'w', encoding='utf-8'): pass py_compile.compile(init_path) # In Py2 ".pyo" files can be generated only like this: subprocess.run( f'{sys.executable} -O -m py_compile "{init_path}"', shell=True, executable='/bin/bash', check=False, ) def rebuild_cache(plugins_dir=PLUGINS_DIR): CACHE_DIR = 'cache' PLUGINS_PATH = 'plugins' # directory with official supported plugins cache_dir = os.path.join(os.path.dirname(__file__), CACHE_DIR) # delete all in cache dir shutil.rmtree(cache_dir) if plugins_dir and os.path.isdir(plugins_dir): # copy all from plugins dir to cache shutil.copytree(plugins_dir, cache_dir) else: os.mkdir(cache_dir) os.chmod(cache_dir, 0o755) init_path = os.path.join(cache_dir, '__init__.py') _reinit_module(init_path) try: import_string, cpname = detectplugin(CACHE_DIR) except PluginImportError as exc: import_string, cpname = None, None print(f'WARNING: {exc}') if cpname is None: shutil.rmtree(cache_dir) os.mkdir(cache_dir) _reinit_module(init_path) import_string, cpname = detectplugin(PLUGINS_PATH, ignore_errors=True) if cpname is None: import_string, cpname = (UNKNOWN_CP_IMPORT, UNKNOWN_CP_NAME) print(f'WARNING: can not detect control panel; the control panel is set to "{cpname}"') # write control panel name to cache file if cpname: # write control panel name to /var/cache/controlpanelname with open(CACHE_CPNAME, 'w', encoding='utf-8') as f: f.write(cpname) # Write plugin import string for fast loading. # Example: # from .plugins import nopanel as api with open(API_LINK_PATH, 'w', encoding='utf-8') as f: f.write(import_string + '\n')	7.57 KB	2026-02-10 15:10
📁 plugins	-	2026-03-19 15:49
📄 utils.py import sys from functools import wraps from clcommon.const import Feature from clcommon.cpapi import is_panel_feature_supported from clcommon.lib.cledition import ( CLEditionDetectionError, ) __all__ = ( "skip_without_lve", "lve_supported_or_exit", ) _SKIP_MESSAGE = ( "Command is skipped as LVE feature is not supported in current environment\n" "Refer to https://docs.cloudlinux.com/solo/introduction/#cloudlinux-os-editions-comparison for more info" ) def skip_without_lve(): try: # we still have some utils that could be run # under user (e.g cloudlinux-selector) if not is_panel_feature_supported(Feature.LVE): print(_SKIP_MESSAGE) sys.exit(0) except CLEditionDetectionError as e: print(f"Error: {e}") sys.exit(1) def lve_supported_or_exit(f): @wraps(f) def inner(args, kwargs): if not is_panel_feature_supported(Feature.LVE): print(_SKIP_MESSAGE) else: return f(args, **kwargs) return inner	1.03 KB	2026-02-10 15:10

Terminal