← Back
Editing: 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) try: with user_context_func(pw_record.pw_uid, pw_record.pw_gid): fd = os.open(ini_file, os.O_RDONLY | os.O_NOFOLLOW) with os.fdopen(fd) 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 FileNotFoundError: continue 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) try: with user_context_func(pw_record.pw_uid, pw_record.pw_gid): fd = os.open(ini_file, os.O_RDONLY | os.O_NOFOLLOW) with os.fdopen(fd) 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 FileNotFoundError: continue except Exception: logger.warning('Failed to read %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)
Save File
Cancel