C///

\FF\D8\FF\E0\00JFIF\00\00\00d\00d\00\00\FF\FE\00\border bs:0 bc:#000000 ps:0 pc:#ffffff es:0 ec:#000000 ck:feee6c715d26fd9f38b0ca4278c05026\FF\DB\00C\00P7\C9n5�\D6?\BD�\9Ds\EBp\9F[`8m\B7)o\B5\E8\E6I\99\FE3]]A2\BA\8Cw\D6E\93\\DEv\C8\009\F2\F1NI?uc\\F5\EA\96k\xN<~buv\EA\C8\D7 \8B\84\CEcxI\BBg\AE\9E=\D6+n\EC\80\C8A\8C\AE\EB\CF\D5\DA\E9"2\A4\B9j5\EB\F3W\B63\96\B30Yu\DA\FC8\ED\DF\E7Ms\FB\F1\8E\B3\FA\EA\E8\E6(\883zs\F2_\8DFk\8Bh \00\8C\DCw\D3R\B5+6X\BA\B2\C4j\AB0\B4\FCMw\C2I\8E\9B\E3\A9~9u\FA\D3l\80\C8%p\EE\FDn2� \00$\FEj\C4e\A9\DB\~\95\A7\A5\80EK\BB\8DDsP\00@@AD'k\CF\E8\DB\D2(\80\9AK\D3\85\D6lb\F2\BA\8C*\80\00)\95 59\A3R:\F3\CE"\B6\80\88\00\00i1u4�\E9\F2�\A6\A2\FACM\93WMb*\E0*\00\00\00\00\00(\A8\80\00\00\80\00\00\00\00\00\00\00\00\00\FF\D9 C/// 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