# coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # """ Main API module that provides class uses as proxy to public vendors methods and converts data to internal objects. """ import os import subprocess from itertools import compress import json import jsonschema import yaml from jsonschema import ValidationError from typing import Optional, List, Dict, Any # NOQA from clcommon.features import ALL_CL_FEATURES from vendors_api.config import integration_scripts from vendors_api.exceptions import ( BadScriptError, MalformedError, ErrorMessage, InternalError, PermissionDenied, BadRequest, NotFound, UnexpectedResult, VendorApiMalformedData ) from vendors_api.models import ( PanelInfo, Databases, Package, User, DomainData, Reseller, Admin, InstalledPHP, ) BASE_DIR = os.path.abspath(os.path.dirname(__file__)) _ERROR_MESSAGE_TO_ERROR = { ErrorMessage.INTERNAL_ERROR: InternalError, ErrorMessage.PERMISSION_DENIED: PermissionDenied, ErrorMessage.BAD_REQUEST: BadRequest, ErrorMessage.NOT_FOUND: NotFound } class PublicApi: """ Proxy to the public universal api for control panels """ @property def _scripts(self): """ Re-read this each time we call api due to lve-stats daemon. """ # already cached inside and refreshed when # integration config changes return integration_scripts() def _execute(self, command, schema_file): # type: (List[str], str) -> Any[List[Dict], Dict] # remove PYTHONPATH and set PYTHONNOUSERSITE to '1' for additional security env = os.environ.copy() env.pop('PYTHONPATH', None) env['PYTHONNOUSERSITE'] = '1' try: with subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, env=env) as p: output, _ = p.communicate() except OSError as e: raise BadScriptError( f"Something is wrong with integration script: `{e}`" ) from e return self._validate(output, schema_file) def _error_key_to_error(self, result, message): # type: (str, str) -> None """ Convert documented api error constants into python exceptions. :param result: error constant :param message: custom message that vendor returned """ error_cls = _ERROR_MESSAGE_TO_ERROR.get(result) if error_cls is None: raise UnexpectedResult( "Unexpected result key: `%(result)s`; message=`%(error)s`", result=result, error=message) raise error_cls(message) def _validate(self, raw_data, schema_path): # type: (str, str) -> Any[List, Dict] """ Checks input for valid json structure :param raw_data: json :return: dict or list """ try: # or {} in order to avoid None deserialized = json.loads(raw_data) or {} except (TypeError, ValueError) as e: raise MalformedError(raw_data) from e try: metadata = deserialized['metadata'] or {} result = metadata['result'] except KeyError as e: raise MalformedError(output=raw_data) from e if result == 'ok': try: schema = self._read_data_schema(schema_path) jsonschema.validate(deserialized['data'], schema) except (ValidationError, KeyError, TypeError) as e: # TypeError in case when something is None # KeyError when dict is not complete raise VendorApiMalformedData(str(e)) from e return deserialized['data'] else: self._error_key_to_error(result, message=metadata.get('message')) def _read_data_schema(self, filename): # type: (str) -> Dict[str, Any] """ Knowing the fact that we store all data schemas in one folder we can easily find their absolute path. We do not handle errors here as normally package contains all it's files. :param filename: schema filename to load :return: dict, jsonschema """ with open(os.path.join(BASE_DIR, 'schemas', filename), encoding='utf-8') as f: data_schema = yaml.load(f.read(), yaml.SafeLoader) return data_schema def panel_info(self): # type: () -> PanelInfo """ Returns the information about the control panel. Necessity: Always Accessed by: All UNIX users Must work inside CageFS also: Yes """ valid_data = self._execute(self._scripts.panel_info, schema_file='panel_info.yaml') result = PanelInfo(valid_data) if not result.supported_cl_features: return result for feature in result.supported_cl_features: if feature in ALL_CL_FEATURES: continue raise ValidationError( f"Feature {feature} is not available, " "please check your panel_info integration script. " f"Available keys are: {', '.join([f.value for f in ALL_CL_FEATURES])}" ) return result def db_info(self): # type: () -> Databases """ Returns the information about databases that are available to the control panel users and are managed by the control panel. Necessity: Only for LVE-Stats Accessed by: admins (UNIX users) Must work inside CageFS also: No """ valid_data = self._execute(self._scripts.db_info, schema_file='db_info.yaml') return Databases(valid_data) def packages(self, owner=None): # type: (Optional[str]) -> List[Package] """ Returns list of abstractions called "package" that represents a group of users that have the same default limits. Necessity: For limits functionality Accessed by: admins (UNIX users) Must work inside CageFS also: No """ args = self._scripts.packages if owner is not None: args += ('--owner', owner) valid_data = self._execute(args, schema_file='packages.yaml') return [Package(x) for x in valid_data] def users(self, owner=None, # type: Optional[str] package_name=None, # type: Optional[str] package_owner=None, # type: Optional[str] filter_names=None, # type: Optional[Any[str, List[str]]] unix_id=None, # type: Optional[int] fields=None # type: Optional[List[str]] ): # type: (...) -> List[User] """ Returns information about UNIX users, created and managed by the control panel. Necessity: Always Accessed by: admins (UNIX users) Must work inside CageFS also: No """ allowed_fields = {'id', 'username', 'owner', 'domain', 'package', 'email', 'locale_code'} if bool(package_name) ^ bool(package_owner): raise ValueError('You can only use package_name ' 'and package_owner in pair') if not isinstance(fields, (list, tuple, type(None))): raise ValueError('fields accept only list of strings') if fields is not None: # get bunch of requested fields that are not allowed by the api not_allowed_fields = set(fields) - allowed_fields if not_allowed_fields: raise ValueError(f"{not_allowed_fields} are not allowed fields") exclusive_groups = [ [owner], [package_name, package_owner], [filter_names], [unix_id], ] # take groups where all args are given used_args = list(compress(exclusive_groups, [all(i) for i in exclusive_groups])) if len(used_args) > 1: raise ValueError( f"You cannot use all these args in one request: {used_args}" ) command = self._scripts.users if owner is not None: command += ('--owner', owner) if package_name and package_owner: command += ('--package-name', package_name, '--package-owner', package_owner) if filter_names and isinstance(filter_names, str): command += ('--username', filter_names) if unix_id: command += ('--unix-id', str(unix_id)) if fields: command += ('--fields', ','.join(fields)) valid_data = self._execute(command, schema_file='users.yaml') result = [User(x) for x in valid_data] # special case when we request for many usernames at once # we do in our code because this can be quite hard for hoster if isinstance(filter_names, (list, tuple)): result = [u for u in result if u.username in filter_names] return result def domains(self, owner=None, name=None, with_php=False): # type: (Optional[str], Optional[str], bool) -> Dict[str, DomainData] """ Returns key-value object, where a key is a domain (or subdomain) and a value is DomainData object Necessity: Selectors, some UI features Accessed by: All UNIX users Must work inside CageFS also: Yes """ if owner and name: raise ValueError('you cannot use both owner and name') command = self._scripts.domains schema = 'domains.yaml' if owner: command += ('--owner', owner) if name: command += ('--name', name) if with_php: command += ('--with-php', ) schema = 'domains_with_php.yaml' valid_data = self._execute(command, schema_file=schema) result = {} for k, v in valid_data.items(): result[k] = DomainData(v) return result def resellers(self, id_=None, filter_names=None): # type: (Optional[int], Optional[Any[str], List[str]]) -> List[Reseller] """ Gives information about resellers who can be users owners in the control panel. Resellers do not obligatory have their own same-name UNIX accounts in the system and could exist only as an account in the control panel. Necessity: Always Accessed by: admins (UNIX users) Must work inside CageFS also: no :param int id_: int, reseller id :param filter_names: name or list of reseller names to return """ if id_ and filter_names: raise ValueError('You cannot use id and name at one call') command = self._scripts.resellers if id_ is not None: command += ('--id', str(id_)) if isinstance(filter_names, str): command += ('--name', filter_names) valid_data = self._execute(command, schema_file='resellers.yaml') result = [Reseller(x) for x in valid_data] # type: List[Reseller] if isinstance(filter_names, (list, tuple)): result = [u for u in result if u.name in filter_names] return result def admins(self, filter_names=None, is_main=None): # type: (Optional[Any[str, List[str]]], Optional[bool]) -> List[Admin] """ Gives information about panel’s administrators, output information about all panel’s administrators who: - could be (or actually are) the owners of the users, listed in users() - could be (or actually are) the owners of the packages, listed in packages() - has UNIX users with the rights to run LVE Manager UI Necessity: Always Accessed by: admins (UNIX users) Must work inside CageFS also: no :param filter_names: name or list of names to return :param is_main: filter output by type of admins: None means no filtering, return all False means only additional admins True means only main admin """ if filter_names and is_main is not None: raise ValueError('unable to use name and is_main at once') command = self._scripts.admins if isinstance(filter_names, str): command += ('--name', filter_names) if is_main is not None: command += ('--is-main', str(is_main).lower()) valid_data = self._execute(command, schema_file='admins.yaml') result = [Admin(x) for x in valid_data] # type: List[Admin] if isinstance(filter_names, (list, tuple)): result = [u for u in result if u.name in filter_names] return result def php(self) -> List[InstalledPHP]: """ Returns list of abstractions called "php" that represents an installed php with it's binary, ini file, modules directory, etc Necessity: For accelerate wp functionality Accessed by: admins (UNIX users) Must work inside CageFS also: No """ args = self._scripts.php valid_data = self._execute(args, schema_file='php.yaml') return [InstalledPHP(x) for x in valid_data] if __name__ == '__main__': # usage example api = PublicApi() print(api.panel_info()) print(api.users(unix_id=123)) print(api.admins()) print(api.resellers()) print(api.domains())