Edit - structure.py

Edit: 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]]

Cancel