FileMgr

✏️ /opt/cloudlinux/venv/lib/python3.11/site-packages/xray/manager/base.py

✕ Tutup

# -*- 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

"""
This module contains classes implementing X-Ray Manager behaviour
and helper functions
"""
import json
import logging
import os
import pwd
import shutil
import signal
import subprocess
import typing
from datetime import datetime, timedelta
from glob import escape as glob_escape, glob
from typing import Iterator, Optional, Tuple

import psutil
from clcommon.const import Feature
from clcommon.cpapi import (
    is_panel_feature_supported,
    get_installed_php_versions
)
from xray.internal.clwpos_safe_imports import php_get_vhost_versions, is_wpos_supported
from clcommon.utils import is_litespeed_running
from clcommon.clpwd import drop_privileges
from secureio import disable_quota

from xray import gettext as _
from ..adviser.clwpos_get import ClWposGetter
from ..apiclient import get_client
from ..continuous.manager import ContinuousManager
from ..internal.exceptions import (
    XRayAPIError, XRayError,
    XRayMissingDomain, XRayManagerExit, XRayManagerExitPHPUnsupported
)
from ..internal.fpm_utils import FPMReloadController
from ..internal.local_counters import open_local_storage
from ..internal.nginx_utils import NginxUserCache
from ..internal.types import DomainInfo, Task, url_split
from ..internal.user_manager import ManageUserAgent
from ..internal.advanced_metrics import AdvancedMetrics
from ..internal.user_plugin_utils import (
    get_xray_exec_user,
    user_mode_restricted,
    nginx_user_cache
)
from ..internal.utils import (
    no_active_tasks,
    switch_schedstats,
    ssa_disabled,
    _cagefsctl_remount,
    set_privileges,
    is_file_recently_modified
)
from ..reconfiguration.global_ini import (
    create_ini_files,
    remove_ini_files,
    is_global_ini_mode,
    remove_global_ini_mode_marker,
    create_global_ini_mode_marker
)

if typing.TYPE_CHECKING:
    from ..apiclient.api_client import APIClient

class BaseManager:
    """
    Basic manager abstract class.
    Implements methods common for all managers.
    Requires managers to implement their custom methods.
    """
    DAEMON_PIDFILE = '/var/run/xray-agent.pid'

VERSIONS = {
        'alt-php54': '/opt/alt/php54/link/conf',
        'alt-php55': '/opt/alt/php55/link/conf',
        'alt-php56': '/opt/alt/php56/link/conf',
        'alt-php70': '/opt/alt/php70/link/conf',
        'alt-php71': '/opt/alt/php71/link/conf',
        'alt-php72': '/opt/alt/php72/link/conf',
        'alt-php73': '/opt/alt/php73/link/conf',
        'alt-php74': '/opt/alt/php74/link/conf',
        'alt-php80': '/opt/alt/php80/link/conf',
        'alt-php81': '/opt/alt/php81/link/conf',
        'alt-php82': '/opt/alt/php82/link/conf',
        'alt-php83': '/opt/alt/php83/link/conf',
        'alt-php84': '/opt/alt/php84/link/conf',
        'alt-php85': '/opt/alt/php85/link/conf',
    }

def __init__(self, system_id: str, phpinfo_mode: bool = False):
        # FIXME: there is no reason to keep class instead of instance
        #        refactor this once it causes issues
        self.api_client_class: type[APIClient] = get_client()
        self.sys_id = system_id
        self.logger = logging.getLogger('manager')
        self.continuous_monitoring = ContinuousManager()
        self.manage_user_agent = ManageUserAgent()
        self.phpinfo_mode = phpinfo_mode
        self.ui_api_client = get_client('lists')(system_id=self.sys_id)

def _get_daemon_pid(self) -> Optional[int]:
        try:
            with open(self.DAEMON_PIDFILE) as f:
                return int(f.read())
        except (OSError, IOError):
            logging.warning('Unable to read daemon pid from pidfile.')

def supported_versions(self) -> dict:
        """
        Get supported PHP versions
        :return: a dict with supported versions
        """
        return self.VERSIONS

def is_version_supported(self, php_version: str) -> bool:
        """
        Check if given PHP version is supported
        :param php_version: PHP version, e.g. 'alt-phpXY'
        :return: True if version is supported, False otherwise
        """
        return php_version in self.supported_versions()

def _ini_path(self, domain_info: DomainInfo) -> str:
        """
        Get a path to directory for additional ini file for based on
        panel-set version
        """
        return self.supported_versions().get(domain_info.panel_php_version)

def get_ini_path(self, domain_info: DomainInfo) -> str:
        """
        Resolve a path to directory for additional ini file
        It depends on version set for domain and on selector
        :param domain_info: a DomainInfo object
        :return: path to directory for ini files
        """
        ini_path = self._ini_path(domain_info)
        try:
            if domain_info.php_ini_scan_dir:
                ini_path = domain_info.php_ini_scan_dir
            elif self.is_selector_enabled(domain_info):
                ini_path = domain_info.selector_ini_path
            else:
                ini_path = domain_info.phpd_location_ini_path or ini_path
        except ValueError:
            # failed to resolve CageFS prefix for user
            pass
        self.logger.info('Ini path resolved as %s', ini_path)
        return ini_path

def get_php_version(self, domain_info: DomainInfo) -> str:
        """
        Get PHP version which serves given domain
        :param domain_info: a DomainInfo object, including user of domain
                            and PHP version set in control panel environment
        :return: real php version of domain (selector or panel one)
        """
        if not domain_info.is_selector_applied and \
                self.is_selector_enabled(domain_info):
            self.logger.info('Selector is enabled for user %s',
                             domain_info.user)
            current_version = domain_info.selector_php_version
        else:
            current_version = domain_info.panel_php_version
        self.logger.info('PHP version detected as %s', current_version)
        return current_version

def is_selector_enabled(self, domain_info: DomainInfo) -> bool:
        """
        Is selector enabled for given domain
        :param domain_info: a DomainInfo object, including user of domain
                            and PHP version set in control panel
        :return: True if selector is enabled for domain, False otherwise
        """
        if not is_panel_feature_supported(Feature.PHP_SELECTOR):
            return False
        if domain_info.selector_php_version is None:
            return False
        return self.panel_specific_selector_enabled(domain_info)

def php_procs_reload(self, domain_info: DomainInfo) -> None:
        """
        Reload FPM service or kill all *php* processes of user
        :param domain_info: a ready-to-use DomainInfo object
        """
        self.reset_criu_imgs(domain_info.name)

if domain_info.panel_fpm:
            self.restart_fpm_service(domain_info)
        elif is_litespeed_running():
            self.gracefully_restart_litespeed(domain_info.user)
        else:
            self.kill_user_php(domain_info.user)

def gracefully_restart_litespeed(self, username):
        """
        Litespeed's graceful restart.
        https://www.litespeedtech.com/support/wiki/doku.php/litespeed_wiki:php:detached_mode#for_a_user
        """
        with drop_privileges(username), disable_quota():
            pw = pwd.getpwnam(username)
            # just  change it's modification date
            try:
                open(os.path.join(pw.pw_dir, '.lsphp_restart.txt'), 'w').close()
            except (FileNotFoundError, PermissionError):
                self.logger.warning('Unable to restart lsws after changes', exc_info=True)

def kill_user_php(self, username: str) -> list:
        """
        Kill all PHP processes, which belong to given username
        :param username: name of user
        :return: list of pids of successfully killed processes
        """
        killed_procs = list()
        for proc in self.user_procs(username):
            if 'php' in proc.info['name']:
                proc.send_signal(signal.SIGHUP)
                killed_procs.append(proc.info['pid'])
        return killed_procs

@staticmethod
    def user_procs(user_name: str) -> Iterator[psutil.Process]:
        """
        Generator yields processes, which belong to given user_name.
        Processes are checked using UID, not user_name
        :param user_name: user name
        :return: a generator object
        """
        uid_by_name = pwd.getpwnam(user_name).pw_uid
        attrs = ['name', 'pid', 'uids']
        for p in psutil.process_iter(attrs):
            if uid_by_name in (p.info['uids'].real, p.info['uids'].effective):
                yield p

def reset_criu_imgs(self, domain: str) -> None:
        """
        Reset criu images if any found for given domain in order to
        enable/disable X Ray correctly
        """
        if not domain:
            return
        criu_imgs_dir = glob(f'/var/run/mod_lsapi/*{glob_escape(domain)}_criu_imgs')
        if criu_imgs_dir:
            shutil.rmtree(criu_imgs_dir[0])
            self.logger.info('criu images in %s dropped', criu_imgs_dir[0])

def _domain_info_by_url(self, url: str) -> DomainInfo:
        """
        DomainInfo retrieving based on url
        """
        domain_name, _ = url_split(url)
        # get_domain_info includes validation of domain existence
        return self.get_domain_info(domain_name)

def get_domain_info(self, domain_name: str) -> DomainInfo:
        """
        Retrieve information about given domain from control panel environment
        Required to be implemented by child classes
        :param domain_name: name of domain
        :return: a DomainInfo object
        """
        raise NotImplementedError(
            _('Manager should implement retrieving domain info!'))

def panel_specific_selector_enabled(self, domain_info: DomainInfo) -> bool:
        """
        Check if selector is enabled specifically for panel
        Required to be implemented by child classes
        :param domain_info: a DomainInfo object
        :return: True if yes, False otherwise
        """
        raise NotImplementedError(
            _('Manager should implement specific panel check for selector enabled!'))

def fpm_service_name(self, dom_info: DomainInfo) -> str:
        """
        Get FPM service name for particular panel
        :param dom_info: DomainInfo object
        :return: name of FPM service
        """
        raise NotImplementedError(
            _('Manager should implement FPM service name retrieving!'))

def restart_fpm_service(self, dom_info: DomainInfo) -> None:
        """
        Restart FPM service for particular version
        :param dom_info: DomainInfo object
        """
        fpm_serv = self.fpm_service_name(dom_info)
        try:
            subprocess.run(
                ['/sbin/service', fpm_serv, 'reload'],
                capture_output=True,
                text=True,
                check=True)
            self.logger.info('Service %s reloaded', fpm_serv)
        except (OSError, ValueError, subprocess.SubprocessError) as e:
            self.logger.error('Failed to reload FPM service',
                              extra={'err': str(e),
                                     'info': dom_info})
        else:
            FPMReloadController(fpm_serv).save_latest_reload()

@staticmethod
    def prepare_wpos_info_path(username: str) -> Tuple[str, int]:
        """Resolve path for user and prepare directory if needed"""
        _upwd = pwd.getpwnam(username)
        _path = f'/var/clwpos/uids/{_upwd.pw_uid}/info.json'
        if not os.path.isdir(os.path.dirname(_path)):
            os.makedirs(os.path.dirname(_path))
            _cagefsctl_remount(username)
        return _path, _upwd.pw_gid

def write_wpos_info(self, user: str) -> None:
        """Prepare and write info.json file for WPOS (AWP) utilities"""
        if not is_wpos_supported():
            return

dest, user_gid = self.prepare_wpos_info_path(user)
        if is_file_recently_modified(dest):
            self.logger.info('File %s modified recently, skip generating',
                             dest)
            return

working_dest = dest + '.tmp'
        to_write = dict(
            vhost_versions=php_get_vhost_versions(user),
            installed_versions=get_installed_php_versions()
        )
        self.logger.debug('Going to write file %s with %s',
                          dest, to_write)
        with set_privileges(target_uid=0, target_gid=user_gid, mask=0o137):
            try:
                with open(working_dest, 'w') as info_json:
                    json.dump(to_write, info_json)
                shutil.move(working_dest, dest)
                self.logger.info('WPOS data written successfully')
            except OSError as e:
                self.logger.error('Unable to write info for WPOS due to %s',
                                  str(e))

@staticmethod
    def response(**kwargs) -> str:
        """
        Create JSON response message with result field == success
        and given keyword arguments in other fields
        :return: json packed string
        """
        initial = {'result': 'success'}
        if kwargs:
            initial.update(kwargs)
        return json.dumps(dict(sorted(initial.items())))

def response_continuous(self, **kwargs) -> 'json str':
        """
        Create JSON response message for continuous actions.
        Extends action definition with 'continuous' addition
        :return: json packed string
        """
        if kwargs.get('action'):
            kwargs['action'] = f"{kwargs['action']} continuous"
        return self.response(**kwargs)

@user_mode_restricted
    def start(self, *, url: str, client_ip: str,
              tracing_by: str, tracing_count: int,
              auto_task: bool = False,
              autotracing: bool = False) -> 'json str':
        """
        Start monitoring of given URL.
        Arguments are only allowed by keyword
        :param url: URL to monitor
        :param client_ip: a client IP address
        :param tracing_by: time or qty
        :param tracing_count: a number of minutes|requests to monitor
        :param auto_task: if a task is an auto created one (x-ray 2.0)
        :param autotracing: task is an auto created with
                            user marked as *autotracing*
        :return: JSON encoded result of start action
        """
        # domain existence validation inside
        domain_info = self._domain_info_by_url(url)
        self.write_wpos_info(domain_info.user)
        real_php_version = self.get_php_version(domain_info)
        if self.is_version_supported(real_php_version):
            ini_files_location = self.get_ini_path(domain_info)
        else:
            self.logger.warning('PHP version is unsupported',
                                extra={'version': real_php_version})
            raise XRayManagerExitPHPUnsupported(url, real_php_version)

# --- ask redis for tracing task id ---
        client = self.api_client_class(system_id=self.sys_id)
        user = '*autotracing*' if autotracing else get_xray_exec_user()
        tracing_task = Task(url=url, client_ip=client_ip,
                            tracing_by=tracing_by,
                            tracing_count=tracing_count,
                            ini_location=ini_files_location,
                            auto_task=auto_task, user=user,
                            domain_owner=domain_info.user)
        tracing_task.task_id = client.create(tracing_task)
        # ---
        tracing_task.add(php_version=real_php_version)
        # --- if time, add cron job
        tracing_task.set_cronjob(system_id=self.sys_id)
        # ---
        # --- update task status in redis db
        client.update(tracing_task.starttime)
        # ---
        self.php_procs_reload(domain_info)
        switch_schedstats(enabled=True)
        try:
            ClWposGetter().post_metadata(domain_info.user, domain_info.name)
        except XRayAPIError as e:
            logging.warning('Failed to send metadata to Smart Advice with: %s',
                            e.reason)
        NginxUserCache(domain_info.user).disable()

# automatically create missing ini files
        # in cagefs and other places if we use global mode
        if is_global_ini_mode():
            create_ini_files()

return self.response(action='start',
                             tracing_task_id=tracing_task.task_id)

def start_auto(self, *, url: str) -> 'json str':
        """
        Start continuous monitoring of given URL (auto task).
        Arguments are only allowed by keyword
        :param url: URL to monitor
        :return: JSON encoded result of start action
        """
        return self.start(url=url, client_ip='*', tracing_by='time',
                          tracing_count=1430, auto_task=True)

def start_autotracing(self, *, url: str,
                          tracing_count: int = 20) -> 'json str':
        """
        Start monitoring of given URL (autotracing task).
        Note, user is overridden as *autotracing*
        Arguments are only allowed by keyword
        :param url: URL to monitor
        :param tracing_count: count of requests to capture
        :return: JSON encoded result of start action
        """
        return self.start(url=url, client_ip='*', tracing_by='request_qty',
                          tracing_count=tracing_count, auto_task=True,
                          autotracing=True)

def stop(self, tracing_task_id: str) -> 'json str':
        """
        Stop monitoring of given task ID
        :param tracing_task_id: an ID of task to stop
        :return: JSON encoded result of stop action
        """
        client = self.api_client_class(system_id=self.sys_id,
                                       tracing_task_id=tracing_task_id)
        tracing_task = client.get_task()

self._request_daemon_storage_flush()
        with open_local_storage(tracing_task.fake_id) as storage:
            tracing_task.update_with_local_data(next_request_id=storage.next_request_id)

try_to_complete_statuses = ['running', 'stopped', 'completed', 'hold']

if tracing_task.status not in try_to_complete_statuses:
            raise XRayManagerExit(
                _("Cannot stop task with status '%s'") % str(tracing_task.status))

# for FPM pre-reload check
        try:
            domain_info = self._domain_info_by_url(tracing_task.url)
            tracing_task.set_domain_owner(domain_info.user)
        except XRayMissingDomain:
            self.logger.info('Stopping task for an already deleted domain')
            domain_info = None
        except XRayError as e:
            # In user mode, the user_mode_verification decorator converts
            # an inner XRayMissingDomain into a generic XRayError so the
            # exception class does not leak "domain exists / does not
            # exist" to user-mode tenants. To preserve graceful cleanup
            # of orphaned task data and cron jobs for an already-deleted
            # domain, the decorator tags the converted exception with a
            # private instance attribute that is not externally
            # observable (XRayError.__str__ does not serialize it).
            # When the marker is present, treat this the same as the
            # XRayMissingDomain case above; otherwise re-raise so the
            # original error-handling contract is unchanged.
            if getattr(e, '_internal_missing_domain', False):
                self.logger.info('Stopping task for an already deleted domain')
                domain_info = None
            else:
                raise

tracing_task.remove()
        # --- if time, remove cron job
        tracing_task.drop_cronjob()
        # ---
        # --- recalculate remaining count
        remaining_count = tracing_task.recalculate_counts()
        # ---
        # --- stop or complete task
        if remaining_count > 0:
            client.stop(remaining_count)
        else:
            self._complete(tracing_task, client)
        # ---
        if domain_info is not None:
            NginxUserCache(domain_info.user).restore()
            self.php_procs_reload(domain_info)
        if no_active_tasks() and ssa_disabled():
            switch_schedstats(enabled=False)
        return self.response(action='stop',
                             tracing_task_id=tracing_task.task_id)

def _request_daemon_storage_flush(self):
        """
        Request daemon to flush it's in-memory storage on disk using SIGUSR2
        and wait for getting SIGUSR2 back that daemon sends when it successfully
        flushed data on disk and we are ready to proceed.
        """
        # don't process SIGUSR2 with default handler
        signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGUSR2})
        daemon_pid = self._get_daemon_pid()

if daemon_pid:
            try:
                os.kill(daemon_pid, signal.SIGUSR2)
            except OSError:
                self.logger.warning('Unable to send daemon signal. Some stats will not be flushed. '
                                    'Maybe daemon is stopped?')

self.logger.info('Waiting for daemon to signal back about flush end.')
        siginfo = signal.sigtimedwait({signal.SIGUSR2}, 15.0)
        if siginfo is None:
            logging.warning('Daemon did not signal back in given timeout. '
                            'Some stats may not be flushed. Continue.')
        # and restore default signal handler
        signal.pthread_sigmask(signal.SIG_DFL, {signal.SIGUSR2})

@user_mode_restricted
    def continue_(self, tracing_task_id: str) -> 'json str':
        """
        Continue monitoring of given task ID
        :param tracing_task_id: an ID of task to continue
        :return: JSON encoded result of continue action
        """
        client = self.api_client_class(system_id=self.sys_id,
                                       tracing_task_id=tracing_task_id)
        tracing_task = client.get_task()
        if tracing_task.status in ('running', 'completed'):
            raise XRayManagerExit(
                _("Cannot continue task with status '%s'") % str(tracing_task.status))

# for FPM pre-reload check
        domain_info = self._domain_info_by_url(tracing_task.url)
        tracing_task.set_domain_owner(domain_info.user)

if '54' in self.supported_versions():
            # for custom panels only
            real_php_version = self.get_php_version(domain_info)
        else:
            # no need in redundant call for non-custom panels
            real_php_version = None

tracing_task.add(php_version=real_php_version)
        # --- if time, add cron job
        tracing_task.set_cronjob(system_id=self.sys_id)
        # ---
        # --- update task status in redis db
        client.update(tracing_task.starttime)
        # ---
        NginxUserCache(domain_info.user).disable()
        self.php_procs_reload(domain_info)
        switch_schedstats(enabled=True)
        return self.response(action='continue',
                             tracing_task_id=tracing_task.task_id)

def complete(self, tracing_task_id: str) -> 'json str':
        """
        Complete given tak ID
        :param tracing_task_id: an ID of task to complete
        :return: JSON encoded result of complete action
        """
        client = self.api_client_class(system_id=self.sys_id,
                                       tracing_task_id=tracing_task_id)
        tracing_task = client.get_task()
        if tracing_task.status in ('running', 'completed'):
            raise XRayManagerExit(
                _("Cannot complete task with status '%s'") % str(tracing_task.status))
        self._complete(tracing_task, client)
        return self.response(action='complete',
                             tracing_task_id=tracing_task.task_id)

@staticmethod
    def _is_to_complete(task: dict) -> bool:
        """
        Returns True if task should be completed.
        False otherwise.
        """
        now = datetime.now()
        tracing_by = task['tracing_by']
        inception = task['starttime'] or task['createtime']
        task_start_time = datetime.fromtimestamp(inception)
        if tracing_by == 'request_qty' and task_start_time + timedelta(days=2) < now:
            return True
        if tracing_by == 'time' and task_start_time + timedelta(minutes=task['tracing_count']) < now:
            return True
        return False

def autocomplete_tasks(self) -> 'json str':
        """
        Gets all server tasks and complete those which match following criteria:
         - traced_by requests_qty: if task started 2 days ago - complete
         - traced_by time: if task runs longer than start time + tracing time - complete
        """
        all_server_tasks = self.ui_api_client.get_task_list().get('result', [])
        self.logger.info('Check tasks for autocompleting...%s', str(all_server_tasks))
        for task_item in all_server_tasks:
            if not self._is_to_complete(task_item):
                continue
            logging.info('Going to complete task with id: %s', str(task_item['tracing_task_id']))
            self.stop(task_item['tracing_task_id'])
        return self.response(action='autocomplete-tasks')

@staticmethod
    def _complete(t_task: Task, client: 'APIClient'):
        """
        Common complete actions:
        - delete cron job,
        - erase request id file
        - send 'complete' status to mongo
        :param t_task: a tracing task object
        :param client: an APIClient object
        """
        t_task.drop_cronjob()
        t_task.erase_request_id_storage()
        client.complete()
        if t_task.auto_task:
            client.share()

def delete(self, tracing_task_id: str) -> 'json str':
        """
        Delete given task ID
        :param tracing_task_id: an ID of task to delete
        :return: JSON encoded result of delete action
        """
        client = self.api_client_class(system_id=self.sys_id,
                                       tracing_task_id=tracing_task_id)
        tracing_task = client.get_task()
        if tracing_task.status == 'running':
            raise XRayManagerExit(
                _("Cannot delete task with status '%s'") % str(tracing_task.status))
        client.delete()
        return self.response(action='delete',
                             tracing_task_id=tracing_task.task_id)

def enable_continuous(self, url: str, email: str) -> 'json str':
        """
        Enable continuous monitoring for given URL
        :param url: URL to monitor
        :param email: email to send reports to
        :return: JSON encoded result of enable action
        """
        # get_domain_info throws an exception in case of non-existent domain
        d_info = self._domain_info_by_url(url)

self.continuous_monitoring.enable(d_info.name, url, email)
        return self.response_continuous(action='enable', url=url)

def disable_continuous(self, url: str) -> 'json str':
        """
        Disable continuous monitoring for given URL
        :param url: URL to monitor
        :return: JSON encoded result of disable action
        """
        domain_name, _ = url_split(url)
        self.continuous_monitoring.disable(domain_name)
        return self.response_continuous(action='disable', url=url)

def start_continuous(self, url: str) -> 'json str':
        """
        Start continuous monitoring for given URL
        :param url: URL to monitor
        :return: JSON encoded result of start action
        """
        domain_name, _ = url_split(url)
        self.continuous_monitoring.start(domain_name)
        return self.response_continuous(action='start', url=url)

def stop_continuous(self, url: str) -> 'json str':
        """
        Stop continuous monitoring for given URL
        :param url: URL to monitor
        :return: JSON encoded result of stop action
        """
        domain_name, _ = url_split(url)
        self.continuous_monitoring.stop(domain_name)
        return self.response_continuous(action='stop', url=url)

def continuous_tracing_list(self) -> 'json str':
        """
        Get list of continuous monitoring tasks
        :return: JSON encoded result of get list action
        """
        tracing_list = self.continuous_monitoring.get_tracing_list()
        return self.response(action='get continuous list',
                             data=tracing_list)

def tasks_list(self) -> 'json str':
        """
        Get list of tasks
        """
        return self.response(action='tasks-list',
                             data=self.ui_api_client.get_task_list())

def requests_list(self, task_id: str) -> 'json str':
        """
        Get list of requests for given tracing task id
        """
        # needed for user verification
        client = self.api_client_class(system_id=self.sys_id,
                                       tracing_task_id=task_id)
        _ = client.get_task()
        return self.response(action='requests-list',
                             data=self.ui_api_client.get_request_list(task_id))

def request_data(self, task_id: str, request_id: int) -> 'json str':
        """
        Get collected statistics for request ID of given tracing task
        """
        # needed for user verification
        client = self.api_client_class(system_id=self.sys_id,
                                       tracing_task_id=task_id)
        _ = client.get_task()
        return self.response(action='request-data',
                             data=self.ui_api_client.get_request_data(
                                 task_id, request_id))

def enable_user_agent(self) -> 'json str':
        """
        Enable X-Ray User Agent:
            start or restart service if it is accidentally already running
        For systemd systems -- start socket unit only
        For SysV -- start the entire service
        :return: JSON encoded result of enable action
        """
        self.manage_user_agent.enable()
        return self.response(action='enable-user-agent')

def disable_user_agent(self) -> 'json str':
        """
        Disable X-Ray User Agent:
             stop the entire service
             or do nothing if it is accidentally not running
        For systemd systems -- also check if socket unit is running
                               and stop it too
        :return: JSON encoded result of disable action
        """
        self.manage_user_agent.disable()
        return self.response(action='disable-user-agent')

def user_agent_status(self) -> 'json str':
        """
        Get status of X-Ray User Agent service
        :return: JSON encoded result of status action
        """
        agent_status = self.manage_user_agent.status()
        return self.response(action='user-agent-status', status=agent_status,
                             user_nginx_cache=nginx_user_cache())

def advanced_metrics(self, args) -> 'json str':
        """
        Advanced metrics tool
        :return: JSON encoded result of status action
        """
        am = AdvancedMetrics()
        if args['enable'] is True:
            am.enable()
        elif args['disable'] is True:
            am.disable()
        elif args['status'] is True:
            status = am.status()
            return self.response(status=status)

return self.response()

def enable_serverwide_mode(self, args) -> 'json str':
        """
        Advanced metrics tool
        :return: JSON encoded result of status action
        """
        create_global_ini_mode_marker()
        create_ini_files()

return self.response()

def disable_serverwide_mode(self, args) -> 'json str':
        """
        Advanced metrics tool
        :return: JSON encoded result of status action
        """
        remove_ini_files()
        remove_global_ini_mode_marker()

return self.response()

Batal

⬅ xray 7 item

Nama	Tipe	Ukuran	Diubah	Perm	Aksi
📁 __pycache__	dir	—	09/07/26 03:13	0755	📂
🐍 __init__.py	py	1.2 KB	17/06/26 19:26	0644	✏️ ⬇
🐍 base.py	py	31.7 KB	17/06/26 19:26	0644	✏️ ⬇
🐍 cpanel.py	py	8.3 KB	17/06/26 19:26	0644	✏️ ⬇
🐍 custom.py	py	6 KB	17/06/26 19:26	0644	✏️ ⬇
🐍 directadmin.py	py	12.9 KB	17/06/26 19:26	0644	✏️ ⬇
🐍 plesk.py	py	7 KB	17/06/26 19:26	0644	✏️ ⬇

✏️ Rename

🔑 Ubah Permission