1#@!#!123s

: / opt / imunify360 / venv / lib64 / python3.11 / site-packages / defence360agent / plugins /
Filename : myimunify.py
back
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

logger = logging.getLogger(__name__)


class MyImunifyPlugin(MessageSink, MessageSource):
    def __init__(self):
        self._previous_myimunify_status = MyImunifyConfig.ENABLED
        self._loop = None
        self._sink = None

    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()