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-19 15:49
📄 __init__.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/LICENCE.TXT #	248 B	2026-02-25 10:26
📁 __pycache__	-	2026-03-19 15:49
📄 admin_config.py # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2025 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT """ Admin-level configuration for website isolation. Stores path constants for global feature flags and per-user mode directories, and provides helpers that implement the CageFS-style per-user directory layout: <basedir>/<prefix>/<username> where prefix is ``uid % 100`` zero-padded to two digits and each user entry is an empty marker file. Directory permissions: ``0o751`` (``drwxr-x--x``). File permissions: ``0o644`` (``-rw-r--r--``). """ import os from ..fs import get_user_prefix WEBSITE_ISOLATION_MARKER = "/opt/cloudlinux/flags/enabled-flags.d/website-isolation.flag" WEBSITE_ISOLATION_AVAILABLE_MARKER = ( "/opt/cloudlinux/flags/available-flags.d/website-isolation.flag" ) # Per-user mode directories (CageFS-style layout): # <dir>/<prefix>/<username> (empty marker file, prefix = uid % 100) # In "Allow All" mode this directory exists and contains denied users (exceptions). ISOLATION_DENIED_DIR = "/etc/cagefs/site-isolation.users.denied" # In "Deny All" mode this directory exists and contains allowed users (exceptions). ISOLATION_ALLOWED_DIR = "/etc/cagefs/site-isolation.users.allowed" DIR_MODE = 0o751 # drwxr-x--x (matches CageFS convention) FILE_MODE = 0o644 # -rw-r--r-- def user_in_dir(dirpath: str, username: str) -> bool: """Check whether username has a marker file inside dirpath.""" prefix = get_user_prefix(username) return os.path.isfile(os.path.join(dirpath, prefix, username)) def add_user_to_dir(dirpath: str, username: str) -> None: """Create an empty marker file for username inside dirpath.""" prefix = get_user_prefix(username) prefix_dir = os.path.join(dirpath, prefix) os.makedirs(prefix_dir, mode=DIR_MODE, exist_ok=True) filepath = os.path.join(prefix_dir, username) open(filepath, "w").close() os.chmod(filepath, FILE_MODE) def remove_user_from_dir(dirpath: str, username: str) -> None: """Remove the marker file for username inside dirpath. Also cleans up the now-empty prefix sub-directory, if applicable. """ prefix = get_user_prefix(username) filepath = os.path.join(dirpath, prefix, username) try: os.remove(filepath) except (IOError, OSError): pass # Remove empty prefix directory prefix_dir = os.path.join(dirpath, prefix) try: os.rmdir(prefix_dir) except (IOError, OSError): pass def list_users_in_dir(dirpath: str) -> set[str]: """Return the set of usernames that have marker files inside dirpath.""" users: set[str] = set() try: for entry in os.listdir(dirpath): prefix_path = os.path.join(dirpath, entry) if not os.path.isdir(prefix_path): continue for username in os.listdir(prefix_path): if os.path.isfile(os.path.join(prefix_path, username)): users.add(username) except (FileNotFoundError, OSError): pass return users	3.07 KB	2026-02-25 10:26
📄 config.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/LICENCE.TXT # import dataclasses import json import os import pathlib from pathlib import Path from clcommon import ClPwd from clcagefslib.io import write_via_tmp DOCROOTS_ISOLATED_BASE = Path("/var/clwebisolate/users") @dataclasses.dataclass class UserConfig: enabled_websites: list[str] = dataclasses.field(default_factory=list) def load_user_config(user: str) -> UserConfig: path = _get_user_config_path(user) if not path.exists(): return UserConfig() try: return UserConfig(**json.loads(path.read_text())) except json.JSONDecodeError: return UserConfig() def save_user_config(user: str, config: UserConfig \| None) -> None: path = _get_user_config_path(user) if not config or not config.enabled_websites: path.unlink(missing_ok=True) return path.parent.mkdir(parents=True, exist_ok=True) write_via_tmp(str(path.parent), str(path), json.dumps(dataclasses.asdict(config), indent=4)) # Set ownership and permissions so user can read their config (but not write) # root:user_group with 0o640 = owner (root) can read/write, group (user_group) can read only # User can read via group membership but cannot write because group has no write permission pw = ClPwd().get_pw_by_name(user) os.chown(path, os.getuid(), pw.pw_gid) # root:user_group os.chmod(path, 0o640) # rw-r----- def _get_user_config_path(user: str) -> pathlib.Path: pw = ClPwd().get_pw_by_name(user) directory = DOCROOTS_ISOLATED_BASE / str(pw.pw_uid) return Path(directory / f"{pw.pw_uid}.json")	1.74 KB	2026-02-25 10:26
📁 crontab	-	2026-03-19 15:49
📄 jail_config.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/LICENCE.TXT # import typing from dataclasses import dataclass, field if typing.TYPE_CHECKING: from .mount_config import IsolatedRootConfig from .mount_types import MountEntry, MountType @dataclass class MountConfig: """ Builds mount configuration for jail.c implementation. Accumulates MountEntry objects, then renders everything at the end. """ uid: int gid: int _mounts: list[MountEntry] = field(default_factory=list, repr=False) @property def _default_opts(self) -> tuple[str, ...]: return f"uid={self.uid}", f"gid={self.gid}", "mode=0750" @property def mounts(self) -> tuple[MountEntry, ...]: """Read-only access to accumulated mount entries.""" return tuple(self._mounts) def add(self, type_: MountType, source: str, target: str = "", options: tuple[str] = tuple()): """Add a single mount entry.""" self._mounts.append( MountEntry( type=type_, source=source, target=target, options=options, ) ) def comment(self, text: str): """Add a comment line for organization.""" self._mounts.append(MountEntry(MountType.COMMENT, text)) def add_overlay(self, overlay: "IsolatedRootConfig"): """ Add mount operations for a directory overlay. Args: overlay: The overlay configuration """ if not overlay.persistent: # Create temporary in-memory storage self._mounts.append( MountEntry( MountType.BIND, "tmpfs", overlay.root_path, ("mkdir",) + self._default_opts ) ) else: # Persistent real directory storage (bind to self with mkdir) self._mounts.append( MountEntry( MountType.BIND, overlay.root_path, overlay.root_path, ("mkdir",) + self._default_opts, ) ) def close_overlay(self, overlay: "IsolatedRootConfig"): """Add the final mount that closes an overlay.""" self._mounts.extend(overlay.mounts) self._mounts.append( MountEntry(MountType.BIND, overlay.root_path, overlay.target, ("recursive",)) ) def render(self, docroot: str) -> str: """Render complete configuration to jail.c syntax.""" lines = [f"[{docroot}]"] lines.extend(m.render() for m in self._mounts) return "\n".join(lines)	2.72 KB	2026-02-25 10:26
📄 jail_config_builder.py #!/opt/cloudlinux/venv/bin/python3 -sbb # -- coding: utf-8 -- # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2025 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # """ Builder for website isolation jail mount configurations. Collects user docroots and isolation settings, then generates the complete jail mount configuration. """ import logging from pathlib import Path from clcommon import ClPwd from clcommon.cpapi import userdomains from clcommon.cpapi.cpapiexceptions import NoPanelUser from ..io import write_via_tmp from . import config, jail_utils from .jail_config import MountConfig from .mount_config import IsolatedRootConfig from .mount_ordering import build_docroot_tree, process_ordered_mounts from .mount_types import MountType class JailMountsConfigBuilder: """ Builder for generating jail mount configuration files. Collects docroots and isolation settings, then generates the mount configuration string for the jail.c implementation. """ def __init__(self, user: str): self.user = user self._all_docroots: set[str] = set() self._isolated_docroots: set[str] = set() self._phpselector_docroots: set[str] = set() def add_docroot(self, docroot: str) -> None: """Register a docroot path for the user.""" self._all_docroots.add(docroot) def enable_isolation(self, docroot: str) -> None: """Mark a docroot as requiring isolation.""" self._isolated_docroots.add(docroot) def enable_phpselector(self, docroot: str) -> None: """Enable per-domain PHP selector for a docroot.""" self._phpselector_docroots.add(docroot) def build(self) -> str: """ Generate the complete mount configuration. Returns: Configuration string in jail.c mount syntax. """ pw = ClPwd().get_pw_by_name(self.user) homedir = pw.pw_dir uid, gid = pw.pw_uid, pw.pw_gid # Build docroot tree once for all isolated docroots tree = build_docroot_tree(self._all_docroots) # Generate config for each isolated docroot generated_configs = [] for docroot in sorted(self._isolated_docroots, key=len): split_storage_base = jail_utils.full_website_path(homedir, docroot) home_overlay = IsolatedRootConfig( root_path=f"{split_storage_base}/home", target=homedir, persistent=True ) # Process ordered mounts for this isolated docroot docroot_mounts = process_ordered_mounts( active_docroot=docroot, tree=tree, uid=uid, gid=gid ) jail_config = MountConfig(uid=uid, gid=gid) # Add storage for the overlay'ed dir jail_config.add_overlay(home_overlay) # open .clwpos directory to make redis.sock available awp_path = f"{homedir}/.clwpos" home_overlay.mount(MountType.BIND, awp_path, awp_path, ("mkdir",)) # Add docroot mounts (from tree processing) # Mount them into already created overlay for mount in docroot_mounts: home_overlay.mount(mount.type, mount.source, mount.target, mount.options) # Apply mounts from isolated root and close target directory jail_config.close_overlay(home_overlay) # Home directory is already overlayed, we can apply per-domain mounts directly jail_config.add(MountType.USER_MOUNTS, "/") # php selector mounts (only when per-domain PHP selector is enabled) if docroot in self._phpselector_docroots: jail_config.add( MountType.BIND, source=f"/etc/cl.selector/{jail_utils.get_website_id(docroot)}", target="/etc/cl.selector" ) jail_config.add( MountType.BIND, source=f"/etc/cl.php.d/{jail_utils.get_website_id(docroot)}", target="/etc/cl.php.d" ) # Override proxyexec token with website specific folder jail_config.add( MountType.BIND, source=f"/var/.cagefs/website/{jail_utils.get_website_id(docroot)}", target="/var/.cagefs", ) generated_configs.append(jail_config.render(docroot)) return "\n".join(generated_configs) def write_jail_mounts_config(user: str, user_config: config.UserConfig \| None) -> None: """ Write or remove the jail mounts configuration file for a user. If user_config is None or has no enabled websites, the config file is removed. Otherwise, builds and writes the mount configuration. Args: user: Username to generate config for user_config: User's isolation configuration, or None to remove config """ jail_config_path = Path(jail_utils.get_jail_config_path(user)) if user_config is None or not user_config.enabled_websites: jail_config_path.unlink(missing_ok=True) return builder = JailMountsConfigBuilder(user) try: domain_to_docroot_map = dict(userdomains(user)) except NoPanelUser: logging.warning("Cannot regenerate mount configuration, no panel user=%s", user) return # add docroot information for isolations for docroot in domain_to_docroot_map.values(): builder.add_docroot(docroot) # add information about which websites should have isolation enabled for domain in user_config.enabled_websites: try: docroot = domain_to_docroot_map[domain] except KeyError: logging.warning("Docroot not found for domain %s", domain) continue builder.enable_isolation(docroot) # PHP Selector is enabled for all isolated websites for domain in user_config.enabled_websites: try: docroot = domain_to_docroot_map[domain] except KeyError: logging.warning("Docroot not found for domain %s", domain) continue builder.enable_phpselector(docroot) result = builder.build() jail_config_path.parent.mkdir(exist_ok=True, mode=0o755) write_via_tmp(str(jail_config_path.parent), str(jail_config_path), result)	6.21 KB	2026-02-25 10:26
📄 jail_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/LICENCE.TXT # import os import pathlib import secrets import shutil from pathlib import Path from clcommon import ClPwd from clcommon.clpwd import drop_privileges from clcagefslib.fs import get_user_var_cagefs_path from clcagefslib.io import write_via_tmp # FNV-1a 64-bit hash constants (must match jail C implementation) _FNV_OFFSET_BASIS = 14695981039346656037 _FNV_PRIME = 1099511628211 def get_jail_config_path(user): cagefs_dir = get_user_var_cagefs_path(user) return pathlib.Path(f"{cagefs_dir}/.cagefs/isolates.mounts") def get_website_id(document_root: str): """ Generates unique id for an isolate website using FNV-1a 64-bit hash. FNV-1a has excellent avalanche properties and distribution. Must match the docroot_hash() function in jail C code. """ hash_value = _FNV_OFFSET_BASIS for char in document_root.encode("utf-8"): hash_value ^= char hash_value = (hash_value * _FNV_PRIME) & 0xFFFFFFFFFFFFFFFF # Keep 64-bit return f"{hash_value:016x}" def create_website_token_directory(user: str, document_root: str): """ Create website token directory structure and files in /var/cagefs. Creates: - /var/cagefs/<user>/.cagefs/website/<website_id>/ - token directory """ pw = ClPwd().get_pw_by_name(user) website_id = get_website_id(document_root) # Token directory in /var/cagefs website_base_dir = Path(get_user_var_cagefs_path(user)) / ".cagefs/website" website_dir = website_base_dir / website_id # mode to prevent directory listing by other users website_base_dir.mkdir(exist_ok=True, parents=True, mode=0o751) # mode to allow user list /var/.cagefs when it's mounted for website website_dir.mkdir(exist_ok=True, mode=0o755) token = _generate_password(32) # create token file read-only for owner, others cannot read it # this replicates behavior of regular token # note: user can still use chown to change it token_file_path = f"{website_dir}/.cagefs.token" write_via_tmp(website_dir, token_file_path, token) os.chmod(token_file_path, 0o400) os.chown(token_file_path, pw.pw_uid, 0) # create file with document root path # that we can use as trusted source in proxyexec docroot_file_path = f"{website_dir}/.cagefs.website" write_via_tmp(website_dir, docroot_file_path, document_root) # only read permissions, without modification # we use this marker as a trusted source of document root # it should not be modifiable by user in any way os.chmod(docroot_file_path, 0o444) os.chown(docroot_file_path, 0, 0) def create_overlay_storage_directory(user: str, document_root: str): """ Create overlay storage directory in user's home. Creates: - <homedir>/.cagefs/websites/<website_id>/ - storage base for overlays Drops privileges to user before creating to ensure proper ownership. """ pw = ClPwd().get_pw_by_name(user) if not Path(pw.pw_dir).exists(): return storage_base = Path(full_website_path(pw.pw_dir, document_root)) with drop_privileges(user): storage_base.mkdir(exist_ok=True, parents=True, mode=0o750) def remove_website_token_directory(user: str, document_root: str): """ Remove website token directory structure and files. """ website_base_dir = Path(get_user_var_cagefs_path(user)) / ".cagefs/website" website_dir = website_base_dir / get_website_id(document_root) if website_dir.exists(): shutil.rmtree(website_dir) def website_suffix_with_hash(docroot): """ Returns path: websites/<document_root_hash> """ return os.path.join("websites", get_website_id(docroot)) def full_website_path(homedir, docroot): """ Returns <homedir>/.cagefs/websites/<document_root_hash> """ return os.path.join(homedir, ".cagefs", website_suffix_with_hash(docroot)) def invalidate_ns_cache(user: str, document_root: str): """ Removes cached namespace from disk """ website_base_dir = Path(get_user_var_cagefs_path(user)) / ".cagefs/website" website_dir = website_base_dir / get_website_id(document_root) (website_dir / ".cagefs.mnt").unlink(missing_ok=True) def _generate_password(length): """ Generate a random password/token using the same algorithm as the C function. Uses cryptographically secure random bytes and converts them to alphanumeric characters. """ if length == 0 or length > 256: raise ValueError("Invalid buffer length requested") charset = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" charset_size = len(charset) # Generate random bytes random_bytes = secrets.token_bytes(length) # Convert bytes to alphanumeric characters result = "".join(charset[b % charset_size] for b in random_bytes) return result	4.93 KB	2026-02-25 10:26
📄 libenter.py # -- coding: utf-8 -- # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2025 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT # """Library functions for executing commands inside CageFS for a site.""" import os import pwd import subprocess from pathlib import Path from typing import List from clcommon import cpapi from .jail_utils import get_website_id, get_user_var_cagefs_path def enter_site(site: str, command: List[str]) -> int: """ Execute a command inside CageFS for a site (document root or domain). User only! Inside cagefs only! :site: Document root or domain name :command: Command and arguments to execute Returns: int: Exit code from the executed command Raises: ValueError: If site cannot be resolved or isolation is not configured """ # Resolve site to document root document_root = site if not document_root.startswith("/"): try: document_root = cpapi.docroot(site)[0] except cpapi.cpapiexceptions.NoDomain: raise ValueError(f"Domain {site} not found") if not os.path.exists(document_root): raise ValueError(f"Document root {document_root} does not exist") docroot_md5 = get_website_id(document_root) inside_cagefs_path = Path("/var/.cagefs/") if inside_cagefs_path.exists(): # we are inside cagefs -> call proxyexec wrapper website_token_path = inside_cagefs_path / f"website/{docroot_md5}/.cagefs.token" if not website_token_path.exists(): raise ValueError(f"Website {site} not found or isolation is not enabled") return subprocess.call( ["/bin/cagefs_enter.proxied", command], env={os.environ, "WEBSITE_TOKEN_PATH": website_token_path}, ) else: # we are outside cagefs -> a bit different path to token jail_path = Path(get_user_var_cagefs_path(pwd.getpwuid(os.getuid()).pw_name)) website_token_path = jail_path / f".cagefs/website/{docroot_md5}/.cagefs.token" return subprocess.call( ["/bin/cagefs_enter", command], env={**os.environ, "WEBSITE_TOKEN_PATH": website_token_path}, )	2.23 KB	2026-02-25 10:26
📄 mount_config.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/LICENCE.TXT # """ Mount configuration builder for website isolation. The code handles all standard behavior (docroot isolation, home overlay, etc). """ import os.path from dataclasses import dataclass, field from .jail_config import MountEntry @dataclass class IsolatedRootConfig: """ Configuration for a directory overlay. Closes access to a directory by mounting a fake/empty directory over it, then selectively exposing only whitelisted paths. Storage is computed as: {storage_base}/{name} """ # Path to the root of this storage (e.g. ~/.clcagefs/website/123/home) root_path: str # Real directory to close target: str # Use temporary tmpfs for storage (default: real directory) persistent: bool = True # List of mounts made inside of this root (dynamically) mounts: list[MountEntry] = field(default_factory=list) def mount(self, type_, source, target, opts: tuple = tuple()): """Mounts whatever asked into the root of isolated storage""" relative_path = os.path.relpath(target, self.target) self.mounts.append(MountEntry(type_, source, f"{self.root_path}/{relative_path}", opts))	1.38 KB	2026-02-25 10:26
📄 mount_ordering.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/LICENCE.TXT # """ Module for processing and ordering mount configurations for website isolation. This module handles the logic for ordering mounts with proper parent-child relationships, remounting parents after children, and adding mkdir attributes when paths are inside tmpfs mounts. Optimization: tmpfs mounts that are inside another tmpfs are skipped because the parent tmpfs already makes them inaccessible. """ from __future__ import annotations from dataclasses import dataclass from .mount_config import MountEntry from .mount_types import MountType @dataclass(frozen=True) class DocrootTree: """Pre-computed tree structure for docroots.""" children: dict[str \| None, tuple[str, ...]] roots: tuple[str, ...] all_docroots: frozenset[str] def build_docroot_tree(all_docroots: set[str]) -> DocrootTree: """ Build the parent-child tree structure from docroots. Call this once and pass the result to process_ordered_mounts() for each active docroot to avoid rebuilding the tree on every call. Args: all_docroots: Set of all docroot paths Returns: DocrootTree structure to pass to process_ordered_mounts() """ def get_immediate_parent(path: str) -> str \| None: """Get the closest parent of path that exists in all_docroots.""" pos = path.rfind("/") while pos > 0: parent = path[:pos] if parent in all_docroots: return parent pos = parent.rfind("/") return None children_lists: dict[str \| None, list[str]] = {p: [] for p in all_docroots} children_lists[None] = [] # roots for path in all_docroots: parent = get_immediate_parent(path) children_lists[parent].append(path) # Sort and convert to tuples (immutable) children = {k: tuple(sorted(v)) for k, v in children_lists.items()} roots = children[None] return DocrootTree(children=children, roots=roots, all_docroots=frozenset(all_docroots)) def process_ordered_mounts( active_docroot: str, tree: DocrootTree, uid: int, gid: int ) -> list[MountEntry]: """ Process docroots and return ordered mounts list. Mounts are generated to a fake home directory structure. The active docroot is mounted from its real path to the corresponding path inside the fake home. Other docroots are hidden with tmpfs mounts in the fake home. Rules: 1. Only one record is marked as "docroot" (the active_docroot) 2. Parents must be mounted before children 3. If parent is mounted first, it must be rw mount with remount to ro after all child records mounted 4. If path is inside another path that we marked as tmpfs, we must add mkdir attribute Args: active_docroot: The docroot that should be marked as "docroot" in the output tree: Pre-computed DocrootTree from build_docroot_tree() uid: User ID for tmpfs mounts gid: Group ID for tmpfs mounts Returns: List of MountEntry in the correct order """ children = tree.children tmpfs_paths = tree.all_docroots - {active_docroot} tmpfs_attrs = (f"uid={uid}", f"gid={gid}", "mode=0750") mounts: list[MountEntry] = [] def visit(path: str, inside_tmpfs: bool): """Process a path and its children recursively.""" is_tmpfs = path in tmpfs_paths path_children = children[path] has_children = bool(path_children) # Skip tmpfs mounts that are inside another tmpfs - they're already hidden # Still need to recurse in case there's an active docroot nested within if is_tmpfs and inside_tmpfs: for child in path_children: visit(child, inside_tmpfs=True) return # Generate mount operation # All mounts into fake_home need mkdir since overlay storage starts empty if path == active_docroot: mounts.append(MountEntry(MountType.BIND, path, path, ("mkdir",))) elif is_tmpfs: opts = ("mkdir",) + tmpfs_attrs if has_children else ("mkdir", "ro") + tmpfs_attrs mounts.append(MountEntry(MountType.BIND, "tmpfs", path, opts)) else: opts = ("mkdir",) if has_children else ("mkdir", "ro") mounts.append(MountEntry(MountType.BIND, path, path, opts)) # Process children child_inside_tmpfs = inside_tmpfs or is_tmpfs for child in path_children: visit(child, child_inside_tmpfs) # On leave: remount if this was a tmpfs parent with children if has_children and is_tmpfs: mounts.append(MountEntry(MountType.BIND, path, path, options=("ro", "remount"))) # Start from roots (paths with no parent in all_docroots) for root in tree.roots: visit(root, inside_tmpfs=False) return mounts	4.96 KB	2026-02-25 10:26
📄 mount_types.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/LICENCE.TXT # from dataclasses import dataclass from enum import Enum, auto class MountType(Enum): """Types of mount operations.""" # Bind mount source to target (source="tmpfs" for tmpfs mounts) BIND = auto() # @USER_MOUNTS directive USER_MOUNTS = auto() # Comment line for organization COMMENT = auto() @dataclass class MountEntry: """A single mount operation.""" type: MountType source: str target: str = "" options: tuple[str, ...] = () def render(self) -> str: """Render to jail.c mount syntax.""" opts = ",".join(self.options) if self.options else "" if self.type == MountType.BIND: parts = [self.source, self.target] if opts: parts.append(opts) return " ".join(parts) elif self.type == MountType.USER_MOUNTS: return f"@USER_MOUNTS {self.source}" elif self.type == MountType.COMMENT: return f"## {self.source}" raise ValueError(f"Unknown mount type: {self.type}")	1.26 KB	2026-02-25 10:26
📄 php.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/LICENCE.TXT # import os import signal import logging import time from pathlib import Path from clcommon.clpwd import drop_privileges import psutil from clcagefslib.webisolation import jail_utils def _term_process(pid, sig): try: os.kill(pid, sig) except OSError as e: logging.error("Failed to terminate process PID %s: %s", pid, str(e)) return False return True def force_process_kill(pids): timeout = time.time() + 5.0 while time.time() < timeout and pids: for pid in list(pids): if not psutil.pid_exists(pid): pids.remove(pid) for pid in list(pids): _term_process(pid, signal.SIGKILL) def _get_website_isolation_docroot(pid: int) -> str \| None: """ Checks if the process is in the isolated environment. Returns path to process document root or None """ website_isolation_marker = Path(f"/proc/{pid}/root/var/.cagefs/.cagefs.website") try: return website_isolation_marker.read_text().strip() except (PermissionError, OSError, FileNotFoundError): return None def reload_processes_with_docroots(username: str, filter_by_docroots: list[str]) -> None: """ If filter_by_docroots is not empty - reload only processes with DOCUMENT_ROOT environment variable matching one of the given docroots. Sends SIGTERM/SIGKILL to all matching processes owned by the given username. If filter_by_docroots is empty - reload all processes owned by the given username with DOCUMENT_ROOT set. """ docroots = set([dr for dr in filter_by_docroots if dr]) signaled_pids = set() logging.debug(f"Requested to reload processes for user '{username}' and docroots: {docroots}") # first of all invalidate the existing cache for document_root in docroots: jail_utils.invalidate_ns_cache(username, document_root) # Iterate through all processes and filter by username first for proc in psutil.process_iter(attrs=["pid", "name", "username"]): current_process = proc if current_process.info.get("username") != username: continue try: env = current_process.environ() except psutil.AccessDenied: logging.warning( "Access denied to process PID %s; skipping", current_process.info.get("pid") ) continue # _get_website_isolation_docroot returns document root for processes # which are already isolated, while DOCUMENT_ROOT in env gives us # processes which should be inside isolation (e.g. to kill during site-isolation-enable) document_root = _get_website_isolation_docroot(proc.pid) or env.get("DOCUMENT_ROOT") if docroots and document_root not in docroots: continue pid = current_process.info.get("pid") if pid is None or pid in signaled_pids: continue with drop_privileges(username): logging.info(f"Terminating process PID {pid} with DOCUMENT_ROOT={document_root}") if not _term_process(pid, signal.SIGTERM): continue signaled_pids.add(pid) with drop_privileges(username): force_process_kill(signaled_pids)	3.39 KB	2026-02-25 10:26
📄 service.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/LICENCE.TXT # import os import shutil import subprocess SERVICE_NAME = "cagefs-docroot-monitor.service" SERVICE_PATH = f"/usr/share/cagefs/service/{SERVICE_NAME}" SYSTEMD_UNIT_PATH = f"/etc/systemd/system/{SERVICE_NAME}" def is_running(): result = subprocess.run( ["/usr/bin/systemctl", "is-active", SERVICE_NAME], capture_output=True, text=True ) return result.returncode == 0 def start_monitoring_service(): if is_running(): return shutil.copy(SERVICE_PATH, SYSTEMD_UNIT_PATH) result = subprocess.run( ["/usr/bin/systemctl", "enable", SERVICE_NAME], capture_output=True, text=True, check=True ) if result.returncode: return result = subprocess.run( ["/usr/bin/systemctl", "daemon-reload"], capture_output=True, text=True, check=True ) if result.returncode: return subprocess.run( ["/usr/bin/systemctl", "start", SERVICE_NAME], capture_output=True, text=True, check=True ) def stop_monitoring_service(): if is_running(): subprocess.run( ["/usr/bin/systemctl", "stop", SERVICE_NAME], capture_output=True, text=True, check=True ) # disable may fail if the unit file was never installed — that's fine subprocess.run( ["/usr/bin/systemctl", "disable", SERVICE_NAME], capture_output=True, text=True ) if os.path.exists(SYSTEMD_UNIT_PATH): os.remove(SYSTEMD_UNIT_PATH) subprocess.run( ["/usr/bin/systemctl", "daemon-reload"], capture_output=True, text=True, check=True )	1.76 KB	2026-02-25 10:26
📄 triggers.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 """ Triggers for website isolation. """ import os import subprocess import logging def trigger_xray_ini_regeneration(user: str, domain: str): """ Call xray regeneration script if it exists. This ensures xray.ini files are created in per-website directories when website isolation is enabled. """ xray_regen_script = "/usr/share/alt-php-xray/cloudlinux-xray-regenerate-ini" if not os.path.isfile(xray_regen_script): # ok when xray is not installed logging.debug("xray regeneration script not found: %s", xray_regen_script) return try: subprocess.run( [xray_regen_script, "--user", user, "--domain", domain], capture_output=True, timeout=30, check=True, ) except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e: logging.warning(f"Failed to regenerate xray ini for {user}: {e.stdout}") def trigger_ssa_ini_regeneration(user: str): """ Call SSA regeneration script if it exists. This ensures clos_ssa.ini files are created in per-website directories when website isolation is enabled. """ ssa_regen_script = "/usr/share/clos_ssa/regenerate_clos_ssa_ini.py" if not os.path.isfile(ssa_regen_script): # ok when ssa is not installed logging.debug("ssa regeneration script not found: %s", ssa_regen_script) return try: subprocess.run( [ssa_regen_script, f"--user={user}"], capture_output=True, timeout=30, check=True ) except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e: logging.warning(f"Failed to regenerate SSA ini for {user}: {e.stdout}")	1.9 KB	2026-02-25 10:26

Terminal