MAINHACK

Viewing / Editing File: cpanel_uploader.py

File is not writable. Editing disabled.
"""
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"
                    )
Cancel / Back
[ MAINHACK ]

Viewing / Editing File: cpanel_uploader.py

Server Info

System Features

User Info
