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 json from urllib.request import urlopen from schema import Schema from .parser import SSAConfigParser from .validation import validate from ..internal.exceptions import SSAError def load_parser() -> SSAConfigParser: """ Create parser and load configuration :return: parser """ conf = SSAConfigParser() conf.read_ssa_conf() return conf def load_validated_parser() -> SSAConfigParser: """ Create parser, load configuration, validate it :return: parser """ conf = load_parser() _ = validate(dict(conf.items())) return conf def load_configuration() -> dict: """ Load and validate configuration """ conf = load_parser() return validate(dict(conf.items())) def load_tunables(external_source: str, scheme: Schema) -> dict: """ Load values from remote JSON config or default values in case of troubles """ cfg_location = f"https://repo.cloudlinux.com/cloudlinux/sources/da/{external_source}" try: with urlopen(cfg_location) as cfg: return validate(json.load(cfg), scheme) except (OSError, json.JSONDecodeError, SSAError): return validate(dict(), scheme)	1.17 KB	2026-02-10 14:37
📁 __pycache__	-	2026-03-21 15:47
📄 parser.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 specific config parser: - has defaults - adds default values for ones missing in config file - allows duplicates, the latest value is loaded - performs value normalization during loading items: - to int - to float - to bool - 'correlation' is case-insensitive and allowed to be on, off, 1, 0, yes, no, true, false - loads default configuration in case of missing config file """ import logging import os import tempfile from configparser import ConfigParser, NoSectionError from typing import Callable, List from ..internal.constants import config_file default_section = 'ssa_conf' logger = logging.getLogger('ssa_configuration_parser') class SSAConfigParser(ConfigParser): """ SSA specific configparser implementation """ def __init__(self): defaults = { 'domains_number': 10, 'urls_number': 5, 'request_number': 10, 'requests_duration': 3.0, 'time': 1, 'correlation': True, 'correlation_coefficient': 0.8, 'ignore_list': '', 'summary_notification_enabled': True } self._comments = { 'domains_number': 'Size of TOP list for slow domains', 'urls_number': 'Size of TOP list for slow urls', 'request_number': 'The threshold value of slow requests number in the period of time to mark URL as a slow one', 'requests_duration': 'The threshold value of request duration in seconds', 'time': 'Period of time in hours required to analyze these requests', 'correlation': 'Flag to enable or disable correlation (On/Off)', 'correlation_coefficient': 'The threshold value of correlation coefficient', 'ignore_list': 'List of URLs or domains that should not be included in a daily report', 'summary_notification_enabled': 'Flag to enable or disable sending daily reports by e-mail (On/Off)' } self.ssa_defaults = defaults.copy() super().__init__(defaults, strict=False) @property def int_options(self) -> tuple: return 'domains_number', 'urls_number', 'request_number', 'time' @property def float_options(self) -> tuple: return 'requests_duration', 'correlation_coefficient' @property def bool_options(self) -> tuple: return 'correlation', 'summary_notification_enabled' def empty_value_fallback(self, meth: Callable, section: str, option: str) -> str: """ If the value is missing in config file, e.g. time= set the default one """ try: val = meth(section, option) except (ValueError, NoSectionError): val = meth(section, option, vars=self.defaults(), fallback=self.ssa_defaults[option]) return val def items(self, section: str = default_section, raw: bool = False, vars: dict = None) -> List[tuple]: """Return a list of (name, value) tuples for each option in a section. Adds special conversions. """ d = self._defaults.copy() try: d.update(self._sections[section]) except KeyError: if section != self.default_section: logger.info('Malformed ssa.conf, section "%s" is missed. ' 'Default will be applied ', section) self.restore_default_conf() orig_keys = list(d.keys()) def value_getter(option): if option in self.bool_options: return self.empty_value_fallback(self.getboolean, section, option) elif option in self.int_options: return self.empty_value_fallback(self.getint, section, option) elif option in self.float_options: return self.empty_value_fallback(self.getfloat, section, option) else: return d[option] return [(option, value_getter(option)) for option in orig_keys] def read_ssa_conf(self): """ Try to read configuration from file, load defaults if file not present """ try: with open(config_file): pass self.read(config_file) except FileNotFoundError: self.read_dict({default_section: self.defaults()}) def write_ssa_conf(self) -> None: """ Try to write current configuration to file, """ temp_file = None try: temp_file = tempfile.NamedTemporaryFile(mode='w', dir=os.path.dirname(config_file), delete=False) # Write the configuration to the temporary file self.write(temp_file, space_around_delimiters=False) # If the writing process completes without any errors, rename the temporary file to the config file os.rename(temp_file.name, config_file) finally: if temp_file is not None: temp_file.close() if os.path.exists(temp_file.name): os.remove(temp_file.name) def override(self, args: dict) -> None: """ Load options from the dict into the main section of the config. """ section = default_section if default_section in self.sections() else self.sections()[0] for option, value in args.items(): self.set(section, option, str(value)) def write(self, fp, space_around_delimiters=True): """ Parent method is overrided in order to: - do not print default section - call its own write_section method instead of parent one. """ if space_around_delimiters: d = ' {} '.format(self._delimiters[0]) else: d = self._delimiters[0] for section in self._sections: self.write_section(fp, section, self._sections[section].items(), d) def write_section(self, fp, section_name, section_items, delimiter): """ The difference between this one and the parent _write_section method is in writing the comments to every option also """ comment_prefix = self._comment_prefixes[0] if self._comment_prefixes else '#' fp.write('[{}]\n'.format(section_name)) for key, value in section_items: value = self._interpolation.before_write(self, section_name, key, value) if value is not None or not self._allow_no_value: value = delimiter + str(value).replace('\n', '\n\t') else: value = '' comment = self._comments.get(key, None) if comment is None: fp.write(f'{key}{value}\n\n') else: fp.write(f'{comment_prefix} {comment}\n{key}{value}\n\n') fp.write('\n') def restore_default_conf(self): """ Clear all sections in config parser, try to copy settings values if they have it. Add new default section and pass all settings to it. """ previous_settings = self.ssa_defaults for section in self.sections(): for option in self.options(section): if option in previous_settings: previous_settings[option] = self.get(section, option) self.remove_section(section) self.add_section(default_section) for option, value in previous_settings.items(): self.set(default_section, option, str(value))	7.76 KB	2026-02-10 14:37
📁 schemes	-	2026-03-21 15:45
📄 validation.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 validation schema for config file loaded values """ from schema import Schema, And, Optional, Use, SchemaError from ..internal.exceptions import SSAError configuration_schema = Schema({ Optional('domains_number'): And(int, lambda n: n >= 0, error='number of domains should be a positive integer'), Optional('urls_number'): And(int, lambda n: n > 0, error='number of URLs should be a positive integer'), Optional('requests_duration'): And(Use(lambda n: float(n)), lambda n: n > 0.0, error='request duration should be positive'), Optional('request_number'): And(int, lambda n: n > 0, error='number of requests should be a positive integer'), Optional('time'): And(int, lambda n: 0 <= n <= 12, error='time should be a positive integer between 1 and 12, or 0'), Optional('correlation'): And(bool, error='bad correlation value, should be On or Off'), Optional('correlation_coefficient'): And(Use(lambda n: float(n)), lambda n: 0.0 < n <= 1.0, error='correlation coefficient could not be negative or greater than 1'), Optional('ignore_list'): Use(lambda s: str(s)), Optional('summary_notification_enabled'): And(bool, error='bad notification value, should be On or Off'), }) def validate(data: dict, scheme: Schema = configuration_schema) -> dict: """ Validate given data. Returns validated data """ try: return scheme.validate(data) except SchemaError as e: raise SSAError(f'Invalid configuration: {str(e)}')	2.05 KB	2026-02-10 14:37

Terminal