aDriv4 website!

Linux ircpanel.novinhost.org 4.18.0-553.34.1.lve.el8.x86_64 #1 SMP Thu Jan 9 16:30:32 UTC 2025 x86_64
Apache
213.232.124.78

File Name : user_plugin_utils.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/LICENSE.TXT

import grp
import json
import logging
import os
import pwd
import socket
import struct
from functools import wraps
from typing import Callable, Optional, Tuple, Any

from clcommon.cpapi import get_main_username_by_uid
from clcommon.lib.cledition import is_cl_solo_edition, is_cl_shared_pro_edition

from xray import gettext as _
from .constants import user_tasks_count, fpm_reload_timeout
from .exceptions import XRayError
from .fpm_utils import FPMReloadController
from .nginx_utils import NginxUserCache

# --------- GLOBALS ---------

logger = logging.getLogger('user_plugin_utils')
_format = '>I'

# --------- FUNCTIONS ---------

def pack_request(_input: Any) -> 'json str':
    """
    Pack input for sending
    """
    return json.dumps(_input)

def unpack_request(byte_command: bytes) -> Any:
    """
    Unpack incoming command
    """
    _command = byte_command.decode()
    logger.info('Command requested => %s', _command)
    return json.loads(_command)

def pack_response(msg: bytes) -> bytes:
    """
    Prefix message with a 4-byte length
    """
    logger.debug('Packing message of %i length', len(msg))
    return struct.pack(_format, len(msg)) + msg

def unpack_response(sock_object: 'socket object') -> bytes:
    """
    Read length-prefixed amount of data from socket
    """
    chunk = 4096
    msg = bytes()
    raw_msglen = sock_object.recv(4)
    if not raw_msglen:
        return sock_object.recv(chunk)
    msglen = struct.unpack(_format, raw_msglen)[0]

while len(msg) != msglen:
        msg += sock_object.recv(chunk)

return msg

def extract_creds(sock_object: 'socket object') -> Tuple[Any, Any, Any]:
    """
    Retrieve credentials from SO_PEERCRED option
    """
    _format = '3i'
    creds = sock_object.getsockopt(socket.SOL_SOCKET,
                                   socket.SO_PEERCRED,
                                   struct.calcsize(_format))
    _pid, _uid, _gid = struct.unpack(_format, creds)
    try:
        user, group = pwd.getpwuid(_uid).pw_name, grp.getgrgid(_gid).gr_name
    except KeyError:
        logger.info('Connected by proc %i of %i:%i',
        _pid, _uid, _gid)
    else:
        logger.info('Connected by proc %i of %i:%i (%s:%s)',
                    _pid, _uid, _gid,
                    user, group)
    return _pid, _uid, _gid

def check_for_root(_uid: int = None) -> bool:
    """
    Check for execution as root | command from root
    """
    if _uid is None:
        _uid = os.geteuid()
    return _uid == 0

def get_xray_exec_user() -> Optional[str]:
    """
    Retrieve the value of XRAYEXEC_UID env and resolve it to username
    """
    proxyuid = os.getenv('XRAYEXEC_UID')
    if proxyuid is not None:
        _proxyuser = get_main_username_by_uid(int(proxyuid))
        logger.info('Got XRAYEXEC_UID: %s (%s), working in USER_MODE',
                    proxyuid, _proxyuser)
        return _proxyuser

def sock_receive(sock_object: 'socket object') -> bytes:
    """
    Read all data from socket object
    """
    data = b''
    while True:
        chunk = sock_object.recv(1024)
        if not chunk:
            logger.debug('All data read, connection ended')
            break
        data += chunk
    return data

def error_response(msg: str) -> 'json str':
    """
    Construct an appropriate formatted response in case of error
    """
    return json.dumps({'result': msg}, ensure_ascii=False)

def nginx_user_cache() -> Optional[bool]:
    """
    Check nginx cache status for current user
    """
    proxyuser = get_xray_exec_user()
    if proxyuser is not None:
        return NginxUserCache(proxyuser).is_enabled

def root_execution_only_check() -> None:
    """
    Check if utility is executed as root and throw error in case if no
    """
    if not check_for_root():
        raise SystemExit(
            error_response(_('Only root is allowed to execute this utility')))

# --------- DECORATORS ---------

def user_mode_verification(func: Callable) -> Callable:
    """
    Decorator aimed to verify domain owner in X-Ray Manager user mode
    Applies to get_domain_info method
    """

def verify(data):
        """
        If exists, check XRAYEXEC_UID against domain owner
        """
        proxyuser = get_xray_exec_user()
        if proxyuser is not None and data.user != proxyuser:
            logger.warning('%s does not belong to user %s', data, proxyuser)
            raise XRayError(_('%s cannot be found') % str(data))

@wraps(func)
    def wrapper(*args, **kwargs):
        """
        Wraps func
        """
        info = func(*args, **kwargs)
        verify(info)
        return info

return wrapper

def user_mode_restricted(func: Callable) -> Callable:
    """
    Decorator aimed to check if user is not hitting limit of running tasks,
    set in X-Ray Manager user mode.
    Applies to start and continue methods.
    Limiting of user's running tasks is applied to Shared PRO only.
    """

def check(*args):
        """
        If XRAYEXEC_UID exists, check if user does not exceed
        limit of running tasks
        """
        # TODO: [unification] ensure is_cl_shared_pro_edition really needed here
        # https://cloudlinux.atlassian.net/browse/XRAY-244 - (seems yes)
        if not is_cl_shared_pro_edition(skip_jwt_check=True):
            return

proxyuser = get_xray_exec_user()
        if proxyuser is not None:
            ui_api_cli_instanse = args[0].ui_api_client
            resp = ui_api_cli_instanse.get_task_list()
            list_of_tasks = resp.get('result')
            if list_of_tasks is not None:
                running_count = len([item for item in list_of_tasks if
                                     item.get('status') == 'running'])
                if running_count >= user_tasks_count:
                    raise XRayError(
                        _('Limit of running tasks is {}. '
                          'You already have {} running task'.format(str(user_tasks_count),
                                                                    str(user_tasks_count))))

@wraps(func)
    def wrapper(*args, **kwargs):
        """
        Wraps func
        """
        check(*args)
        return func(*args, **kwargs)

return wrapper

def with_fpm_reload_restricted(func: Callable) -> Callable:
    """
    Decorator aimed to restrict frequent reloads of FPM service
    Applies to get_domain_info method
    """

def check(*args, data):
        """
        """
        # TODO: [unification] ensure is_cl_solo_edition really needed here
        # https://cloudlinux.atlassian.net/browse/XRAY-244 (seems yes)
        if is_cl_solo_edition(skip_jwt_check=True):
            return

proxyuser = get_xray_exec_user()
        if proxyuser is not None and data.panel_fpm:
            _fpm_service = args[0].fpm_service_name(data)
            if FPMReloadController(_fpm_service).restrict():
                raise XRayError(
                    _('The X-Ray User service is currently busy. Operation is temporarily not permitted. '
                      'Try again in %s minute') % str(fpm_reload_timeout),
                    flag='warning')

@wraps(func)
    def wrapper(*args, **kwargs):
        """
        Wraps func
        """
        info = func(*args, **kwargs)
        check(*args, data=info)
        return info

return wrapper

def username_verification(func: Callable) -> Callable:
    def validate(username: str):
        """
        If exists, check XRAYEXEC_UID against user passed param
        """
        proxyuser = get_xray_exec_user()
        if proxyuser is not None and username != proxyuser:
            raise XRayError(_('Incorrect user for request'))

@wraps(func)
    def wrapper(*args, **kwargs):
        response = func(*args, **kwargs)
        username = kwargs['username']
        validate(username)
        return response

return wrapper

def user_mode_advice_verification(func: Callable) -> Callable:
    """
    Decorator aimed to verify user in X-Ray Smart Advice user mode
    Applies to get_detailed_advice method, which takes part in
    advice_details and advice_apply methods
    """

def verify(data: dict) -> None:
        """
        If exists, check XRAYEXEC_UID against user in metadata of an advice
        """
        proxyuser = get_xray_exec_user()
        try:
            username = data['metadata']['username']
        except KeyError:
            raise XRayError(_('Requested advice cannot be verified'))
        if proxyuser is not None and username != proxyuser:
            raise XRayError(_('Requested advice does not exist'))

@wraps(func)
    def wrapper(*args, **kwargs):
        """
        Wraps func
        """
        advice_info, _ = func(*args, **kwargs)
        verify(advice_info)
        return advice_info, _

return wrapper

Name	Type	Size	Owner/Group	Permission	Last Modified	Actions
.	dir	-	root/root	0755	2025-11-12 08:23:13
..	dir	-	root/root	0755	2025-09-28 12:28:32
__pycache__	dir	-	root/root	0755	2025-09-28 09:03:02
__init__.py	file	550 B	root/root	0644	2025-07-28 04:37:34
advanced_metrics.py	file	1.42 KB	root/root	0644	2025-07-28 04:37:34
clwpos_safe_imports.py	file	2.21 KB	root/root	0644	2025-07-28 04:37:34
constants.py	file	2.15 KB	root/root	0644	2025-07-28 04:37:34
exceptions.py	file	3.69 KB	root/root	0644	2025-07-28 04:37:34
fpm_utils.py	file	2.25 KB	root/root	0644	2025-07-28 04:37:34
local_counters.py	file	3.37 KB	root/root	0644	2025-07-28 04:37:34
nginx_utils.py	file	7.11 KB	root/root	0644	2025-07-28 04:37:34
phpinfo_utils.py	file	6 KB	root/root	0644	2025-07-28 04:37:34
types.py	file	23.51 KB	root/root	0644	2025-07-28 04:37:34
user_limits.py	file	4.51 KB	root/root	0644	2025-07-28 04:37:34
user_manager.py	file	5.85 KB	root/root	0644	2025-07-28 04:37:34
user_plugin_utils.py	file	8.85 KB	root/root	0644	2025-07-28 04:37:34
utils.py	file	27.33 KB	root/root	0644	2025-07-28 04:37:34

For more Private tool come to me!

https://t.me/aDriv4vz