cpanel_uploader.py

"""
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",
                    )

docroot: /home/induvedam/public_html

curdir: /opt/imunify360/venv/lib/python3.11/site-packages/im360/plugins

	.	4096
	..	4096
X	__init__.py	0
	__pycache__	4096
X	coraza_ff_watcher.py	3066
X	cpanel_uploader.py	2777
X	export_wblist.py	7058
X	fgw.py	2071
X	lfd.py	3511
X	modsec_ruleset_checker.py	5031
X	ossec_rules_checker.py	1803
X	pam_manager.py	6747
X	php_immunity.py	1438
X	remoteip_install.py	1053
X	repeater.py	1187
X	send_server_config.py	4606
X	service_manager.py	5298
X	startup_actions.py	2469
X	strategy_getter.py	2046
X	waf_rules_configurator.py	2535
X	webshield_feature_flags.py	1148
X	whitelist_current_user.py	4184