✏️ Editing: plugins.py Read Only

import asyncio
import logging

from defence360agent import utils
from defence360agent.contracts import messages, plugins

logger = logging.getLogger(__name__)

class IDSAwareMessageSink(plugins.MessageSink):
    """A message sink that accepts messages taking into account the
    current IDS (strategy).

"""

STRATEGY = None  # no IDS preference by default

@plugins.expect(messages.MessageType.StrategyChange)
    async def on_strategy_change(self, message):
        """De/activate the plugin on strategy change if necessary."""
        # - activate sink if it has no IDS preference
        # - activate sink if it can work with the current IDS (as
        #   indicated by STRATEGY)
        # - deactivate sink if it can't work with the current IDS
        if self.STRATEGY is None or message.strategy == self.STRATEGY:
            if not self.is_active():
                try:
                    await self.activate()
                except asyncio.CancelledError:
                    raise
                except Exception:
                    logger.exception("Failed to activate %r plugin", self)
                    utils.fail_agent_service()
                else:
                    logger.info("Plugin %r is activated", self)
        elif self.is_active():
            try:
                await self.deactivate()
            except (Exception, asyncio.CancelledError):
                logger.exception("Failed to deactivate %r plugin", self)
                utils.fail_agent_service()
            else:
                logger.info("Plugin %r is deactivated", self)

async def process_message(self, message):
        if self.is_active() or isinstance(
            message, messages.MessageType.StrategyChange
        ):
            return await super().process_message(message)

async def shutdown(self):
        if self.is_active():
            try:
                await self.deactivate()
            except (Exception, asyncio.CancelledError):
                logger.exception("Failed to deactivate %r plugin", self)
                # NOTE: don't propagate errors on shutdown
            else:
                logger.info("Plugin %r is deactivated", self)
        return await super().shutdown()

async def activate(self):
        """This method will be called for plugins with specified strategy
        when this strategy becomes active.
        """
        self._mark_as_active()

async def deactivate(self):
        """This method will be called for plugins with specified strategy
        when this strategy becomes inactive.
        """
        self._mark_as_inactive()

def is_active(self):
        """Indicates if plugin is active."""
        return getattr(self, "_is_active", False)

def _mark_as_active(self):
        self._is_active = True

def _mark_as_inactive(self):
        self._is_active = False

🔐 Sid Gifari File Manager Pro