FM / crontab

/opt/cloudlinux/venv/lib/python3.11/site-packages/clcagefslib/webisolation/crontab/

/opt/cloudlinux/venv/lib/python3.11/site-packages/clcagefslib/webisolation/crontab

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
constants.py	2406	0644	edit dl rm
libhooks.py	6994	0644	edit dl rm
parser.py	7745	0644	edit dl rm
processor.py	8837	0644	edit dl rm
structure.py	4370	0644	edit dl rm
utils.py	1819	0644	edit dl rm
__init__.py	549	0644	edit dl rm

Edit: /opt/cloudlinux/venv/lib/python3.11/site-packages/clcagefslib/webisolation/crontab/structure.py (4370B)

# -*- 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 EnvAssignmentLine:
    """Represents a `name = value` environment-assignment line per crontab(5)."""

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, EnvAssignmentLine, BeginWebsite, EndWebsite
]

@dataclass
class CrontabStructure:
    """Structure representing parsed crontab entries."""

global_records: List[CrontabEntry]
    docroot_sections: Dict[str, List[CrontabEntry]]

cmd: