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-27 02:38
📄 Readme.md # Overview This is a short overview of what is going on in this tool. ## Idea cloudlinux-reconfigure automatically updates system configuration according to "specification" of different editions. Right now "specifications" are defined in `spec/` package as python scripts with the only method `provision(...)`. There are some limitations that developers should stick to: - `provision` method must be able to configure server no matter what edition was installed previously. This means, e.g. if we make transition between `Admin` and `Solo` the latter is responsible for erasing `cloudlinux-admin-meta` package before installation of the `cloudlinux-solo-meta` ## Future Here are couple notes for the future ### Provisioning responsibilities Maybe we can add additional `deprovision` method which will be responsible for the cleanup of the edition that we are leaving. This will shift the responsibility e.g. of removing meta packages to the module which performed provisioning previously. On the other hand, it will increase migration time. ### Project structure We can consider grouping actions not by the `edition`, but by the `project`, e.g. ``` spec/ accelerate_wp/ provision(edition, ...) deprovision(edition, ...) cagefs/ provision(edition, ...) deprovision(edition, ...) lve-utils/ provision(edition, ...) deprovision(edition, ...) ``` ### Ansible? We can use ansible playbooks instead of python to describe provisioning. User-friendly playbooks with wide possibilities and also options like dry-run which allow us to draw nice-looking progress for user.	1.61 KB	2026-02-17 10:40
📄 __init__.py # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # https://cloudlinux.com/docs/LICENCE.TXT # import logging import inspect import os import sys from dataclasses import dataclass from clcommon.utils import is_ubuntu, run_command, ExternalProgramFailed def _get_default_args(func): """ Get dict with key-value pairs of default argument values """ signature = inspect.signature(func) return { k: v.default for k, v in signature.parameters.items() if v.default is not inspect.Parameter.empty } def _get_args_dict(func, args, kwargs): """ Get all func arguments including defaults, args and kwargs """ func_args = {} func_args.update(_get_default_args(func)) func_args.update(dict(zip(func.__code__.co_varnames, args))) func_args.update(kwargs) return func_args def task(name): """ Simple wrapper that logs execution begin and end """ def decorator(func): def wrapped_func(args, kwargs): func_args = _get_args_dict(func, args, kwargs) logging.info(name.format(func_args)) try: func(args, kwargs) except Exception: logging.exception('FAILED to %s', name.format(func_args)) raise return wrapped_func return decorator def setup_logger(logger_name, log_file): """ Logger setup for all modules """ app_logger = logging.getLogger(logger_name) app_logger.setLevel(logging.DEBUG) try: fh = logging.FileHandler(log_file) except IOError: pass else: fh.formatter = logging.Formatter( '[%(levelname)s \| %(asctime)s]: %(message)s') fh.level = logging.DEBUG app_logger.addHandler(fh) for handler in app_logger.handlers: if isinstance(handler, logging.StreamHandler) and handler.stream == sys.stderr: handler.formatter = logging.Formatter('[cloudlinux-customizer]: %(message)s') handler.level = logging.INFO return app_logger @dataclass class RunResult: """ Describes process call result """ exitcode: int stdout: str stderr: str def run(cmd, fail_ok=False): """ Just a wrapper above run_command that provides some cool logging of what happens in subprocess """ logging.debug('Running %s', ' '.join(cmd)) env_data = os.environ.copy() # Force C locale to ensure consistent output across all systems env_data['LC_ALL'] = 'C' if is_ubuntu(): # Prevent debconf from asking questions env_data['DEBIAN_FRONTEND'] = 'noninteractive' try: exitcode, stdout, stderr = run_command( cmd, env_data=env_data, return_full_output=True, ) logging.debug('Stdout: %s', stdout) logging.debug('Stderr: %s', stderr) return RunResult(exitcode, stdout, stderr) except ExternalProgramFailed as e: if not fail_ok: logging.error(str(e)) return RunResult(-1, None, None)	3.09 KB	2026-02-17 10:40
📁 __pycache__	-	2026-03-27 02:38
📄 awp.py # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # https://cloudlinux.com/docs/LICENCE.TXT # import logging from clconfigure import run, task from clcommon.clwpos_lib import configure_accelerate_wp from clcommon.const import Feature from clcommon.cpapi import is_panel_feature_supported from clcommon.utils import is_ubuntu STATUS_ALLOWED_FOR_ALL = 'allowed_for_all' SUITE_AWP_FREE = 'accelerate_wp' @task("Enabling Accelerate WP") def enable_accelerate_wp_free(): """ Set AccelerateWP free suite """ if is_ubuntu(): logging.warning('AccelerateWP setup upon new server is skipped: Ubuntu is not supported') return if not is_panel_feature_supported(Feature.WPOS): logging.warning('AccelerateWP setup upon new server is skipped: Control Panel is not supported') return try: configure_accelerate_wp() except Exception: logging.exception('AccelerateWP configuration upon new server setup failed') # see reasoning in CLPRO-1924 logging.info('Turning off notifications and automatic WordPress plugin installation') run(['/usr/bin/cloudlinux-awp-admin', 'set-options', '--smart-advice-notifications', 'off', '--smart-advice-wordpress-plugin', 'off', '--smart-advice-reminders', 'off']) @task("Changing Accelerate WP suite {suite} status to {status}") def set_suite_status(suite: str, status: str): """ Set AccelerateWP suite to desired state. """ if status == STATUS_ALLOWED_FOR_ALL: run(['/usr/bin/cloudlinux-awp-admin', 'set-suite', '--suites', suite, '--allowed-for-all']) else: raise NotImplementedError(f"Currently not implemented status {status}")	1.75 KB	2026-02-17 10:40
📄 cagefs.py # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2024 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # https://cloudlinux.com/docs/LICENCE.TXT # import functools import logging import os from clcommon.const import Feature from clcommon.cpapi import is_panel_feature_supported from clconfigure import run, task from clconfigure.packages import install_package CAGEFSCTL = "/usr/sbin/cagefsctl" def _skip_without_cagefs(func): @functools.wraps(func) def decorator(args, kwargs): if not is_panel_feature_supported(Feature.CAGEFS): logging.debug("Feature 'CageFS' is not supported. Skipping...") return if not os.path.exists(CAGEFSCTL): logging.debug("CageFS is not installed. Skipping...") return return func(args, **kwargs) return decorator @_skip_without_cagefs @task("Cleaning CageFS environment") def clean_cagefs_without_lve_environment(): if is_panel_feature_supported(Feature.LVE): logging.debug("LVE feature is supported. Skipping...") return result = run([CAGEFSCTL, "--clean-without-lve-environment"]) if result.exitcode != 0: logging.warning("Unable to clean non-LVE CageFS enviroment. stdout: %s, stderr: %s", result.stdout, result.stderr) @_skip_without_cagefs @task("Reconfiguring CageFS") def reconfigure_cagefs(): install_package("cagefs", reinstall=True) _rebuild_apache_directadmin() def _rebuild_apache_directadmin(): da_bin = "/usr/local/directadmin/directadmin" if not os.path.exists(da_bin): return result = run([da_bin, "build", "apache"]) if result.exitcode != 0: logging.warning("Error during rebuilding Apache. stdout: %s, stderr: %s", result.stdout, result.stderr)	1.81 KB	2026-02-17 10:40
📁 cli	-	2026-03-27 02:38
📄 clpro.py # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # https://cloudlinux.com/docs/LICENCE.TXT # import logging from clconfigure import run, task @task("Disabling autotracing") def disable_autotracing(): result = run(['/usr/sbin/cloudlinux-autotracing', 'disable', '--all']) if result.exitcode != 0: logging.warning('Unable to disable autotracing. stdout: %s, stderr: %s', result.stdout, result.stderr) @task("Reconfiguration of Shared Pro exclusive features") def trigger_manage_clplus(): result = run(['/usr/share/cloudlinux/cl_plus/manage_clplus']) if result.exitcode != 0: logging.warning('Unable to reconfigure Shared Pro-only features. stdout: %s, stderr: %s', result.stdout, result.stderr)	868 B	2026-02-17 10:40
📄 files.py # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2024 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # https://cloudlinux.com/docs/LICENCE.TXT # import subprocess from pathlib import Path from clcommon.lib.consts import CL_EDITION_FILE_FOR_USERS def remove_cl_edition_file(): """ Remove the file that contains the CloudLinux edition for end-user access. See CLOS-2411 for more info. """ Path(CL_EDITION_FILE_FOR_USERS).unlink(missing_ok=True) def update_cl_edition_file(): """ Populate edition cache again with new edition. """ with subprocess.Popen( ['/usr/bin/cldetect', '--detect-edition'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) as proc: stdout, _ = proc.communicate() if proc.returncode != 0: return with open(CL_EDITION_FILE_FOR_USERS, 'wb') as f: f.write(stdout)	933 B	2026-02-17 10:40
📄 limits.py # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # https://cloudlinux.com/docs/LICENCE.TXT # import logging import pathlib from clconfigure import run, task from clconfigure.packages import install_package from clconfigure.services import ( STATE_FAILED, STATE_MASKED, STATE_RUNNING, STATE_STOPPED, STATE_UNMASKED, STATE_ENABLED, STATE_DISABLED, get_service_state, set_service_state ) def initialize_lvestats(): # NOTE(vlebedev): We need to trigger 'lve-stats' scriptlets to properly initialized it's database stuff. install_package('lve-stats', reinstall=True) enable_lve_services() @task("Changing default limits state to '{desired_state}'") def set_default_limits_state(desired_state): """ Brings default limits to given state (unlimited \| default). May be executed more that once, does't crash on future calls """ if desired_state == 'unlimited': run(['lvectl', 'set', 'default', '--unlimited']) else: raise RuntimeError('Another states temporary unavailable') def disable_lve_services(): """ Turn off all lve-related services. """ for state in [STATE_STOPPED, STATE_DISABLED, STATE_MASKED]: for service in ['lve', 'lve_namespaces', 'lvestats', 'lvectl']: set_service_state(state, service) def enable_lve_services(): """ Turn on all lve-related services. """ for state in [STATE_UNMASKED, STATE_RUNNING, STATE_ENABLED]: for service in ['lve', 'lve_namespaces', 'lvestats', 'lvectl']: set_service_state(state, service) @task("Unloading lve module") def unload_lve_module(): """ Restart services that still use lve device and unload lve module then """ if not pathlib.Path('/dev/lve').exists(): # already unloaded return # Check if lve device is held by any process res = run(['lsof', '/dev/lve']) res_stdout_lines = res.stdout.split() if res.stdout is not None else [] services = [ service for service in ['mysqld', 'mysql', 'mariadb', 'httpd', 'apache2'] if any(line.startswith(service) for line in res_stdout_lines) ] if services: run(['systemctl', 'restart', *services]) res = run(['modprobe', '-rf', 'kmodlve']) if pathlib.Path('/sys/module/kmodlve').exists(): logging.warning('Failed to unload the lve module. Please reboot the server.') def apply_workaround_lve_failed(): """ Apply workaround for the case When `systemctl stop lve` makes it transition to "failed" state instead of "inactive" """ if get_service_state('lve') != STATE_FAILED: return set_service_state(STATE_UNMASKED, 'lve') set_service_state(STATE_RUNNING, 'lve') set_service_state(STATE_STOPPED, 'lve') set_service_state(STATE_MASKED, 'lve')	2.86 KB	2026-02-17 10:40
📄 linksafe.py # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # https://cloudlinux.com/docs/LICENCE.TXT # from concurrent.futures import ThreadPoolExecutor from typing import Iterator from clcommon.utils import is_ubuntu from clconfigure import task, run from clconfigure.packages import install_package # Files of the packages listed below are not supposed # to allow creation of symlinks to them PACKAGES_TO_EXCLUDE = { 'alt-python27-cllib', 'alt-php-xray', 'alt-php-ssa', } def initialize_linksafe(): # NOTE(vlebedev): Also we need to trigger linksafe scriptlets to properly initialize relevant UNIX groups. # Otherwise this problem may occur: # https://cloudlinux.zendesk.com/hc/en-us/articles/ # 115004139173-What-does-Error-failed-to-create-symlink-13-Permission-denied-mean install_package('cloudlinux-linksafe', reinstall=True) assign_linksafe_group() @task("Assigning linksafe group to files of 'alt-' packages residing under /opt/alt") def assign_linksafe_group(concurrency: int = 5): with ThreadPoolExecutor(max_workers=concurrency) as tp: for package in _installed_packages_iter( "alt-php", "alt-python", "alt-ruby", ): if package not in PACKAGES_TO_EXCLUDE: tp.submit(run, ['/usr/bin/cl-linksafe-apply-group', package]) def _installed_packages_iter(patterns: str) -> Iterator[str]: if is_ubuntu(): result = run(["dpkg-query", "-W", "-f", "${Package}\n", patterns]) else: result = run(["rpm", "-qa", *patterns]) lines = result.stdout.strip().splitlines() for line in lines: yield line	1.75 KB	2026-02-17 10:40
📄 lve.py # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2024 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # https://cloudlinux.com/docs/LICENCE.TXT # import logging import pathlib from clconfigure import run, task @task("Unloading lve module") def unload_lve_module(): """ Restart services that still use lve device and unload lve module then """ # Check if lve device is held by any process res = run(['lsof', '/dev/lve']) res_stdout_lines = res.stdout.split() if res.stdout is not None else [] services = [ service for service in ['mysqld', 'mariadb', 'httpd'] if any(line.startswith(service) for line in res_stdout_lines) ] if services: run(['systemctl', 'restart', *services]) res = run(['modprobe', '-rf', 'kmodlve']) if pathlib.Path('/sys/module/kmodlve').exists(): logging.warning('Failed to unload the lve module. Please reboot the server.')	965 B	2026-02-17 10:40
📄 packages.py # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # https://cloudlinux.com/docs/LICENCE.TXT # import logging import os import re from clcommon.utils import is_ubuntu from clconfigure import task, run STATE_REMOVED = 'uninstalled' STATE_INSTALLED = 'installed' @task("Erasing package '{package_name}'") def erase_package(package_name): if is_ubuntu(): run(['dpkg', '--purge', '--force-depends', package_name]) else: run(['rpm', '--erase', '--nodeps', package_name]) current_state = get_package_state(package_name) logging.info("Checking package '%s' state again... package is now in state '%s'", package_name, current_state) if STATE_REMOVED != current_state: raise RuntimeError("Failed to do required actions") @task("Changing package '{package_name}' state to '{desired_state}'") def set_package_state(desired_state, package_name): """ Brings package to given state (installed \| uninstalled). May be executed more than once, doesn't crash on future calls """ current_state = get_package_state(package_name) logging.debug("Checking package '%s' state... package is in state '%s'", package_name, current_state) if desired_state == current_state: logging.debug("No actions needed for package '%s'", package_name) return logging.info('State does not match target, changing...') if desired_state == STATE_REMOVED: if is_ubuntu(): run(['apt-get', 'purge', '-y', package_name]) else: run(['yum', 'remove', '-y', package_name]) elif desired_state == STATE_INSTALLED: if is_ubuntu(): run(['apt-get', 'install', '-y', package_name]) else: run(['yum', 'install', '-y', package_name]) else: raise NotImplementedError() current_state = get_package_state(package_name) logging.info("Checking package '%s' state again... package is now in state '%s'", package_name, current_state) if desired_state != current_state: raise RuntimeError("Failed to do required actions") @task("(Re)installing package '{package_name}'") def install_package(package_name: str, reinstall: bool = False): state = get_package_state(package_name) if state == STATE_INSTALLED and reinstall: cmd = ['apt-get' if is_ubuntu() else 'yum', 'reinstall', '-y', package_name] elif state == STATE_REMOVED: cmd = ['apt-get' if is_ubuntu() else 'yum', 'install', '-y', package_name] else: return run(cmd) def get_package_state(package): """ Gets current package state. Either installed or uninstalled. """ if is_ubuntu(): resp = run(['dpkg', '-s', package]) if resp.exitcode == 0 and 'Status: install ok installed' in resp.stdout: return STATE_INSTALLED return STATE_REMOVED resp = run(['rpm', '-qv', package]) return STATE_INSTALLED if resp.exitcode == 0 else STATE_REMOVED def get_list_installed_alt_phps(): """ Gets installed alt-phps return: list ['php44', 'php54', 'php80'] """ return [php for php in os.listdir('/opt/alt') if re.match(r'^php\d+$', php)]	3.18 KB	2026-02-17 10:40
📄 services.py # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # https://cloudlinux.com/docs/LICENCE.TXT # import logging from clconfigure import task, run STATE_MASKED = 'masked' STATE_STOPPED = 'stopped' STATE_UNMASKED = 'unmasked' STATE_RUNNING = 'running' STATE_FAILED = 'failed' STATUS_STOPPED = 'inactive' STATUS_RUNNING = 'active' STATE_ENABLED = 'enabled' STATE_DISABLED = 'disabled' def get_service_state(service, kind='is-active'): """ Get information about systemctl service state (either activity or enability) """ result = run(['systemctl', kind, service + '.service']) state = result.stdout.strip() return state @task("Changing service '{service_name}' state to '{desired_state}'") def set_service_state(desired_state, service_name): """ Brings service to given state. May be executed more that once, does't crash on future calls """ current_state = get_service_state(service_name, kind='is-enabled') current_status = get_service_state(service_name) logging.debug("Checking service '%s' state... package is in state='%s' running='%s'" "", service_name, current_state, current_status) action = None if desired_state == STATE_MASKED and current_state != STATE_MASKED: action = 'mask' elif desired_state == STATE_UNMASKED and current_state == STATE_MASKED: action = 'unmask' elif desired_state == STATE_STOPPED and current_status != STATUS_STOPPED: action = 'stop' elif desired_state == STATE_DISABLED and current_state != STATE_DISABLED: action = 'disable' elif desired_state == STATE_RUNNING and current_status != STATUS_RUNNING: action = 'start' elif desired_state == STATE_ENABLED and current_state != STATE_ENABLED: action = 'enable' if action is None: logging.debug("No actions needed for service '%s'", service_name) else: run(['systemctl', action, service_name + '.service']) current_state = get_service_state(service_name) current_status = get_service_state(service_name) logging.debug("Checking service '%s' state again... service now is in state='%s' running='%s'" "", service_name, current_state, current_status)	2.29 KB	2026-02-17 10:40
📁 spec	-	2026-03-27 02:38
📄 watcher.py # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # https://cloudlinux.com/docs/LICENCE.TXT # import logging import os import re import subprocess import time from clcommon.lib.cledition import CLEditionDetectionError from lve_utils import PKG_VERSION PKG_VERSION_TINY = re.sub(r"\.el\w(h?)\.", ".elX.", PKG_VERSION) # https://cl.sentry.cloudlinux.com/settings/cloudlinux_os/projects/userland SENTRY_DSN = "https://9713d1296f804031b058b8f2d789d7ac:8ddacae32d8246cf8b25cf826bf3fc0a@cl.sentry.cloudlinux.com/12" LAST_KNOWN_EDITION_CACHE_PATH = "/var/lve/.edition.previous" SYSTEMID_FILE_PATH = "/etc/sysconfig/rhn/systemid" # how much time should past after registration # before we assume that server is "old" _SECONDS_IN_DAY = 24 * 60 * 60 SYSTEMID_NEW_TIME = 3 * _SECONDS_IN_DAY def _reconfigure_server(current_edition, is_pre_check): """ Run reconfiguration script """ subprocess.run( ["/usr/sbin/cloudlinux-customizer", "preconfigure" if is_pre_check else "reconfigure", "-t", current_edition], check=True, text=True, stderr=subprocess.STDOUT, ) # _on_* handlers below are different for the future additional actions # that will be added with edition switch support def _on_edition_changed(previous_edition, current_edition, is_pre_check): """ Called when saved edition does not match current one. """ _reconfigure_server(current_edition, is_pre_check) def _on_server_newly_registered(current_edition, is_pre_check): """ Called when saved server id detected as one registered recently """ _reconfigure_server(current_edition, is_pre_check) def _get_registration_time(): """ Assume that last registration time is the date of systemid modification. """ try: return os.path.getmtime(SYSTEMID_FILE_PATH) except FileNotFoundError: return None def _get_last_check_time(): """ Assume that last check time is equal to the date of last cache modification. """ try: return os.path.getmtime(LAST_KNOWN_EDITION_CACHE_PATH) except FileNotFoundError: return None def check(new_edition: str, is_pre_check=False): """ Compares current edition with latest saved in file and @param new_edition: provides information about new edition that server is using Depending on is_pre_check this edition can either be already applied or just planning to be installed @param is_pre_check: True means that current check is executed BEFORE new jwt is actually saved in jwt.token file False means that registration is complete and jwt.token file was updated with new edition token """ try: with open(LAST_KNOWN_EDITION_CACHE_PATH, encoding="utf-8") as f: last_edition = f.read() except FileNotFoundError: last_edition = None registration_time = _get_registration_time() is_edition_changed = last_edition is not None and last_edition != new_edition # system_id file exists and it is younger than 3d is_registration_fresh = registration_time is not None and registration_time >= (time.time() - SYSTEMID_NEW_TIME) last_check_time = _get_last_check_time() # edition cache file does not exist yet # or system_id exists but edition cache was modified before registration # and edition (RE-REGISTRATION CASE) is changing is_last_check_before_registration = (last_check_time is None) or ( registration_time is not None and last_check_time < registration_time and is_edition_changed ) logging.info( "Checking edition registration data: registration time=%s, last edition=%s, new edition=%s", str(registration_time), str(last_edition), str(new_edition), ) if is_last_check_before_registration and is_registration_fresh: logging.info( "New server registration detected, last check time=%s, registration time=%s", str(last_check_time), str(registration_time), ) _on_server_newly_registered(new_edition, is_pre_check) elif is_edition_changed: logging.info("Edition change detected, last edition=%s, new edition=%s", str(last_edition), str(new_edition)) _on_edition_changed(last_edition, new_edition, is_pre_check) else: logging.info("Nothing to do, requested edition=%s already configured", str(new_edition)) # keep old edition in file until post hook executes if not is_pre_check: logging.info('Saving new edition="%s"', str(new_edition)) save_edition(edition=new_edition) def save_edition(edition): """ Stores current edition value in cache file """ try: with open(LAST_KNOWN_EDITION_CACHE_PATH, "w", encoding="utf-8") as f: f.write(edition) except CLEditionDetectionError as e: logging.warning("Unable to detect current edition, error=%s", e)	4.94 KB	2026-02-17 10:40

Terminal