Na}{

File Content: `strategy_getter.py`

"""Plugin that detects whether IDS has set on non-resident agent start."""
import asyncio
import contextlib
import logging

from defence360agent import utils

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

logger = logging.getLogger(__name__)


class StrategyGetter(plugins.MessageSource):
    """Send StrategyChange message on non-resident agent start."""

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

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

    async def shutdown(self):
        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 = strategy.Strategy.get()
        except asyncio.CancelledError:
            pass
        except Exception as e:
            logger.error("Failed to get 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(
            "Got new strategy: %s -> %s",
            strategy.Strategy.current,
            new_strategy,
        )
        strategy.Strategy.current = new_strategy

        await self._sink.process_message(
            messages.MessageType.StrategyChange(strategy=new_strategy)
        )

Name	Size	Permission
__pycache__	---	0755
__init__.py	0 bytes	0644
cagefs.py	5265 bytes	0644
cpanel_uploader.py	2777 bytes	0644
export_wblist.py	3854 bytes	0644
fgw.py	1747 bytes	0644
lfd.py	3508 bytes	0644
modsec_ruleset_checker.py	5031 bytes	0644
ossec_rules_checker.py	1803 bytes	0644
pam_manager.py	6829 bytes	0644
php_immunity.py	1438 bytes	0644
remoteip_install.py	1053 bytes	0644
repeater.py	1187 bytes	0644
send_server_config.py	4534 bytes	0644
service_manager.py	4756 bytes	0644
startup_actions.py	2483 bytes	0644
strategy_getter.py	2068 bytes	0644
waf_rules_configurator.py	2535 bytes	0644
whitelist_current_user.py	2817 bytes	0644

File Content: strategy_getter.py

File Content: `strategy_getter.py`