ＳＫＹＳＨＥＬＬＭＡＮＡＧＥＲ

📁 ＳＫＹＳＨＥＬＬＭＡＮＡＧＥＲ PHP v8.1.34

Path:root/home/terracebizon/public_html/wp-includes/js/tinymce/themes/

Name	Size	Perm	Actions
📁 inlite	-	0755	🗑️ 🏷️
📁 modern	-	0755	🗑️ 🏷️
📄 config.php	6.83 KB	0444	🗑️ 🏷️ ⬇️ ✏️
📄 error_log	17907.08 KB	0644	🗑️ 🏷️ ⬇️ ✏️

Edit: utils.py

# 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 helpful utility functions for X-Ray Manager
"""

# PEP 563: keep annotations unevaluated. The X | None / list[...] forms below
# are PEP 604/585 syntax, which is only valid at runtime on Python 3.10+/3.9+.
# This package installs into the cloudlinux-venv interpreter, and the spec only
# floors that at `cloudlinux-venv >= 1.0.2` -- it does not pin a Python version,
# and the package still builds for CL7. Deferring annotation evaluation keeps the
# modern syntax readable without raising the runtime floor.
from __future__ import annotations

import dbm
import errno
import fcntl
import html
import logging
import os
import platform
import pwd
import re
import shelve
import shlex
import shutil
import stat
import subprocess
import time
import xml.etree.ElementTree as ET
from collections.abc import Callable
from contextlib import contextmanager
from datetime import date, timedelta
from functools import wraps
from glob import glob
from socket import AF_UNIX, SOCK_STREAM, fromfd, socket

import sentry_sdk
from clcommon.clpwd import drop_privileges
from clcommon.const import Feature
from clcommon.cpapi import get_cp_description, getCPName, is_panel_feature_supported, is_wp2_environment
from clcommon.lib.cledition import get_cl_edition_readable
from clcommon.ui_config import UIConfig
from clcommon.utils import get_rhn_systemid_value
from sentry_sdk.integrations.atexit import AtexitIntegration
from sentry_sdk.integrations.logging import LoggingIntegration

from xray import gettext as _
from xray.internal.clwpos_safe_imports import php_get_vhost_versions_user

from .constants import agent_file, jwt_token_location, local_tasks_storage, logging_level, sentry_dsn, user_agent_sock
from .exceptions import XRayError, XRayManagerExit

logger = logging.getLogger('utils')
subprocess_errors = (OSError, ValueError, subprocess.SubprocessError)

def chmod_created_dir(path: str, mode: int) -> None:
    """Set ``mode`` on a directory we just created, without following symlinks.

``mkdir``'s mode argument is masked by the ambient umask, so a directory
    that must have an exact mode has to be chmod'ed after the fact. Doing that
    with a path-based ``os.chmod`` is unsafe here: these call sites run as root
    with no privilege drop, and ``os.chmod`` re-resolves the name at syscall
    time. A tenant who controls the parent can win the mkdir -> chmod window,
    replace the new directory with a symlink, and redirect the root chmod onto
    a root-owned target.

Pin the inode instead. ``O_NOFOLLOW`` refuses a symlink swapped in for the
    final component outright, and reading the fd back through ``/proc/self/fd``
    catches an ancestor component that was swapped, because that changes the
    canonical path. ``fchmod`` then acts on the pinned inode and never on a
    re-resolved name -- the same idiom ``unified_write``/``unified_erase`` use
    in ``xray.internal.types``.

Anything unexpected (the directory vanished, was replaced by a symlink,
    moved) is logged and skipped rather than raised: the mode is a hardening
    detail, not a precondition of the operation that created the directory.

Caller contract: ``path`` must be the CANONICAL path of the directory that
    was just created -- i.e. already ``os.path.realpath``'d, and resolved before
    the mkdir, not after. Passing a path that merely happens to reach the right
    inode through a symlink is not an error, but the identity check below will
    not match and the mode will be skipped.
    """
    _chmod_pinned_dir(path, mode)

def repair_dir_mode(path: str, damaged_mode: int, target_mode: int) -> None:
    """Upgrade a directory that an earlier release left at ``damaged_mode``.

The mode-pinning above only runs when a directory is CREATED, so it does
    nothing for hosts whose directories were already created wrong -- and the
    0.6-51/0.6-52 crons ran under umask 077 for weeks, so that is most affected
    hosts rather than an edge case (ZD 290458). Unlike xray.so, xray.ini and the
    skeleton ini -- which are rewritten wholesale on the next run and therefore
    self-heal -- a directory is only ever created once.

The repair is deliberately narrow: it fires only when the current mode is
    EXACTLY ``damaged_mode``, the signature the umask-077 runs produced. An
    administrator's or tenant's deliberate 0750 (or anything else) is left
    alone, so this cannot quietly loosen a directory nobody asked us to touch.

Same caller contract and same symlink handling as ``chmod_created_dir``.
    """
    # Cheap pre-filter so the healthy majority of hosts -- where the directory
    # is already correct -- pay one stat instead of the whole open/readlink/
    # fstat/fchmod sequence on every call. It cannot introduce a TOCTOU: the
    # authoritative check is still the fstat on the PINNED fd inside
    # _chmod_pinned_dir, which is the same inode fchmod acts on. This stat only
    # decides whether it is worth looking.
    try:
        if stat.S_IMODE(os.stat(path).st_mode) != damaged_mode:
            return
    except OSError:
        return
    _chmod_pinned_dir(path, target_mode, only_if_mode=damaged_mode)

def _chmod_pinned_dir(path: str, mode: int, only_if_mode: int | None = None) -> None:
    """Shared worker for the two functions above: chmod through a pinned fd.

``only_if_mode`` makes the chmod conditional on the pinned inode's CURRENT
    mode, read via ``fstat`` on the same fd that ``fchmod`` acts on -- so the
    check and the change cannot be separated by a race.
    """
    try:
        dir_fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC)
    except OSError as e:
        logger.warning('Not setting mode on directory', extra={'path': path, 'err': str(e)})
        return
    try:
        pinned = os.readlink(f'/proc/self/fd/{dir_fd}')
        if pinned != path:
            logger.warning(
                'Directory changed under us; not setting mode',
                extra={'path': path, 'pinned': pinned},
            )
            return
        if only_if_mode is not None:
            current = stat.S_IMODE(os.fstat(dir_fd).st_mode)
            if current != only_if_mode:
                return
            logger.info(
                'Repairing directory mode left by the umask-077 crons',
                extra={'path': path, 'from_mode': oct(current), 'to_mode': oct(mode)},
            )
        os.fchmod(dir_fd, mode)
    except OSError as e:
        logger.warning('Failed to set mode on directory', extra={'path': path, 'err': str(e)})
    finally:
        os.close(dir_fd)

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

def skeleton_update(func: Callable) -> Callable:
    """
    Decorator aimed to update ini file in cagefs-skeleton
    Applies to task.add nd task.remove
    """

def update(*args):
        """
        Copy ini file to cagefs-skeleton
        Action takes place for cPanel ea-php only
        """
        original_ini = os.path.join(args[0].ini_location, 'xray.ini')
        if original_ini.startswith('/opt/cpanel') and glob('/usr/share/cagefs'):
            skeleton_ini = os.path.join('/usr/share/cagefs/.cpanel.multiphp', original_ini[1:])
        elif original_ini.startswith('/usr/local') and glob('/usr/share/cagefs-skeleton'):
            skeleton_ini = os.path.join('/usr/share/cagefs-skeleton', original_ini[1:])
            # Resolve BEFORE creating. chmod_created_dir compares the pinned fd's
            # canonical path against the path it was handed, so handing it a
            # non-canonical one silently skips the mode pin on any host where the
            # skeleton tree is reached through a symlink. Resolving afterwards
            # would be worse than useless: it would re-follow whatever the name
            # points at by then, which is the very swap that check exists to catch.
            skeleton_dir = os.path.realpath(os.path.dirname(skeleton_ini))
            if not os.path.exists(skeleton_dir):
                os.mkdir(skeleton_dir)
                # mkdir()'s mode is masked by the ambient umask; this directory
                # is mirrored into every user's CageFS jail, so it has to stay
                # traversable whatever umask the caller ran under (ZD 290458).
                chmod_created_dir(skeleton_dir, 0o755)
            else:
                # Already-broken hosts: this dir is root-owned and mirrored into
                # every jail, so at 0700 jailed PHP cannot traverse it and the
                # xray.ini inside stays unreachable however correct its own mode
                # is. It is created once and never rewritten, so it cannot
                # self-heal the way the ini files do.
                repair_dir_mode(skeleton_dir, 0o700, 0o755)
        else:
            return

if not os.path.exists(original_ini):
            if os.path.lexists(skeleton_ini):
                if os.path.islink(skeleton_ini):
                    logger.warning('Refusing to unlink symlink in cagefs-skeleton', extra={'xray_ini': skeleton_ini})
                    return
                try:
                    os.unlink(skeleton_ini)
                except OSError as e:
                    logger.warning(
                        'Failed to unlink ini in cagefs-skeleton', extra={'xray_ini': skeleton_ini, 'err': str(e)}
                    )
        else:
            try:
                if os.path.islink(skeleton_ini):
                    logger.warning('Refusing to copy over symlink in cagefs-skeleton', extra={'xray_ini': skeleton_ini})
                    return
                # Read source first so a transient source-read failure
                # cannot truncate-then-fail the existing skeleton file.
                with open(original_ini, 'rb') as src:
                    src_bytes = src.read()
                fd = os.open(skeleton_ini, os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW, 0o644)
                try:
                    os.write(fd, src_bytes)
                    # open()'s mode argument only applies on creation and is
                    # masked by the ambient umask. fchmod pins 0644 so the
                    # skeleton copy stays readable by the jailed PHP whatever
                    # umask the caller ran under (ZD 290458).
                    os.fchmod(fd, 0o644)
                finally:
                    os.close(fd)
            except OSError as e:
                logger.warning(
                    'Failed to copy ini into cagefs-skeleton', extra={'xray_ini': original_ini, 'err': str(e)}
                )

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

return wrapper

def dbm_storage_update(func: Callable) -> Callable:
    """
    Decorator aimed to update DBM storage with fake_id:real_id mapping
    Applies to task.add nd task.remove
    """

def update(*args):
        """
        Update DBM storage contents
        """
        task_instance = args[0]
        with dbm_storage(local_tasks_storage) as task_storage:
            task_storage[task_instance.fake_id] = task_instance.task_id

def remove(*args):
        """
        Remove task from DBM storage
        """
        with dbm_storage(local_tasks_storage) as task_storage:
            try:
                del task_storage[args[0].fake_id.encode()]
            except KeyError:
                # ignore absence of item during removal
                pass

@wraps(func)
    def wrapper(*args, **kwargs):
        """
        Wraps func
        """
        # add task id into DBM storage as early as possible
        try:
            if func.__name__ == 'add':
                update(*args)
        except RuntimeError as e:
            raise XRayError(str(e))

try:
            result = func(*args, **kwargs)
        except Exception:
            # cleanup recently added task id from DBM storage in case of
            # any accidental fails during add procedure
            if func.__name__ == 'add':
                remove(*args)
            raise

# during task removal cleanup task from DBM storage as late as possible
        try:
            if func.__name__ == 'remove':
                remove(*args)
        except RuntimeError as e:
            raise XRayError(str(e))

return result

return wrapper

def check_jwt(func: Callable) -> Callable:
    """
    Decorator aimed to validate given JWT token
    """

def check():
        """
        Check if retrieved JWT token is valid
        """
        is_xray_supported()

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

return wrapper

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

def timestamp() -> int:
    """
    Get current epoch timestamp as int
    :return: timestamp as int
    """
    return int(time.time())

def prev_date() -> date:
    """
    Pick a yesterday date
    :return: a datetime.date object
    """
    return date.today() - timedelta(days=1)  # noqa: DTZ011 - report windows are local-date by design

def date_of_timestamp(ts: int) -> date:
    """
    Get the datetime.date object for given int timestamp
    :param ts: timestamp
    :return: datetime.date object
    """
    return date.fromtimestamp(ts)  # noqa: DTZ012 - naive local date, matches prev_date()

def get_formatted_date() -> str:
    """
    Get a formatted representation of yesterday date
    :return: str date in the form of dd/mm/YYYY
    """
    return prev_date().strftime("%d/%m/%Y")

def get_html_formatted_links(links: list[dict]) -> str:
    """
    HTML formatted links.

Both the domain (interpolated into the visible <a> body) and the
    link (interpolated into the href attribute) are passed through
    html.escape(..., quote=True) so HTML-special characters in either
    slot cannot break attribute parsing or inject tags. The href is
    also explicitly double-quoted — without quotes, whitespace in the
    link value would let any subsequent token be parsed as a new
    attribute (e.g. ``onmouseover=...``). Defense-in-depth: today the
    sole caller (continuous.tracing.generate_mail) passes
    url_split(task.url)[0] and task.shared_link, both of which are
    derived from controlled sources, but the template must remain
    safe for any future caller.
    """
    html_item = '<p>{num}) <a href="{link}">{domain}</a></p>'
    return '\n'.join(
        [
            html_item.format(num=i, link=html.escape(v, quote=True), domain=html.escape(k, quote=True))
            for i, link_map in enumerate(links, 1)
            for k, v in link_map.items()
        ]
    )

def get_text_formatted_links(links: list[dict]) -> str:
    """
    Formatted links
    """
    text_item = '{num}) {dom}: {link}'
    return '\n'.join(
        [text_item.format(num=i, dom=k, link=v) for i, link_map in enumerate(links, 1) for k, v in link_map.items()]
    )

def read_sys_id() -> str:
    """
    Obtain system ID from /etc/sysconfig/rhn/systemid
    :return: system ID without ID- prefix
    """
    try:
        tree = ET.parse('/etc/sysconfig/rhn/systemid')
        root = tree.getroot()
        whole_id = root.find(".//member[name='system_id']/value/string").text
        with sentry_sdk.configure_scope() as scope:
            scope.set_tag("system_id", whole_id)
        return whole_id.lstrip('ID-')
    except (OSError, ET.ParseError) as e:
        raise XRayError(_('Failed to retrieve system_id')) from e

def write_sys_id(sys_id: str, agent_system_id_path: str = agent_file) -> None:
    """
    Write system_id into file /usr/share/alt-php-xray/agent_sys_id
    """
    fd = os.open(agent_system_id_path, os.O_CREAT | os.O_WRONLY | os.O_TRUNC | os.O_NOFOLLOW, 0o600)
    try:
        os.write(fd, sys_id.encode())
    finally:
        os.close(fd)

def read_agent_sys_id() -> str:
    """
    Read system_id saved by agent during its initialization
    """
    try:
        with open(agent_file) as agent_sysid_file:
            return agent_sysid_file.read().strip()
    except OSError as e:
        logger.info("Failed to retrieve agent's system_id, returning real one", extra={'err': str(e)})
        return read_sys_id()
        # raise XRayError("Failed to retrieve agent's system_id") from e

def is_xray_supported() -> bool | None:
    """Raise XRayError in case of detected non-supported edition"""
    is_supported = is_panel_feature_supported(Feature.XRAY)
    if not is_supported:
        current_edition = get_cl_edition_readable()
        current_panel = getCPName()
        logger.info(
            'Current CloudLinux edition: %s or Control Panel: %s is not supported by X-Ray',
            str(current_edition),
            str(current_panel),
        )
        raise XRayManagerExit(
            _('Current CloudLinux edition: {} or Control Panel: {} is not supported by X-Ray').format(
                current_edition, current_panel
            )
        )
    return True

@check_jwt
def read_jwt_token() -> str:
    """
    Obtain jwt token from /etc/sysconfig/rhn/jwt.token
    :return: token read
    """
    try:
        with open(jwt_token_location) as token_file:
            return token_file.read().strip()
    except OSError:
        raise XRayError(_('JWT file %s read error') % str(jwt_token_location))

def pkg_version(filepath: str) -> str | None:
    """Get version of package from file. alt-php-xray supported"""
    try:
        with open(filepath) as v_file:
            version = v_file.read().strip()
    except OSError:
        return
    # remove dist suffix
    return '.'.join(version.split('.')[:2]) or '0.0-0'

def xray_version() -> str | None:
    """Get version of alt-php-xray package"""
    return pkg_version('/usr/share/alt-php-xray/version')

def sentry_init() -> None:
    """
    Initialize Sentry client
    shutdown_timeout=0 disables Atexit integration as stated in docs:
    'it’s easier to disable it by setting the shutdown_timeout to 0'
    https://docs.sentry.io/platforms/python/default-integrations/#atexit
    On the other hand, docs say, that
    'Setting this value too low will most likely cause problems
    for sending events from command line applications'
    https://docs.sentry.io/error-reporting/configuration/?platform=python#shutdown-timeout
    """

def add_info(event: dict, hint: dict) -> dict:
        """
        Add extra data into sentry event
        :param event: original event
        :param hint: additional data caught
        :return: updated event
        """
        event['extra'].update({'xray.version': '0.6-53.el8'})

# sentry_sdk defaults event['server_name'] to socket.gethostname();
        # overwrite it with the stable, non-identifying system_id so the raw
        # hostname (host PII) is never broadcast on outbound events (F-12).
        event['server_name'] = get_rhn_systemid_value("system_id")

extra_data = event.get('extra', {})
        fingerprint = extra_data.get('fingerprint', None)

if fingerprint:
            event['fingerprint'] = [fingerprint]

return event

def set_tags(sentry_scope):
        cp_description = get_cp_description()
        cp_version = cp_description.get('version') if cp_description else None
        cp_name = cp_description.get('name') if cp_description else None
        cp_product = 'WP2' if is_wp2_environment() else None

tags = (
            ('Control Panel Name', cp_name),
            ('Control Panel Version', cp_version),
            ('Control Panel Product', cp_product),
            ('kernel', platform.release()),
            ('CloudLinux version', get_rhn_systemid_value("os_release")),
            ('Cloudlinux edition', get_cl_edition_readable()),
            ('Architecture', get_rhn_systemid_value("architecture")),
        )
        # set_tags does not work in current version of sentry_sdk
        # https://github.com/getsentry/sentry-python/issues/1344
        for tag in tags:
            sentry_scope.set_tag(*tag)

def nope(pending, timeout) -> None:
        pass

sentry_logging = LoggingIntegration(level=logging.INFO, event_level=logging.WARNING)
    xray_ver = xray_version() or 'alt-php-xray@0.6-53.el8'
    silent_atexit = AtexitIntegration(callback=nope)
    sentry_sdk.init(
        dsn=sentry_dsn,
        before_send=add_info,
        release=xray_ver,
        max_value_length=10000,
        # Override the SDK default of socket.gethostname() so the
        # raw hostname is not shipped as event['server_name'] (F-12).
        server_name=get_rhn_systemid_value("system_id"),
        integrations=[sentry_logging, silent_atexit],
    )
    with sentry_sdk.configure_scope() as scope:
        scope.user = {
            # Correlate only by the stable, non-identifying system_id; avoid
            # broadcasting the host's live IP / hostname / local username.
            "id": get_rhn_systemid_value("system_id")
        }
        try:
            set_tags(scope)
        except Exception as e:  # noqa: BLE001 - Sentry setup must never break the caller
            logger.debug('Failed to set Sentry tags: %s', str(e))

def configure_logging(logname: str, level=logging_level) -> str | None:
    """
    Configure logging and Sentry
    :param logname: path to log
    :return: logpath
    """
    levels = {
        'debug': logging.DEBUG,
        'info': logging.INFO,
        'warning': logging.WARNING,
        'error': logging.ERROR,
        'critical': logging.CRITICAL,
    }

sentry_init()
    try:
        handlers = [logging.FileHandler(filename=logname)]
        if level == 'debug':
            handlers.append(logging.StreamHandler())
        logging.basicConfig(
            level=levels.get(level, logging.INFO),
            format='%(asctime)s [%(threadName)s:%(name)s] %(message)s',
            datefmt='%m/%d/%Y %I:%M:%S %p',
            handlers=handlers,
        )
    except OSError:
        # dummy logging
        logging.basicConfig(handlers=[logging.NullHandler()])
        return

try:
        os.chmod(logname, 0o600)
    except PermissionError:
        pass
    return logname

_safe_username_pattern = re.compile(r'^[a-zA-Z0-9_][a-zA-Z0-9._-]{0,31}$')

def validate_system_user(username: str) -> None:
    """Validate that username is a real system user.

Raises ValueError with a clear message if username is empty,
    has an invalid format, or does not exist in the system user database.
    """
    if not username:
        raise ValueError('username must not be empty')
    if not _safe_username_pattern.match(username):
        raise ValueError(f'Invalid username: {username!r}')
    try:
        pwd.getpwnam(username)
    except KeyError:
        raise ValueError(f'system user does not exist: {username!r}') from None

def build_clwpos_user_cmd(username: str, clwpos_args: list) -> list:
    """Build subprocess argv for /usr/bin/clwpos-user invocation.

Non-CageFS: wraps in sudo -u <user> bash -c with shell-quoted args.
    CageFS: passes args directly via cagefs_enter_user argv.
    """
    validate_system_user(username)
    if not is_panel_feature_supported(Feature.CAGEFS):
        safe_parts = [shlex.quote(str(a)) for a in clwpos_args]
        inner_cmd = '/usr/bin/clwpos-user ' + ' '.join(safe_parts)
        return ['sudo', '-u', username, '-s', '/bin/bash', '-c', inner_cmd]
    else:
        return ['/sbin/cagefs_enter_user', username, '/usr/bin/clwpos-user'] + clwpos_args

def safe_move(src: str, dst: str) -> None:
    """
    Move file with error catching
    :param src: source
    :param dst: destination
    """
    try:
        shutil.move(src, dst)
    except OSError as e:
        raise XRayError(_('Failed to move file {} to {}: {}').format(src, dst, str(e))) from e

def create_socket(sock_location: str) -> socket:
    """
    Create world-writable socket in given sock_location
    or reuse existing one
    :param sock_location: socket address
    :return: socket object
    """
    listen_fds = int(os.environ.get("LISTEN_FDS", "0"))
    if listen_fds == 0:
        with umask_0():
            try:
                # sock.close does not remove the file
                os.unlink(sock_location)
            except FileNotFoundError:
                pass
            sockobj = socket(AF_UNIX)
            sockobj.bind(sock_location)
            sockobj.listen()
    else:
        sockobj = fromfd(3, AF_UNIX, SOCK_STREAM)
        sockobj.listen()
    return sockobj

def get_current_cpu_throttling_time(lve_id: int) -> int:
    """
    Retrieve current value of CPU throttled time.
    Return 0 in case of failures
    """
    if not is_panel_feature_supported(Feature.LVE):
        return 0
    marker = 'throttled_time'
    stat_file = f'/sys/fs/cgroup/cpu,cpuacct/lve{lve_id}/cpu.stat'
    try:
        with open(stat_file) as stat_values:
            for value in stat_values:
                if value.startswith(marker):
                    logger.debug('%s', value)
                    return int(value.strip().split(marker)[-1].strip())
    except OSError as e:
        logger.error('Failed to open %s: %s', stat_file, str(e))
    return 0

def _selectorctl_get_version(username: str) -> tuple | None:
    """
    'selectorctl -u username --user-current' command
    :param username: name of user
    :return: tuple(stdout, stderr) or None if command fails
    """
    _selectorctl = '/usr/bin/selectorctl'
    if not os.path.isfile(_selectorctl):
        return None
    try:
        result = subprocess.run(
            [_selectorctl, '-u', username, '--user-current'], capture_output=True, text=True, check=True
        )
        return result.stdout.strip(), result.stderr.strip()
    except subprocess.CalledProcessError as e:
        logger.warning('Failed to get selectorctl user-current', extra={'err': str(e)})
    except subprocess_errors as e:
        logger.error('selectorctl --user-current failed: %s', str(e))

def cagefsctl_get_prefix(username: str) -> str | None:
    """
    'cagefsctl --get-prefix username' command
    :param username: name of user
    :return: cagefsctl prefix for given username
            or None if command fails
    """
    _cagefsctl = '/usr/sbin/cagefsctl'
    if not os.path.isfile(_cagefsctl):
        return None
    try:
        result = subprocess.run([_cagefsctl, '--getprefix', username], capture_output=True, text=True, check=True)
        return result.stdout.strip()
    except subprocess.CalledProcessError as e:
        logger.warning('Failed to get cagefsctl prefix', extra={'err': str(e)})
    except subprocess_errors as e:
        logger.error('cagefsctl --getprefix failed: %s', str(e))

def _cagefsctl_remount(username: str | None = None) -> None:
    """
    'cagefsctl --remount username' or 'cagefsctl --remount-all' command
    :param username: name of user or None (for remount-all)
    """
    _cagefsctl = '/usr/sbin/cagefsctl'
    if not os.path.isfile(_cagefsctl):
        return
    if username is None:
        args = [_cagefsctl, '--wait-lock', '--remount-all']
    else:
        args = [_cagefsctl, '--remount', username]
    try:
        subprocess.run(args, check=True, capture_output=True)
        logger.info('Remounted %s', username)
    except subprocess.CalledProcessError as e:
        logger.warning('Failed to remount cagefs', extra={'err': str(e)})
    except subprocess_errors as e:
        logger.error('cagefsctl --remount failed: %s', str(e))

def _is_cagefs_enabled(username: str) -> bool:
    """
    'cagefsctl --user-status username' command
    :param username: name of user
    :return: True if user has Enabled status, False otherwise
    """
    _cagefsctl = '/usr/sbin/cagefsctl'
    if not os.path.isfile(_cagefsctl):
        return False
    try:
        result = subprocess.run([_cagefsctl, '--user-status', username], capture_output=True, text=True, check=False)
        return 'Enabled' in result.stdout.strip()
    except subprocess_errors as e:
        logger.error('cagefsctl --user-status failed: %s', str(e))

def _is_selector_phpd_location_set() -> bool:
    """
    Check if there is php.d.location = selector
    set in /etc/cl.selector/symlinks.rules
    """
    try:
        with open('/etc/cl.selector/symlinks.rules') as rules_file:
            contents = rules_file.read()
    except OSError:
        return False
    return 'selector' in contents

def no_active_tasks() -> bool:
    """Check if there are no active tasks (== empty task storage)"""
    with dbm_storage(local_tasks_storage) as task_storage:
        return len(task_storage.keys()) == 0

def switch_schedstats(enabled: bool) -> None:
    """
    Switch on/off throttle statistics gathering by kmodlve
    :param enabled: True or False
    """
    if not is_panel_feature_supported(Feature.LVE):
        # do nothing if there is no LVE feature
        return

try:
        with open('/proc/sys/kernel/sched_schedstats', mode='wb', buffering=0) as f:
            f.write(b'1' if enabled else b'0')
    except OSError as e:
        logger.info('Failed to set sched_schedstats to %s: %s', enabled, str(e))

def is_xray_app_available() -> bool:
    """
    Check if end-users have access to X-Ray UI of End-User plugin
    """
    return UIConfig().get_param('hideXrayApp', 'uiSettings') is False

def is_xray_user_agent_active() -> bool:
    """Check if User Agent is listening"""
    with socket(AF_UNIX, SOCK_STREAM) as s:
        try:
            s.connect(user_agent_sock)
        except (ConnectionError, OSError):
            return False
    return True

def ssa_disabled() -> bool:
    """Check if SSA is disabled by its internal flag-file"""
    return not os.path.isfile('/usr/share/clos_ssa/ssa_enabled')

def is_file_recently_modified(filepath: str) -> bool:
    """Check is file was modified during the last day"""
    # 86400sec == 1day
    try:
        return timestamp() - os.stat(filepath).st_mtime < 86400
    except OSError:
        return False

def get_user_php_version(user):
    with drop_privileges(user):
        result = php_get_vhost_versions_user()
    return result

# --------- CONTEXT MANAGERS ---------

@contextmanager
def filelock(fd) -> None:
    """
    Context manager for locking given file object
    :param fd: а file object providing a fileno() method
    """
    # try to lock file with waiting for lock to be released
    # will be executed as __enter__
    for _attempt in range(120):
        try:
            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
            logger.info('File %s locked', fd)
            break
        except OSError as e:
            logger.info('Failed to lock: %s', str(e))
            # raise on unrelated IOErrors
            if e.errno not in (errno.EAGAIN, errno.EACCES):
                raise
            time.sleep(0.5)
    else:
        raise XRayError(_('Failed to lock at all. Exiting thread'), flag='warning')
    try:
        yield
    finally:
        # release lock
        # will be executed as __exit__
        fcntl.flock(fd, fcntl.LOCK_UN)
        logger.info('File %s unlocked', fd)

@contextmanager
def dbm_storage(filename: str, is_shelve: bool = False):
    """
    Context manager for waiting for lock to be released for DBM file storage,
    either plain DBM or a Shelf object
    (desired return value is controlled by _shelve_instance flag)
    :param filename: a DBM file to open
    :param is_shelve: if a shelve file should be opened instead of plain DBM
    """
    _file = os.path.basename(filename)
    _err = None
    for _attempt in range(100):
        try:
            if is_shelve:
                storage = shelve.open(filename)  # noqa: SIM115 - closed by dbm_storage()
            else:
                storage = dbm.open(filename, 'c')  # noqa: SIM115 - closed by dbm_storage()
            logger.debug('Storage %s opened', _file)
            break
        except dbm.error as e:
            logger.info('[#%i] Failed to open storage %s: %s', _attempt, _file, e)
            _err = e
            time.sleep(0.3)
    else:
        raise RuntimeError(f'Failed to open {_file} storage: {_err}')
    try:
        yield storage
    finally:
        storage.close()
        logger.debug('Storage %s closed', _file)

@contextmanager
def umask_0(mask: int = 0) -> None:
    """
    Context manager for dropping umask
    """
    prev = os.umask(mask)
    yield
    os.umask(prev)

@contextmanager
def set_privileges(
    target_uid: int | None = None,
    target_gid: int | None = None,
    target_path='.',
    mask: int | None = None,
    with_check=True,
) -> None:
    """
    Context manager to drop privileges during some operation
    and then restore them back.
    If target_uid or target_gid are given, use input values.
    Otherwise, stat target_uid and target_gid from given target_path.
    If no target_path given, use current directory.
    Use mask if given.
    :param target_uid: uid to set
    :param target_gid: gid to set
    :param target_path: directory or file to stat for privileges,
                       default -- current directory
    :param mask: umask to use
    :param with_check: check the result of switching privileges
    """
    prev_uid = os.getuid()
    prev_gid = os.getgid()

permission_issue_message = _('Unable to execute required operation: permission issue')
    try:
        stat_info = os.stat(target_path)
    except OSError:
        stat_info = None

if target_uid is None:
        if stat_info is None:
            target_uid = prev_uid
        else:
            target_uid = stat_info.st_uid
    if target_gid is None:
        if stat_info is None:
            target_gid = prev_gid
        else:
            target_gid = stat_info.st_gid

if mask is not None:
        prev = os.umask(mask)

if prev_gid != target_gid:
        os.setegid(target_gid)
        logger.debug('Dropped GID privs to %s', target_gid)
        if with_check and os.getegid() != target_gid:
            # break operation if privileges dropping failed
            raise XRayError(permission_issue_message)
    if prev_uid != target_uid:
        os.seteuid(target_uid)
        logger.debug('Dropped UID privs to %s', target_uid)
        if with_check and os.geteuid() != target_uid:
            if prev_gid != target_gid:
                # check if GID should be restored
                os.setegid(prev_gid)
            # break operation if privileges dropping failed
            raise XRayError(permission_issue_message)

# Wrap the yield in try/finally so the restoration block ALWAYS runs,
    # even if the with-block body raises. seteuid/setegid are process-wide
    # (POSIX, not thread-local) — a missed restore on the exception path
    # would leak the dropped UID/GID to every other thread in the long-
    # lived xray-user-agent daemon. The sibling helper user_context() in
    # this module uses the same try/finally idiom (see below).
    try:
        yield
    finally:
        if prev_uid != target_uid:
            os.seteuid(prev_uid)
            logger.debug('Restored UID privs to %s', prev_uid)
        if prev_gid != target_gid:
            os.setegid(prev_gid)
            logger.debug('Restored GID privs to %s', prev_gid)

if mask is not None:
            os.umask(prev)

@contextmanager
def user_context(uid, gid):
    """
    Dive into user context by dropping permissions
    to avoid most of the security issues.

Does not cover cagefs case because it also requires nsenter,
    which is only available with execve() call in our system
    """
    try:
        os.setegid(gid)
        os.seteuid(uid)

yield
    finally:
        os.seteuid(0)
        os.setegid(0)

def retry_on_exceptions(max_retries, exceptions_to_retry):
    """
    Decorator to retry method on specific exceptions
    """

def decorator(func):
        def wrapper(*args, **kwargs):
            retries = 0
            exception = ValueError(_('Request to website failed even after %s retries.') % str(max_retries))
            while retries < max_retries:
                try:
                    return func(*args, **kwargs)
                except tuple(exceptions_to_retry) as e:
                    retries += 1
                    logger.warning('Retry to request website, exception: %s', str(e))
                    exception = e
                    time.sleep(1)  # Wait for 1 second before retrying
            raise exception

return wrapper

return decorator

📁 ＳＫＹＳＨＥＬＬ ＭＡＮＡＧＥＲ PHP v8.1.34

Edit: utils.py

📁 ＳＫＹＳＨＥＬＬＭＡＮＡＧＥＲ PHP v8.1.34