ꦫꦣꦺꦤ꧀ꦄꦤ꧀ꦠꦱꦺꦤ

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

import asyncio
import time
from logging import getLogger
from pathlib import Path

from defence360agent.contracts.config import Core, ConfigFile
from defence360agent.contracts.messages import MessageType
from defence360agent.contracts.plugins import MessageSink, MessageSource
from defence360agent.subsys.panels.hosting_panel import HostingPanel
from defence360agent.utils import create_task_and_log_exceptions, retry_on
from imav.malwarelib.subsys.malware import HackerTrapHitsSaver
from im360.subsys.modsec_cache_dir import create_modsec_cache_directory
from im360.subsys.whitelist_rbl import (
    create_rbl_whitelist,
    ensure_rbl_whitelist,
)

logger = getLogger(__name__)

class StartupActions(MessageSink, MessageSource):
    PANEL_NAME_PATH = Path(Core.GLOBAL_CONFDIR) / "panel-name.txt"
    RETRY_TIMEOUT = 5 * 60  # 5 minutes

def _create_tasks(self):
        async def sleep_on_error(*_):
            await asyncio.sleep(self.RETRY_TIMEOUT)

tasks = [
            create_task_and_log_exceptions(
                self._loop, HackerTrapHitsSaver.reset_sa_hits
            ),
            create_task_and_log_exceptions(
                self._loop, create_modsec_cache_directory
            ),
            create_task_and_log_exceptions(self._loop, create_rbl_whitelist),
            create_task_and_log_exceptions(self._loop, ensure_rbl_whitelist),
            create_task_and_log_exceptions(
                self._loop,
                retry_on(
                    Exception,
                    on_error=sleep_on_error,
                    max_tries=3,
                )(self.update_panel_name),
            ),
            create_task_and_log_exceptions(
                self._loop,
                self._check_config,
            ),
        ]
        return tasks

async def create_sink(self, loop):
        pass

async def _check_config(self):
        # notify about the update on start
        message = MessageType.ConfigUpdate(
            conf=ConfigFile(), timestamp=time.time()
        )
        await self._sink.process_message(message)

async def create_source(self, loop, sink):
        self._loop = loop
        self._sink = sink
        self._tasks = self._create_tasks()

async def update_panel_name(self):
        self.PANEL_NAME_PATH.write_text(HostingPanel().NAME)

async def shutdown(self):
        for task in self._tasks:
            task.cancel()
        await asyncio.gather(*self._tasks, return_exceptions=True)

[-] repeater.py [edit]
[-] whitelist_current_user.py [edit]
[-] modsec_ruleset_checker.py [edit]
[-] startup_actions.py [edit]
[-] cpanel_uploader.py [edit]
[-] service_manager.py [edit]
[+] __pycache__
[-] fgw.py [edit]
[-] php_immunity.py [edit]
[-] send_server_config.py [edit]
[-] lfd.py [edit]
[-] ossec_rules_checker.py [edit]
[-] remoteip_install.py [edit]
[-] strategy_getter.py [edit]
[-] __init__.py [edit]
[-] pam_manager.py [edit]
[-] waf_rules_configurator.py [edit]
[-] export_wblist.py [edit]
[+] ..