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 # -- 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 # """ Crontab business logic for website isolation support. """ from .processor import ( process_list, process_save, ) from .libhooks import ( migrate_crontab_docroot, cleanup_stale_crontab_sections, ) __all__ = [ "process_list", "process_save", "migrate_crontab_docroot", "cleanup_stale_crontab_sections", ]	549 B	2026-02-25 10:26
📁 __pycache__	-	2026-03-19 15:49
📄 constants.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 # """Constants and regex patterns for crontab parsing.""" import re # Path to the isolation wrapper script ISOLATION_WRAPPER = "/usr/sbin/cagefs_enter_site" # Environment variable that indicates website isolation is active DOCUMENT_ROOT_ENV = "PROXYEXEC_DOCUMENT_ROOT" # Pattern to match crontab schedule fields (5 fields for standard cron) # Matches: minute hour day month weekday CRON_SCHEDULE_PATTERN = re.compile(rb"^(\S+\s+\S+\s+\S+\s+\S+\s+\S+)\s+(.)$") # Markers for website cron sections WEBSITE_CRON_BEGIN_PATTERN = re.compile(rb"^##\s+WEBSITE\s+CRON\s+BEGIN\s+(.+)$") WEBSITE_CRON_END_PATTERN = re.compile(rb"^##\s+WEBSITE\s+CRON\s+END\s$")	864 B	2026-02-25 10:26
📄 libhooks.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 # """Hook integration functions for crontab operations.""" import logging import shlex import subprocess from clcommon.cpapi import userdomains from .constants import ISOLATION_WRAPPER from .parser import parse_crontab_structure, write_crontab_structure from .structure import ParsedCrontabLine CRONTAB_BIN = "/usr/bin/crontab" logger = logging.getLogger(__name__) def migrate_crontab_docroot( username: str, old_docroot: str, new_docroot: str, ) -> bool: """ Migrate crontab entries from old document root to new document root. When a document root path changes (e.g., domain rename, docroot move), this function updates all crontab entries that reference the old path to use the new path instead. Args: username: Unix username whose crontab to migrate old_docroot: Previous document root path new_docroot: New document root path Returns: bool: True if migration was successful, False if no changes needed or error Raises: Exception: If crontab operations fail critically """ # Get current crontab for the user list_result = subprocess.run( [CRONTAB_BIN, "-u", username, "-l"], capture_output=True, ) # No existing crontab - nothing to migrate if list_result.returncode != 0: logger.debug( "No crontab to migrate for user %s (returncode: %d)", username, list_result.returncode, ) return False existing_data = list_result.stdout if not existing_data: logger.debug("Empty crontab for user %s, nothing to migrate", username) return False # Parse existing crontab into structure structure = parse_crontab_structure(existing_data) # Check if old docroot section exists if old_docroot not in structure.docroot_sections: logger.debug( "No crontab entries found for old docroot %s for user %s", old_docroot, username, ) return False # Migrate: move entries from old docroot to new docroot old_entries = structure.docroot_sections.pop(old_docroot) # Update wrapper commands in the entries to use new docroot migrated_entries = [] for entry in old_entries: if isinstance(entry, ParsedCrontabLine): # Get the clean command (without wrapper) clean_command = entry.get_clean_command() has_newline = clean_command.endswith(b"\n") command_str = clean_command.rstrip(b"\n").decode("utf-8", errors="replace") # Rebuild with new docroot quoted_command = shlex.quote(command_str) prefix_parts = shlex.join([ISOLATION_WRAPPER, new_docroot, "bash", "-c"]) wrapped_command_str = f"{prefix_parts} {quoted_command}" wrapped_command = wrapped_command_str.encode("utf-8") if has_newline: wrapped_command += b"\n" migrated_entries.append( ParsedCrontabLine(schedule=entry.schedule, command=wrapped_command) ) else: # Keep comments as-is migrated_entries.append(entry) # Add migrated entries to new docroot section (merge with existing if any) if new_docroot in structure.docroot_sections: structure.docroot_sections[new_docroot].extend(migrated_entries) else: structure.docroot_sections[new_docroot] = migrated_entries # Write updated crontab modified_data = write_crontab_structure(structure) save_result = subprocess.run( [CRONTAB_BIN, "-u", username, "-"], input=modified_data, capture_output=True, ) if save_result.returncode != 0: logger.error( "Failed to save migrated crontab for user %s: %s", username, save_result.stderr.decode("utf-8", errors="replace"), ) return False logger.info( "Successfully migrated crontab entries for user %s from %s to %s", username, old_docroot, new_docroot, ) return True def cleanup_stale_crontab_sections( username: str, ) -> bool: """ Remove crontab sections for docroots that no longer exist. After a domain is deleted, this function cleans up crontab entries for docroots that are no longer valid (not in the current domain list). Args: username: Unix username whose crontab to clean up Returns: bool: True if cleanup was performed, False if no changes needed Raises: Exception: If crontab operations fail critically """ # Get current valid docroots from userdomains valid_docroots = {docroot for _, docroot in userdomains(username)} # Get current crontab for the user list_result = subprocess.run( [CRONTAB_BIN, "-u", username, "-l"], capture_output=True, ) # No existing crontab - nothing to clean if list_result.returncode != 0: logger.debug( "No crontab to clean for user %s (returncode: %d)", username, list_result.returncode, ) return False existing_data = list_result.stdout if not existing_data: logger.debug("Empty crontab for user %s, nothing to clean", username) return False structure = parse_crontab_structure(existing_data) stale_docroots = [ docroot for docroot in structure.docroot_sections.keys() if docroot not in valid_docroots ] if not stale_docroots: logger.debug( "No stale crontab sections found for user %s", username, ) return False for stale_docroot in stale_docroots: logger.info( "Removing stale crontab section for user %s, docroot: %s", username, stale_docroot, ) del structure.docroot_sections[stale_docroot] modified_data = write_crontab_structure(structure) save_result = subprocess.run( [CRONTAB_BIN, "-u", username, "-"], input=modified_data, capture_output=True, ) if save_result.returncode != 0: logger.error( "Failed to save cleaned crontab for user %s: %s", username, save_result.stderr.decode("utf-8", errors="replace"), ) return False logger.info( "Successfully cleaned up %d stale crontab section(s) for user %s: %s", len(stale_docroots), username, ", ".join(stale_docroots), ) return True	6.6 KB	2026-02-25 10:26
📄 parser.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 # """Parsing and serialization functions for crontab entries.""" from .constants import ( CRON_SCHEDULE_PATTERN, WEBSITE_CRON_BEGIN_PATTERN, WEBSITE_CRON_END_PATTERN, ) from .structure import ( BeginWebsite, CommentLine, CrontabEntry, CrontabStructure, EndWebsite, ParsedCrontabLine, ) def parse_crontab_line(line: bytes) -> CrontabEntry: """ Parse a crontab line into a cron entry, comment, or website marker. Args: line: A single line from crontab output (bytes) """ stripped = line.strip() # Check for website cron begin marker begin_match = WEBSITE_CRON_BEGIN_PATTERN.match(stripped) if begin_match: docroot_bytes = begin_match.group(1).strip() docroot = docroot_bytes.decode("utf-8", errors="replace") return BeginWebsite(docroot=docroot) # Check for website cron end marker if WEBSITE_CRON_END_PATTERN.match(stripped): return EndWebsite() # Handle empty lines and comments if not stripped: return CommentLine(content=line) if stripped.startswith(b"#"): return CommentLine(content=line) # Try to match cron schedule + command schedule_match = CRON_SCHEDULE_PATTERN.match(stripped) if not schedule_match: return CommentLine(content=line) schedule = schedule_match.group(1) command = schedule_match.group(2) if line.endswith(b"\n"): command += b"\n" return ParsedCrontabLine( schedule=schedule, command=command, ) def parse_crontab_structure(input_data: bytes) -> CrontabStructure: """ Parse crontab file into structured format with user records and docroot sections. Supports both old format (with isolation wrapper prefixes) and new format (with section markers). Args: input_data: bytes containing crontab entries Returns: CrontabStructure: Structure containing global_records and docroot_sections """ lines = input_data.splitlines(keepends=True) crontab_records: dict[ str \| None, list[CrontabEntry], ] = {} current_docroot: str \| None = None for line in lines: parsed = parse_crontab_line(line) if isinstance(parsed, BeginWebsite): current_docroot = parsed.docroot continue if isinstance(parsed, EndWebsite): current_docroot = None continue crontab_records.setdefault(current_docroot, []).append(parsed) global_records = crontab_records.pop(None, []) return CrontabStructure( global_records=global_records, docroot_sections=crontab_records, ) def write_crontab_structure(structure: CrontabStructure) -> bytes: """ Write crontab structure to bytes in new format. Format: - User records at the top - Then docroot sections with markers: ## WEBSITE CRON BEGIN /path/to/docroot ... entries ... ## WEBSITE CRON END Args: structure: CrontabStructure with user_records and docroot_sections Returns: bytes: Crontab content in new format """ result_parts = entries_to_str_list(structure.global_records) # Add docroot sections (preserve insertion order) for docroot in structure.docroot_sections.keys(): entries = structure.docroot_sections[docroot] if not entries: continue result_parts.append(f"## WEBSITE CRON BEGIN {docroot}\n".encode("utf-8")) result_parts.extend(entries_to_str_list(entries)) result_parts.append(b"## WEBSITE CRON END\n") return b"".join(result_parts) def entries_to_str_list(entries: list[CrontabEntry], without_wrapper: bool = False) -> list[bytes]: """ Converts parsed crontab entries into a string list. """ lines = [] for entry in entries: if isinstance(entry, ParsedCrontabLine): line = ( entry.schedule + b" " + (entry.get_clean_command() if without_wrapper else entry.command) ) if not line.endswith(b"\n"): line += b"\n" lines.append(line) elif isinstance(entry, CommentLine): lines.append(entry.content) return lines	4.38 KB	2026-02-25 10:26
📄 processor.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 # """Processing functions for crontab operations.""" import os import pwd import shlex import subprocess import sys from typing import BinaryIO, Callable, Optional from clcagefslib.domain import is_isolation_enabled from .constants import ISOLATION_WRAPPER from .parser import ( parse_crontab_structure, write_crontab_structure, entries_to_str_list, ) from .structure import ParsedCrontabLine from .utils import get_document_root CRONTAB_BIN = "/usr/bin/crontab" def process_list(stdout: Optional[BinaryIO] = None, stderr: Optional[BinaryIO] = None) -> int: """ Process CRONTAB_LIST command. Runs 'crontab -l' to get current crontab entries. When isolation is active (PROXYEXEC_DOCUMENT_ROOT is set), only shows entries for the current document root. Removes isolation prefixes from output. Args: stdout: Output stream buffer (defaults to sys.stdout.buffer) stderr: Error stream buffer (defaults to sys.stderr.buffer) Returns: int: Exit code from crontab command, or 1 on error """ stdout = stdout or sys.stdout.buffer stderr = stderr or sys.stderr.buffer result = subprocess.run( [CRONTAB_BIN, "-l"], capture_output=True, ) # early exit in case site isolation is not turned on if not is_isolation_enabled(pwd.getpwuid(os.getuid()).pw_name): stdout.write(result.stdout) stderr.write(result.stderr) return result.returncode if result.returncode != 0: # Pass through stderr from crontab stderr.write(result.stderr) return result.returncode document_root = get_document_root() # Parse structure and pick entries to show structure = parse_crontab_structure(result.stdout) if document_root is not None: entries_to_show = structure.docroot_sections.get(document_root, []) else: entries_to_show = structure.global_records # Convert selected entries to bytes, removing wrapper prefixes if isolation is active result_parts = entries_to_str_list(entries_to_show, without_wrapper=bool(document_root)) output_data = b"".join(result_parts) stdout.write(output_data) return 0 def process_save( stdin: Optional[BinaryIO] = None, stdout: Optional[BinaryIO] = None, stderr: Optional[BinaryIO] = None, run_func: Optional[Callable] = None, ) -> int: """ Process CRONTAB_SAVE command. Reads crontab entries from stdin. If isolation is active: 1. Gets the current full crontab 2. Removes entries for the current document root 3. Adds new entries to the current document root section 4. Merges and saves the result in new format If isolation is not active: 1. Gets the current full crontab 2. Replaces user records section with new entries 3. Preserves all docroot sections 4. Saves in new format This ensures entries for other document roots are preserved. Args: stdin: Input stream buffer (defaults to sys.stdin.buffer) stdout: Output stream buffer (defaults to sys.stdout.buffer) stderr: Error stream buffer (defaults to sys.stderr.buffer) run_func: Function to run subprocess (defaults to subprocess.run) Returns: int: Exit code from crontab command, or 1 on error """ stdin = stdin or sys.stdin.buffer stdout = stdout or sys.stdout.buffer stderr = stderr or sys.stderr.buffer run_func = run_func or subprocess.run input_data = stdin.read() document_root = get_document_root() if is_isolation_enabled(pwd.getpwuid(os.getuid()).pw_name): # Get current crontab to preserve entries list_result = run_func( [CRONTAB_BIN, "-l"], capture_output=True, ) # No existing crontab or error - start fresh # Note: returncode 1 typically means "no crontab for user", which is acceptable existing_data = list_result.stdout if list_result.returncode == 0 else b"" # Parse existing crontab into structure existing_structure = parse_crontab_structure(existing_data) # Parse new input (filtered data, no section markers) # input_data is already bytes from stdin.read() new_structure = parse_crontab_structure(input_data) # Extract entries and add wrapper prefixes for docroot sections parsed_entries = [] for entry in new_structure.global_records: if isinstance(entry, ParsedCrontabLine) and document_root: # Decode command to string for shlex.quote() command_bytes = entry.command has_newline = command_bytes.endswith(b"\n") command_str = command_bytes.rstrip(b"\n").decode("utf-8", errors="replace") # Use shlex.quote() to properly quote the command for bash -c quoted_command = shlex.quote(command_str) # Build the wrapped command: wrapper path docroot bash -c "quoted_command" # Use shlex.join() for the prefix to handle paths with spaces, then add quoted command prefix_parts = shlex.join([ISOLATION_WRAPPER, document_root, "bash", "-c"]) wrapped_command_str = f"{prefix_parts} {quoted_command}" wrapped_command = wrapped_command_str.encode("utf-8") if has_newline: wrapped_command += b"\n" parsed_entries.append( ParsedCrontabLine(schedule=entry.schedule, command=wrapped_command) ) else: # Keep as-is for comments and when document_root is None (may have wrapper prefixes) parsed_entries.append(entry) if document_root: # Isolation active: replace docroot section with new entries existing_structure.docroot_sections[document_root] = parsed_entries else: existing_structure.global_records = parsed_entries # Write in new format modified_data = write_crontab_structure(existing_structure) else: # pass through modified_data = input_data result = run_func( [CRONTAB_BIN, "-"], input=modified_data, capture_output=True, ) # Pass through any output from crontab if result.stdout: stdout.write(result.stdout) if result.stderr: stderr.write(result.stderr) return result.returncode	6.5 KB	2026-02-25 10:26
📄 structure.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 # """Data structures for crontab representation.""" import logging import shlex from dataclasses import dataclass from typing import Dict, List, Optional, Union from .constants import ISOLATION_WRAPPER logger = logging.getLogger(__name__) @dataclass class CommentLine: """Represents a comment or empty line in crontab.""" content: bytes @dataclass class BeginWebsite: """Represents a website cron begin marker.""" docroot: str @dataclass class EndWebsite: """Represents a website cron end marker.""" pass @dataclass class ParsedCrontabLine: """Represents a parsed crontab entry.""" schedule: bytes command: bytes # Command with wrapper prefix if present def _parse_wrapper_command(self) -> Optional[tuple[str, str]]: """ Parse command to extract docroot and clean command if it has isolation wrapper. Returns: Optional[tuple[str, str]]: (docroot, clean_command) if wrapper found, None otherwise """ # Quick check: only attempt parsing if command starts with wrapper # avoid expensive shlex.split() for non-wrapped command command_str = self.command.rstrip(b"\n").decode("utf-8", errors="replace") if not command_str.startswith(ISOLATION_WRAPPER): return None try: parts = shlex.split(command_str) # Check if command matches expected format: # [ISOLATION_WRAPPER, docroot, "bash", "-c", command] if ( len(parts) >= 5 and parts[0] == ISOLATION_WRAPPER and parts[2] == "bash" and parts[3] == "-c" ): docroot = parts[1] # The command is everything after "bash -c", which shlex.split() already unescaped clean_command = parts[4] if len(parts) == 5 else " ".join(parts[4:]) return docroot, clean_command else: # Command starts with wrapper but doesn't match expected format logger.error( "Failed to parse wrapper command: command doesn't match expected format. " "Expected: [%s, docroot, 'bash', '-c', command]. Got: %s", ISOLATION_WRAPPER, command_str[:200] if len(command_str) > 200 else command_str, ) except (ValueError, IndexError) as e: # shlex.split() failed (e.g., unclosed quotes) logger.error( "Failed to parse wrapper command with shlex.split(): %s. Command: %s", e, command_str[:200] if len(command_str) > 200 else command_str, ) return None def get_docroot(self) -> Optional[str]: """ Extract docroot from command if it has isolation wrapper prefix. Returns: Optional[str]: The document root if command has wrapper prefix, None otherwise """ result = self._parse_wrapper_command() return result[0] if result else None def get_clean_command(self) -> bytes: """ Get command without wrapper prefix. Returns: bytes: Command without wrapper prefix, or original command if no wrapper """ result = self._parse_wrapper_command() if result: clean_cmd_str = result[1] clean_cmd = clean_cmd_str.encode("utf-8") # Preserve line ending (always \n on Linux) if self.command.endswith(b"\n"): if not clean_cmd.endswith(b"\n"): clean_cmd += b"\n" return clean_cmd return self.command # Type alias for crontab entry types CrontabEntry = Union[ParsedCrontabLine, CommentLine, BeginWebsite, EndWebsite] @dataclass class CrontabStructure: """Structure representing parsed crontab entries.""" global_records: List[CrontabEntry] docroot_sections: Dict[str, List[CrontabEntry]]	4.11 KB	2026-02-25 10:26
📄 utils.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 # """Utility functions for crontab operations.""" import os from .constants import DOCUMENT_ROOT_ENV def get_document_root() -> str \| None: """ Get the document root from environment variable. Returns: Optional[str]: The document root path if PROXYEXEC_DOCUMENT_ROOT is set, None otherwise. """ return os.environ.get(DOCUMENT_ROOT_ENV)	598 B	2026-02-25 10:26

Terminal