✏️ Editing: cpanel_uploader.py Read Only

"""
cPanel upload hook manager plugin.

It enables/disables cPanel upload hook on an imunify360 config change.
"""
import logging
from functools import lru_cache

from defence360agent.contracts.config import Malware as Config
from defence360agent.contracts.config import SystemConfig
from defence360agent.contracts.messages import MessageType
from defence360agent.contracts.plugins import MessageSink, expect
from defence360agent.utils import CheckRunError, await_for, check_run, retry_on
from im360.subsys.panels.cpanel import cPanel
from im360.subsys.panels.hosting_panel import HostingPanel

logger = logging.getLogger(__name__)

MANAGE_HOOKS_CMD = "/usr/local/cpanel/bin/manage_hooks"
HOOK_PATH = "/usr/libexec/imunify360/cpanel_fileman_hook"

class CpanelUploadHookManager(MessageSink):
    def __init__(self):
        self._current_cpanel_scan_status = None

async def create_sink(self, loop):
        """MessageSink method"""

@lru_cache(maxsize=1)
    def is_supported(self) -> bool:
        return HostingPanel().NAME == cPanel.NAME

async def is_installed(self) -> bool:
        hooks = (await check_run([MANAGE_HOOKS_CMD, "list"])).decode()
        return HOOK_PATH in hooks

@retry_on(CheckRunError, max_tries=2, on_error=await_for(seconds=2))
    async def _reset_hook(self, enabled):
        action = "add" if enabled else "del"
        await check_run([MANAGE_HOOKS_CMD, action, "script", HOOK_PATH])

@expect(MessageType.ConfigUpdate)
    async def update_hook(self, message):
        # expect to get ConfigUpdate every time on start up (at least)
        # see ConfigWatcher plugin for details
        if isinstance(message["conf"], SystemConfig) and self.is_supported():
            enabled = Config.CPANEL_SCAN_ENABLED
            if self._current_cpanel_scan_status != enabled:
                installed = await self.is_installed()
                if installed != enabled:  # need to update
                    try:
                        await self._reset_hook(enabled)
                    except CheckRunError as exc:
                        logger.error(
                            "Error occured during update cpanel hook: %s", exc
                        )
                    else:
                        self._current_cpanel_scan_status = enabled
                        logger.info(
                            "cPanel uploader hook %sinstalled successfully",
                            "" if enabled else "un",
                        )
                else:  # already installed/removed
                    self._current_cpanel_scan_status = enabled
                    logger.info(
                        "cPanel uploader hook %sinstalled already",
                        "" if enabled else "un",
                    )

🔐 Sid Gifari File Manager Pro