← Back
Editing: 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]]
Save File
Cancel