byp4a55 professor6T9

Linux premium133.web-hosting.com 4.18.0-553.44.1.lve.el8.x86_64 #1 SMP Thu Mar 13 14:29:12 UTC 2025 x86_64
LiteSpeed
162.0.232.69

File Name : 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())

Name	Type	Size	Owner/Group	Permission	Last Modified	Actions
.	dir	-	root/root	0755	2026-03-03 09:04:35
..	dir	-	root/root	0755	2026-03-03 09:06:27
schemas	dir	-	root/root	0755	2026-03-03 09:04:35
Readme.md	text/x-script.python	4.13 KB	root/root	0644	2026-02-13 12:40:39
__init__.py	text/x-script.python	413 B	root/root	0644	2026-02-13 12:40:39
config.py	text/x-script.python	2.52 KB	root/root	0644	2026-02-13 12:40:39
exceptions.py	text/x-script.python	3.01 KB	root/root	0644	2026-02-13 12:40:39
models.py	text/x-script.python	7.38 KB	root/root	0644	2026-02-13 12:40:39
parser.py	text/x-script.python	13.48 KB	root/root	0644	2026-02-13 12:40:39

Team Anon Force

https://t.me/Professor6T9x