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-27 02:38
📄 Readme.md # List of projects where vendor's API is used - x-ray: @dkavchuk @skokhan - userland: @oshyshatskyi @rprilipskii, @alutsiuk PLEASE, ADD YOUR PROJECT AND CONTACT PERSON HERE SO WE CAN NOTIFY YOU ABOUT CHANGES # Introduction Vendor's api is a new proxy level between control panel data and CloudLinux utilities. The complete integration guide for vendors is located [here](https://docs.cloudlinux.com/control_panel_integration/#introduction) and must be kept up-to date with api implementation on our side. The original ControlPanel API (cpapi) is located at clcommon/cpapi was written years ago and gave ability for CloudLinux developers to write proxy scripts that translate control panel data in python objects. Unfortunately, cpapi is not easy to extend by third-party devs: they must know python well and also update scripts when we change internal python version in cllib. As a result of long discussion, Vendors API layer was added. Unlike cpapi, it uses executable scripts as data source. E.g. vendor writes `db_info` executable and we parse it's request to create python objects: /scripts/db_info { "data": { "mysql": { "access": { "login": "root", "password": "Cphxo6z", "host": "localhost", "port": 3306 }, "mapping": { } } }, "metadata": { "result": "ok" } } The complete list of scripts is located [here](https://docs.cloudlinux.com/control_panel_integration/#the-list-of-the-integration-scripts). ## Preparation Unfortunately we don't have already working bundled scripts that you can install in your testing environment, but we can well-detailed response examples in [docs](https://docs.cloudlinux.com/control_panel_integration/#the-list-of-the-integration-scripts) which you can copy-paste to simple bash script and use them to check your code locally. We would really appeciate if someone will create testing package with some mocked integration scripts. ## Vendors API usage in CloudLinux code Vendors API may be used directly using simple import: from vendors_api import PublicApi api = PublicApi() api.db_info() -> Databases object \ in projects where original cpapi is used we added proxy layer that transforms vendor's api objects into cpapi format (see cpapi/plugins/vendors.py). That is temporary decision while we don't move our code completely to the new api. ONLY THINGS DECLARED IN vendors_api.\_\_init__ ARE AVAILABLE FOR PUBLIC USAGE ## Before modifying vendors API Before adding new data or fields, you must deside whether your changes will break backward compatibility. It's always recommended to leave some fallbacks in our code in order not to force vendors to update their scripts. In case if you add some major changes or even break compatibility, you MUST bump api version in spec file: Provides: public_cp_vendors_api = 1.3 That gives ability for vendor's to conflict new version while their scripts are not ready. ## Adding new fields and scripts Vendors api is highly covered with json schemas checks and python typing to make it catch all possible vendor errors on the early stage. Let's imagine that you are trying to add new possible field to `db_info` script. In order to do that, you must modify python representation first. Go to the `parser.py` and check what data type is returned. In our case it is `Databases` object: def db_info(self): # type: () -> Databases Find the model in `models.py`, modify it's \_\_slots\_\_ to keep new field. Next, modify `__init__` to process new fields and save them to model. That's all for python part. Next, you must modify script jsonschema that checks responses validity. It's located in `schemas/db_info.yaml` for our case: ```text --- type: object properties: mysql: type: - object - "null" properties: ... ``` Don't forget to cover your logic with unittests: - test_vendors_api.py - test_vendors_api_cpapi.py - test_vendors_api_models.py - test_vendors_api_validation.py # Review and merging Please ask @oshyshatskyi, @avolkov or @alutsik about reviewing and merging your change.	4.13 KB	2026-02-10 15:10
📄 __init__.py # coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # from vendors_api.config import integration_scripts, CONFIG_PATH from vendors_api.parser import PublicApi # add other public things here __all__ = ( 'integration_scripts', 'CONFIG_PATH', 'PublicApi' )	413 B	2026-02-10 15:10
📁 __pycache__	-	2026-03-19 15:49
📄 config.py # coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # """ Wrapper over integration config which provides typing. """ import os import shlex from configparser import ConfigParser from collections import namedtuple from vendors_api.exceptions import NotImplementedByVendor CONFIG_PATH = '/opt/cpvendor/etc/integration.ini' integration_scripts_cache = None config_last_modify_time = None class ScriptsCMD(namedtuple('VendorScriptsPath', [ 'panel_info', 'db_info', 'packages', 'users', 'domains', 'resellers', 'admins', 'php', ])): """ Wrapper over namedtuple that raises exception when we access property that has None value """ def __getattribute__(self, item): """ Raise error if command is not implemented, but we try to get it in code """ obj = super().__getattribute__(item) if obj is None: raise NotImplementedByVendor(item) return obj def _split_path_or_none(cmd): """ Taking command line as input, convert it into subprocess args If cmd is None -> return None """ if cmd is None: return None return tuple(shlex.split(cmd)) def integration_scripts(): global config_last_modify_time global integration_scripts_cache modify_time = os.path.getmtime(CONFIG_PATH) if integration_scripts_cache is None or \ config_last_modify_time != modify_time: config = _read_config_file() scripts = config['integration_scripts'] integration_scripts_cache = ScriptsCMD( panel_info=_split_path_or_none(scripts.get('panel_info')), db_info=_split_path_or_none(scripts.get('db_info')), packages=_split_path_or_none(scripts.get('packages')), users=_split_path_or_none(scripts.get('users')), domains=_split_path_or_none(scripts.get('domains')), resellers=_split_path_or_none(scripts.get('resellers')), admins=_split_path_or_none(scripts.get('admins')), php=_split_path_or_none(scripts.get('php')), ) config_last_modify_time = modify_time return integration_scripts_cache def _read_config_file(): """ Parse config file and return result as dictionary. """ parser = ConfigParser(interpolation=None, strict=False) parser.read(CONFIG_PATH) return {s: dict(parser.items(s)) for s in parser.sections()}	2.52 KB	2026-02-10 15:10
📄 exceptions.py # coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # """ Module contains various exceptions raised by this api. """ from clcommon import FormattedException class ErrorMessage: """ Documented constants that can be received from vendors scripts """ INTERNAL_ERROR = "InternalAPIError" PERMISSION_DENIED = "PermissionDenied" BAD_REQUEST = "BadRequest" NOT_FOUND = "NotFound" class VendorScriptError(FormattedException): """ Any errors related to vendors scripts. """ def __init__(self, message, **context): super().__init__({ 'message': message, 'context': context }) class UnexpectedResult(VendorScriptError): """ An error that is raised in case of any unexpected results returned by vendors script. """ pass class MalformedError(VendorScriptError): """ An error that is raised in case of malformed output that we cannot parse """ def __init__(self, output): super().__init__( 'vendor api returned invalid response that ' 'cannot be parsed with json, output: %(output)s', output=output ) class BadScriptError(VendorScriptError): """ An error that is raised when we cannot run vendor's script for some reason. """ pass class NotImplementedByVendor(VendorScriptError): """ Raised when we try to start script that is not implemented by vendor """ def __init__(self, command): super().__init__( "Command `%(command)s` is not implemented " "in integration scripts. Please, implement it and try again.", command=command ) class _VendorApiFailed(VendorScriptError): """ Bunch of api errors returned by vendors scripts """ def __init__(self, message): super().__init__(message=message) class InternalError(_VendorApiFailed): """ n case when data is temporarily unavailable due to internal errors in the integration script (database unavailable or something like that) """ pass class PermissionDenied(_VendorApiFailed): """ In case when data is unavailable due to restricted access of a user that called the script """ pass class BadRequest(_VendorApiFailed): """ In case when known to be false arguments are passed to the utility. """ pass class NotFound(_VendorApiFailed): """ In case when during data filtering the target entity doesn't exist in the control panel """ pass class VendorApiMalformedData(VendorScriptError): """ Raised when vendor script returns bad data. """ def __init__(self, error): super().__init__( "Integration script returned invalid response " "which is not accepted by internal schema check. See error below.\n" "%(error)s", error=error )	3.01 KB	2026-02-10 15:10
📄 models.py # coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # from typing import NamedTuple, Dict, List, Optional # NOQA class _Base: __slots__ = ('_received_fields', '_api') def __init__(self, opts): """ Initializes class by given dictionary and api object. :type opts: dict """ # required for subclasses to know # which fields we really received from # API and which are just None # e.g. case when we asked vendor's script to # return username & package and tried to access # 'id' somewhere (see __getattribute__ implementation) self._received_fields = set(opts.keys()) def __getattribute__(self, item): """ When parsing data in __init__, we save list of received fields. When accesing any of those fields, we must check that is was received first. This is needed for dynamic instances of "user" which can be different depending on 'fields' argument passed to the integration script. :type item: str :return: object or raise exception """ if item == '__slots__' or item not in self.__slots__ or item in self._received_fields: return object.__getattribute__(self, item) raise AttributeError(f'{item} is not set, but used in code') def __repr__(self): class_name = self.__class__.__name__ fields_dict = {k: getattr(self, k) for k in self._received_fields} return f'{class_name} ({fields_dict})' def __eq__(self, other): if self.__slots__ != other.__slots__: return False # models with different scopes cannot be equal if self._received_fields != other._received_fields: return False # loop and check all values for slot in set(self.__slots__) & self._received_fields: if getattr(self, slot) != getattr(other, slot): return False return True def __ne__(self, other): return not self == other class PanelInfo(_Base): __slots__ = ( 'name', 'version', 'user_login_url', 'supported_cl_features' ) _DEPRECATED_FEATURE_UPGRADES = { 'wpos': 'accelerate_wp' } def __init__(self, opts): # optional field default value opts.setdefault('supported_cl_features', None) self.name: str = opts['name'] self.version: str = opts['version'] self.user_login_url: str = opts['user_login_url'] self.supported_cl_features: Optional[Dict[str, bool]] = \ self._upgrade_feature_names(opts['supported_cl_features']) super().__init__(opts) def _upgrade_feature_names(self, features: Dict[str, bool]): """ Automatically convert old feature names into new ones that we defined in this class. e.g. feature 'wpos' is now called 'accelerate_wp' """ if features is None: return features new_features = features.copy() for old_name, new_name in self._DEPRECATED_FEATURE_UPGRADES.items(): if old_name not in new_features: continue new_features[new_name] = new_features[old_name] del new_features[old_name] return new_features DbAccess = NamedTuple('DbAccess', [ ('login', str), ('password', str), ('host', str), ('port', str), ]) DbInfo = NamedTuple('DBInfo', [ ('access', DbAccess), ('mapping', Dict[str, List[str]]) ]) class Databases(_Base): __slots__ = ( 'mysql', ) def __init__(self, opts): self.mysql = None # type: Optional[DbInfo] mysql_raw = opts.get('mysql') if mysql_raw is not None: access = mysql_raw['access'] self.mysql = DbInfo( access=DbAccess(access), mapping=mysql_raw['mapping'] ) super().__init__(opts) class Package(_Base): __slots__ = ( 'name', 'owner', ) def __init__(self, opts): self.owner = opts['owner'] # type: str self.name = opts['name'] # type: str super().__init__(opts) class User(_Base): __slots__ = ( 'id', 'username', 'owner', 'domain', 'package', 'email', 'locale_code', ) def __init__(self, opts): self.id = opts.get('id') # type: int self.username = opts.get('username') # type: str self.owner = opts.get('owner') # type: str if opts.get('package'): self.package = Package(opts.get('package')) # type: Package else: self.package = None self.email = opts.get('email') # type: str self.domain = opts.get('domain') # type: str self.locale_code = opts.get('locale_code') # type: str super().__init__(opts) class InstalledPHP(_Base): __slots__ = ( 'identifier', 'version', 'modules_dir', 'dir', 'bin', 'ini' ) def __init__(self, opts): self.identifier = opts.get('identifier') # type: str self.version = opts.get('version') # type: str self.modules_dir = opts.get('modules_dir') # type: str self.dir = opts.get('dir') # type: str self.bin = opts.get('bin') # type: str self.ini = opts.get('ini') # type: str super().__init__(opts) class PHPConf(NamedTuple): """ An object representing structure of input PHP configuration for a domain """ version: str # PHP version XY ini_path: str # path to directory with additional ini files is_native: bool = False # is PHP version set in native.conf fpm: Optional[str] = None # FPM service name handler: Optional[str] = None # current handler name php_version_id: Optional[str] = None # php version identifier class DomainData(_Base): __slots__ = [ 'owner', 'document_root', 'is_main', 'php', ] def __init__(self, opts): # optional field default value opts.setdefault('php', None) self.owner = opts['owner'] # type: str self.document_root = opts['document_root'] # type: str self.is_main = opts['is_main'] # type: bool self.php = None # type: Optional[PHPConf] php_conf = opts['php'] if php_conf is not None: self.php = PHPConf(php_conf) super().__init__(opts) class Reseller(_Base): __slots__ = [ 'id', 'name', 'locale_code', 'email', ] def __init__(self, opts): self.id = opts['id'] # type: str self.name = opts['name'] # type: str self.locale_code = opts['locale_code'] # type: str self.email = opts['email'] # type: Optional[str] super().__init__(opts) class Admin(_Base): __slots__ = [ 'name', 'unix_user', 'locale_code', 'email', 'is_main', ] def __init__(self, opts): self.name = opts['name'] # type: str self.unix_user = opts['unix_user'] # type: Optional[str] self.locale_code = opts['locale_code'] # type: str self.email = opts['email'] # type: Optional[str] self.is_main = opts['is_main'] # type: bool super().__init__(opts)	7.38 KB	2026-02-10 15:10
📄 parser.py # coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # """ 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())	13.48 KB	2026-02-10 15:10
📁 schemas	-	2026-03-19 15:49

Terminal