MAINHACK

Viewing / Editing File: strategy_checker.py

File is not writable. Editing disabled.
"""Plugin that detects whether IDS has changed."""
import asyncio
import contextlib
import logging

from defence360agent import utils
from defence360agent.contracts import plugins, sentry, messages
from im360.contracts import config
from im360.internals import strategy

logger = logging.getLogger(__name__)


class IDSSensor(plugins.MessageSource):
    """Send StrategyChange message when IDS changes."""

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.__task = None

    async def create_source(self, loop, sink):
        # implement MessageSource.create_source()
        self._loop = loop
        self._sink = sink
        self.__task = loop.create_task(self._detect_strategy_change())

    async def shutdown(self):
        # implement BasePlugin.shutdown()
        if self.__task is None or self.__task.done():
            return
        self.__task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await self.__task

    @utils.abort_agent_on(Exception)
    async def _detect_strategy_change(self):
        """Check whether IDS has changed periodically."""
        with contextlib.suppress(asyncio.CancelledError):
            while True:
                await self._refresh_strategy()
                await asyncio.sleep(
                    config.Subsys.THIRD_PARTY_IDS_CHECK_TIMEOUT
                )

    async def _refresh_strategy(self):
        try:
            new_strategy = await strategy.Strategy.detect()
        except asyncio.CancelledError:
            pass
        except Exception as e:
            logger.error("Failed to detect strategy, %s", e)
        else:
            if new_strategy != strategy.Strategy.current:
                await self._on_strategy_changed(new_strategy)

    async def _on_strategy_changed(self, new_strategy):
        logger.info(
            "Strategy changed: %s -> %s",
            strategy.Strategy.current,
            new_strategy,
        )

        sentry.set_strategy(new_strategy)

        strategy.Strategy.current = new_strategy

        await self._sink.process_message(
            messages.MessageType.StrategyChange(strategy=new_strategy)
        )
Cancel / Back
[ MAINHACK ]

Viewing / Editing File: strategy_checker.py

Server Info

System Features

User Info
