byp4a55 professor6T9

Linux das102.truehost.cloud 4.18.0-553.8.1.lve.el8.x86_64 #1 SMP Thu Jul 4 16:24:39 UTC 2024 x86_64
LiteSpeed
156.232.88.2

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

"""
This module contains helpful utility functions for SSA Agent
"""
import datetime
import dbm
import logging
import os
import platform
import re
import xml.etree.ElementTree as ET
from collections import namedtuple
from contextlib import contextmanager
from datetime import date, timedelta
from distutils.version import LooseVersion
from socket import socket, fromfd, AF_UNIX, SOCK_STREAM, AF_INET, AF_INET6, \
    SOCK_DGRAM
from typing import Optional, Union
from urllib.parse import urlparse

import sentry_sdk
from sentry_sdk.integrations.atexit import AtexitIntegration
from sentry_sdk.integrations.logging import LoggingIntegration

from clcommon.const import Feature
from clcommon.cpapi import (
    is_panel_feature_supported,
    get_cp_description,
    is_throttling_supported,
    is_wp2_environment
)
from clcommon.utils import get_kmodlve_module_version, get_username
from clcommon.lib.cledition import get_cl_edition_readable
from clcommon.lib.network import get_hostname
from clcommon.utils import get_rhn_systemid_value

from .constants import sentry_dsn
from .exceptions import SSAError

logger = logging.getLogger('utils')
URL = namedtuple('URL', ['domain_name', 'uri_path'])

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

def url_split(url: str) -> URL:
    """
    Split URL into domain_name and uripath including query string
    :param url: URL of format protocol://domain/path;parameters?query#fragment
    :return: namedtuple URL(domain_name, uripath)
    """
    fragments = urlparse(url)
    qs = f'?{fragments.query}' if fragments.query else ''
    uri = f'{fragments.path}{qs}' if fragments.path else '/'
    # logger.info('Parsed %s into %s:%s', url, fragments.netloc, uri)
    return URL(fragments.netloc.replace('www.', ''), uri)

def ssa_version() -> Optional[str]:
    """Get version of alt-php-ssa package"""
    return pkg_version('/usr/share/clos_ssa/version')

def xray_version() -> Optional[str]:
    """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({'ssa.version': '0.4-18.el8'})
        return event

def try_get_ip(address_family, private_ip):
        """
        address_family - we can choose constants represent the address
                           (and protocol) families
                           (AF_INET for ipv4 and AF_INET6 for ipv6)
        private_ip - specify some private ip address. For instance:
                     ipv4 -> 10.255.255.255 or ipv6 -> fc00::
        """
        try:
            with socket(address_family, SOCK_DGRAM) as s:
                s.connect((private_ip, 1))
                return s.getsockname()[0]
        except Exception:
            logger.info('Cannot retrieve IP address')

def get_ip():
        """
        We are trying to get an IPv4 or IPv6 address.
        In case of failure we'll return 127.0.0.1
        """
        ipversions = (AF_INET, '10.255.255.255'), (AF_INET6, 'fc00::')
        for addr_fam, priv_ip in ipversions:
            ip = try_get_ip(addr_fam, priv_ip)
            if ip:
                return ip
        return '127.0.0.1'

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 = (('alt-php-xray', xray_version() or 'UNKNOWN'),
                ('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")),
                ('ip_address', get_ip()),
                ('username', get_username())
                )
        # 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

ssa_ver = ssa_version() or "alt-php-ssa@0.4-18.el8"

sentry_logging = LoggingIntegration(level=logging.INFO,
                                        event_level=logging.WARNING)
    silent_atexit = AtexitIntegration(callback=nope)
    sentry_sdk.init(dsn=sentry_dsn, before_send=add_info,
                    release=ssa_ver,
                    max_value_length=10000,
                    integrations=[sentry_logging, silent_atexit])
    with sentry_sdk.configure_scope() as scope:
        scope.user = {
            "id": get_rhn_systemid_value("system_id") or get_ip() or get_hostname() or get_username()
        }
        try:
            set_tags(scope)
        except Exception:
            pass

def set_logging_into_file(fname: str, as_error: bool = False) -> str:
    """
    Try to configure logging into given fname
    If as_error True, log the exception as ERROR, otherwise -- as INFO
    """
    try:
        logging.basicConfig(filename=fname, level=logging.INFO,
                            format='%(asctime)s %(message)s',
                            datefmt='%m/%d/%Y %I:%M:%S %p')
        try:
            os.chmod(fname, 0o666)
        except PermissionError:
            pass
        return fname
    except OSError as e:
        logger.log(logging.ERROR if as_error else logging.INFO,
                   'No logging configuration applied: %s',
                   str(e))

def configure_logging(logname: str) -> str:
    """
    Configure logging
    :param logname: path to log
    :return: logpath
    """
    sentry_init()
    if set_logging_into_file(logname) is None:
        try:
            os.makedirs(os.path.dirname(logname))
        except Exception as e:
            logger.warning('Failed to create logdir %s', str(e))
            return ''
        logname = set_logging_into_file(logname, as_error=True)
    return logname

def create_socket(sock_location: str) -> 'socket object':
    """
    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():
            sockobj = socket(AF_UNIX)
            sockobj.bind(sock_location)
            sockobj.listen()
    else:
        sockobj = fromfd(3, AF_UNIX, SOCK_STREAM)
        sockobj.listen()
    return sockobj

def previous_day_date() -> str:
    """
    Returns date of previous day in a format "day.month.year"
    """
    yesterday = date.today() - timedelta(days=1)
    return yesterday.strftime('%d.%m.%Y')

def format_date(datestr: str, formatstr='%d.%m.%Y') -> str:
    """
    Convert date to format YYYY-mm-dd
    """
    _date = datetime.datetime.strptime(datestr, formatstr)
    return _date.strftime("%Y-%m-%d")

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:
        logger.warning('Failed to retrieve system_id: %s', str(e))

def duration_cast(duration: int) -> float:
    """
    Cast duration from microseconds to seconds leaving 2 digits after point
    """
    return float(format(duration/1000000, '0.2f'))

def pkg_version(filepath: str) -> Optional[str]:
    """Get version of package from file. alt-php-ssa/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 is_xray_version_supported() -> bool:
    """
    Check version of alt-php-xray package.
    Autotracing in X-Ray is supported since 0.4-1
    """
    version_number = xray_version()
    if version_number is None:
        # no xray installed
        return False
    try:
        return LooseVersion(version_number) >= LooseVersion('0.4-1')
    except (TypeError, AttributeError):
        return False

def is_xray_user_agent_active() -> bool:
    """Check if User Agent is listening"""
    user_agent_sock = '/opt/alt/php-xray/run/xray-user.sock'
    with socket(AF_UNIX, SOCK_STREAM) as s:
        try:
            s.connect(user_agent_sock)
        except (ConnectionError, OSError):
            return False
    return True

def no_xray_active_tasks() -> bool:
    """Check if there are no active X-Ray tasks (== empty task storage)"""
    xray_tasks_storage = '/usr/share/alt-php-xray/tasks'
    if not os.path.isfile(xray_tasks_storage):
        # no tasks file usually means no X-Ray at all, thus no tasks
        return True

try:
        with dbm.open(xray_tasks_storage, 'c') as xray_tasks:
            return len(xray_tasks.keys()) == 0
    except dbm.error:
        # in case we somehow failed to open the file, but it exists,
        # suppose there are active tasks
        return False

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

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

def is_io_throttling_detection_available() -> bool:
    """
    Check kmodlve module version or kernel version in order to determine
    if it provides the detection of IO throttling
    """
    kmod_min_version = LooseVersion('2.0-23')
    kernel_min_version = LooseVersion('1.5.58')
    kmod_current = get_kmodlve_module_version()
    if kmod_current:
        return compare_versions_ge(kmod_current, kmod_min_version)
    return compare_versions_ge(extract_kernel_version(), kernel_min_version)

def extract_kernel_version() -> str:
    """
    Get kernel version in the form of "major-minor" from current platform
    """
    pattern = re.compile(r'lve([0-9]+(\.[0-9]+)+)\.el')
    release = platform.release()
    try:
        version_num = pattern.search(release).group(1)
    except AttributeError:
        return '0.0.0'
    return version_num

def compare_versions_ge(first_version: str, second_version: Union[str, LooseVersion]) -> bool:
    """
    Comparing two versions using the greater or equal operator.
    """
    try:
        return LooseVersion(first_version) >= second_version
    except TypeError as e:
        raise SSAError(
            f'Unable to compare required versions: unexpected versions format "{first_version}" and "{second_version}"')

def is_kernel_version_supported():
    """
    General check of kernel support (IO throttling availability is required)
    """
    if not is_throttling_supported():
        # no throttling at all
        return True
    return is_io_throttling_detection_available()

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

@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, target_gid: int = None,
                   target_path='.', mask: int = 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()

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 SSAError(
                'Unable to execute required operation: permission issue')
    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 SSAError(
                'Unable to execute required operation: permission issue')
    yield
    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)

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

# for easy mocking in tests where
# we don't need singleton
IS_SINGLETON_ENABLED = True

def singleton(some_cls):
    class __Singleton:
        """
        A singleton wrapper class. Its instances would be created
        for each decorated class.
        """

def __init__(self, _cls):
            self._wrapped = _cls
            self._instance = None

def __call__(self, *args, **kwargs):
            """Returns a single instance of decorated class"""
            if self._instance is None or not IS_SINGLETON_ENABLED:
                self._instance = self._wrapped(*args, **kwargs)
            return self._instance

return __Singleton(some_cls)

Name	Type	Size	Owner/Group	Permission	Last Modified	Actions
.	dir	-	root/root	0755	2026-02-19 04:58:04
..	dir	-	root/root	0755	2026-02-02 09:10:51
__pycache__	dir	-	root/root	0755	2026-02-03 04:02:47
__init__.py	text/plain	171 B	root/root	0644	2025-12-10 11:09:40
constants.py	text/x-script.python	918 B	root/root	0644	2025-12-10 11:09:40
exceptions.py	text/x-script.python	1.26 KB	root/root	0644	2025-12-10 11:09:40
mailer.py	text/x-script.python	5.39 KB	root/root	0644	2025-12-10 11:09:40
tools.py	text/x-script.python	2.26 KB	root/root	0644	2025-12-10 11:09:40
utils.py	text/x-script.python	15.53 KB	root/root	0644	2025-12-10 11:09:40

Team Anon Force

https://t.me/Professor6T9x