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
📄 __init__.py # -- coding: utf-8 --	23 B	2026-02-10 15:10
📁 __pycache__	-	2026-03-19 15:49
📄 backward_plugin.py # General class, implementing common methods, using all cpapi plugins by default import os from clcommon.features import Feature from clcommon.utils import exec_utility from collections import defaultdict from clcommon import ClPwd from clcommon.clcustomscript import lvectl_custompanel_script from clcommon.cpapi.GeneralPanel import GeneralPanelPluginV1 from clcommon.cpapi.cpapiexceptions import CPAPIExternalProgramFailed # This flag means that we use enchanced custom long script. # With keys, which aren't described in official documentation. USE_ENCHANCED_CUSTOM_LONG_SCRIPT = os.environ.get('USE_ENCHANCED_CUSTOM_LONG_SCRIPT') == '1' class PanelPlugin(GeneralPanelPluginV1): def __init__(self): super().__init__() self._custom_script_name = lvectl_custompanel_script() self._cp_name = "Unknown" def getCPName(self): """ Return panel name :return: """ return self._cp_name def get_cp_description(self): """ Retrieve panel name and it's version :return: dict: { 'name': 'panel_name', 'version': 'panel_version', 'additional_info': 'add_info'} or None if can't get info """ if self._custom_script_name is not None: return {'name': self._cp_name, 'version': '0', 'additional_info': None} # panel detect fail if custom script not found return None # proxy to avoid arguments-differ def cpinfo(self, cpuser=None, keyls=('cplogin', 'package', 'mail', 'reseller', 'dns', 'locale'), search_sys_users=True): return self._cpinfo(cpuser, keyls=keyls, search_sys_users=search_sys_users) def _cpinfo(self, cpuser=None, keyls=('cplogin', 'package', 'mail', 'reseller', 'dns', 'locale', 'uid'), search_sys_users=True, raise_exc=False): """ Retrives specified info about panel users :param str\|unicode\|list\|tuple\|None cpuser: user login :param keyls: List or cortege of data which is necessary to obtain the user, values can 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 userid - uid :param search_sys_users: :param raise_exc: hack for clquota that raises exception in case of exit code != 0 :return: returns a tuple of tuples of data in the same sequence as specified keys in keylst Examples: cpinfo('cltest1') (('cltest1', 'default', '', 'root', 'cltest1.com', 'en'),) cpinfo() (('res1usr1', 'res1_pack1', '', 'res1', 'res1usr1.com', 'en'), ('res1', 'default', '', 'root', 'res1.com', 'en'), ('res2', 'default', '', 'res2', 'res2.com', 'en'), ('cltest1', 'default', '', 'root', 'cltest1.com', 'en'), ('system', 'undefined', None, 'root', '', 'en')) :rtype: tuple """ try: if USE_ENCHANCED_CUSTOM_LONG_SCRIPT: all_users_data = super().list_users(raise_exc=raise_exc) else: # all_users_data Example: # {1000: 'Package1', 1001: 'BusinessPackage'} all_users_data = self.list_all(raise_exc=raise_exc) except (OSError, IOError, IndexError, AttributeError, CPAPIExternalProgramFailed): all_users_data = {} cl_pwd = ClPwd() # Build full users data dict users_dict = defaultdict(dict) for uid, value in all_users_data.items(): try: user_name = cl_pwd.get_names(uid)[0] except ClPwd.NoSuchUserException: # user with uid absent in system, use some fictive name # We use : in username, because system does not allow this symbol in username user_name = f'::{uid}::' if isinstance(value, dict): package = value['package'] reseller = value['reseller'] else: package = value reseller = '' try: users_dict[user_name] = {'cplogin': user_name, 'package': package, 'mail': None, 'reseller': reseller, 'dns': None, 'locale': None, 'uid': uid, } except Exception: # we are ignoring all errors pass out_list = [] user_name_list = [] if isinstance(cpuser, str): user_name_list = [cpuser] elif isinstance(cpuser, (list, tuple)): user_name_list = list(cpuser) elif cpuser is None: user_name_list = list(users_dict.keys()) for user_name in user_name_list: # for case if users_dict has fictive user names if user_name not in users_dict: continue user_data = users_dict[user_name] user_data_list = [] for data_key in keyls: user_data_list.append(user_data[data_key]) out_list.append(tuple(user_data_list)) return tuple(out_list) # inherited implementation does not work because # old plugin does not contain resellers def reseller_package_by_uid(self, user_id): """ Retrieves reseller name and package name by uid :param user_id: User id :return: Cortege: (Reseller_name, Package_name) """ package_name = '' try: _, stdout = exec_utility(self._custom_script_name, ['--userid='+str(user_id)]) package_name = stdout.split('\n').pop(0) except (OSError, IOError, IndexError, AttributeError): pass return '', package_name # This is reseller method. # Key `--list-resellers-packages` of custom long script isn't # described in official documentation, so we use this method # only in tests and implement the key only in tests # for real client's custom long script we return empty result def resellers_packages(self, raise_exc=False): """ Return dictionary, contains available resellers packages, grouped by resellers :return: Dictionary. Example: {'res1': ['BusinessPackage', 'UltraPackage', 'Package'], 'res2': ['SimplePackage', 'Package'] } """ if not USE_ENCHANCED_CUSTOM_LONG_SCRIPT: return {} try: return super().resellers_packages(raise_exc=raise_exc) except (OSError, IOError, IndexError, AttributeError, CPAPIExternalProgramFailed): return {} # This is reseller method. # Key `--list-users` of custom long script isn't # described in official documentation, so we use this method # only in tests and implement the key only in tests # for real client's custom long script we return empty result def resellers(self): """ Generates a list of resellers in the control panel :return: list/tuple of cpusers registered in the control panel :rtype: tuple :raise: NotSupported """ if not USE_ENCHANCED_CUSTOM_LONG_SCRIPT: return tuple() resellers_list = [value['reseller'] for value in self.list_users().values()] return tuple(set(resellers_list)) def get_admin_emails_list(self): """ Gets admin emails list :rtype: List :return: List: ['admin1@mail.com', 'admin2@mail.com' ] """ return [] def docroot(self, domain): """ Return document root for domain :return: Cortege: (document_root, owner) """ return () def userdomains(self, 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 [] def homedirs(self): """ 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 [] def reseller_users(self, resellername=None): """ Return reseller users :param resellername: reseller name; autodetect name if None :rtype: List :return list[str]: user names list """ return [] def reseller_domains(self, resellername=None): """ Return reseller users and their main domains :param resellername: reseller name; autodetect name if None :rtype: List :return dict[str, str]: pairs user <==> domain """ return {} def get_user_login_url(self, domain): """ Get login url for current panel; :type domain: str :rtype: str :return: Panel login URL """ return "" def get_reseller_id_pairs(self): """ Get dict reseller => id Optional method for panels without hard link reseller <=> system user :rtype: dict[str,int] - {'res1': id1} :return: """ return {} # This is reseller method. # Key `--list-reseller-users` of custom long script isn't # described in official documentation, so we use this method # only in tests and implement the key only in tests # for real client's custom long script we return empty result def get_reseller_users(self, reseller): if not USE_ENCHANCED_CUSTOM_LONG_SCRIPT: return {} try: return super().get_reseller_users(reseller) except (OSError, IOError, IndexError, AttributeError, CPAPIExternalProgramFailed): return {} def list_users(self, raise_exc=False): """ Return list of <userid package reseller> triples :rtype: dict :return: Dictionary. Example: {1000: {'reseller': '', 'package': 'Package1'}, 1001: {'reseller': '', 'package': 'BusinessPackage'}, } """ packages_users = defaultdict(dict) users_info = self._cpinfo(keyls=('uid', 'reseller', 'package'), raise_exc=True) for uid, reseller, package in users_info: packages_users[uid] = {'reseller': reseller, 'package': package} return packages_users def get_unsupported_cl_features(self) -> list[Feature]: """ Return list of cloudlinux features that can be used on current control panel. """ # empty list mostly for historical reasons return []	10.87 KB	2026-02-10 15:10
📄 cpanel.py # -- coding: utf-8 -- import json import logging import os import pwd import subprocess import sys import urllib.error import urllib.parse import urllib.request from collections import OrderedDict from configparser import ConfigParser, NoOptionError, NoSectionError from traceback import format_exc from urllib.parse import urlparse import requests import yaml from clcommon import ClPwd from clcommon.clconfpars import load as loadconfig from clcommon.clconfpars import load_fast from clcommon.cpapi.cpapiexceptions import ( CPAPIExternalProgramFailed, CpApiTypeError, DuplicateData, EncodingError, NoDBAccessData, NoDomain, NotSupported, ParsingError, ) from clcommon.cpapi.GeneralPanel import ( CPAPI_CACHE_STORAGE, DomainDescription, GeneralPanelPluginV1, PHPDescription, ) from clcommon.cpapi.pluginlib import getuser from clcommon.features import Feature from clcommon.lib.whmapi_lib import WhmApiError, WhmApiRequest, WhmNoPhpBinariesError from clcommon.utils import ExternalProgramFailed, get_file_lines, get_modules_dir_for_alt_php, grep, run_command __cpname__ = 'cPanel' DBMAPPING_SCRIPT = os.path.join(os.path.dirname(sys.executable), "cpanel-dbmapping") UAPI = '/usr/bin/uapi' logger = logging.getLogger(__name__) # WARN: Probably will be deprecated for our "official" plugins. # See pluginlib.detect_panel_fast() def detect(): return os.path.isfile('/usr/local/cpanel/cpanel') CPANEL_DB_CONF = '/root/.my.cnf' CPANEL_USERPLANS_PATH = '/etc/userplans' CPANEL_DATABASES_PATH = '/var/cpanel/databases/' CPANEL_USERS_DIR = '/var/cpanel/users/' CPANEL_RESELLERS_PATH = '/var/cpanel/resellers' CPANEL_USERDATADOMAINS_PATH = '/etc/userdatadomains;/var/cpanel/userdata/{user}/cache' CPANEL_USERDATAFOLDER_PATH = '/var/cpanel/userdata/{user}' CPANEL_ACCT_CONF_PATH = '/etc/wwwacct.conf' CPANEL_USEROWNERS_FILE = '/etc/trueuserowners' SYSCONF_CLOUDLINUX_PATH = '/etc/sysconfig/cloudlinux' CPANEL_CONFIG = '/var/cpanel/cpanel.config' USERCONF_PARAM_MAP = { 'dns': 'dns', 'package': 'plan', 'reseller': 'owner', 'mail': 'contactemail', 'locale': 'locale', 'cplogin': 'user' } SUPPORTED_CPANEL_CPINFO = ('cplogin', 'package', 'mail', 'reseller', 'dns', 'locale') def db_access(_conf_path=CPANEL_DB_CONF): access = {} reg_data_config = ConfigParser( allow_no_value=True, interpolation=None, strict=False ) opened_files = reg_data_config.read(_conf_path) if not opened_files: raise NoDBAccessData( f"Cannot find database access data for localhost. No such file {_conf_path}" ) # Options in MySQL config files can be double- or single-quoted, so we strip() them try: if reg_data_config.has_option(section="client", option="password"): access["pass"] = reg_data_config.get( section="client", option="password", raw=True ).strip("\"'") else: access["pass"] = reg_data_config.get( section="client", option="pass", raw=True ).strip("\"'") access["login"] = reg_data_config.get( section="client", option="user", raw=True ).strip("\"'") except (NoOptionError, NoSectionError) as err: raise NoDBAccessData( "Cannot find database access data for localhost from config " f"file {_conf_path}; {err.message}" ) from err access["db"] = "mysql" return access def cpusers(_userplans_file=CPANEL_USERPLANS_PATH): """ Parse the file /etc/userplans, which contains the pairs of user-plan :param _userplans_file: path to the user's plans file :return: list of the non-system users """ with open(_userplans_file, encoding='utf-8') as stream: users_list = [line.split(':')[0].strip() for line in stream if not line.startswith('#') and line.count(':') == 1 and len(line.strip()) > 3] return tuple(users_list) def resellers(_resellers_path=CPANEL_RESELLERS_PATH): if not os.path.isfile(_resellers_path): # on a clean system, this file may not be return tuple() with open(_resellers_path, encoding='utf-8') as stream: # Example of file # res1res1:add-pkg,add-pkg-ip,add-pkg-shell # res1root:add-pkg,add-pkg-ip,add-pkg-shell,allow-addoncreate,allow-emaillimits-pkgs # r: resellers_list = [line.split(':', 1)[0].strip() for line in stream if not line.startswith('#') and (':' in line) and len(line.strip()) > 1] return tuple(resellers_list) def admins(): return {'root'} def is_reseller(username, _resellers_path=CPANEL_RESELLERS_PATH): """ Check if given user is reseller; :param _resellers_path: for testing only :type username: str :rtype: bool """ return any(cplogin == username for cplogin in resellers(_resellers_path)) def dblogin_cplogin_pairs(cplogin_lst=None, with_system_users=False): """ Get mapping between system and DB users @param cplogin_lst :list: list with usernames for generate mapping @param with_system_users :bool: add system users to result list or no. default: False """ # initialize results list results = [] # accept only list and tuple parameters uid_list = [] for username in (cplogin_lst or []): try: uid_list.append(str(pwd.getpwnam(username).pw_uid)) except KeyError: # no user exists - skip it uid_list.append("-1") # generate system command params = [DBMAPPING_SCRIPT] if not with_system_users: params.append("--nosys") params += uid_list with subprocess.Popen( params, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) as p: output, _ = p.communicate() # output format: "DBuser user UID" for line in output.split("\n"): line = line.strip() if line: results.append(line.split()[:2]) return tuple(results) def cpinfo(cpuser=None, keyls=('cplogin', 'package', 'mail', 'reseller', 'dns', 'locale'), _cpanel_users_dir=CPANEL_USERS_DIR, quiet=True, search_sys_users=True): returned = [] if isinstance(cpuser, str): cpusers_list = [cpuser] elif isinstance(cpuser, (list, tuple)): cpusers_list = tuple(cpuser) elif cpuser is None: cpusers_list = cpusers(_userplans_file=CPANEL_USERPLANS_PATH) else: raise CpApiTypeError(funcname='cpinfo', supportedtypes='str\|unicode\|list\|tuple', received_type=type(cpuser).__name__) for cpuser in cpusers_list: user_config_path = os.path.join(_cpanel_users_dir, cpuser) if not os.path.exists(user_config_path): if not quiet: sys.stderr.write(f'WARNING: Can not load data to the user "{cpuser}"; ' 'Perhaps there is no such user in cPanel') continue # ignore bad symbols in config here # because of LVEMAN-1150 (which also seems # being already fixed by cpanel) cpuser_data = loadconfig(user_config_path, ignore_bad_encoding=True) user_data = [] for data_key in keyls: data = cpuser_data.get(USERCONF_PARAM_MAP.get(data_key)) # USERCONF_PARAM_MAP.get('cplogin') == user # check if user tag in user config and if tag is missing - use file name as user name if data_key == 'cplogin' and data is None: data = os.path.basename(user_config_path) user_data.append(data) returned.append(tuple(user_data)) if 'mail' in keyls: # checking the presence of an additional e-mail additional_mail = cpuser_data.get('contactemail2') if additional_mail: user_data[list(keyls).index('mail')] = additional_mail user_data_tuple = tuple(user_data) if user_data_tuple not in returned: returned.append(tuple(user_data)) return tuple(returned) def get_admin_email(_conf1=None, _conf2=None, _hostname=None): """ :param str\|None _conf1: for testing :param str\|None _conf2: for testing :param str\|None _hostname: for testing :return: """ # 1. Try to get admin email from /etc/sysconfig/cloudlinux lines = [] try: lines = get_file_lines(_conf1 or SYSCONF_CLOUDLINUX_PATH) except (OSError, IOError): pass for line in lines: if line.startswith('EMAIL'): parts = line.split('=') if len(parts) == 2 and '@' in parts[1].strip(): return parts[1].strip() # 2. Try to get admin email from /etc/wwwacct.conf lines = [] try: lines = get_file_lines(_conf2 or CPANEL_ACCT_CONF_PATH) except (OSError, IOError): pass host = '' for line in lines: if line.startswith('CONTACTEMAIL'): s = line.replace('CONTACTEMAIL', '').strip() if s: return s if line.startswith('HOST'): s = line.replace('HOST', '').strip() if s: host = s if host: return 'root@' + host # Admin email not found in system files, use common address from clcommon.cpapi.plugins.universal import get_admin_email # pylint: disable=import-outside-toplevel return get_admin_email(_hostname=_hostname) def _parse_userdatadomains(_path, parser, quiet=True): if '{user}' in _path: call_as_user = pwd.getpwuid(os.geteuid()).pw_name _path = _path.replace('{user}', call_as_user) path_list = _path.split(';') for path_ in path_list: if not os.path.exists(path_): continue try: with open(path_, encoding='utf-8') as stream: # example line: # test.russianguns.ru: russianguns==root==sub== # russianguns.ru==/home/russianguns/fla==192.168.122.40:80======0 for i, line in enumerate(stream): if not line.strip(): # ignore the empty string continue if line.count(': ') != 1: if not quiet: sys.stderr.write(f'Can\'t parse {i} line in file "{path_}"; line was ignored\n') continue domain, domain_raw_data = line.split(': ') domain_data = domain_raw_data.strip().split('==') parser(path_, domain, domain_data) except IOError as e: if not quiet: sys.stderr.write(f"Can't open file {path_} [{e}]\n") continue def _parse_userdataaliases(_path, quiet=True): path_list = _path.split(';') aliases = [] for path_ in path_list: if not os.path.exists(path_): continue try: with open(path_, encoding='utf-8') as stream: # example line: # test.russianguns.ru: russianguns==root==sub== # russianguns.ru==/home/russianguns/fla==192.168.122.40:80======0 for i, line in enumerate(stream): if not line.strip(): # ignore the empty string continue if "serveralias" not in line: continue aliases += line.replace("serveralias: ", '').strip().split(' ') except IOError as e: if not quiet: sys.stderr.write(f"Can't open file {path_} [{e}]\n") continue return aliases def useraliases(cpuser, domain, _path=CPANEL_USERDATAFOLDER_PATH, quiet=True): # use dict to avoid duplicates if '{user}' in _path: _path = _path.replace('{user}', cpuser) _path = os.path.join(_path, domain) aliases = _parse_userdataaliases(_path, quiet=quiet) return list(aliases) def docroot(domain, _path=CPANEL_USERDATADOMAINS_PATH, quiet=True): domain = domain.strip() pathes = set() result = {'docroot_path': None, 'user': None} def parser(path, d, domain_data): pathes.add(path) if d == domain: result['docroot_path'] = domain_data[4] result['user'] = domain_data[0] _parse_userdatadomains(_path, parser, quiet=quiet) if not (result['docroot_path'] is None or result['user'] is None): return result['docroot_path'], result['user'] watched = '; '.join( ['Can\'t find record "%(d)s" in file "%(p)s"' % {'d': domain, 'p': p} for p in pathes] ) raise NoDomain(f"Can't obtain document root for domain '{domain}'; {watched}") # User name to domain cache # Example: # { 'user1': [('user1.org', '/home/user1/public_html'), # ('mk.user1.com.user1.org', '/home/user1/public_html/www.mk.user1.com')] } _user_to_domains_map_cpanel = {} def userdomains(cpuser, _path=CPANEL_USERDATADOMAINS_PATH, quiet=True): # If user present in cache, take data from it if cpuser in _user_to_domains_map_cpanel: return _user_to_domains_map_cpanel[cpuser] # use dict to avoid duplicates domains_tmp = OrderedDict() domains = OrderedDict() def parser(path, d, domain_data): user_ = domain_data[0] document_root = domain_data[4] # Update cache if user_ in _user_to_domains_map_cpanel: user_data = _user_to_domains_map_cpanel[user_] else: user_data = [] if 'main' == domain_data[2]: # insert main domain to 1st position in list if (d, document_root) not in user_data: user_data.insert(0, (d, document_root)) else: if (d, document_root) not in user_data: user_data.append((d, document_root)) _user_to_domains_map_cpanel[user_] = user_data if user_ == cpuser: if 'main' == domain_data[2]: # main domain must be first in list domains.update({d: document_root}) else: domains_tmp.update({d: document_root}) _parse_userdatadomains(_path, parser, quiet=quiet) domains.update(domains_tmp) return list(domains.items()) def _load_main_user_file(cpuser: str, _path=CPANEL_USERDATAFOLDER_PATH, quiet=True) -> dict: """ Private helper to load and parse the main file for a given user. E.g. `/var/cpanel/userdata/{user}/main` :param cpuser: The cPanel username. :param _path: A path template containing the "{user}" placeholder. :param quiet: If False, errors are printed. :return: The parsed YAML data as a dictionary. :raises ParsingError: If an error occurs while parsing. """ if '{user}' in _path: _path = _path.replace('{user}', cpuser) main_file = os.path.join(_path, "main") try: with open(main_file, "rt", encoding="utf-8") as f: data = yaml.safe_load(f) return data except IOError as e: if not quiet: sys.stderr.write(f"Can't open file {main_file} [{e}]\n") return {} except yaml.YAMLError as e: raise ParsingError(str(e)) from e def useraddondomains(cpuser: str, _path=CPANEL_USERDATAFOLDER_PATH, quiet=True) -> dict: """ Return all addon domains for the given cPanel user. It loads the user's main file at `/var/cpanel/userdata/{user}/main` and parses the YAML content, and then extracts the addon_domains mapping. :param cpuser: The cPanel username. :param _path: The path template for the userdata folder :param quiet: If False, errors might be printed. :return: A dict mapping addon domain names to their server names. For example: { "aurapoolshop.com.au": "aurapoolshop.thepoolandyardman.com.au" } :raises ParsingError: If the main file cannot be parsed. """ main_data = _load_main_user_file(cpuser, _path=_path, quiet=quiet) return main_data.get("addon_domains", {}) def domain_owner(domain, _path=CPANEL_USERDATADOMAINS_PATH, quiet=True): users_list = [] def parser(path, d, domain_data): if d == domain: users_list.append(domain_data[0]) _parse_userdatadomains(_path, parser, quiet=quiet) if len(users_list) > 1: raise DuplicateData( f"domain {domain} belongs to few users: [{','.join(users_list)}]" ) if len(users_list) == 0: return None return users_list[0] def homedirs(_sysusers=None, _conf_path = CPANEL_ACCT_CONF_PATH): """ Detects and returns list of folders contained the home dirs of users of the cPanel :param str\|None _sysusers: for testing :param str\|None _conf_path: for testing :return: list of folders, which are parent of home dirs of users of the panel """ HOMEDIR = 'HOMEDIR ' HOMEMATCH = 'HOMEMATCH ' homedirs = [] users_homedir = '' users_home_match = '' if os.path.exists(_conf_path): lines = get_file_lines(_conf_path) for line in lines: if line.startswith(HOMEDIR): users_homedir = line.split(HOMEDIR)[1].strip() elif line.startswith(HOMEMATCH): users_home_match = line.split(HOMEMATCH)[1].strip() if users_homedir: homedirs.append(users_homedir) clpwd = ClPwd() users_dict = clpwd.get_user_dict() # for testing only if isinstance(_sysusers, (list, tuple)): class pw: def __init__(self, name, dir): self.pw_name = name self.pw_dir = dir users_dict = {} for (name,dir) in _sysusers: users_dict[name] = pw(name, dir) for user_data in users_dict.values(): userdir = user_data.pw_dir if os.path.exists(userdir + '/public_html') or os.path.exists(userdir + '/www'): homedir = os.path.dirname(userdir) if users_home_match and homedir.find('/'+users_home_match) == -1: continue if homedir not in homedirs: homedirs.append(homedir) return homedirs def _reseller_users_parser(json_string): try: json_serialized = json.loads(json_string) result = json_serialized['result'] return [item['user'] for item in result['data']] except (KeyError, ValueError, TypeError) as e: raise ParsingError(str(e)) from e def _reseller_users_json(reseller_name=None): """ Call UAPI and get json string; :type reseller_name: str \| None :raises: ParsingError, CPAPIExternalProgramFailed :rtype: str """ reseller_name = reseller_name or getuser() # Attention!! /usr/bin/uapi utility may works unstable. See PTCLLIB-95 for details cmd = [UAPI, 'Resellers', 'list_accounts', '--output=json'] # root user MUST specify --user, and reseller CAN'T do that if reseller_name != getuser() or getuser() == "root": cmd.append(f'--user={urllib.parse.quote(reseller_name)}') try: json_string = run_command(cmd=cmd, return_full_output=True)[1] # take only std_out, ignore std_err except ExternalProgramFailed as e: raise CPAPIExternalProgramFailed(str(e)) from e return json_string def reseller_users(resellername): """ Return reseller users :param resellername: reseller name; return empty list if None :return list[str]: user names list """ # Attention!! /usr/bin/uapi utility may works unstable. See PTCLLIB-95 for details # json_string = _reseller_users_json(resellername) # return _reseller_users_parser(json_string) # So we read reseller's user list from /etc/trueuserowners # /etc/trueuserowners example: # #userowners v1 # cltest1: root # res: res # res1: root # res2: root # res2user1: res2 # res2usr1: res2 # resnew: resnew # resnew1: resnew1 # rn1user1: resnew1 if resellername is None: return [] result = [] userowner_file_data = get_file_lines(CPANEL_USEROWNERS_FILE) if resellername is not None: for line in grep(rf'\: {resellername}$', fixed_string=False, match_any_position=True, multiple_search=True, data_from_file=userowner_file_data): splitted_line = line.strip().split(': ') result.append(splitted_line[0]) return result def _reseller_user_domains_parser(json_string): try: json_serialized = json.loads(json_string) result = json_serialized['result'] users_data = {} for item in result['data']: users_data[item['user']] = item['domain'] return users_data except (KeyError, ValueError, TypeError) as e: raise ParsingError(str(e)) from e def reseller_domains(reseller_name=None): """ Get dict[user, domain] Attention!! This function may work unstable. See PTCLLIB-95 for details. :param reseller_name: reseller's name :rtype: dict[str, str\|None] :raises DomainException: if cannot obtain domains """ json_string = _reseller_users_json(reseller_name) return _reseller_user_domains_parser(json_string) def get_user_login_url(domain): return f'http://{domain}:2083' def is_no_php_binaries_on_cpanel(): """ Checks that there are no installed php binaries only for cpanel """ try: WhmApiRequest('php_get_installed_versions').call() except WhmNoPhpBinariesError: return True return False class PanelPlugin(GeneralPanelPluginV1): DEFAULT_LOCALE = 'en' BAD_CODING_ERROR_CODE = 48 HTTPD_CONFIG_FILE = '/etc/apache2/conf/httpd.conf' def __init__(self): super().__init__() def invalidate_cpapi_cache(self): """ Goes through all panel caches and invalidates it if needed """ method_marker_pairs = (('_get_php_version_id_to_handler_map', ['/etc/cpanel/ea4/php.conf']), ('_get_vhosts_php_versions', ['/etc/userdatadomains', '/etc/cpanel/ea4/php.conf'])) for pair in method_marker_pairs: method, markers = pair[0], pair[1] cache_file = os.path.join(CPAPI_CACHE_STORAGE, method + '.cache') if self.is_cache_valid(cache_file, markers): # cache is up to dated -> nothing to do continue # cache is outdated -> rewrite data = getattr(self, method)() self.rewrite_cpapi_cache(data, cache_file) def _run_long_script(self, args): """ Processes decoding errors from long script which mean that cpanel wrote something bad to config file (most likely LVEMAN-1150) :param args: arguments to pass :return: stdout, stderr """ with subprocess.Popen( [self._custom_script_name] + args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) as p: out, err = p.communicate() returncode = p.returncode if returncode == self.BAD_CODING_ERROR_CODE: raise EncodingError( "Problem with encoding in %(script)s file, error is: '%(error)s'.", script=self._custom_script_name, error=err ) return out, err, returncode def getCPName(self): """ Return panel name :return: """ return __cpname__ def get_cp_description(self): """ Retrieve panel name and it's version :return: dict: { 'name': 'panel_name', 'version': 'panel_version', 'additional_info': 'add_info'} or None if can't get info """ try: with subprocess.Popen( ['/usr/local/cpanel/cpanel', '-V'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) as p: out, _ = p.communicate() return {'name': __cpname__, 'version': out.split()[0], 'additional_info': None} except Exception: return None def db_access(self): """ 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 db_access() def cpusers(self): """ Generates a list of cpusers registered in the control panel :return: list of cpusers registered in the control panel :rtype: tuple """ return cpusers() def resellers(self): """ Generates a list of resellers in the control panel :return: list of cpusers registered in the control panel :rtype: tuple """ return resellers() def is_reseller(self, username): """ Check if given user is reseller; :type username: str :rtype: bool """ return is_reseller(username) def dblogin_cplogin_pairs(self, cplogin_lst=None, with_system_users=False): """ Get mapping between system and DB users @param cplogin_lst :list: list with usernames for generate mapping @param with_system_users :bool: add system users to result list or no. default: False """ return dblogin_cplogin_pairs(cplogin_lst, with_system_users) def cpinfo(self, cpuser=None, keyls=('cplogin', 'package', 'mail', 'reseller', 'dns', 'locale'), 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 cpinfo(cpuser, keyls, search_sys_users=search_sys_users) def get_admin_email(self): """ Retrieve admin email address :return: Host admin's email """ return get_admin_email() @staticmethod def useraliases(cpuser, domain): """ Return aliases from user domain :param str\|unicode cpuser: user login :param str\|unicode domain: :return list of aliases """ return useraliases(cpuser, domain) def docroot(self, domain): """ Return document root for domain :param str\|unicode domain: :return Cortege: (document_root, owner) """ return docroot(domain) def userdomains(self, cpuser): """ Return domain and document root pairs for control panel user first domain is main domain :param str\|unicode cpuser: user login :return list of tuples (domain_name, documen_root) """ return userdomains(cpuser) def useraddondomains(self, 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 server names :rtype: dict """ return useraddondomains(cpuser) def homedirs(self): """ Detects and returns list of folders contained the home dirs of users of the cPanel :return: list of folders, which are parent of home dirs of users of the panel """ return homedirs() def reseller_users(self, resellername=None): """ Return reseller users :param resellername: reseller name; autodetect name if None :return list[str]: user names list """ return reseller_users(resellername) def reseller_domains(self, resellername=None): """ Get dict[user, domain] Attention!! This function may work unstable. See PTCLLIB-95 for details. :param reseller_name: reseller's name :rtype: dict[str, str\|None] :raises DomainException: if cannot obtain domains """ return reseller_domains(resellername) def get_user_login_url(self, domain): """ Get login url for current panel; :type domain: str :rtype: str """ return get_user_login_url(domain) def domain_owner(self, domain): """ Return domain's owner :param domain: Domain/sub-domain/add-domain name :rtype: str :return: user name or None if domain not found """ return domain_owner(domain) def get_system_php_info(self): try: default_version = WhmApiRequest( 'php_get_system_default_version').call()['version'] except WhmApiError as e: raise CPAPIExternalProgramFailed(e) from e return { 'default_version_id': default_version } def get_domains_php_info(self): """ Returns info about domains: username, php_version, handler_type For each domain we detect handler and php_version _get_php_version_id_to_handler_map() returns data of installed versions, so if the version of some domain was removed we can`t detect the handler. In such case we set handler_type to None. Otherwise we detect handler and set it to handler_type :rtype dict """ php_version_to_handler_map = self._get_php_version_id_to_handler_map() php_settings_per_vhost = self._get_vhosts_php_versions() domains_php_info = {} for domain_info in php_settings_per_vhost: php_version_id = domain_info['version'] if php_version_id not in list(php_version_to_handler_map.keys()): logger.error("Unable to find php %s in handlers map %s. ", php_version_id, php_version_to_handler_map, extra={ 'php_version_id': php_version_id, 'php_version_to_handler_map': php_version_to_handler_map }) handler_type = None else: handler_type = 'fpm' if domain_info['php_fpm'] \ else php_version_to_handler_map[php_version_id] domains_php_info[domain_info['vhost']] = DomainDescription( username=domain_info['account'], php_version_id=php_version_id, handler_type=handler_type, display_version=php_version_id ) return domains_php_info @staticmethod def get_installed_php_versions(): """ Get the list of PHP version installed in panel :return: list """ try: # ['alt-php56', 'alt-php72', 'ea-php74'] php_versions = WhmApiRequest('php_get_installed_versions').call()['versions'] except (KeyError, WhmApiError) as e: logger.error('CPAPI: Could not get list of installed PHP versions: %s', e) # todo: consider changing this to return exceptions return [] else: php_description = [] for php_name in php_versions: if php_name.startswith('alt-'): php_root_dir = f'/opt/{php_name.replace("-", "/")}/' php_description.append(PHPDescription( identifier=php_name, version=f'{php_name[-2]}.{php_name[-1]}', dir=os.path.join(php_root_dir), modules_dir=os.path.join(php_root_dir, get_modules_dir_for_alt_php()), bin=os.path.join(php_root_dir, 'usr/bin/php'), ini=os.path.join(php_root_dir, 'link/conf/default.ini'), )) elif php_name.startswith('ea-'): php_root_dir = f'/opt/cpanel/{php_name}/root/' php_description.append(PHPDescription( identifier=php_name, version=f'{php_name[-2]}.{php_name[-1]}', modules_dir=os.path.join(php_root_dir, 'usr/lib64/php/modules/'), dir=os.path.join(php_root_dir), bin=os.path.join(php_root_dir, 'usr/bin/php'), ini=os.path.join(php_root_dir, 'etc/php.ini'), )) else: # unknown php, skip continue return php_description @staticmethod @GeneralPanelPluginV1.cache_call(panel_parker=['/etc/userdatadomains', '/etc/cpanel/ea4/php.conf']) def _get_vhosts_php_versions(): """ See https://documentation.cpanel.net/display/DD/WHM+API+1+Functions+-+php_get_vhost_versions :rtype: dict """ try: return WhmApiRequest('php_get_vhost_versions').call()['versions'] except WhmApiError as e: raise CPAPIExternalProgramFailed(e) from e @staticmethod @GeneralPanelPluginV1.cache_call(panel_parker=['/etc/cpanel/ea4/php.conf']) def _get_php_version_id_to_handler_map(): """ Returns dict with info about php version and it`s current handler: {'ea-php56': 'cgi', 'ea-php72': 'suphp', 'alt-php51': 'suphp', 'alt-php52': 'suphp' ...} Using cpanel whmapi request Tries to get all handlers or if there is problem with some handler - gets handlers one by one As a result information could be incomplete if some handlers are not available See https://documentation.cpanel.net/display/DD/WHM+API+1+Functions+-+php_get_handlers :rtype: dict """ try: handlers = WhmApiRequest('php_get_handlers').call()['version_handlers'] except WhmApiError as e: logger.error("Unable to get information about php handlers, " "falling back to per-handler data gathering. " "Error happened: %s", e, extra={ 'error_message': e.message, 'error_context': e.context }) handlers = PanelPlugin._get_handler_info_for_each_version() return { php['version']: php['current_handler'] for php in handlers } @staticmethod def _get_handler_info_for_each_version(): """ Gets handler data from each version one by one, so that data can still be collected even when one of the installed versions is broken. :rtype: list """ handlers = [] installed_php_versions = PanelPlugin.get_installed_php_versions() for version in installed_php_versions: # {'version_handlers': [{'available_handlers': ['cgi', 'none'], 'version': 'ea-php72', # 'current_handler': None}]} try: version_handler = \ WhmApiRequest('php_get_handlers').with_arguments( version=version['identifier'] ).call()['version_handlers'][0] handlers.append(version_handler) except (KeyError, WhmApiError) as e: logger.error('CPAPI: Could not get data for PHP version: %s', e) continue return handlers def get_admin_locale(self): cpanel_config = load_fast(CPANEL_CONFIG) try: server_locale = cpanel_config['server_locale'] if server_locale: return server_locale return PanelPlugin.DEFAULT_LOCALE except KeyError: return PanelPlugin.DEFAULT_LOCALE @staticmethod def get_apache_connections_number(): """ Retrieves Apache's connections number :return: tuple (conn_num, message) conn_num - current connections number, 0 if error message - OK/Trace """ # curl http://127.0.0.1/whm-server-status?auto \| grep "Total Accesses" try: url = 'http://127.0.0.1/whm-server-status?auto' response = requests.get(url, timeout=5) if response.status_code != 200: return 0, f"GET {url} response code is {response.status_code}" s_response = response.content.decode('utf-8') s_response_list = s_response.split('\n') out_list = list(grep("Total Accesses", data_from_file=s_response_list)) # out_list example: ['Total Accesses: 200'] s_total_accesses = out_list[0].split(':')[1].strip() return int(s_total_accesses), 'OK' except Exception: return 0, format_exc() @staticmethod def get_apache_ports_list(): """ Retrieves active httpd's ports from httpd's config :return: list of apache's ports """ # cat /etc/apache2/conf/httpd.conf \| grep Listen _httpd_ports_list = [] try: lines = get_file_lines(PanelPlugin.HTTPD_CONFIG_FILE, unicode_errors_handle='surrogateescape') except (OSError, IOError): return None lines = [line.strip() for line in lines] for line in grep('Listen', match_any_position=False, multiple_search=True, data_from_file=lines): # line examples: # Listen 0.0.0.0:80 # Listen [::]:80 try: value = int(line.split(':')[-1]) if value not in _httpd_ports_list: _httpd_ports_list.append(value) except (IndexError, ValueError): pass if not _httpd_ports_list: _httpd_ports_list.append(80) return _httpd_ports_list @staticmethod 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 """ # cat /etc/apache2/conf/httpd.conf \| grep MaxRequestWorkers # MaxRequestWorkers 150 try: lines = get_file_lines(PanelPlugin.HTTPD_CONFIG_FILE, unicode_errors_handle='surrogateescape') mrw_list = list(grep('MaxRequestWorkers', match_any_position=False, data_from_file=lines)) if len(mrw_list) != 1: return 0, 'MaxRequestWorkers directive is absent or multiple in httpd\'s config' parts = mrw_list[0].split(' ') if len(parts) == 2: return int(parts[1]), 'OK' return 0, f'httpd config line syntax error. Line is \'{mrw_list[0]}\'' except (OSError, IOError, IndexError, ValueError): return 0, format_exc() @staticmethod def get_user_emails_list(username: str, domain: str) -> str: # "acct" : [ # { # "has_backup" : 0, # "email" : "bla@cloudlinux.com, blabla@gmail.com" # } # ] emails = \ WhmApiRequest('listaccts').with_arguments(want='email', searchmethod='exact', search=username, searchtype='user').call()['acct'][0] user_emails = emails['email'] if user_emails == 'unknown': return '' return user_emails @staticmethod def panel_login_link(username): link = WhmApiRequest('create_user_session').with_arguments(user=username, service='cpaneld').call()['url'] if not link: return '' # https://77.79.198.14:2083/cpsess3532861743/login/?session=stackoverflow%3ascBEPeVeSXqZgMLs%.. -> # https://77.79.198.14:2083/ parsed = urlparse(link) return f'{parsed.scheme}://{parsed.netloc}/' @staticmethod def panel_awp_link(username): link = PanelPlugin.panel_login_link(username).rstrip("/") if len(link) == 0: return '' return f'{link}/cpsess0000000000/frontend/paper_lantern/lveversion/wpos.live.pl' def get_server_ip(self): try: with open('/var/cpanel/mainip', encoding='utf-8') as f: return f.read().strip() except FileNotFoundError as e: raise NotSupported( 'Unable to detect main ip for this server. ' 'Contact CloudLinux support and report the issue.' ) from e def suspended_users_list(self): # [{'time': 'Tue Mar 26 12:41:31 2024', 'owner': 'root', 'is_locked': 0, # 'unixtime': 1711456891, 'reason': 'Unknown', 'user': 'susp2'}, # {'is_locked': 0, 'time': 'Tue Mar 26 12:18:53 2024', 'owner': # 'root', 'reason': 'Unknown', 'user': 'susp', 'unixtime': 1711455533}] suspended_info = WhmApiRequest('listsuspended').call()['account'] return [item['user'] for item in suspended_info] def get_unsupported_cl_features(self) -> list[Feature]: return []	41.19 KB	2026-02-10 15:10
📄 directadmin.py # -- coding: utf-8 -- """ CloudLinux API for DirectAdmin control panel """ import glob import os import re import subprocess import sys import syslog from traceback import format_exc from typing import Dict, List, Tuple # NOQA from urllib.parse import urlparse import requests from clcommon.clconfpars import ( WebConfigMissing, WebConfigParsingError, apache_conf_parser, load_fast, nginx_conf_parser, openlitespeed_conf_parser, read_unicode_file_with_decode_fallback, ) from clcommon.clconfpars import load as loadconfig from clcommon.clpwd import ClPwd from clcommon.cpapi.cpapicustombin import ( _docroot_under_user_via_custom_bin, get_domains_via_custom_binary, ) from clcommon.cpapi.cpapiexceptions import ( CpApiTypeError, NoDBAccessData, NoDomain, NoPanelUser, ParsingError, ReadFileError, ) from clcommon.cpapi.GeneralPanel import ( DomainDescription, GeneralPanelPluginV1, PHPDescription, ) from clcommon.cpapi.plugins.universal import ( get_admin_email as universal_get_admin_email, ) from clcommon.features import Feature from clcommon.utils import ( ExternalProgramFailed, find_module_param_in_config, get_file_lines, get_modules_dir_for_alt_php, grep, ) __cpname__ = 'DirectAdmin' DA_DIR = '/usr/local/directadmin' DA_CONF = os.path.join(DA_DIR, 'conf/directadmin.conf') DA_DATA_DIR = os.path.join(DA_DIR, 'data') DA_DB_CONF = os.path.join(DA_DIR, 'conf/mysql.conf') DA_USERS_PATH = os.path.join(DA_DATA_DIR, 'users') DA_OPT_PATH = os.path.join(DA_DIR, 'custombuild', 'options.conf') USER_CONF = 'user.conf' DOMAINOWNERS = '/etc/virtual/domainowners' ADMIN_DIR = os.path.join(DA_DATA_DIR, 'admin') RESELLERS_LIST = os.path.join(ADMIN_DIR, 'reseller.list') ADMINS_LIST = os.path.join(ADMIN_DIR, 'admin.list') USER_PATTERN = re.compile(rf'.+/(.+)/{re.escape(USER_CONF)}') # WARN: Probably will be deprecated for our "official" plugins. # See pluginlib.detect_panel_fast() def detect(): return os.path.isfile('/usr/local/directadmin/directadmin') or \ os.path.isfile('/usr/local/directadmin/custombuild/build') def db_access(): access = {} try: login_data = loadconfig(DA_DB_CONF) access['login'] = login_data['user'] access['pass'] = login_data['passwd'] except IOError as err: raise NoDBAccessData( 'Can not open file with data to database access; ' + str(err) ) from err except KeyError as err: raise NoDBAccessData( f'Can not get database access data from file {DA_DB_CONF}' ) from err return access def cpusers(): match_list = [USER_PATTERN.match(path) for path in glob.glob(os.path.join(DA_USERS_PATH, '', USER_CONF))] users_list = [match.group(1) for match in match_list if match] return tuple(users_list) def resellers(): with open(RESELLERS_LIST, encoding='utf-8') as f: resellers_list = [line.strip() for line in f] return tuple(resellers_list) def admins(): with open(ADMINS_LIST, encoding='utf-8') as f: admins_list = [line.strip() for line in f] return set(admins_list) def dblogin_cplogin_pairs(cplogin_lst=None, with_system_users=False): from clcommon.cpapi.plugins.universal import _dblogin_cplogin_pairs # pylint: disable=import-outside-toplevel access = db_access() data = _dblogin_cplogin_pairs(cplogin_lst=cplogin_lst, access=access) if with_system_users: data += tuple(get_da_user(DA_USERS_PATH).items()) return data def get_da_user(path, quiet=True): users = {} cur_dir = os.getcwd() os.chdir(path) dir_list = glob.glob('./') for user_dir in dir_list: if os.path.isdir(user_dir): file_domains = path + '/' + user_dir + '/domains.list' try: with open(file_domains, encoding='utf-8') as f: if len(f.readline()) > 0: user_name = user_dir[2:] users[user_name] = user_name except IOError: if not quiet: sys.stderr.write("No file " + file_domains) os.chdir(cur_dir) return users def cpinfo(cpuser=None, keyls=('cplogin', 'package', 'mail', 'reseller', 'dns', 'locale'), search_sys_users=True): returned = [] if isinstance(cpuser, str): cpusers_list = [cpuser] elif isinstance(cpuser, (list, tuple)): cpusers_list = tuple(cpuser) elif cpuser is None: cpusers_list = cpusers() else: raise CpApiTypeError(funcname='cpinfo', supportedtypes='str\|unicode\|list\|tuple', received_type=type(cpuser).__name__) def _get_reseller(config): if config.get('usertype') == 'reseller': return config.get('username') return config.get('creator') _user_conf_map = {'cplogin': lambda config: config.get('username'), 'package': lambda config: config.get('package'), 'mail': lambda config: config.get('email'), 'reseller': lambda config: _get_reseller(config), 'dns': lambda config: config.get('domain'), 'locale': lambda config: config.get('language')} keyls_ = [_user_conf_map[key] for key in keyls] for username in cpusers_list: user_conf_file = os.path.join(DA_USERS_PATH, username, USER_CONF) if os.path.exists(user_conf_file): user_config = load_fast(user_conf_file) returned.append([key(user_config) for key in keyls_]) return returned def _docroot_under_root(domain): # type: (str) -> Tuple[str, str] """ Old method for getting doc_root for domain under root Method reads DA config :return: (doc_root, username) cortege """ user_name = None # Load /etc/virtual/domainowners _domain_to_user_map = _load_domains_owners() if domain in _domain_to_user_map: user_name = _domain_to_user_map[domain] else: # Domain not in main domains - search in subdomains # Group domains by owner to avoid repeated userdomains() calls per user owner_to_main_domains = {} for main_domain, owner in _domain_to_user_map.items(): owner_to_main_domains.setdefault(owner, []).append(main_domain) # Iterate by unique owners (not all domains) for owner in owner_to_main_domains: # get subdomains for this user subdomains = _get_subdomains(userdomains(owner), _domain_to_user_map) if domain in subdomains: # Subdomain found user_name = owner break if user_name is None: domains_list = [] else: domains_list = userdomains(user_name) for d in domains_list: if domain in d: return d[1], user_name def _docroot_under_user_old_mechanism(domain): # type: (str) -> Tuple[str, str] """ Old method for getting doc_root for domain under user Method parses /home/<username>/domains directory :return: (doc_root, username) cortege """ clpwd = ClPwd() user_pw = clpwd.get_pw_by_uid(os.getuid())[0] list_domains_and_doc_roots = _get_domains_list_as_user(user_pw.pw_dir) for domain_data in list_domains_and_doc_roots: if domain_data['server_name'] == domain: return domain_data['document_root'], user_pw.pw_name def docroot(domain): # type: (str) -> Tuple[str, str] """ Retrieves document root for domain :param domain: Domain to determine doc_root :return: Cortege: (doc_root, domain_user) """ res = None domain = domain.strip() uid = os.getuid() euid = os.geteuid() if euid == 0 and uid == 0: res = _docroot_under_root(domain) else: res = _docroot_under_user_via_custom_bin(domain) # If there was successful result, res object will have # (doc_root, domain_user) format. If there wasn't found any correct # doc_roots, res will be None. if res is not None: return res raise NoDomain(f"Can't obtain document root for domain '{domain}'") def _is_nginx_installed(): """ Check if nginx is installed via custombuild; """ config = loadconfig(DA_CONF) return bool(int(config.get('nginx', 0)) or int(config.get('nginx_proxy', 0))) def _is_openlitespeed_installed(): """ Check if OpenLiteSpeed is installed via custombuild; """ config = loadconfig(DA_CONF) return bool(int(config.get('openlitespeed', 0))) def _get_domains_list_as_root(user_path): """ Get domains list for user from httpd or nginx config as root :param user_path: path to DA directory of user's profile :return: parsed httpd or nginx config :rtype: list """ try: if _is_nginx_installed(): httpd_conf = nginx_conf_parser(os.path.join(user_path, 'nginx.conf')) elif _is_openlitespeed_installed(): httpd_conf = openlitespeed_conf_parser(os.path.join(user_path, 'openlitespeed.conf')) else: httpd_conf = apache_conf_parser(os.path.join(user_path, 'httpd.conf')) except WebConfigParsingError as e: raise ParsingError(e.message) from e except WebConfigMissing: return [] return httpd_conf def _get_domains_list_as_user(user_home): # type: (str) -> List[Dict[str, str, bool]] """ Get domains list for user from ~/domains directory as user. Method DOESN'T search subdomains, because it's almost impossible detect by user's folders without privileges escalation :param user_home: path to user home :return: list of dictionaries {'server_name': 'domain', 'document_root': 'doc_root', 'ssl': False} """ domains_dir = 'domains' doc_root_dir = 'public_html' domains_list = [] domains_path = os.path.join(user_home, domains_dir) # Searching main domains # All directories of main domains are saved in ~/domains directory for domain_dir in os.listdir(domains_path): domain_path = os.path.join(domains_path, domain_dir) doc_root_path = os.path.join(domains_path, domain_dir, doc_root_dir) if os.path.isdir(domain_path) and os.path.isdir(doc_root_path): domains_list.append({ 'server_name': domain_dir, 'document_root': doc_root_path, 'ssl': False, }) else: continue return domains_list def userdomains(cpuser, as_root=False): # type: (str, bool) -> List[Tuple[str, str]] """ Get user's domains list :return list: domain names Example: [('cltest1.com', '/home/cltest1/domains/cltest1.com/public_html'), ('mk.cltest1.com', '/home/cltest1/domains/cltest1.com/public_html/mk'), ('cltest11.com', '/home/cltest1/domains/cltest11.com/public_html') ] """ domains_list = [] user_path = os.path.join(DA_USERS_PATH, cpuser) euid = os.geteuid() # old method to get list of user's domains main_domain_path = '' if not os.path.exists(user_path): return [] user_home = os.path.expanduser('~' + cpuser) public_path = os.path.join(user_home, 'public_html') if os.path.exists(public_path) and os.path.islink(public_path): main_domain_path = os.path.realpath(public_path) if euid == 0 or as_root: httpd_conf = _get_domains_list_as_root(user_path) for domain in httpd_conf: if domain['ssl'] is True: continue # Put main domain in start of list if domain['server_name'] in main_domain_path: domains_list.insert(0, (domain['server_name'], domain['document_root'])) else: domains_list.append((domain['server_name'], domain['document_root'])) return domains_list # this case works the same as above but through the rights escalation binary wrapper # call path: here -> binary -> python(diradmin euid) -> userdomains(as_root=True) -> print json result to stdout rc, res = get_domains_via_custom_binary() if rc == 0: return res elif rc == 11: raise NoPanelUser(f'User {cpuser} not found in the database') else: raise ExternalProgramFailed(f'Failed to get userdomains: {res}') def homedirs(): """ Detects and returns list of folders contained the home dirs of users of the DirectAdmin :return: list of folders, which are parent of home dirs of users of the panel """ home_dirs = set() clpwd = ClPwd() users_dict = clpwd.get_user_dict() for user_name, pw_user in list(users_dict.items()): conf_file = os.path.join(DA_USERS_PATH, user_name, USER_CONF) if os.path.exists(conf_file): home_dir = os.path.dirname(pw_user.pw_dir) home_dirs.add(home_dir) return list(home_dirs) def domain_owner(domain): """ Return domain's owner :param domain: Domain/sub-domain/add-domain name :return: user name or None if domain not found """ return _load_domains_owners().get(domain, None) @GeneralPanelPluginV1.cache_call(panel_parker=[DOMAINOWNERS]) def _load_domains_owners() -> Dict[str, str]: """ Get domain<->user map from /etc/virtual/domainowners file """ # 1. Load DA data file try: domains_lines = read_unicode_file_with_decode_fallback(DOMAINOWNERS).splitlines() except (OSError, IOError) as e: raise ReadFileError(str(e)) from e # 2. File loaded successfully, parse data and fill dictionaries _domain_to_user_map = {} for line_ in domains_lines: line_ = line_.strip() # pass empty line if not line_: continue domain_, user_ = line_.split(':') domain_ = domain_.strip() user_ = user_.strip() # Fill domain to user map _domain_to_user_map[domain_] = user_ return _domain_to_user_map def reseller_users(resellername): """ Return list of reseller users :param resellername: reseller name; return empty list if None :return list[str]: user names list """ if resellername is None: return [] all_users_dict = ClPwd().get_user_dict() users_list_file = os.path.join(DA_USERS_PATH, resellername, 'users.list') try: with open(users_list_file, encoding='utf-8') as users_list: users_list = [item.strip() for item in users_list] users_list.append(resellername) # performing intersection to avoid returning non-existing users # that are still present in config file for some reason return list(set(all_users_dict) & set(users_list)) except (IOError, OSError): return [] def reseller_domains(resellername=None): """ Get pairs user <=> domain for given reseller; Empty list if cannot get or no users found; :type resellername: str :return list[tuple[str, str]]: tuple[username, main_domain] """ if resellername is None: return [] users = reseller_users(resellername) return dict(cpinfo(users, keyls=('cplogin', 'dns'))) def get_admin_email(): admin_user_file = os.path.join(DA_USERS_PATH, 'admin', USER_CONF) try: cnf = loadconfig(admin_user_file) return cnf.get('email', universal_get_admin_email()) except FileNotFoundError: return universal_get_admin_email() def is_reseller(username): """ Check if given user is reseller; :type username: str :rtype: bool :raise: ParsingError, ReadFileError """ user_config = os.path.join(DA_USERS_PATH, username, USER_CONF) if os.path.exists(user_config): try: return loadconfig(user_config)['usertype'] == 'reseller' except IndexError as e: raise ParsingError('User config exists, but no usertype given') from e return False def get_user_login_url(domain): return f'http://{domain}:2222' def _get_da_php_config(): """ Return map (PHP_DA_CODE:{PHP_HANDLER, PHP_VERSION}) :return: """ _php_da_map = {} try: php_cfg = loadconfig(DA_OPT_PATH) except (IOError, OSError): return None # iterate through custombuild options.conf php_mode and php_release options i = 1 while f'php{i}_mode' in php_cfg and f'php{i}_release' in php_cfg: _php_da_map[str(i)] = {} _php_da_map[str(i)]['handler_type'] = php_cfg[f'php{i}_mode'] _php_da_map[str(i)]['php_version_id'] = php_cfg[f'php{i}_release'] i += 1 return _php_da_map def _get_php_code_info_for_domain(domain, owner): """ Return php code from domain config :param domain: :param owner: :return: string '1' or '2' - php code in DA """ domain_config_file = os.path.join(DA_USERS_PATH, str(owner), 'domains', str(domain) + '.conf') try: domain_config = loadconfig(domain_config_file) except (IOError, OSError): return '1' domain_php = domain_config.get('php1_select') # None - DA custombuild has only one php version # '0' - it means that user selected default version PHP of DA custombuild if domain_php is None or domain_php == '0': domain_php = '1' return domain_php def _get_subdomains(all_domains, mapped_all_domains): subdomains = [] for domain in all_domains: if domain[0] in mapped_all_domains.keys(): continue subdomains.append(domain[0]) return subdomains def get_domains_php_info(): """ Return php version information for each domain :return: domain to php info mapping Example output: {'cltest.com': {'handler_type': 'mod_php', 'php_version_id': '7.1', 'username': 'cltest'}, 'cltest2.com': {'handler_type': 'fastcgi', 'php_version_id': '7.3', 'username': 'kek_2'}, 'cltest3.com': {'handler_type': 'suphp', 'php_version_id': '5.5', 'username': 'cltest3'}, 'omg.kek': {'handler_type': 'php-fpm', 'php_version_id': '5.2', 'username': 'cltest'}} :rtype: dict[str, dict] """ # returns only main domains map_domain_user = _load_domains_owners() result_map = {} php_da_map = _get_da_php_config() if php_da_map is None: return result_map owner_to_domains: dict[str, list[str]] = {} for domain, owner in map_domain_user.items(): owner_to_domains.setdefault(owner, []).append(domain) for owner, domains in owner_to_domains.items(): all_domains_in_httpd_file = userdomains(owner) # get safely to not break something to other teams try: subdomains = _get_subdomains(all_domains_in_httpd_file, map_domain_user) except Exception: subdomains = [] for domain in domains: php_info_code = _get_php_code_info_for_domain(domain, owner) if php_info_code not in php_da_map \ or php_da_map[php_info_code]['php_version_id'] == 'no': # 'no' means that php_release specified in user's config # does not exist in custombuild options.conf php_info_code = '1' php_info = php_da_map[php_info_code] try: domain_aliases = _useraliases(owner, domain) except Exception: domain_aliases = [] # https://forum.directadmin.com/threads/sub-domain-different-php-version.58426/ # subdomain version should be the same as main domain for domain_entity in [domain] + subdomains + domain_aliases: result_map[domain_entity] = DomainDescription( username=owner, php_version_id=php_info['php_version_id'], handler_type=php_info['handler_type'], display_version=f'php{php_info["php_version_id"].replace(".", "")}' ) return result_map def _get_installed_alt_php_versions(): """ Gets installed alt-phpXY - could be chosen via CloudLinux PHP Selector w/o being compiled via custombuild """ installed_list = [] alt_phps_directory = '/opt/alt/' pattern = re.compile(r'^php\d+$') for item in os.listdir(alt_phps_directory): item_path = os.path.join(alt_phps_directory, item) # Check if the item is a directory and its name matches the pattern if os.path.isdir(item_path) and pattern.match(item) and os.path.exists(f'{item_path}/usr/bin/php'): version = item.replace('php', '') installed_list.append(PHPDescription( identifier=f'alt-{item}', version=f'{version[:1]}.{version[1:]}', dir=f'{item_path}/', modules_dir=os.path.join(item_path, get_modules_dir_for_alt_php()), bin=os.path.join(item_path, 'usr/bin/php'), ini=os.path.join(item_path, 'link/conf/default.ini'), )) return installed_list def _get_da_php_extension_dir(directadmin_php_dir): return subprocess.run( [f'{directadmin_php_dir}bin/php-config', '--extension-dir'], text=True, capture_output=True, check=False, ).stdout def _get_compiled_custombuild_versions(): """ Gets compiled phpXY - could be chosen via DirectAdmin PHP Selector """ php_da_map = _get_da_php_config() if php_da_map is None: return [] # {'1': {'handler_type': 'php-fpm', 'php_version_id': '7.4'}, # '2': {'handler_type': 'php-fpm', 'php_version_id': '8.0'}, # '3': {'handler_type': 'php-fpm', 'php_version_id': 'no'}, # '4': {'handler_type': 'php-fpm', 'php_version_id': 'no'}} installed_php_data = php_da_map.values() installed_list = [] # obtain php version compiled via custombuild: phpXY for version_info in installed_php_data: version = version_info['php_version_id'] if version == 'no': continue directadmin_php_dir = f'/usr/local/php{version.replace(".", "")}/' if not os.path.exists(directadmin_php_dir): continue modules_dir_path = _get_da_php_extension_dir(directadmin_php_dir) if modules_dir_path: modules_dir_path = modules_dir_path.strip() installed_list.append(PHPDescription( identifier=f'php{version.replace(".", "")}', version=version, dir=os.path.join(directadmin_php_dir), modules_dir=modules_dir_path, bin=os.path.join(directadmin_php_dir, 'bin/php'), ini=os.path.join(directadmin_php_dir, 'lib/php.ini'), )) return installed_list def _get_aliases(path): """ Parse user aliases file and return data """ if not os.path.exists(path): return [] data = [] try: with open(path, encoding='utf-8') as f: data = f.readlines() except IOError as e: syslog.syslog(syslog.LOG_WARNING, f'Can`t open file "{path}" due to : "{e}"') return [record.strip().split('=')[0] for record in data] def _useraliases(cpuser, domain): """ Return aliases from user domain :param str\|unicode cpuser: user login :param str\|unicode domain: :return list of aliases """ path = f'/usr/local/directadmin/data/users/{cpuser}/domains/{domain}.pointers' data = _get_aliases(path) return data class PanelPlugin(GeneralPanelPluginV1): HTTPD_CONFIG_FILE = '/etc/httpd/conf/httpd.conf' HTTPD_MPM_CONFIG = '/etc/httpd/conf/extra/httpd-mpm.conf' HTTPD_INFO_CONFIG = '/etc/httpd/conf/extra/httpd-info.conf' def __init__(self): super().__init__() self.ADMINS_LIST = os.path.join(ADMIN_DIR, 'admin.list') def getCPName(self): """ Return panel name :return: """ return __cpname__ def get_cp_description(self): """ Retrieve panel name and it's version :return: dict: { 'name': 'panel_name', 'version': 'panel_version', 'additional_info': 'add_info'} or None if can't get info """ try: with subprocess.Popen( ['/usr/local/directadmin/directadmin', 'v'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) as p: out, _ = p.communicate() # output may differ (depending on version): # 'Version: DirectAdmin v.1.642' # 'DirectAdmin v.1.643 55acaa256ec6ed99b9aaec1050de793b298f62b0' # 'DirectAdmin 1.644 55acaa256ec6ed99b9aaec1050de793b298f62b0' version_words = (word.lstrip('v.') for word in out.split()) def _is_float(s): return s.replace('.', '').isdigit() version = next(filter(_is_float, version_words), '') return {'name': __cpname__, 'version': version, 'additional_info': None} except Exception: return None def db_access(self): """ 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 db_access() def cpusers(self): """ Generates a list of cpusers registered in the control panel :return: list of cpusers registered in the control panel :rtype: tuple """ return cpusers() def resellers(self): """ Generates a list of resellers in the control panel :return: list of cpusers registered in the control panel :rtype: tuple """ return resellers() def is_reseller(self, username): """ Check if given user is reseller; :type username: str :rtype: bool """ return is_reseller(username) # unlike admins(), this method works fine in post_create_user # hook; looks like directadmin updates admins.list a little bit later # then calls post_create_user.sh def is_admin(self, username): """ Return True if username is in admin names :param str username: user to check :return: bool """ user_conf_file = os.path.join(DA_USERS_PATH, username, USER_CONF) if not os.path.exists(user_conf_file): return False user_config = load_fast(user_conf_file) return user_config['usertype'] == 'admin' def dblogin_cplogin_pairs(self, cplogin_lst=None, with_system_users=False): """ Get mapping between system and DB users @param cplogin_lst :list: list with usernames for generate mapping @param with_system_users :bool: add system users to result list or no. default: False """ return dblogin_cplogin_pairs(cplogin_lst, with_system_users) def cpinfo(self, cpuser=None, keyls=('cplogin', 'package', 'mail', 'reseller', 'dns', 'locale'), 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 cpinfo(cpuser, keyls, search_sys_users=search_sys_users) def get_admin_email(self): """ Retrieve admin email address :return: Host admin's email """ return get_admin_email() def docroot(self, domain): """ Return document root for domain :param str\|unicode domain: :return Cortege: (document_root, owner) """ return docroot(domain) @staticmethod def useraliases(cpuser, domain): return _useraliases(cpuser, domain) def userdomains(self, cpuser): """ Return domain and document root pairs for control panel user first domain is main domain :param str\|unicode cpuser: user login :return list of tuples (domain_name, documen_root) """ return userdomains(cpuser) def homedirs(self): """ Detects and returns list of folders contained the home dirs of users of the cPanel :return: list of folders, which are parent of home dirs of users of the panel """ return homedirs() def reseller_users(self, resellername=None): """ Return reseller users :param resellername: reseller name; autodetect name if None :return list[str]: user names list """ return reseller_users(resellername) def reseller_domains(self, resellername=None): """ Get dict[user, domain] :param reseller_name: reseller's name :rtype: dict[str, str\|None] :raises DomainException: if cannot obtain domains """ return reseller_domains(resellername) def get_user_login_url(self, domain): """ Get login url for current panel; :type domain: str :rtype: str """ return get_user_login_url(domain) def admins(self): """ List all admins names in given control panel :return: list of strings """ return admins() def domain_owner(self, domain): """ Return domain's owner :param domain: Domain/sub-domain/add-domain name :rtype: str :return: user name or None if domain not found """ return domain_owner(domain) def get_domains_php_info(self): """ Return php version information for each domain :return: domain to php info mapping :rtype: dict[str, dict] """ return get_domains_php_info() @staticmethod def _get_da_skin_name(): """ Retrieve current DA skin name :return: Current DA skin name. None if unknown """ config = loadconfig(DA_CONF) # starting from DA 1.664 `docsroot` option was replaced by `system_skin` if 'system_skin' in config: return config['system_skin'] # grep '^docsroot=' /usr/local/directadmin/conf/directadmin.conf \| cut -d/ -f4 docsroot = config.get('docsroot', None) # docsroot like './data/skins/evolution' if docsroot is None: return None return docsroot.split('/')[-1] @staticmethod def get_encoding_name(): """ Retrieve encoding name, used for package/reseller names :return: """ enhanced_skin_config = os.path.join(DA_DIR, "data/skins/enhanced/lang/en/lf_standard.html") default_encoding = 'utf8' current_skin = PanelPlugin._get_da_skin_name() if current_skin == 'enhanced': # For enchanced skin we read encoding from its config # :LANG_ENCODING=iso-8859-1 see LU-99 for more info skin_config = loadconfig(enhanced_skin_config) # Option in file is 'LANG_ENCODING', but key is lowercase return skin_config.get('lang_encoding', default_encoding) return default_encoding def get_unsupported_cl_features(self) -> list[Feature]: return [ Feature.RUBY_SELECTOR, ] @staticmethod def get_apache_ports_list() -> list[int]: """ Retrieves active httpd's ports from httpd's config :return: list of apache's ports """ # cat /etc/apache2/conf/httpd.conf \| grep Listen _httpd_ports_list = [] try: lines = get_file_lines(PanelPlugin.HTTPD_CONFIG_FILE) except (OSError, IOError): return None lines = [line.strip() for line in lines] for line in grep('Listen', match_any_position=False, multiple_search=True, data_from_file=lines): # line examples: # Listen 0.0.0.0:80 # Listen [::]:80 try: value = int(line.split(' ')[1]) if value not in _httpd_ports_list: _httpd_ports_list.append(value) except (IndexError, ValueError): pass if not _httpd_ports_list: _httpd_ports_list.append(80) return _httpd_ports_list @staticmethod def _get_active_web_server_params() -> Tuple[str, str]: """ Determines active web server from options.conf, directive 'webserver' :return: tuple (active_web_server_name, apache_active_module_name) active_web_server_name: 'apache', 'nginx', 'nginx_apache', 'litespeed', 'openlitespeed', etc apache_active_module_name: 'prefork', 'event', 'worker' (None, None) if DA options.conf read/parse error """ web_server_name = None apache_active_module_name = None try: # cat /usr/local/directadmin/custombuild/options.conf \| grep webserver # webserver=apache # webserver can be: apache, nginx, nginx_apache, litespeed, openlitespeed. options_lines = get_file_lines(DA_OPT_PATH) grep_result_list = list(grep('^apache_mpm\|^webserver', fixed_string=False, match_any_position=False, multiple_search=True, data_from_file=options_lines)) # grep_result_list example: ['webserver=apache\n', 'apache_mpm=auto\n'] for line in grep_result_list: line_parts = line.strip().split('=') if line_parts[0] == 'webserver': web_server_name = line_parts[1] if line_parts[0] == 'apache_mpm': apache_active_module_name = line_parts[1] # modules are 'prefork', 'event', 'worker'. 'auto' == 'worker' if apache_active_module_name == 'auto': apache_active_module_name = 'worker' except (OSError, IOError, IndexError): pass return web_server_name, apache_active_module_name def _get_max_request_workers_for_module(self, apache_module_name: str) -> Tuple[int, str]: """ Determine MaxRequestWorkers directive value for specified apache module. Reads config file /etc/httpd/conf/extra/httpd-mpm.conf :param apache_module_name: Current apache's module name: 'prefork', 'event', 'worker' :return: tuple (max_req_num, message) max_req_num - Maximum request apache workers number or 0 if error message - OK/Error message """ try: return find_module_param_in_config(self.HTTPD_MPM_CONFIG, apache_module_name, 'MaxRequestWorkers') except (OSError, IOError, IndexError, ValueError): return 0, format_exc() def get_apache_max_request_workers(self) -> Tuple[int, str]: """ 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/Error message """ web_server_name, apache_active_module_name = self._get_active_web_server_params() if web_server_name is None or apache_active_module_name is None: return 0, f"There was error during read/parse {DA_OPT_PATH}. Apache collector will not work" if web_server_name != "apache": return 0, f"DA is configured for web server '{web_server_name}'; but 'apache' is needed. " \ "Apache collector will not work" return self._get_max_request_workers_for_module(apache_active_module_name) def _get_httpd_status_uri(self) -> str: """ Determine apache mod_status URI from /etc/httpd/conf/extra/httpd-info.conf config :return Apache mod_status URI or None if error/not found """ location_uri = None try: # # grep -B 2 'SetHandler server-status' /etc/httpd/conf/extra/httpd-info.conf # # <Location /server-status> # SetHandler server-status info_lines = get_file_lines(self.HTTPD_INFO_CONFIG) location_directive = '<Location' location_line = None for line in info_lines: line = line.strip() if line.startswith(location_directive): # Location directive found, save it location_line = line continue if line.startswith('SetHandler server-status') and location_line: # server-status found, Extract URI from Location directive start tag location_uri = location_line.replace(location_directive, '').replace('>', '').strip() break except (OSError, IOError): pass return location_uri def get_apache_connections_number(self): """ Retrieves Apache's connections number (from apache's mod_status) :return: tuple (conn_num, message) conn_num - current connections number, 0 if error message - OK/Trace """ web_server_name, _ = self._get_active_web_server_params() if web_server_name is None: return 0, f"There was error during read/parse {DA_OPT_PATH}. Apache collector will not work" if web_server_name != "apache": return 0, f"DA is configured for web server '{web_server_name}'; but 'apache' is needed. " \ "Apache collector will not work" try: # curl localhost/server-status?auto \| grep "Total Accesses" # Total Accesses: 25 location_uri = self._get_httpd_status_uri() if location_uri is None: return 0, "Can't found mod_status URI in configs" url = f'http://127.0.0.1{location_uri}?auto' response = requests.get(url, timeout=5) if response.status_code != 200: return 0, f"GET {url} response code is {response.status_code}" s_response = response.content.decode('utf-8') s_response_list = s_response.split('\n') out_list = list(grep("Total Accesses", data_from_file=s_response_list)) # out_list example: ['Total Accesses: 200'] s_total_accesses = out_list[0].split(':')[1].strip() return int(s_total_accesses), 'OK' except Exception: return 0, format_exc() @staticmethod def get_installed_php_versions(): """ Returns installed alt-php(s) on the server compiled phpXY via custombuild and alt-phpXY has different paths also user could choose version via PHP selector which was not compiled with custombuild (will be absent in DA configs) """ return _get_installed_alt_php_versions() + _get_compiled_custombuild_versions() def get_server_ip(self): ip_list_file = '/usr/local/directadmin/data/admin/ip.list' if not os.path.exists(ip_list_file): return '' with open(ip_list_file, encoding='utf-8') as f: ips = f.readlines() if not ips: return '' return ips[0].strip() @staticmethod def get_user_emails_list(username: str, domain: str): user_conf = f'/usr/local/directadmin/data/users/{username}/user.conf' if not os.path.exists(user_conf): return '' user_conf = load_fast(user_conf) return user_conf.get('email', '') @staticmethod def panel_login_link(username): generated_login = subprocess.run(['/usr/local/directadmin/directadmin', '--create-login-url', f'user={username}'], capture_output=True, text=True, check=False).stdout # http://server-206-252-237-2.da.direct:2222/api/login/url?key=0SrJm1CNAIh34w4Fk8Kp4ohypUFp_pMm if len(generated_login) == 0: return '' parsed = urlparse(generated_login) return f'{parsed.scheme}://{parsed.netloc}/' @staticmethod def panel_awp_link(username): link = PanelPlugin.panel_login_link(username).rstrip("/") if len(link) == 0: return '' return f'{link}/evo/user/plugins/awp#/' def suspended_users_list(self): all_users = cpusers() suspended_users = [] for user in all_users: user_conf_file = os.path.join(DA_USERS_PATH, user, USER_CONF) if not os.path.exists(user_conf_file): continue user_config = load_fast(user_conf_file) if user_config.get('suspended') == 'yes': suspended_users.append(user) return suspended_users	40.74 KB	2026-02-10 15:10
📄 interworx.py # -- coding: utf-8 -- import configparser import os import re from typing import Dict # NOQA from clcommon.cpapi.cpapiexceptions import NotSupported from clcommon.cpapi.GeneralPanel import GeneralPanelPluginV1 from clcommon.cpapi.plugins.universal import _dblogin_cplogin_pairs from clcommon.features import Feature __cpname__ = 'InterWorx' PARSE_URI_PATTERN = re.compile(r'(.+)://(.+):(.+)@.*') IWORX_INI = '/usr/local/interworx/iworx.ini' # WARN: Probably will be deprecated for our "official" plugins. # See pluginlib.detect_panel_fast() def detect(): return os.path.isfile(IWORX_INI) def db_access(_conf_path=IWORX_INI): access = {} data_config = configparser.ConfigParser(strict=False) data_config.read(_conf_path) mysql_uri = data_config.get(section='mysql', option='rootdsn').strip('\"') parse_result = PARSE_URI_PATTERN.search(mysql_uri) (db, login, passwd) = (None, None, None) if parse_result: (db, login, passwd) = parse_result.group(1, 2, 3) access['login'] = login access['pass'] = passwd access['host'] = data_config.get(section='mysql', option='default_server').strip('\"') access['db'] = db return access def cpusers(): raise NotSupported({ 'message': '%(action)s is not currently supported.', 'context': {'action': 'Getting all users registered in the Control Panel'} }) def dblogin_cplogin_pairs(cplogin_lst=None, with_system_users=False): access = db_access() data = _dblogin_cplogin_pairs(cplogin_lst=cplogin_lst, access=access) return data def get_user_login_url(domain): return f'https://{domain}:2443' class PanelPlugin(GeneralPanelPluginV1): def __init__(self): super().__init__() def getCPName(self): """ Return panel name :return: """ return __cpname__ def get_cp_description(self): """ Retrieve panel name and it's version :return: dict: { 'name': 'panel_name', 'version': 'panel_version', 'additional_info': 'add_info'} or None if can't get info """ try: with open(IWORX_INI, "r", encoding='utf-8') as f: out = f.read() version = out.split('version')[4].replace('\n', '').replace('=', '').replace('"', '').split('[')[0] return {'name': __cpname__, 'version': version, 'additional_info': None} except Exception: return None def db_access(self): """ 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 db_access() @staticmethod def useraliases(cpuser, domain): """ Return aliases from user domain :param str\|unicode cpuser: user login :param str\|unicode domain: :return list of aliases """ return [] def cpusers(self): """ Generates a list of cpusers registered in the control panel :return: list of cpusers registered in the control panel :rtype: tuple """ return cpusers() def get_user_login_url(self, domain): """ Get login url for current panel; :type domain: str :rtype: str """ return get_user_login_url(domain) def get_unsupported_cl_features(self) -> list[Feature]: return [ Feature.RUBY_SELECTOR, Feature.PYTHON_SELECTOR, Feature.NODEJS_SELECTOR, Feature.RESELLER_LIMITS, Feature.WPOS, ]	3.67 KB	2026-02-10 15:10
📄 ispmanager.py # -- coding: utf-8 -- import re import os import grp import subprocess from typing import Dict # NOQA from clcommon import mysql_lib from clcommon.cpapi.cpapiexceptions import NoDBAccessData, NotSupported from clcommon.cpapi.GeneralPanel import GeneralPanelPluginV1 from clcommon.features import Feature __cpname__ = 'ISPManager' def _is_5_version(): return os.path.isfile('/usr/local/mgr5/sbin/mgrctl') # WARN: Probably will be deprecated for our "official" plugins. # See pluginlib.detect_panel_fast() def detect(): return os.path.isfile('/usr/local/ispmgr/bin/ispmgr') or _is_5_version() ISP_DB_CONF = '/usr/local/ispmgr/etc/ispmgr.conf' ISP5_DB_CONF = '/usr/local/mgr5/etc/ispmgr.conf.d/db.conf' SECTION_PATTERN = r'(\S+) "([^"]+)" {([^}]+)}' KEYWORDS_PATTERN = r'(\S+)\s+(\S+)' def conf_pars(sectype, secname=None, seckeys=None, path=ISP_DB_CONF): """ /usr/local/ispmgr/etc/ispmgr.conf parser :param sectype: Type sector for example: Service or DbServer or Account :param secname: Name sector. May be different :param seckeys: Name key for retrieving and filtering :param path: path to config file default /usr/local/ispmgr/etc/ispmgr.conf :return: list """ seckeys_filter = {} seckeys_extracted = None if seckeys: seckeys_extracted = [] for key_val in seckeys: key_val_splited = key_val.split() if len(key_val_splited) == 2: seckeys_filter.update(dict([key_val_splited])) seckeys_extracted.append(key_val_splited[0]) elif len(key_val_splited) == 1: seckeys_extracted.append(key_val_splited[0]) with open(path, encoding='utf-8') as f: result_list = [] for stype, sname, sbody in re.findall(SECTION_PATTERN, f.read()): blst = re.findall(KEYWORDS_PATTERN, sbody) if stype == sectype and secname in (None, secname): result = dict([(k, v) for k, v in blst if seckeys_extracted is None or k in seckeys_extracted]) if set(seckeys_filter.items()).issubset(set(result.items())): result_list.append(result) return result_list def _db_access_5(): try: with open(ISP5_DB_CONF, encoding='utf-8') as db_conf: cnf = dict(re.findall(KEYWORDS_PATTERN, db_conf.read())) return {'pass': cnf['DBPassword'], 'login': cnf['DBUser'], 'host': cnf['DBHost'], 'db': 'mysql'} except IOError as e: raise NoDBAccessData(f'Can not open config file {KEYWORDS_PATTERN}') from e except IndexError as e: raise NoDBAccessData( f'Can not find database access data in config file {KEYWORDS_PATTERN}' ) from e def db_access(_conf_path=ISP_DB_CONF): if _is_5_version(): # ISP Manager 5 return _db_access_5() # ISP Manager 4 access = {} access_list = conf_pars(sectype='DbServer', seckeys=('Hostname', 'Password', 'Type mysql', 'User'), path=_conf_path) for access_from_conf in access_list: try: access['pass'] = access_from_conf['Password'] access['login'] = access_from_conf['User'] access['host'] = access_from_conf['Hostname'] access['db'] = 'mysql' return access except KeyError: pass raise NoDBAccessData( f'Can not find database access data for localhost in config file {_conf_path}' ) def _dbname_dblogin_pairs(access): dbhost = access.get('host', 'localhost') dblogin = access['login'] dbpass = access['pass'] sql = 'SELECT db.Db, db.User FROM db GROUP BY db.User, db.Db' connector = mysql_lib.MySQLConnector(host=dbhost, user=dblogin, passwd=dbpass, db='mysql') with connector.connect() as db: return db.execute_query(sql) def cpusers(): raise NotSupported({ 'message': '%(action)s is not currently supported.', 'context': {'action': 'Getting all users registered in the Control Panel'} }) def _dbname_cplogin_pairs_iter(cplogin_lst=None, _conf_path=ISP_DB_CONF): """ Extract (database name <=> control panel login) pairs from ISPmanager config file :param cplogin_lst: :param _conf_path: :return: """ grpid_login_dict = dict([(grp_tuple[2], grp_tuple[0]) for grp_tuple in grp.getgrall()]) with open(_conf_path, encoding='utf-8') as f: for line_numb, line in enumerate(f): if line.startswith('DbAssign '): line_splited = line.split() if len(line_splited) >= 3: dbname, user_uid = line_splited[2:] try: cplogin = grpid_login_dict.get(int(user_uid)) if cplogin is None: print(f'WARNING: can not find group name with id {user_uid}; ' f'line {line_numb} in file {_conf_path}') except ValueError: # if can not convert user_uid to int print(f'WARNING: can not pars line {line_numb} in file {_conf_path}') cplogin = None if cplogin and (cplogin_lst is not None or cplogin in cplogin_lst): yield dbname, cplogin def get_user_login_url(domain): return f'https://{domain}:1500' class PanelPlugin(GeneralPanelPluginV1): def __init__(self): super().__init__() def getCPName(self): """ Return panel name :return: """ return __cpname__ def get_cp_description(self): """ Retrieve panel name and it's version :return: dict: { 'name': 'panel_name', 'version': 'panel_version', 'additional_info': 'add_info'} or None if can't get info """ #################################################################### # ISP Manager v4 and v5 (master and node) check # ISPmanager v5 check and detect its type - master/node # ISP Manager support recommendation: # Node has package ispmanager-node-common-5.56.0-3.el6.x86_64 # Master has packages ispmanager-business-common-5.56.0-3.el6.x86_64 and # ispmanager-business-5.56.0-3.el6.x86_64, may have ispmanager-node-common-5.56.0-3.el6.x86_64 def _get_isp_manager_v4_description(): """ Handle ISP Manager v4 specific version retrieval. """ with subprocess.Popen( ['/usr/local/ispmgr/bin/ispmgr', '-v'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) as proc: out, _ = proc.communicate() version = out.replace('\n', '').split()[1] return {'name': __cpname__, 'version': version, 'additional_info': None} def _get_isp_manager_v5_description(): """ Handle ISP Manager v5 specific checks and version retrieval. """ packages = [ ('ispmanager-business-common', 'Master'), ('ispmanager-node-common', 'Node'), ('ispmanager-lite-common', 'Lite'), ] for package, node_type in packages: version = _query_package_version(package) if version is not None: return { 'name': 'ISP Manager', 'version': version, 'additional_info': node_type, } raise RuntimeError('Failed to detect version of ISP manager') def _query_package_version(package_name): """ Query the version of a package using rpm. """ with subprocess.Popen( ['/bin/rpm', '-q', '--queryformat', '[%{VERSION}]', package_name], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) as proc: version, _ = proc.communicate() if proc.returncode != 0 or not version: return None return version.strip() try: if os.path.isfile('/usr/local/mgr5/sbin/mgrctl'): # ISP Manager v5 return _get_isp_manager_v5_description() else: # ISPmanager v4 return _get_isp_manager_v4_description() except Exception: return None def db_access(self): """ 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 db_access() @staticmethod def useraliases(cpuser, domain): """ Return aliases from user domain :param str\|unicode cpuser: user login :param str\|unicode domain: :return list of aliases """ return [] def cpusers(self): """ Generates a list of cpusers registered in the control panel :return: list of cpusers registered in the control panel :rtype: tuple """ return cpusers() def get_user_login_url(self, domain): """ Get login url for current panel; :type domain: str :rtype: str """ return get_user_login_url(domain) def get_unsupported_cl_features(self) -> list[Feature]: return [ Feature.RUBY_SELECTOR, Feature.PYTHON_SELECTOR, Feature.NODEJS_SELECTOR, Feature.RESELLER_LIMITS, Feature.WPOS, ]	9.64 KB	2026-02-10 15:10
📄 nopanel.py # -- coding: utf-8 -- from clcommon.cpapi.const import UNKNOWN_CP_NAME from clcommon.cpapi.cpapiexceptions import NotSupported from clcommon.cpapi.GeneralPanel import GeneralPanelPluginV1 from clcommon.features import Feature __cpname__ = UNKNOWN_CP_NAME def get_cp_description(): """ Retrieve panel name and it's version :return: dict: { 'CPName': 'panel_name', 'CPVer': 'panel_version', 'CPAdd': 'add_info'} or None if can't get info """ return None def db_access(): raise NotSupported('Unable to receive data. Not supported without a control panel') def cpusers(): """ return all users list registered on Control Panel :return: tuple registered users """ raise NotSupported({ 'message': '%(action)s is not currently supported.', 'context': {'action': 'Getting all users registered in the control panel'} }) def dblogin_cplogin_pairs(cplogin_lst=None, with_system_users=False): raise NotSupported('Unable to receive data. Not supported without a control panel') def get_admin_email(_hostname=None): """ Gets admin email :param str\|None _hostname: for testing :return: admin's email :rtype: string """ from clcommon.cpapi.plugins.universal import get_admin_email # pylint: disable=import-outside-toplevel return get_admin_email(_hostname=_hostname) class PanelPlugin(GeneralPanelPluginV1): def __init__(self): super().__init__() def getCPName(self): """ Return panel name :return: """ return __cpname__ def get_cp_description(self): """ Retrieve panel name and it's version :return: dict: { 'CPName': 'panel_name', 'CPVer': 'panel_version', 'CPAdd': 'add_info'} or None if can't get info """ return None def get_admin_email(self): """ Retrieve admin email address :return: Host admin's email """ return get_admin_email() def db_access(self): raise NotSupported('Unable to receive data. Not supported without a control panel') def cpusers(self): """ return all users list registered on Control Panel :return: tuple registered users """ raise NotSupported({ 'message': '%(action)s is not currently supported.', 'context': {'action': 'Getting all users registered in the control panel'} }) def dblogin_cplogin_pairs(self, cplogin_lst=None, with_system_users=False): raise NotSupported('Unable to receive data. Not supported without a control panel') def admin_packages(self, raise_exc=False): """ Return list of available admin's packages :return: List of packages. For example ['BusinessPackage', 'Package2'] """ return [] def reseller_package_by_uid(self, user_id): """ Retrieves reseller name and package name by uid :param user_id: User id :return: Cortege: (Reseller_name, Package_name) """ return () def get_uids_list_by_package(self, package_name, reseller_name=None): """ Retrieves uid list for package :param package_name: Package name :param reseller_name: Reseller name. None for admin's package :return: List of uids Example: ['1000', '1002', '1006', '1007', '1008'] """ return [] def cpinfo(self, cpuser=None, keyls=('cplogin', 'package', 'mail', 'reseller', 'dns', 'locale'), search_sys_users=True): """ Retrives specified info about panel users :param str\|unicode\|list\|tuple\|None cpuser: user login :param keyls: List or cortege of data which is necessary to obtain the user, values can 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 userid - uid :param search_sys_users: :return: returns a tuple of tuples of data in the same sequence as specified keys in keylst Examples: cpinfo('cltest1') (('cltest1', 'default', '', 'root', 'cltest1.com', 'en'),) cpinfo() (('res1usr1', 'res1_pack1', '', 'res1', 'res1usr1.com', 'en'), ('res1', 'default', '', 'root', 'res1.com', 'en'), ('res2', 'default', '', 'res2', 'res2.com', 'en'), ('cltest1', 'default', '', 'root', 'cltest1.com', 'en'), ('system', 'undefined', None, 'root', '', 'en')) :rtype: tuple """ return () def resellers_packages(self, raise_exc=False): """ Return dictionary, contains available resellers packages, grouped by resellers :return: Dictionary. Example: {'res1': ['BusinessPackage', 'UltraPackage', 'Package'], 'res2': ['SimplePackage', 'Package'] } """ return {} def get_admin_emails_list(self): """ Gets admin emails list :rtype: List :return: List: ['admin1@mail.com', 'admin2@mail.com' ] """ return [] def docroot(self, domain): """ Return document root for domain :return: Cortege: (document_root, owner) """ return "", "" @staticmethod def useraliases(cpuser, domain): """ Return aliases from user domain :param str\|unicode cpuser: user login :param str\|unicode domain: :return list of aliases """ return [] def userdomains(self, 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 [] def homedirs(self): """ 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 [] def reseller_users(self, resellername=None): """ Return reseller users :param resellername: reseller name; autodetect name if None :rtype: List :return list[str]: user names list """ return [] def reseller_domains(self, resellername=None): """ Return reseller users and their main domains :param resellername: reseller name; autodetect name if None :return dict[str, str]: pairs user <==> domain """ return {} def get_user_login_url(self, domain): """ Get login url for current panel; :type domain: str :rtype: str :return: Panel login URL """ return "" def get_reseller_id_pairs(self): """ Get dict reseller => id Optional method for panels without hard link reseller <=> system user :rtype: dict[str,int] - {'res1': id1} :return: """ return {} def get_unsupported_cl_features(self) -> list[Feature]: """ Return list of CloudLinux features that can be used on the current control panel. """ # PHP_SELECTOR can work without a control panel, other selectors can't, # because they need actual info from cpapi functions like userdomains, not empty lists return_value = [ Feature.RUBY_SELECTOR, Feature.PYTHON_SELECTOR, Feature.NODEJS_SELECTOR, Feature.RESELLER_LIMITS, Feature.WPOS, ] return return_value	7.75 KB	2026-02-10 15:10
📄 plesk.py # -- coding: utf-8 -- import os import re import time import xml.etree.ElementTree as ETree from collections import defaultdict from functools import wraps from traceback import format_exc from typing import Any, AnyStr, Dict, List, Optional, Tuple, Union # NOQA from urllib.parse import urlparse from clcommon import ClPwd, mysql_lib from clcommon.clfunc import uid_max from clcommon.cpapi.cpapicustombin import _docroot_under_user_via_custom_bin, get_domains_via_custom_binary from clcommon.cpapi.cpapiexceptions import DuplicateData, NoDomain, NoPackage, NoPanelUser, NotSupported from clcommon.cpapi.GeneralPanel import DomainDescription, GeneralPanelPluginV1, PHPDescription from clcommon.features import Feature from clcommon.utils import ExternalProgramFailed, find_module_param_in_config, get_modules_dir_for_alt_php, run_command PSA_SHADOW_PATH = "/etc/psa/.psa.shadow" SUPPORTED_CPINFO = {'cplogin', 'package', 'mail', 'reseller', 'dns', 'locale'} UID_MAX = uid_max() __cpname__ = 'Plesk' # WARN: Probably will be deprecated for our "official" plugins. # See pluginlib.detect_panel_fast() def detect(): return os.path.isfile('/usr/local/psa/version') def db_access(_pass_path=PSA_SHADOW_PATH): access = {} access['login'] = 'admin' with open(_pass_path, 'r', encoding='utf-8') as f: access['pass'] = f.read().strip() return access def query_sql(query, data=None, _access=None, _dbname='psa', as_dict=False): """ Return the result of a Plesk database query :param query: SQL query string with possible parameters :param data: arguments for the SQL parameter insertion :param _access: database authentication data :param _dbname: the name of the database :param as_dict: controls the format of the output data :type query: str :type _access: dict :type as_dict: bool :return: Tuple of rows according to the query in the format specified by as_dict :rtype: tuple(tuple) or tuple(dict) """ # Example of returned data: # >>> query_sql('SELECT login from sys_users') # ((u'cltest',), (u'cltest3',), (u'user2',), (u'user1tst',)) # >>> query_sql('SELECT login from sys_users', as_dict=True) # ({'login': u'cltest'}, # {'login': u'cltest3'}, # {'login': u'user2'}, # {'login': u'user1tst'}) access = _access or db_access() dbhost = access.get('host', 'localhost') dblogin = access['login'] dbpass = access['pass'] connector = mysql_lib.MySQLConnector(host=dbhost, user=dblogin, passwd=dbpass, db=_dbname, use_unicode=True, charset='utf8', as_dict=as_dict) with connector.connect() as db: return db.execute_query(query, args=data) def cpusers(_access=None, _dbname='psa'): cpusers_lst = [fetched_one[0] for fetched_one in cpinfo(keyls=('cplogin', ))] return cpusers_lst def resellers(): sql = "SELECT clients.login FROM clients WHERE clients.type='reseller'" return [cplogin for (cplogin, ) in query_sql(sql)] def admins(): sql = "SELECT clients.login FROM clients WHERE clients.type='admin'" return set([cplogin for (cplogin, ) in query_sql(sql)]) def is_reseller(username): sql = "SELECT clients.type FROM clients WHERE clients.login=%s" try: return query_sql(sql, (username,))[0][0] == 'reseller' except IndexError: return False def _sys_users_info(sys_login, keyls): # type: (Any[str, None], Tuple[str]) -> List[Tuple] # Templates.name can be None and it is ok mapping = { 'cplogin': 'sys_users.login AS cplogin', 'mail': 'clients.email AS email', 'reseller': 'reseller.login AS reseller', 'dns': 'domains.name AS dns', 'locale': 'clients.locale AS local', 'package': 'Templates.name AS package' } select_query = ', '.join([mapping[key] for key in keyls]) sql = rf"""SELECT {select_query} FROM sys_users JOIN hosting ON hosting.sys_user_id=sys_users.id JOIN domains ON hosting.dom_id=domains.id AND domains.webspace_id=0 JOIN clients ON clients.id=domains.cl_id JOIN clients reseller ON reseller.id=domains.vendor_id LEFT JOIN Subscriptions ON Subscriptions.object_type = "domain" AND domains.id = Subscriptions.object_id LEFT JOIN PlansSubscriptions ON PlansSubscriptions.subscription_id = Subscriptions.id LEFT JOIN Templates AS Templates ON Templates.id = PlansSubscriptions.plan_id AND "domain" = Templates.type """ # make query like "where x in (%s, %s, %s, ...)" if isinstance(sys_login, (list, tuple)): placeholders = ','.join(['%s'] * len(sys_login)) sql += rf" WHERE sys_users.login IN ({placeholders})" users = query_sql(sql, data=sys_login) return users def _resellers_info(sys_login, keyls): # type: (Any[str, None], Tuple[str]) -> List[Tuple] # items with 'NULL' are not available for this panel mapping = { 'cplogin': 'clients.login AS cplogin', 'mail': 'clients.email AS email', 'reseller': 'NULL as reseller', 'dns': 'NULL as dns', 'locale': 'clients.locale AS local', 'package': 'NULL as package' } select_query = ', '.join([mapping[key] for key in keyls]) sql = f"SELECT {select_query} FROM clients WHERE clients.type IN (\"reseller\", \"admin\")" # make query like "where x in (%s, %s, %s, ...)" if isinstance(sys_login, (list, tuple)): placeholders = ','.join(['%s'] * len(sys_login)) sql += rf" AND clients.login IN ({placeholders})" users = query_sql(sql, data=sys_login) return users def cpinfo(cpuser=None, keyls=('cplogin', 'package', 'mail', 'reseller', 'dns', 'locale'), search_sys_users=True): """ Get info about user[s] or about reseller[s]. :param str\|None cpuser: get info about specified login, None for all :param list\|tuple keyls: keys to return :param bool search_sys_users: work with sys users or with resellers :rtype: tuple[tuple] """ if isinstance(cpuser, str): cpuser = [cpuser] # just for developers for key in keyls: if key not in SUPPORTED_CPINFO: raise NotSupported(f'Key {key} is not supported for this control panel. ' f'Available keys: {SUPPORTED_CPINFO}') if search_sys_users: return _sys_users_info(cpuser, keyls) return _resellers_info(cpuser, keyls) def get_admin_email(args, kwargs): try: return query_sql(r"SELECT val FROM misc WHERE param='admin_email';")[0][0] except IndexError: return None def docroot_basic(domain): # type: (str) -> Any[None, Tuple[str, str]] sql = r""" SELECT hosting.www_root, sys_users.login FROM hosting JOIN domains ON hosting.dom_id=domains.id JOIN sys_users ON hosting.sys_user_id=sys_users.id WHERE domains.name=%s """ try: return query_sql(sql, data=(domain,))[0] except IndexError as e: raise NoDomain(f'Cannot obtain document root for {domain}') from e def docroot(domain): # type: (str) -> Any[None, Tuple[str, str]] res = None domain = domain.strip() uid = os.getuid() euid = os.geteuid() if euid == 0 and uid == 0: res = docroot_basic(domain) else: res = _docroot_under_user_via_custom_bin(domain) # If there was successful result, res object will have # (doc_root, domain_user) format. If there wasn't found any correct # doc_roots, res will be None. if res is not None: return res raise NoDomain(f"Can't obtain document root for domain '{domain}'") def reseller_users(resellername): """ Return reseller users :param resellername: reseller name; return empty list if None :return list[str]: user names list """ if resellername is None: return [] sql = """ SELECT sys_users.login FROM clients as reseller JOIN domains ON domains.vendor_id=reseller.id JOIN hosting ON hosting.dom_id=domains.id JOIN sys_users ON hosting.sys_user_id=sys_users.id WHERE domains.webspace_id=0 AND reseller.login=%s; """ return [sys_login for (sys_login,) in query_sql(sql, data=(resellername,))] def memoize(f): cache = {'userdomains_map': {}} @wraps(f) def wrapper(cpuser, args, *kwargs): if cpuser not in cache['userdomains_map']: cache['userdomains_map'] = f(cpuser, args, *kwargs) return cache['userdomains_map'][cpuser] return wrapper @memoize def userdomains_basic(cpuser, _access=None, _dbname='psa'): """ Return domains of given user :param str cpuser: Username :param str _dbname: Database name where is located data :return: List of domains pairs such as (domain_name, None) to be suitable for domain_lib, starting from a main domain. :rtype: list of tuples :raises NoPanelUser: User is not found in Plesk database. """ # WARN: ORDER BY columns must be present in SELECT columns for newer Mysql # webspace_id == 0 is main domain sql = r""" SELECT DISTINCT su.login, d.name, h.www_root, d.webspace_id FROM domains as d, hosting as h, sys_users as su WHERE h.sys_user_id = su.id AND h.dom_id = d.id ORDER BY d.webspace_id ASC; """ # data: # ( # (u'customer1', u'customer1.org', 10L), # (u'customer1', u'mk.customer1.org.customer1.org', 10L) # ) data = query_sql(sql, as_dict=True, _access=_access) # _user_to_domains_map: # { 'user1': [('user1.org', '/var/www/vhosts/user1.com/httpdocs'), # ('mk.user1.org', '/var/www/vhosts/user1.com/mk.user1.org')] } _user_to_domains_map = defaultdict(list) for data_dict in data: _user_to_domains_map[data_dict['login']].append( (data_dict['name'], data_dict['www_root'])) if cpuser not in _user_to_domains_map: raise NoPanelUser( f'User {cpuser} not found in the database') return _user_to_domains_map def userdomains(cpuser, _access=None, _dbname='psa', as_root=False): """ Return domains of given user :param str cpuser: Username :param str _dbname: Database name where is located data :return: List of domains pairs such as (domain_name, None) to be suitable for domain_lib, starting from a main domain. :rtype: list of tuples :raises NoPanelUser: User is not found in Plesk database. """ euid = os.geteuid() if euid == 0 or _access or as_root: return userdomains_basic(cpuser, _access, _dbname) # this case works the same as above but through the rights escalation binary wrapper # call path: here -> binary -> python(diradmin euid) -> userdomains(as_root=True) -> print json result to stdout rc, res = get_domains_via_custom_binary() if rc == 0: return res elif rc == 11: raise NoPanelUser(f'User {cpuser} not found in the database') else: raise ExternalProgramFailed(f'Failed to get userdomains: {res}') def domain_owner(domain, _access=None, _dbname='psa'): """ Return domain owner :param str domain: Domain/sub-domain/add-domain name :param str _dbname: Database name where is located data :return: user name or None if domain not found :rtype: str """ sql = r""" SELECT DISTINCT `su`.`login` FROM `sys_users` `su`, `hosting` `h`, `domains` `d`, `domains` `sd` WHERE `h`.`sys_user_id`=`su`.`id` AND `h`.`dom_id`=`d`.`id` AND (`d`.`name`=%s OR `d`.`id`=`sd`.`webspace_id` AND `sd`.`name`=%s)""" users_list = [u[0] for u in query_sql(sql, (domain, domain))] # FIXME: how this possible? if len(users_list) > 1: raise DuplicateData( f"domain {domain} belongs to few users: [{','.join(users_list)}]" ) if len(users_list) == 0: return None return users_list[0] def dblogin_cplogin_pairs(cplogin_lst=None, with_system_users=False): raise NotSupported('Getting binding credentials in the database to the user name in the system is not currently ' 'supported.') def homedirs(_sysusers=None, _cpusers=None): """ Detects and returns list of folders contained the home dirs of users of the Plesk :param str\|None _sysusers: for testing :param str\|None _cpusers: for testing :return: list of folders, which are parent of home dirs of users of the panel """ homedirs = [] if _cpusers is None: try: results = cpusers() except NoPackage: results = None else: results = _cpusers users = [] if results is not None: users = [line[0] for line in results] # Plesk assumes MIN_UID as 10000 clpwd = ClPwd(10000) users_dict = clpwd.get_user_dict() # for testing only if isinstance(_sysusers, (list, tuple)): class pw: def __init__(self, name, dir): self.pw_name = name self.pw_dir = dir users_dict = {} for (name, dir) in _sysusers: users_dict[name] = pw(name, dir) for user_name, user_data in users_dict.items(): if len(users) and user_name not in users: continue homedir = os.path.dirname(user_data.pw_dir) if homedir not in homedirs: homedirs.append(homedir) return homedirs def get_user_login_url(domain): return f'https://{domain}:8443' def get_reseller_id_pairs(): """ Plesk has no user associated with reseller, but we need some id for out internal purposes. Let's take it from database. """ sql = """SELECT clients.login, clients.id + %s FROM clients WHERE clients.type='reseller'""" return dict(query_sql(sql, data=[UID_MAX])) def reseller_domains(resellername): # type: (str) -> Dict[str, str] if not resellername: return {} sql = r"""SELECT sys_users.login AS cplogin, domains.name AS dns FROM sys_users JOIN hosting ON hosting.sys_user_id=sys_users.id JOIN domains ON hosting.dom_id=domains.id AND domains.webspace_id=0 JOIN clients reseller ON reseller.id=domains.vendor_id WHERE reseller.login=%s """ users = query_sql(sql, data=[resellername]) return dict(users) def _extract_xml_value(xml_string, key): """ Plesk stores some information in simple xml formatted strings. """ try: elem = ETree.fromstring(xml_string).find(key) except ETree.ParseError: return None else: return elem.text if elem is not None else None def get_domains_php_info(): """ Plesk stores the information about the handler in xml format. Return the php version info for each domain. Example output: {'cltest.com': {'handler_type': 'fpm', 'php_version_id': 'plesk-php71-fpm', 'username': 'cltest'},` 'cltest2.com': {'handler_type': 'fastcgi', 'php_version_id': 'x-httpd-lsphp-custom', 'username': 'kek_2'}, 'cltest3.com': {'handler_type': 'fastcgi', 'php_version_id': 'plesk-php56-fastcgi', 'username': 'cltest3'}, 'omg.kek': {'handler_type': 'fastcgi', 'php_version_id': 'plesk-php71-fastcgi', 'username': 'cltest'}} :rtype: dict[str, dict] """ sql = r""" SELECT sys_users.login, d.name, h.php_handler_id, handlers.value FROM domains AS d JOIN hosting AS h ON h.dom_id=d.id JOIN sys_users ON h.sys_user_id=sys_users.id JOIN (SELECT ServiceNodeEnvironment. FROM ServiceNodeEnvironment WHERE (serviceNodeId = '1' AND section = 'phphandlers')) AS handlers ON handlers.name=h.php_handler_id WHERE h.php='true' """ # Php hanlder info xml example: # # <?xml version="1.0" encoding="UTF-8"?> # <handler> # <id>plesk-php71-fpm</id> # <type>fpm</type> # <typeName>FPM application</typeName> # <version>7.1</version> # <fullVersion>7.1.22</fullVersion> # <displayname>7.1.22</displayname> # <path>/opt/plesk/php/7.1/sbin/php-fpm</path> # <clipath>/opt/plesk/php/7.1/bin/php</clipath> # <phpini>/opt/plesk/php/7.1/etc/php.ini</phpini> # <custom>true</custom> # <registered>true</registered> # <service>plesk-php71-fpm</service> # <poold>/opt/plesk/php/7.1/etc/php-fpm.d</poold> # <outdated /> # </handler> domains_php_info = query_sql(sql) # yep, vendor php_handler_id has only "fpm/cgi/fastcgi" w/o version, so additional bicycle needed vendor_version_ids = ['cgi', 'fastcgi', 'fpm', 'x-httpd-lsphp-custom'] php_versions = {} for username, domain, php_handler_id, handler_xml in domains_php_info: display_version = php_handler_id if php_handler_id not in vendor_version_ids \ else f'vendor-php{_extract_xml_value(handler_xml, "version")}'.replace('.', '') def _cast(handler_name: str, version_id: str) -> str: if handler_name == 'fpm': return 'php-fpm' elif 'x-httpd-lsphp' in version_id: return 'lsapi' return handler_name handler = _extract_xml_value(handler_xml, key='type') or 'unknown' handler = _cast(handler, php_handler_id) # transform different php variations into some normal form display_version = display_version\ .replace('-dedicated', '')\ .replace('-fpm', '')\ .replace('-fastcgi', '')\ .replace('x-httpd-lsphp-', 'alt-php') php_versions[domain] = DomainDescription( username=username, php_version_id=display_version, # not a typo handler_type=handler, display_version=display_version ) return php_versions def get_main_username_by_uid(uid: int) -> str: """ Get "main" panel username by uid. :param uid: uid :return Username or 'N/A' if user not found """ if uid == 0: return 'root' try: _clpwd = ClPwd() pwd_list = _clpwd.get_pw_by_uid(uid) if os.geteuid() == 0: for user_pwd in pwd_list: username = user_pwd.pw_name try: userdomains(username) return username except NoPanelUser: pass else: # Under user cycle implemented in suid binary, see scripts/plesk_suid_caller.py username = pwd_list[0].pw_name userdomains(username) return username except (NoPanelUser, ClPwd.NoSuchUserException): pass return 'N/A' class PanelPlugin(GeneralPanelPluginV1): def __init__(self): super().__init__() self.HTTPD_MPM_CONFIG = '/etc/httpd/conf.modules.d/01-cgi.conf' # Defaults of MaxRequestWorkers for all possible mpm modules self.MPM_MODULES = { "prefork": 256, "worker": 400, "event": 400 } # Vars for httpd modules caching self.httpd_modules_ts = 0 self.httpd_modules = "" def getCPName(self): """ Return panel name :return: """ return __cpname__ def get_cp_description(self): """ Retrieve panel name and it's version :return: dict: { 'name': 'panel_name', 'version': 'panel_version', 'additional_info': 'add_info'} or None if can't get info """ try: with open("/usr/local/psa/version", "r", encoding="utf-8") as f: out = f.read() return {'name': __cpname__, 'version': out.split()[0], 'additional_info': None} except Exception: return None def db_access(self): """ 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 db_access() def cpusers(self): """ Generates a list of cpusers registered in the control panel :return: list of cpusers registered in the control panel :rtype: tuple """ return cpusers() def resellers(self): """ Generates a list of resellers in the control panel :return: list of cpusers registered in the control panel :rtype: tuple """ return resellers() def is_reseller(self, username): """ Check if given user is reseller; :type username: str :rtype: bool """ return is_reseller(username) def dblogin_cplogin_pairs(self, cplogin_lst=None, with_system_users=False): """ Get mapping between system and DB users @param cplogin_lst :list: list with usernames for generate mapping @param with_system_users :bool: add system users to result list or no. default: False """ return dblogin_cplogin_pairs(cplogin_lst, with_system_users) def cpinfo(self, cpuser=None, keyls=('cplogin', 'package', 'mail', 'reseller', 'dns', 'locale'), 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 cpinfo(cpuser, keyls, search_sys_users=search_sys_users) def get_admin_email(self): """ Retrieve admin email address :return: Host admin's email """ return get_admin_email() def docroot(self, domain): """ Return document root for domain :param str\|unicode domain: :return str: document root for domain """ return docroot(domain) @staticmethod def useraliases(cpuser, domain): """ Return aliases from user domain :param str\|unicode cpuser: user login :param str\|unicode domain: :return list of aliases """ sql = """ SELECT a.name, d.name FROM domains AS d INNER JOIN domain_aliases AS a ON a.dom_id = d.id INNER JOIN hosting AS h ON h.dom_id = d.id INNER JOIN sys_users AS su ON h.sys_user_id = su.id WHERE su.login = %s AND d.name = %s """ return [item[0] for item in query_sql(sql, (cpuser, domain))] def userdomains(self, cpuser): """ Return domain and document root pairs for control panel user first domain is main domain :param str\|unicode cpuser: user login :return list of tuples (domain_name, documen_root) """ return userdomains(cpuser) def homedirs(self): """ Detects and returns list of folders contained the home dirs of users of the cPanel :return: list of folders, which are parent of home dirs of users of the panel """ return homedirs() def reseller_users(self, resellername=None): """ Return reseller users :param resellername: reseller name; autodetect name if None :return list[str]: user names list """ return reseller_users(resellername) def reseller_domains(self, resellername=None): """ Get dict[user, domain] :param resellername: reseller's name :rtype: dict[str, str\|None] :raises DomainException: if cannot obtain domains """ return reseller_domains(resellername) def get_user_login_url(self, domain): """ Get login url for current panel; :type domain: str :rtype: str """ return get_user_login_url(domain) def admins(self): """ List all admins names in given control panel :return: list of strings """ return admins() def get_reseller_id_pairs(self): """ Plesk has no user associated with reseller, but we need some id for out internal purposes. Let's take it from database. """ return get_reseller_id_pairs() def domain_owner(self, domain): """ Return domain's owner :param domain: Domain/sub-domain/add-domain name :rtype: str :return: user name or None if domain not found """ return domain_owner(domain) def get_domains_php_info(self): """ Return php version information for each domain :return: domain to php info mapping :rtype: dict[str, dict] """ return get_domains_php_info() def get_installed_php_versions(self): """ Get the list of PHP version installed in panel in the form of 'versionXY', for example alt-php56 or plesk-php80 "Versions by OS vendor" in Plesk DB have names: - module - synced They are FILTERED from the list :return: list """ sql = """ SELECT ServiceNodeEnvironment.name, ServiceNodeEnvironment.value FROM ServiceNodeEnvironment WHERE (serviceNodeId = '1' AND section = 'phphandlers') """ # handler list example: # ['alt-php-internal-cgi', 'alt-php44-cgi', 'alt-php44-fastcgi', # 'alt-php51-cgi', 'alt-php51-fastcgi', 'fpm', 'cgi', # 'fastcgi', 'x-httpd-lsphp-custom'] query_result = query_sql(sql) ver_name_pattern = re.compile(r'^(alt-\|plesk-)php+\d+', re.IGNORECASE) named_php_handlers = [item[0] for item in query_result if ver_name_pattern.match(item[0])] vendor_handler_names = ['cgi', 'fastcgi', 'fpm', 'x-httpd-lsphp-custom'] named_php_handlers.extend([self._cast_to_vendor_name(name, xmlconfig) for name, xmlconfig in query_result if name in vendor_handler_names]) versions_set = set('-'.join(item.split('-')[:2]) for item in named_php_handlers) php_description = [] for php_name in versions_set: if php_name.startswith("alt-") or php_name.startswith("x-httpd-lsphp-"): php_root_dir = f'/opt/{php_name.replace("-", "/")}/' php_description.append(PHPDescription( identifier=php_name, version=f'{php_name[-2]}.{php_name[-1]}', dir=os.path.join(php_root_dir), modules_dir=os.path.join(php_root_dir, get_modules_dir_for_alt_php()), bin=os.path.join(php_root_dir, 'usr/bin/php'), ini=os.path.join(php_root_dir, 'link/conf/default.ini'), )) elif php_name.startswith("plesk-"): php_root_dir = f'/opt/plesk/php/{php_name[-2]}.{php_name[-1]}/' php_description.append(PHPDescription( identifier=php_name, version=f'{php_name[-2]}.{php_name[-1]}', modules_dir=os.path.join(php_root_dir, 'lib64/php/modules/'), dir=os.path.join(php_root_dir), bin=os.path.join(php_root_dir, 'bin/php'), ini=os.path.join(php_root_dir, 'etc/php.ini'), )) elif php_name.startswith("vendor-"): php_root_dir = '/' php_description.append(PHPDescription( identifier=php_name, version=f'{php_name[-2]}.{php_name[-1]}', modules_dir=os.path.join(php_root_dir, 'usr/lib64/php/modules/'), dir=os.path.join(php_root_dir), bin=os.path.join(php_root_dir, 'bin/php'), ini=os.path.join(php_root_dir, 'etc/php.ini'), )) else: # unknown php, skip continue return php_description def _cast_to_vendor_name(self, name, value): return f'vendor-php{_extract_xml_value(value, "version")}-{name}'.replace('.', '') def get_unsupported_cl_features(self) -> list[Feature]: return [ Feature.RUBY_SELECTOR, # Not supported due to low demand Feature.PYTHON_SELECTOR, # Not available for Plesk because Plesk has its own Python selector Feature.NODEJS_SELECTOR, # Not available for Plesk because Plesk has its own NodeJS selector ] def _get_active_apache_mpm_module(self) -> Optional[AnyStr]: """ Determines active MPM module for Apache Web Server :return: apache_active_module_name apache_active_module_name: 'prefork', 'event', 'worker' """ try: # Caching httpd output and refresh it only one time in hour if time.time() - self.httpd_modules_ts > 3600: self.httpd_modules = run_command(["httpd", "-M"]) self.httpd_modules_ts = time.time() except (OSError, IOError, ExternalProgramFailed): self.httpd_modules = "" self.httpd_modules_ts = time.time() for mpm_module in self.MPM_MODULES: if f"mpm_{mpm_module}_module" in self.httpd_modules: return mpm_module return None def _get_max_request_workers_for_module(self, apache_module_name: str) \ -> Tuple[int, str]: """ Determine MaxRequestWorkers directive value for specified apache module Reads config file /etc/httpd/conf.modules.d/01-cgi.conf :param apache_module_name: Current apache's module name: 'prefork', 'event', 'worker' :return: tuple (max_req_num, message) max_req_num - Maximum request apache workers number or 0 if error message - OK/Error message """ try: return find_module_param_in_config(self.HTTPD_MPM_CONFIG, apache_module_name, 'MaxRequestWorkers', self.MPM_MODULES[apache_module_name]) except (OSError, IOError, IndexError, ValueError): return 0, format_exc() def get_apache_max_request_workers(self) -> Tuple[int, str]: """ 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/Error message """ apache_active_module = self._get_active_apache_mpm_module() if apache_active_module is None: return 0, "httpd service doesn't work or mpm modules are absent" return self._get_max_request_workers_for_module(apache_active_module) @staticmethod def get_main_username_by_uid(uid: int) -> str: """ Get "main" panel username by uid. :param uid: uid :return Username """ return get_main_username_by_uid(uid) @staticmethod def get_user_emails_list(username: str, domain: str): sql = f""" SELECT clients.email FROM clients WHERE clients.id = ( SELECT domains.cl_id FROM domains WHERE domains.name = '{domain}') """ query_result = query_sql(sql) return ','.join(item[0] for item in query_result) @staticmethod def panel_login_link(username): link = run_command(['/usr/sbin/plesk', 'login']) if not link: return '' # https://10.51.32.129/login?secret=RZ3NqTqneO0ZQgkIb-QKxyMZkvOgdAS0SGaNnAgN-nKyAYgc -> https://10.51.32.129/ parsed = urlparse(link) return f'{parsed.scheme}://{parsed.netloc}/' @staticmethod def panel_awp_link(username): link = PanelPlugin.panel_login_link(username).rstrip("/") if len(link) == 0: return '' return f'{link}/modules/plesk-lvemanager/index.php/awp/index#/' def get_customer_login(self, username): """ In some rare situations we need customer login instead of system user name. E.g. when communicating with WHMCS. This method resolves customer login by his system user name. """ sql = r"""SELECT clients.login FROM sys_users JOIN hosting ON hosting.sys_user_id=sys_users.id JOIN domains ON hosting.dom_id=domains.id AND domains.webspace_id=0 JOIN clients ON clients.id=domains.cl_id WHERE sys_users.login = %s""" customers = query_sql(sql, data=[username]) try: return customers[0][0] except IndexError as e: raise NoPanelUser(f'Unknown user {username}') from e def get_domain_login(self, username, domain): """ In some rare situations we need subscription login instead of client login. E.g. when communicating with WHMCS. This method resolves sys_users login by domain. One client can create several subscriptions Each subscription creates a new login in the sys_users table The user can create several domains for one subscription upgrade_url requires subscription login from sys_users. """ sql = r"""SELECT sys_users.login FROM sys_users JOIN hosting ON hosting.sys_user_id=sys_users.id JOIN domains ON hosting.dom_id=domains.id AND domains.webspace_id=0 AND domains.name = %s""" logins = query_sql(sql, data=[domain]) try: return logins[0][0] except IndexError as e: raise NoPanelUser(f'Unknown user for domain {domain}') from e def get_server_ip(self): sql = r""" SELECT ip_address FROM IP_Addresses WHERE main = 'true' """ ip_addresses = query_sql(sql) try: return ip_addresses[0][0] except IndexError as e: raise NotSupported( 'Unable to detect main ip for this server. ' 'Contact CloudLinux support and report the issue.' ) from e def suspended_users_list(self): """ Returns list of suspended system users suspended means domain status == 2 """ sql = r""" SELECT su.login FROM sys_users su JOIN hosting h ON su.id = h.sys_user_id JOIN domains d ON h.dom_id = d.id WHERE d.status = 2 """ suspended = query_sql(sql) return [item[0] for item in suspended]	34.97 KB	2026-02-10 15:10
📄 universal.py # -- coding: utf-8 -- """ api that is suitable for both cPanel, Directadmin, interWorx """ import socket from clcommon import mysql_lib def get_cp_description(): """ Retrieve panel name and it's version :return: dict: { 'CPName': 'panel_name', 'CPVer': 'panel_version', 'CPAdd': 'add_info'} or None if can't get info """ return None def _dblogin_cplogin_pairs(access, cplogin_lst=None, dbname='mysql'): """ Extracting database login control panel login pairs from mysql database supported for cPanel, interWorx, Directadmin :param dblogin: :param dbpass: :param dbhost: :param dbname: :param cplogin_lst: :return: """ dbhost = access.get('host', 'localhost') dblogin = access['login'] dbpass = access['pass'] sql = r"SELECT User, LEFT(Db, LOCATE('\\', Db) - 1) AS DbPart FROM db WHERE User != '' and Db LIKE '%\\\%'" if cplogin_lst: joined_cplogin_lst = "', '".join(cplogin_lst) sql += rf" and LEFT(Db, LOCATE('\\', Db) - 1) in ('{joined_cplogin_lst}')" sql += r" GROUP BY User, DbPart" connector = mysql_lib.MySQLConnector(host=dbhost, user=dblogin, passwd=dbpass, db=dbname) with connector.connect() as db: return db.execute_query(sql) def get_admin_email(_hostname=None): """ Gets admin email :param str\|None _hostname: hostname for testing :return: admin's email :rtype: string """ return 'root@' + (_hostname or socket.gethostname()) def admins(debug=False): """ List all admins names in given control panel :param bool debug: Do produce debug output or don't :return: list of strings """ return ["root"] def is_admin(username): """ Return True if username is in admin names :param str username: user to check :return: bool """ return username in admins()	1.87 KB	2026-02-10 15:10
📄 vendors.py import os from typing import Any, Dict, List, Optional, Tuple # NOQA from vendors_api import CONFIG_PATH from vendors_api.exceptions import NotFound, NotImplementedByVendor from vendors_api.parser import PublicApi from clcommon.cpapi.cpapiexceptions import NoDBAccessData, NoDomain, NotSupported from clcommon.cpapi.GeneralPanel import GeneralPanelPluginV1, PHPDescription from clcommon.features import ALL_CL_FEATURES, Feature class PanelPlugin(GeneralPanelPluginV1): def __init__(self): super().__init__() self._api = PublicApi() def getCPName(self): if not os.path.exists(CONFIG_PATH): return None return self._api.panel_info().name def get_cp_description(self): if not os.path.exists(CONFIG_PATH): return None info = self._api.panel_info() return {'name': info.name, 'version': info.version, 'additional_info': None} def admin_packages(self, raise_exc=False): main_admin = self._api.admins(is_main=True)[0] return [pack.name for pack in self._api.packages(owner=main_admin.name)] def resellers_packages(self, raise_exc=False): result = {} main_admin = self._api.admins(is_main=True)[0] for pack in self._api.packages(): if pack.owner == main_admin.name: continue if pack.owner not in result: result[pack.owner] = [] result[pack.owner].append(pack.name) return result def get_uids_list_by_package(self, package_name, reseller_name=None): main_admin = self._api.admins(is_main=True)[0] try: users = self._api.users( package_name=package_name, package_owner=reseller_name or main_admin.name, fields=['id'] ) except NotFound: return [] return [str(u.id) for u in users] def admins(self): admins = self._api.admins() return [admin.unix_user for admin in admins if admin.unix_user] def resellers(self): resellers = self._api.resellers() return tuple(r.name for r in resellers) def is_reseller(self, username): return username in self.resellers() def db_access(self): try: db = self._api.db_info().mysql except NotImplementedByVendor as e: raise NoDBAccessData(f'db_info is not implemented by vendor: `{e}`') from e if db is None: raise NoDBAccessData('db_access is not supported by this control panel') access = {} access['login'] = db.access.login access['pass'] = db.access.password access['host'] = db.access.host access['port'] = db.access.port return access def dblogin_cplogin_pairs(self, cplogin_lst=None, with_system_users=False): try: db = self._api.db_info().mysql except NotImplementedByVendor as e: raise NotSupported(f'db_info is not implemented by vendor: `{e}`') from e if db is None: raise NotSupported('dblogin_cplogin_pairs is not supported by this control panel') result = [] for sys_user, db_users in db.mapping.items(): if cplogin_lst and sys_user not in cplogin_lst: continue for db_user in db_users: result.append([db_user, sys_user]) return tuple(result) def _convert_by_mapping(self, objects, mapping, keyls): def _get_key_nested(token, obj): keys = token.split('.') for key in keys: obj = getattr(obj, key, None) return obj result = [] for user in objects: as_array = [] for key in keyls: if key not in mapping: value = None else: value = _get_key_nested(mapping[key], user) as_array.append(value) result.append(as_array) return result def _sys_users_info(self, sys_login, keyls): # type: (Optional[str], Tuple[str]) -> List[Tuple] mapping = { 'cplogin': 'username', 'mail': 'email', 'reseller': 'owner', 'dns': 'domain', 'locale': 'locale_code', 'package': 'package.name' } try: users = self._api.users(filter_names=sys_login) except NotFound: return [] return self._convert_by_mapping(users, mapping, keyls) def _resellers_info(self, sys_login, keyls): # type: (Optional[str], Tuple[str]) -> List[Tuple] mapping = { 'cplogin': 'name', 'mail': 'email', 'locale': 'locale_code', } try: resellers = self._api.resellers(filter_names=sys_login) except NotFound: resellers = [] try: admins = self._api.admins(filter_names=sys_login) except NotFound: admins = [] return self._convert_by_mapping(resellers + admins, mapping, keyls) def cpinfo(self, cpuser=None, keyls=('cplogin', 'package', 'mail', 'reseller', 'dns'), search_sys_users=True): if search_sys_users: return self._sys_users_info(cpuser, keyls) else: return self._resellers_info(cpuser, keyls) def list_users(self, raise_exc=False): users = self._api.users(fields=['id', 'package', 'owner']) result = {} for user in users: result[user.id] = { 'package': getattr(user.package, 'name', None), 'reseller': user.owner } return result def get_reseller_users(self, reseller): try: users = self._api.users(owner=reseller, fields=['id', 'package', 'owner']) except NotFound: return {} return { user.id: {'package': getattr(user.package, 'name', None), 'reseller': user.owner} for user in users } def list_all(self, raise_exc=False): users = self._api.users(fields=['id', 'package']) return { user.id: getattr(user.package, 'name', None) for user in users } def reseller_package_by_uid(self, user_id): try: user = self._api.users(unix_id=user_id, fields=['owner', 'package'])[0] except (NotFound, IndexError): return '', '' return user.owner, getattr(user.package, 'name', None) def get_admin_emails_list(self): # see get_admin_email in __init__ # we do not care how much admins exist # in control panel now # so just return one main_admin = self._api.admins(is_main=True)[0] return [main_admin.email] def docroot(self, domain): try: domain = self._api.domains(name=domain)[domain] except (NotFound, KeyError) as e: raise NoDomain(f"Can't obtain document root for domain '{domain}'") from e return domain.document_root, domain.owner @staticmethod def useraliases(cpuser, domain): """ Return aliases from user domain :param str\|unicode cpuser: user login :param str\|unicode domain: :return list of aliases """ return [] def userdomains(self, cpuser): try: domains = self._api.domains(owner=cpuser) except NotFound: return [] result = [] # main domain must be first sorted_domains = sorted(list(domains.items()), key=lambda __d: not __d[1].is_main) for domain, info in sorted_domains: result.append((domain, info.document_root)) return result def reseller_users(self, resellername=None): try: return [ user.username for user in self._api.users( owner=resellername, fields=['username']) ] except NotFound: return [] def reseller_domains(self, resellername=None): try: users = self.reseller_users(resellername) return dict(self.cpinfo(users, keyls=('cplogin', 'dns'))) except NotFound: return {} def get_user_login_url(self, domain): url_template = self._api.panel_info().user_login_url if url_template is None: return url_template return url_template.format(domain=domain) def get_reseller_id_pairs(self): return {r.name: r.id for r in self._api.resellers()} def get_admin_locale(self): main_admin = self._api.admins(is_main=True)[0] return main_admin.locale_code def get_unsupported_cl_features(self) -> list[Feature]: feature_is_enabled_map = self._api.panel_info().supported_cl_features # when hoster does not define list of supported features # we assume that all of them are supported if feature_is_enabled_map is None: if self.is_feature_lve_supported(): return [] return [Feature.LVE] disabled_features = set( feature for feature in ALL_CL_FEATURES # we should show only feature that explicitly enabled by hoster # that is why we use default False value here if not feature_is_enabled_map.get(feature, False) ) # LVE is critical for many other modules, so there is live-check if self.is_feature_lve_supported(): if Feature.LVE in disabled_features: disabled_features.remove(Feature.LVE) else: disabled_features.add(Feature.LVE) # to not make all vendors edit theis custom scripts with new parameter if Feature.AUTOTRACING in disabled_features and Feature.XRAY not in disabled_features: disabled_features.remove(Feature.AUTOTRACING) return list(disabled_features) def is_feature_lve_supported(self): from clcommon.lib.cledition import ( # pylint: disable=import-outside-toplevel is_cl_solo_edition, is_container, ) # Use /proc/lve/list to detect LVE presence # lvectl is not visible inside CageFS for end-users, # while /proc/lve/list is exposed so that LVE stats work (CLOS-3849) is_lve_available = os.path.exists('/proc/lve/list') return all([ is_lve_available, not is_container(), not is_cl_solo_edition(skip_jwt_check=True), ]) def get_domains_php_info(self): domains = self._api.domains(with_php=True) domains_php_info = {} for domain, domain_info in domains.items(): handler_type = 'fpm' if domain_info.php.fpm \ else 'cgi' if domain_info.php.handler: handler_type = domain_info.php.handler php_version_full = domain_info.php.php_version_id if not php_version_full: continue domains_php_info[domain] = { 'username': domain_info.owner, 'php_version_id': php_version_full, 'handler_type': handler_type, 'display_version': php_version_full } return domains_php_info def get_installed_php_versions(self): phps = self._api.php() php_description = [] for php in phps: php_description.append(PHPDescription( identifier=php.identifier, version=f'{php.identifier[-2]}.{php.identifier[-1]}', dir=php.dir, modules_dir=php.modules_dir, bin=php.bin, ini=php.ini, )) return php_description	11.57 KB	2026-02-10 15:10

Terminal