Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:45
📄 __init__.py import os staging_flag_file = '/opt/cloudlinux/staging_mode' def staging() -> bool: """ Is staging enabled """ return os.path.isfile(staging_flag_file)	171 B	2026-02-10 14:37
📁 __pycache__	-	2026-03-21 15:47
📄 constants.py from . import staging agent_log = '/var/log/clos_ssa/agent.log' agent_sock = '/opt/alt/clos_ssa/run/ssa.sock' agent_file = '/usr/share/clos_ssa/agent.file' config_file = '/usr/share/clos_ssa/ssa.conf' storage_file = '/usr/share/clos_ssa/storage' flag_file = '/usr/share/clos_ssa/ssa_enabled' autotracing_stats_file = '/usr/share/clos_ssa/autotracer_stats.latest' manager_log = '/var/log/clos_ssa/manager.log' autotracing_log = '/var/log/clos_ssa/autotracing.log' report_path = '/usr/share/clos_ssa/reports' mail_template_location = '/usr/share/clos_ssa/mail_templates' if staging(): stat_server = '' # mute sending to ClickHouse by default for staging s_key = 'a28704900cb548d29004fca032417efe' s_postfix = '10' else: stat_server = 'stat-api.cloudlinux.com' s_key = 'f24458e2b44b4b6f88e08e9debeff6b2' s_postfix = '6' sentry_dsn = f'https://{s_key}@cl.sentry.cloudlinux.com/{s_postfix}'	915 B	2026-02-10 14:37
📄 exceptions.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 custom exception classes """ import json class SSAError(Exception): """ SSA internal exception class. Adds an 'error' status and stores information about occurred exception. Provides JSON encoded info about occurred exception. """ def __init__(self, message: str, *, flag: str = 'error', extra: dict = None): super().__init__(message) self.status = 'error' self.reason = message self.extra_data = extra self.type_flag = flag def __str__(self) -> 'json str': if self.type_flag == 'warning': msg = { 'result': 'success', 'warning': self.reason } else: msg = { 'result': self.reason } return json.dumps(msg) class SSAAgentError(SSAError): """ SSA Agent exception class """ class SSAManagerError(SSAError): """ SSA Manager exception class """ class SSAMailerError(SSAError): """ SSA Mailer exception class """	1.26 KB	2026-02-10 14:37
📄 mailer.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 SSA Mailer class and helper tools """ import logging import os import smtplib import subprocess from configparser import ConfigParser, SectionProxy from email.message import EmailMessage from socket import gethostname from clcommon.lib.cledition import is_cl_solo_edition from jinja2 import Environment, FileSystemLoader from clcommon.lib.network import get_ip_addr from .constants import mail_template_location from .exceptions import SSAMailerError from .utils import duration_cast, format_date def _get_server_ip(hostname): return get_ip_addr(hostname) or 'UNKNOWN' def render_report_table(report_data: dict) -> tuple: """ Render an HTML table for e-mail report and a full e-mail message """ env = Environment( loader=FileSystemLoader(mail_template_location), trim_blocks=True, lstrip_blocks=True ) env.filters['duration_cast'] = duration_cast table_template = env.get_template('report_table.html') report_table = table_template.render(domains=report_data['domains'], # TODO: [unification] make mail template same across all editions is_solo=is_cl_solo_edition( skip_jwt_check=True)) mail_template = env.get_template('mail.html') hostname = gethostname() mail = mail_template.render(date=format_date(report_data['date']), report_table=report_table, hostname=hostname, ip_addr=_get_server_ip(hostname)) return report_table, mail class Mailer: """ Class contains SSA e-mail send logic """ def __init__(self): self.logger = logging.getLogger('mailer') self._sender = None @property def mail_server(self) -> tuple: """ Local mail server address """ return ('localhost',) @property def sender(self) -> str: """ 'From' mail address """ return f"slow-site-analyzer@{gethostname()}" @staticmethod def read_template(name: str) -> SectionProxy: """ Get preformatted data for e-mail by name of template """ tmpl = f'{mail_template_location}/{name}.ini' if os.path.exists(tmpl): config = ConfigParser(interpolation=None) config.read(tmpl) return config['data'] raise SSAMailerError( f'Failed to find template {name} in {mail_template_location}') def _smtp_send(self, message: EmailMessage) -> None: """ Send preformatted e-mail via localhost SMTP """ self.logger.info('Try to send via smtp') try: with smtplib.SMTP(self.mail_server) as server: result = server.send_message(message) self.logger.info('Send result: %s', result) except smtplib.SMTPException as e: raise SSAMailerError(f'smtp mailing failed: {str(e)}') except (ConnectionError, OSError) as e: raise SSAMailerError(f'smtp connection failed: {str(e)}') def _console_send(self, message: EmailMessage) -> None: """ Send preformatted e-mail via sendmail utility """ self.logger.info('Try to send via sendmail utility') cmd = ["/usr/sbin/sendmail", "-t", "-oi"] try: subprocess.run(cmd, input=message.as_string(), capture_output=True, text=True, check=True) except (OSError, ValueError, subprocess.CalledProcessError) as e: raise SSAMailerError(f'sendmail utility failed with {str(e)}') def _send(self, mail: EmailMessage) -> None: """ Try to send mail via localhost smtp server, if fails -- try to use sendmail utility """ try: self._smtp_send(mail) except SSAMailerError as e: self.logger.error(str(e)) try: self._console_send(mail) except SSAMailerError as e: self.logger.error(str(e)) self.logger.critical( 'Both smtp and sendmail failed to send message to %s', mail['To']) def send_simple_mail(self, recipient: str, template: str = 'ssa_report', kwargs) -> None: """ Create a message from given template, including additional kwargs, and send it to given recipient """ self._send(self._message(recipient, template, kwargs)) def _message(self, recipient: str, template: str, *kwargs) -> EmailMessage: """ Form a message """ data = self.read_template(template) msg = EmailMessage() msg['Subject'] = data['subject'] % kwargs msg['From'] = self.sender msg['To'] = recipient msg.set_content(data['text'] % kwargs) msg.add_alternative(data['html'] % kwargs, subtype='html') self.logger.info('Generated mail --> %s', msg.as_string()) return msg	5.39 KB	2026-02-10 14:37
📄 tools.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 logging import signal import sys from typing import Union, Iterable, Callable from .utils import switch_schedstats, no_xray_active_tasks from ..modules.processor import RequestProcessor logger = logging.getLogger('tools') def log_on_receive(signum: int, frame) -> None: """ Signal handler to log received signal """ logger.info('Received sig: %i (%s)', signum, signum) def sigterm_received(signum: int, frame) -> None: """ Signal handler to dump collected statistics and switch off schedstats """ logger.info('Received sig: %i', signum) RequestProcessor().flush_buffer() logger.info('Dumped successfully after %i signal, exiting...', signum) # disable throttling detection kernel mechanism # if there is no X-Ray tracing tasks active if no_xray_active_tasks(): logger.info('No X-Ray tasks running, switching schedstats off') switch_schedstats(enabled=False) sys.exit(0) def register_a_signal(signum: int, sighandler: Callable) -> None: """ Register given sighandler for given signum """ try: signal.signal(signum, sighandler) logger.info('Registered %s', signum) except (OSError, ValueError) as e: logger.info('Skipped %s because of %s', signum, e) def register_sigterm(): """ Register a dump handler for SIGTERM signal """ register_a_signal(signal.SIGTERM, sigterm_received) def register_signals(signals: Union[int, Iterable] = None, handler: Callable = log_on_receive) -> None: """ Register given handler for given signal or range of signals If no signals given, add given handler for all available signals """ if signals is None: # register all available signals for sig in signal.Signals: register_a_signal(sig, handler) else: if isinstance(signals, int): # register single signal register_a_signal(signals, handler) else: # register range of signals for sig in signals: register_a_signal(sig, handler)	2.26 KB	2026-02-10 14:37
📄 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-21.el9'}) 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-21.el9" 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, backlog: int = 2048) -> 'socket object': """ Create world-writable socket in given sock_location or reuse existing one :param sock_location: socket address :param backlog: maximum number of queued connections (default 2048 for high-traffic servers) :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(backlog) else: sockobj = fromfd(3, AF_UNIX, SOCK_STREAM) sockobj.listen(backlog) 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: 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)	15.66 KB	2026-02-10 14:37

Terminal