File Manager

X7ROOT File Manager

Current Path: /opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/plugins

opt / imunify360 / venv / lib / python3.11 / site-packages / defence360agent / plugins /

📁 ..
📄 __init__.py(0 B)
📁 __pycache__
📄 accumulate.py(3.53 KB)
📄 analyst_cleanup_update.py(5.53 KB)
📄 backup_info_sender.py(3.06 KB)
📄 checkpoint.py(1.23 KB)
📄 client.py(9.7 KB)
📄 config_merger.py(828 B)
📄 config_watcher.py(1.89 KB)
📄 event_hook_executor.py(777 B)
📄 event_monitor.py(3.32 KB)
📄 event_monitor_message_processor.py(6.33 KB)
📄 files_recurring_update.py(1.09 KB)
📄 icontact_sender.py(4.36 KB)
📄 idle_time_out.py(1.21 KB)
📄 lve_utils_install.py(1.58 KB)
📄 myimunify.py(1.93 KB)
📄 ping.py(536 B)
📄 send_domain_list.py(2.79 KB)
📄 send_server_config.py(10.37 KB)

Editing: myimunify.py

import logging

from defence360agent.contracts.config import MyImunifyConfig
from defence360agent.contracts.messages import MessageType
from defence360agent.contracts.plugins import (
    MessageSink,
    MessageSource,
    expect,
)
from defence360agent.myimunify.model import update_users_protection
from defence360agent.subsys.panels import hosting_panel
from defence360agent.subsys.persistent_state import load_state, save_state

logger = logging.getLogger(__name__)

class MyImunifyPlugin(MessageSink, MessageSource):
    def __init__(self):
        self._previous_myimunify_status = (
            load_state("MyImunifyPlugin").get("myimunify_enabled")
            or MyImunifyConfig.ENABLED
        )
        self._loop = None
        self._sink = None

async def shutdown(self):
        save_state(
            "MyImunifyPlugin",
            {"myimunify_enabled": self._previous_myimunify_status},
        )

async def create_sink(self, loop):
        pass

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

async def _update_myimunify_users(self):
        existing_users = await hosting_panel.HostingPanel().get_users()
        await update_users_protection(self._sink, existing_users, False)

@expect(MessageType.ConfigUpdate)
    async def on_config_update(self, message: MessageType.ConfigUpdate):
        myimunify_enabled = MyImunifyConfig.ENABLED
        previous_status = self._previous_myimunify_status
        # We're also triggering additional MessageType.ConfigUpdate messages
        # so we must update previous_status before triggering new one
        self._previous_myimunify_status = myimunify_enabled
        if myimunify_enabled and not previous_status:
            await self._update_myimunify_users()

if myimunify_enabled != previous_status:
            await hosting_panel.HostingPanel().switch_ui_config(
                myimunify_enabled=myimunify_enabled
            )

X7ROOT File Manager

Editing: myimunify.py

Upload File

Create Folder