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
📄 __init__.py	0 B	2026-02-10 14:37
📁 __pycache__	-	2026-03-21 15:47
📄 agent.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 contains classes implementing SSA Agent behaviour """ import atexit import json import logging from concurrent.futures import ThreadPoolExecutor from threading import current_thread from .internal.constants import agent_sock from .internal.exceptions import SSAError from .internal.utils import create_socket from .modules.processor import RequestProcessor # Maximum number of concurrent worker threads for handling requests. # Limits memory usage on high-traffic servers. MAX_WORKERS = 50 class SimpleAgent: """ SSA Simple Agent class """ def __init__(self): self.logger = logging.getLogger('agent') self.request_processor = RequestProcessor() self.executor = ThreadPoolExecutor(max_workers=MAX_WORKERS) atexit.register(self._shutdown) # start serving incoming connections self.listen() def _shutdown(self): """Gracefully shutdown the thread pool executor.""" self.executor.shutdown(wait=False) def listen(self) -> None: """ Start listening socket """ _socket = create_socket(agent_sock) while True: connection, address = _socket.accept() self.executor.submit(self.handle, connection) self.logger.debug('[ThreadPool] Submitted task') def handle(self, connection: 'socket object') -> None: """ Handle incoming connection :param connection: socket object usable to send and receive data on the connection """ fileobj = connection.makefile(errors='ignore') try: input_data = self.read_input(fileobj) self.request_processor.handle(input_data) except SSAError as e: self.logger.error('Handled exception in [%s]: %s', current_thread().name, str(e)) pass connection.close() def read_input(self, fileio: 'file object') -> dict: """ Read input data and return decoded json :param fileio: a file-like object providing read method """ data = fileio.read() self.logger.info('[%s] I received %i bytes: %s', current_thread().name, len(data.encode()), data) if data: return json.loads(data.strip(), strict=False) else: return {}	2.55 KB	2026-02-10 14:37
📁 autotracing	-	2026-03-21 15:45
📄 clos_ssa_ini.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 """ Shared constants and helpers for clos_ssa.ini file management. """ import pwd import re INI_FILE_NAME = 'clos_ssa.ini' # Base user locations for clos_ssa.ini files INI_USER_LOCATIONS_BASE = ( dict(path='/var/cagefs///etc/cl.php.d/alt-php[0-9][0-9]', user=lambda path: pwd.getpwnam(path.split('/')[4])), ) # Per-website locations used when website isolation is enabled INI_USER_LOCATIONS_WEBSITE_ISOLATION = ( dict(path='/var/cagefs///etc/cl.php.d/*/alt-php[0-9][0-9]', user=lambda path: pwd.getpwnam(path.split('/')[4])), ) # Directories to exclude EXCLUDE_DIR_PATHS = ( 'php44', 'php51', 'php52', 'php53', r'php\d+-imunify', 'php-internal' ) def is_excluded_path(dir_path: str) -> bool: """Check if given path is in exclude list.""" for pattern in EXCLUDE_DIR_PATHS: if re.search(pattern, dir_path): return True return False def extract_php_version(path: str) -> str: """Extract PHP version number from path (e.g., 'alt-php80' -> '80').""" match = re.search(r'php(\d{2})', path) return match.group(1) if match else None	1.29 KB	2026-02-10 14:37
📄 cloudlinux-autotracing.py #!/opt/cloudlinux/venv/bin/python3 -sbb # -- 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 """ Main cloudlinux-autotracing utility script """ from ssa.ssa_utils.run_autotracing import run if __name__ == '__main__': run()	382 B	2026-02-10 14:37
📄 cloudlinux-ssa-agent.py #!/opt/cloudlinux/venv/bin/python3 -sbb # -- 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 """ Main cloudlinux-ssa-agent utility script """ from ssa.ssa_utils.run_agent import run if __name__ == '__main__': run()	374 B	2026-02-10 14:37
📄 cloudlinux-ssa-manager.py #!/opt/cloudlinux/venv/bin/python3 -sbb # -- 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 """ Main cloudlinux-ssa-manager utility script """ from ssa.ssa_utils.run_manager import run if __name__ == '__main__': run()	378 B	2026-02-10 14:37
📁 configuration	-	2026-03-21 15:45
📄 db.py #!/opt/cloudlinux/venv/bin/python3 -sbb # coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2020 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # import os import contextlib import sqlite3 from datetime import datetime, timedelta from sqlalchemy import ( Column, Boolean, DateTime, Integer, String, create_engine, event, func, text ) from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.engine.reflection import Inspector from sqlalchemy.orm import Session from sqlalchemy.orm.session import close_all_sessions from sqlalchemy.exc import DatabaseError SSA_DB = '/var/lve/ssa.db' OLD_SSA_DB = SSA_DB + '.old' RETENTION_TIME_DAYS = 1 Base = declarative_base() class RequestResult(Base): """ Describes processed request stored in database file. E.g. { "timestamp": "1650008727", "url": "http://mydomain.com/index.php", "duration": 162077, "hitting_limits": false, "throttled_time": 0, "io_throttled_time": 0, "wordpress": true } Note: created_at, updated_at is saved in local TZ format """ __tablename__ = 'scrape_result' id = Column(Integer, primary_key=True) domain = Column(String, index=True, nullable=False) path = Column(String, index=True, nullable=False) timestamp = Column(Integer, nullable=False) duration = Column(Integer, nullable=False) is_slow_request = Column(Boolean, nullable=False) hitting_limits = Column(Boolean, nullable=False) throttled_time = Column(Integer, nullable=False) io_throttled_time = Column(Integer, nullable=False) wordpress = Column(Boolean, nullable=False) # Index on created_at speeds up cleanup_old_data queries significantly created_at = Column(DateTime(timezone=True), server_default=func.now(), index=True) updated_at = Column(DateTime(timezone=True), onupdate=func.now(), server_default=func.now()) def cleanup_old_data(engine): """ Removes outdated records from database, saving disk space. Runs VACUUM after deletion to reclaim disk space. """ n_days_ago = datetime.today() - timedelta(days=RETENTION_TIME_DAYS) with session_scope(engine) as session: session.query(RequestResult)\ .filter(RequestResult.created_at < n_days_ago)\ .delete() # Run VACUUM to reclaim disk space after deleting old records vacuum_database(engine) def vacuum_database(engine): """ Run VACUUM command to reclaim disk space after deletions. VACUUM must be run outside of a transaction, so we use raw connection. """ # Get a raw connection from the pool raw_conn = engine.raw_connection() original_isolation_level = raw_conn.isolation_level try: # Set isolation_level to None to enable autocommit mode (required for VACUUM) raw_conn.isolation_level = None cursor = raw_conn.cursor() cursor.execute("VACUUM") cursor.close() except Exception: # VACUUM failure is not critical - database will still work, just won't reclaim space pass finally: # Always restore original isolation_level before returning connection to pool raw_conn.isolation_level = original_isolation_level raw_conn.close() def ensure_created_at_index(engine): """ Ensure that created_at index exists on existing databases. This handles migration for databases created before the index was added. Uses CREATE INDEX IF NOT EXISTS for atomicity and race condition safety. """ INDEX_NAME = 'ix_scrape_result_created_at' TABLE_NAME = 'scrape_result' COLUMN_NAME = 'created_at' try: # Use engine.begin() for auto-commit on context exit (SQLAlchemy 1.x compatible) with engine.begin() as conn: # Use IF NOT EXISTS for atomic operation - safe for concurrent access conn.execute(text( f"CREATE INDEX IF NOT EXISTS {INDEX_NAME} ON {TABLE_NAME}({COLUMN_NAME})" )) except DatabaseError: # Database is malformed, table doesn't exist, or locked - skip silently # Index creation is best-effort optimization, not critical for operation pass def create_db_if_not_exist(engine): if not is_db_present(engine): Base.metadata.create_all(engine) def is_db_present(engine): if not os.path.isfile(SSA_DB): return False database_inspection = Inspector.from_engine(engine) tables = [table for table in database_inspection.get_table_names()] return len(tables) > 0 def setup_wal_mode(dbapi_con, con_record): dbapi_con.execute('PRAGMA journal_mode = WAL') def _setup_database(readonly): connection_string = f'file:{SSA_DB}' if readonly: connection_string = f'{connection_string}?mode=ro' creator = lambda: sqlite3.connect(connection_string, uri=True) engine = create_engine( 'sqlite:////', creator=creator, echo=False, ) event.listen(engine, 'connect', setup_wal_mode) create_db_if_not_exist(engine) # Ensure created_at index exists for existing databases (migration) if not readonly: ensure_created_at_index(engine) return engine def setup_database(readonly=False): return _setup_database(readonly) def restore_database(engine): """ Restore database by establish connections to old and new databases, merge data to new one if possible and delete old one. """ if os.path.exists(SSA_DB): # Closing all sessions to ensure that no sessions is using database during replacing close_all_sessions() os.replace(SSA_DB, OLD_SSA_DB) new_engine = setup_database() # Dispose of the existing engine to close and refresh all connections, ensuring it connects to the new database engine.dispose() old_engine = create_engine(f'sqlite:////{OLD_SSA_DB}') try: with session_scope(old_engine) as session_old, session_scope(new_engine) as session_new: # Check if old database is able to read and it make sense to try save unharmed data session_old.query(RequestResult).first() merge_unharmed_data_from_database(session_old, session_new) except DatabaseError: pass for path in (OLD_SSA_DB, SSA_DB + "-wal", SSA_DB + "-shm"): if os.path.exists(path): os.remove(path) def merge_unharmed_data_from_database(session_old, session_new): """ Scrape all unharmed records from malformed database and merge them into new database. """ offset = 0 batch_size = 10 while True: query = session_old.query(RequestResult).offset(offset).limit(batch_size) try: records_to_save = query.all() if not records_to_save: break for record in records_to_save: session_new.merge(record) except DatabaseError: for pos_in_batch in range(batch_size): try: record_to_save = query.offset(offset + pos_in_batch).first() if not record_to_save: break session_new.merge(record_to_save) except DatabaseError: pass session_new.commit() offset += batch_size def is_malformed_database(engine): """ Try integrity check of database file to see if it is malformed. If database unable to execute it, will also count as malformed. """ if os.path.exists(OLD_SSA_DB) and os.path.exists(SSA_DB): os.remove(OLD_SSA_DB) try: with session_scope(engine) as db: result = db.execute(text("PRAGMA integrity_check")) errors = result.fetchall() return errors[0][0] != 'ok' except DatabaseError: return True @contextlib.contextmanager def session_scope(engine) -> Session: """ Provide a transactional scope around a series of operations. """ session = Session(bind=engine) try: yield session session.commit() except: session.rollback() raise finally: session.close()	8.07 KB	2026-02-10 14:37
📁 internal	-	2026-03-21 15:45
📄 manager.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 classes implementing SSA Manager behaviour """ import json import logging import os import subprocess from contextlib import contextmanager from glob import iglob from secureio import disable_quota from typing import Tuple from .configuration import load_validated_parser, load_configuration from .internal.constants import flag_file from .internal.exceptions import SSAManagerError from .internal.utils import ssa_version from .modules.autotracer import AutoTracer from .modules.decision_maker import DecisionMaker from .clos_ssa_ini import ( INI_FILE_NAME, INI_USER_LOCATIONS_BASE, is_excluded_path, ) from .website_isolation import ( copy_inis_to_website_isolation_paths, remove_inis_from_website_isolation_paths, regenerate_inis_for_user as _regenerate_inis_for_user_isolation, ) class Manager: """ SSA Manager class. """ def __init__(self): self.logger = logging.getLogger('manager') self.ini_file_name = INI_FILE_NAME # Module search patterns for different distributions # CloudLinux uses lib64, Ubuntu uses lib/x86_64-linux-gnu self.module_patterns_with_usr = [ 'usr/lib64/php/modules/clos_ssa.so', # CloudLinux alt-php, cPanel 'usr/lib/x86_64-linux-gnu/php/modules/clos_ssa.so', # Ubuntu alt-php, cPanel ] self.module_patterns_no_usr = [ 'lib64/php/modules/clos_ssa.so', # CloudLinux Plesk 'lib/x86_64-linux-gnu/php/modules/clos_ssa.so', # Reserved for future use ] self.module_glob_pattern_directadmin = 'lib/php/extensions//clos_ssa.so' # DirectAdmin dynamic path self.wildcard_ini_locations = ( '/opt/alt/php[0-9][0-9]/link/conf', '/opt/cpanel/ea-php[0-9][0-9]/root/etc/php.d', '/opt/plesk/php/[0-9].[0-9]/etc/php.d', '/usr/local/php[0-9][0-9]/lib/php.conf.d', '/usr/share/cagefs/.cpanel.multiphp/opt/cpanel/ea-php[0-9][0-9]/root/etc/php.d', '/usr/share/cagefs-skeleton/usr/local/php[0-9][0-9]/lib/php.conf.d' ) # same as above, but ini which are writeable by users self.wildcard_ini_user_locations = INI_USER_LOCATIONS_BASE self.subprocess_errors = ( OSError, ValueError, subprocess.SubprocessError ) @staticmethod def response(args, kwargs) -> str: """ Form a success json response with given kwargs """ raw_response = {'result': 'success'} raw_response.update({k: v for k, v in kwargs.items()}) return json.dumps(raw_response) @property def _enabled(self) -> bool: """ Is SSA enabled """ return os.path.isfile(flag_file) @property def _restart_required_settings(self) -> set: """ Configuration settings required Request Processor restart """ return {'requests_duration', 'ignore_list'} @property def solo_filtered_settings(self) -> set: return {'correlation', 'correlation_coefficient', 'request_number', 'time', 'domains_number'} def _restart_required(self, settings: dict) -> set: """ SSA Agent requires restart in case of changing these configuration: - requests_duration - ignore_list """ return self._restart_required_settings.intersection(settings) def run_service_utility(self, command: str, check_retcode=False) -> subprocess.CompletedProcess: """ Run /sbin/service utility to make given operation with SSA Agent service :command: command to invoke :check_retcode: whether to run with check or not :return: subprocess info about completed process """ try: result = subprocess.run(['/sbin/service', 'ssa-agent', command], capture_output=True, text=True, check=check_retcode) self.logger.info(f'ssa-agent {command} succeeded') except subprocess.CalledProcessError as e: self.logger.error( 'SSA Agent %s failed with code %s: %s', str(e.cmd), str(e.returncode), str(e.stdout), extra={'cmd': e.cmd, 'retcode': e.returncode, 'stdout': e.stdout, 'stderr': e.stderr}) raise SSAManagerError( f'SSA Agent {e.cmd} failed with code {e.returncode}: {e.stdout or e.stderr}') except self.subprocess_errors as e: self.logger.error('Failed to run %s command for SSA Agent', str(command), extra={'err': str(e)}) raise SSAManagerError( f'Failed to run {command} for SSA Agent: {e}') return result def set_config(self, args: dict) -> str: """ Change SSA config and restart it. :args: dict to override current option values :return: JSON encoded result of the action """ config = load_validated_parser() config.override(args) try: config.write_ssa_conf() except OSError as e: self.logger.error('Failed to update SSA config file', extra={'err': str(e)}) raise SSAManagerError(f'Failed to update SSA config file: {e}') if self._restart_required(args): self.run_service_utility('restart', check_retcode=True) return self.response() def get_config(self) -> str: """ Get current SSA config. :return: JSON encoded current config """ full_config = load_configuration() return self.response(config=full_config) def get_ssa_status(self) -> str: """ Get current status of SSA. :return: JSON encoded current status """ status = 'enabled' if self._enabled else 'disabled' return self.response(ssa_status=status) def enable_ssa(self) -> str: """ Enable SSA: - add clos_ssa extension for each PHP version on server - add clos_ssa extension into cagefs for each user and each ver - start SSA Agent (if it is not already started) - restart Apache (etc.) and FPM, reset CRIU images - create flag_file indicating that SSA is enabled successfully :return: JSON encoded current status """ if not self._enabled: self.generate_inis() self.start_ssa_agent() self.create_flag() return self.get_ssa_status() def disable_ssa(self) -> str: """ Disable SSA: - remove clos_ssa extension for each PHP version on server - remove clos_ssa extension from cagefs for each user and each ver - stop SSA Agent - restart Apache (etc.) and FPM, reset CRIU images - remove flag_file indicating that SSA is enabled :return: JSON encoded current status """ if self._enabled: self.remove_clos_inis() self.stop_ssa_agent() self.remove_flag() return self.get_ssa_status() def get_stats(self) -> str: """ Get SSA statistics. Includes: - config values - version - SSA status (enabled\|disabled) - SSA Agent status (active\|inactive) :return: JSON encoded current statistics """ _config = {key: str(value).lower() for key, value in load_configuration().items()} return self.response( config=_config, version=ssa_version(), status='enabled' if self._enabled else 'disabled', agent_status=self.status_ssa_agent(), autotracing=AutoTracer().get_stats() ) def unused_dir_path(self, dir_path: str) -> bool: """ Check if directory path should be excluded. """ return is_excluded_path(dir_path) def _find_module_in_root(self, php_root: str, patterns: list) -> str: """ Search for clos_ssa.so module in php_root using a list of patterns. Returns the first found module path, or the first pattern as expected path if none exist. """ for pattern in patterns: module_path = os.path.join(php_root, pattern) if os.path.exists(module_path): return module_path # If no module found, return the first pattern as expected path return os.path.join(php_root, patterns[0]) if patterns else '' def get_module_path(self, ini_path: str) -> str: """ Determine the path to clos_ssa.so module based on ini_path. Returns the expected module path, or empty string if not found. """ # For CloudLinux alt-php: /opt/alt/phpXX/link/conf # CloudLinux: /opt/alt/phpXX/usr/lib64/php/modules/clos_ssa.so # Ubuntu: /opt/alt/phpXX/usr/lib/x86_64-linux-gnu/php/modules/clos_ssa.so if ini_path.startswith('/opt/alt/php') and '/link/conf' in ini_path: php_root = ini_path.split('/link/conf')[0] return self._find_module_in_root(php_root, self.module_patterns_with_usr) # For CageFS cPanel multiphp: /usr/share/cagefs/.cpanel.multiphp/opt/cpanel/ea-phpXX/root/etc/php.d # Check this BEFORE regular cPanel (more specific path) # This is typically a symlink/mount to cagefs-skeleton, check the skeleton location # /usr/share/cagefs/.cpanel.multiphp/opt/... → /usr/share/cagefs-skeleton/opt/... if ini_path.startswith('/usr/share/cagefs/.cpanel.multiphp/opt/cpanel/ea-php') and '/root/etc/php.d' in ini_path: skeleton_path = ini_path.replace('/usr/share/cagefs/.cpanel.multiphp', '/usr/share/cagefs-skeleton') php_root = skeleton_path.split('/etc/php.d')[0] return self._find_module_in_root(php_root, self.module_patterns_with_usr) # For cPanel EA4: /opt/cpanel/ea-phpXX/root/etc/php.d # cPanel uses lib64 on both CloudLinux and Ubuntu if ini_path.startswith('/opt/cpanel/ea-php') and '/root/etc/php.d' in ini_path: php_root = ini_path.split('/etc/php.d')[0] return self._find_module_in_root(php_root, self.module_patterns_with_usr) # For Plesk: /opt/plesk/php/X.X/etc/php.d # CloudLinux: /opt/plesk/php/X.X/lib64/php/modules/clos_ssa.so if ini_path.startswith('/opt/plesk/php/') and '/etc/php.d' in ini_path: php_root = ini_path.split('/etc/php.d')[0] return self._find_module_in_root(php_root, self.module_patterns_no_usr) # For CageFS skeleton DirectAdmin: /usr/share/cagefs-skeleton/usr/local/phpXX/lib/php.conf.d # Check this BEFORE regular DirectAdmin (more specific path) # DirectAdmin copies modules to CageFS skeleton with dynamic extension_dir # Path example: /usr/share/cagefs-skeleton/usr/local/php83/lib/php/extensions/no-debug-non-zts-20230831/clos_ssa.so if ini_path.startswith('/usr/share/cagefs-skeleton/usr/local/php') and '/lib/php.conf.d' in ini_path: php_root = ini_path.split('/lib/php.conf.d')[0] # Use glob to find module in dynamic extensions directory possible_paths = list(iglob(os.path.join(php_root, self.module_glob_pattern_directadmin))) if possible_paths: return possible_paths[0] return '' # For DirectAdmin: /usr/local/phpXX/lib/php.conf.d # DirectAdmin uses modules copied from alt-php to dynamic extension_dir # Path example: /usr/local/php83/lib/php/extensions/no-debug-non-zts-20230831/clos_ssa.so if ini_path.startswith('/usr/local/php') and '/lib/php.conf.d' in ini_path: php_root = ini_path.split('/lib/php.conf.d')[0] # Use glob to find module in dynamic extensions directory possible_paths = list(iglob(os.path.join(php_root, self.module_glob_pattern_directadmin))) if possible_paths: return possible_paths[0] return '' # For CageFS user paths: /var/cagefs/user/home/etc/cl.php.d/alt-phpXX # CageFS users see modules from skeleton, not from global /opt/alt/ if ini_path.startswith('/var/cagefs/') and '/etc/cl.php.d/alt-php' in ini_path: php_ver = ini_path.split('/etc/cl.php.d/alt-php')[1].split('/')[0] skeleton_root = f'/usr/share/cagefs-skeleton/opt/alt/php{php_ver}' return self._find_module_in_root(skeleton_root, self.module_patterns_with_usr) # Default fallback - return empty string if pattern not recognized return '' def existing_paths(self) -> Tuple[Tuple[int, int], str]: """ Generator of existing paths (matching known wildcard locations) for additional ini files Returns tuple of (uid, gid) and path. """ for location in self.wildcard_ini_locations: for dir_path in iglob(location): if self.unused_dir_path(dir_path): continue yield (0, 0), dir_path for location in self.wildcard_ini_user_locations: for dir_path in iglob(location['path']): if self.unused_dir_path(dir_path): continue try: pw_record = location['user'](dir_path) except: self.logger.info('Unable to get information about user ' 'owning %s directory (maybe he`s already terminated?), ' 'skip updating', dir_path) continue else: yield (pw_record.pw_uid, pw_record.pw_gid), dir_path @contextmanager def _user_context(self, 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 generate_single_ini(self, uid: int, gid: int, ini_path: str) -> None: """ Enable SSA extension for single ini_path (given) """ # Check if module exists before generating ini file module_path = self.get_module_path(ini_path) if not module_path: self.logger.warning('Cannot determine module path for %s, skipping ini generation', ini_path) return if not os.path.exists(module_path): self.logger.info('Module %s does not exist, skipping ini generation for %s', module_path, ini_path) # Remove ini file if it exists but module is missing ini_file_path = os.path.join(ini_path, self.ini_file_name) if os.path.exists(ini_file_path): try: with self._user_context(uid, gid): os.unlink(ini_file_path) self.logger.info('Removed ini file %s (module not found)', ini_file_path) except Exception as e: self.logger.warning('Failed to remove ini file %s: %s', ini_file_path, str(e)) return path = os.path.join(ini_path, self.ini_file_name) with self._user_context(uid, gid), \ disable_quota(), \ open(path, 'w') as ini: self.logger.info('Generating %s file...', path) ini.write('extension=clos_ssa.so\n') def generate_inis(self) -> None: """ Place clos_ssa.ini into each existing Additional ini path, including cagefs ones and per-website directories """ self.logger.info('Generating clos_ssa.ini files...') for (uid, gid), ini_path in self.existing_paths(): try: self.generate_single_ini(uid, gid, ini_path) except PermissionError: self.logger.info('Unable to update file %s, ' 'possible permission misconfiguration', ini_path) continue except Exception as e: self.logger.error('Exception on generating clos_ssa.ini: "%s", error: "%s"', ini_path, str(e)) continue # Also copy to per-website directories if website isolation is enabled copy_inis_to_website_isolation_paths(self._user_context) self.logger.info('Finished!') def find_clos_inis(self) -> Tuple[Tuple[int, int], str]: """ Generator function searching for clos_ssa.ini files in all existing Additional ini paths Returns tuple of (uid, gid) and path. """ for (uid, gid), ini_path in self.existing_paths(): for name in os.listdir(ini_path): if self.ini_file_name not in name: continue yield (uid, gid), os.path.join(ini_path, name) def remove_clos_inis(self) -> None: """ Remove all gathered clos_ssa.ini files, including per-website directories """ self.logger.info('Removing clos_ssa.ini files...') for (uid, gid), clos_ini in self.find_clos_inis(): try: with self._user_context(uid, gid): os.unlink(clos_ini) except Exception as e: self.logger.exception('Exception on removing clos_ssa.ini: "%s", error: "%s"', clos_ini, str(e)) continue # Also remove from per-website directories remove_inis_from_website_isolation_paths(self._user_context) self.logger.info('Finished!') def start_ssa_agent(self) -> None: """ Start SSA Agent service or restart it if it is accidentally already running """ agent_status = self.run_service_utility('status') if agent_status.returncode: self.run_service_utility('start', check_retcode=True) else: self.run_service_utility('restart', check_retcode=True) def stop_ssa_agent(self) -> None: """ Stop SSA Agent service or do nothing if it is accidentally not running """ agent_status = self.run_service_utility('status') if not agent_status.returncode: self.run_service_utility('stop', check_retcode=True) def status_ssa_agent(self) -> str: """ Get SSA Agent status: active or inactive """ try: self.run_service_utility('status', check_retcode=True) except SSAManagerError: return 'inactive' return 'active' def create_flag(self) -> None: """ Create a flag file indicating successful enablement """ with open(flag_file, 'w'): pass self.logger.info(f'Flag file {flag_file} created') def remove_flag(self) -> None: """ Remove a flag file indicating enablement """ try: os.unlink(flag_file) self.logger.info(f'Flag file {flag_file} removed') except OSError as e: self.logger.warning( f'Flag file {flag_file} removal failed: {str(e)}') def get_report(self) -> str: """ Get last report. :return: JSON encoded report """ report = DecisionMaker().get_json_report() return self.response(report) def regenerate_inis(self) -> None: """ Regenerates clos_ssa inis while SSA is enabled """ if self._enabled: self.generate_inis() def regenerate_inis_for_user(self, user: str) -> None: """ Regenerate clos_ssa.ini files for a specific user's website isolation directories. This is called by cagefsctl when enabling website isolation for a user. Only creates per-website ini files if base per-user ini exists. :param user: Username to regenerate ini files for """ if not self._enabled: return _regenerate_inis_for_user_isolation(user, self._user_context) def initialize_manager() -> 'Manager instance': """ Factory function for appropriate manager initialization :return: appropriate manager instance """ return Manager()	20.43 KB	2026-02-10 14:37
📁 modules	-	2026-03-21 15:45
📁 ssa_utils	-	2026-03-21 15:45
📄 website_isolation.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 """ Website isolation support for SSA (clos_ssa.ini) files. This module provides functions to manage clos_ssa.ini files in per-website directories when CageFS website isolation is enabled. """ import logging import os import subprocess from glob import iglob from secureio import disable_quota from .clos_ssa_ini import ( INI_FILE_NAME, INI_USER_LOCATIONS_BASE, INI_USER_LOCATIONS_WEBSITE_ISOLATION, is_excluded_path, extract_php_version, ) # Try to import website isolation check from securelve (cagefs) try: from clcagefslib.domain import is_website_isolation_allowed_server_wide, is_isolation_enabled except ImportError: def is_website_isolation_allowed_server_wide(): return False def is_isolation_enabled(user): return False logger = logging.getLogger(__name__) def copy_inis_to_website_isolation_paths(user_context_func) -> None: """ Copy clos_ssa.ini files from base user paths to per-website directories. :param user_context_func: Context manager function for user permissions """ if not is_website_isolation_allowed_server_wide(): return # Collect all base ini files: {(user, php_ver): (content, uid, gid)} base_ini_files = {} for location in INI_USER_LOCATIONS_BASE: for dir_path in iglob(location['path']): if is_excluded_path(dir_path): continue try: pw_record = location['user'](dir_path) except: logger.debug("Cannot get pw_record for path: %s", dir_path) continue ini_file = os.path.join(dir_path, INI_FILE_NAME) if not os.path.exists(ini_file): continue try: with open(ini_file) as f: php_ver = extract_php_version(dir_path) if php_ver: base_ini_files[(pw_record.pw_name, php_ver)] = ( f.read(), pw_record.pw_uid, pw_record.pw_gid ) except Exception: logger.warning('Failed to read %s', ini_file) continue if not base_ini_files: return created_ini = set() # Copy to per-website directories for location in INI_USER_LOCATIONS_WEBSITE_ISOLATION: for dir_path in iglob(location['path']): if is_excluded_path(dir_path): continue try: pw_record = location['user'](dir_path) except: logger.debug("Cannot get pw_record for path: %s", dir_path) continue if not is_isolation_enabled(pw_record.pw_name): continue php_ver = extract_php_version(dir_path) if not php_ver: continue key = (pw_record.pw_name, php_ver) if key not in base_ini_files: continue content, uid, gid = base_ini_files[key] ini_file = os.path.join(dir_path, INI_FILE_NAME) if not os.path.exists(os.path.dirname(ini_file)): continue try: with user_context_func(uid, gid), disable_quota(): with open(ini_file, 'w') as f: f.write(content) created_ini.add(pw_record.pw_name) except Exception as e: logger.warning('Failed to create %s: %s', ini_file, str(e)) continue for username in created_ini: _regenerate_user_website_isolation(username) def remove_inis_from_website_isolation_paths(user_context_func) -> None: """ Remove clos_ssa.ini files from all per-website directories. :param user_context_func: Context manager function for user permissions """ if not is_website_isolation_allowed_server_wide(): return removed_ini = set() for location in INI_USER_LOCATIONS_WEBSITE_ISOLATION: for dir_path in iglob(location['path']): if is_excluded_path(dir_path): continue try: pw_record = location['user'](dir_path) except: continue ini_file = os.path.join(dir_path, INI_FILE_NAME) if os.path.exists(ini_file): try: with user_context_func(pw_record.pw_uid, pw_record.pw_gid): os.unlink(ini_file) removed_ini.add(pw_record.pw_name) except Exception as e: logger.warning('Failed to remove %s: %s', ini_file, str(e)) continue for username in removed_ini: _regenerate_user_website_isolation(username) def _regenerate_user_website_isolation(user: str) -> None: """ Needed to terminate php processes to immediately apply clos_ssa.ini creation/deletion. """ try: subprocess.run( ["/usr/sbin/cagefsctl", "--site-isolation-regenerate", user], capture_output=True, check=True, text=True ) except subprocess.CalledProcessError as e: logger.warning( "Failed to trigger cagefsctl site isolation regeneration for %s: %s", user, e.stdout ) def regenerate_inis_for_user(user: str, user_context_func) -> None: """ Regenerate clos_ssa.ini files for a specific user's website isolation directories. This is called by cagefsctl when enabling website isolation for a user. Only creates per-website ini files if base per-user ini exists. :param user: Username to regenerate ini files for :param user_context_func: Context manager function for user permissions """ if not is_website_isolation_allowed_server_wide(): return logger.info('Regenerating clos_ssa.ini for user %s website isolation...', user) # First, collect existing base ini files for this user: {php_ver: (content, uid, gid)} base_ini_files = {} for location in INI_USER_LOCATIONS_BASE: for dir_path in iglob(location['path']): if is_excluded_path(dir_path): continue try: pw_record = location['user'](dir_path) if pw_record.pw_name != user: continue except: logger.debug("Cannot get pw_record for path: %s", dir_path) continue ini_file = os.path.join(dir_path, INI_FILE_NAME) if not os.path.exists(ini_file): continue try: with open(ini_file) as f: php_ver = extract_php_version(dir_path) if php_ver: base_ini_files[php_ver] = (f.read(), pw_record.pw_uid, pw_record.pw_gid) except Exception: logger.warning('Failed to create %s', ini_file) continue if not base_ini_files: logger.info('No base clos_ssa.ini files found for user %s', user) return # Copy to per-website directories for location in INI_USER_LOCATIONS_WEBSITE_ISOLATION: for dir_path in iglob(location['path']): if is_excluded_path(dir_path): continue try: pw_record = location['user'](dir_path) if pw_record.pw_name != user: continue except: logger.debug("Cannot get pw_record for path: %s", dir_path) continue php_ver = extract_php_version(dir_path) if not php_ver: continue if php_ver not in base_ini_files: continue content, uid, gid = base_ini_files[php_ver] ini_file = os.path.join(dir_path, INI_FILE_NAME) if not os.path.exists(os.path.dirname(ini_file)): continue try: with user_context_func(uid, gid), disable_quota(): with open(ini_file, 'w') as f: f.write(content) logger.info('Created %s', ini_file) except Exception as e: logger.warning('Failed to create %s: %s', ini_file, str(e)) continue logger.info('Finished regenerating for user %s!', user)	8.39 KB	2026-02-10 14:37

Terminal