ꦫꦣꦺꦤ꧀ꦄꦤ꧀ꦠꦱꦺꦤ

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

from logging import getLogger

from defence360agent.api import inactivity
from defence360agent.contracts.config import SimpleRpc
from defence360agent.contracts.plugins import MessageSink
from defence360agent.utils import clip, fail_agent_service, recurring_check

logger = getLogger(__name__)

class IdleTimeOutCheck(MessageSink):
    async def create_sink(self, loop):
        self._loop = loop
        if SimpleRpc.SOCKET_ACTIVATION:
            inactivity.track.reset_timer()
            self._task = loop.create_task(
                recurring_check(
                    period=clip(
                        SimpleRpc.INACTIVITY_TIMEOUT // 5, low=1, high=60
                    ),
                )(
                    self._check_timeout,
                )()
            )
        else:
            self._task = None

async def shutdown(self):
        if self._task:
            self._task.cancel()
            # CancelledError is handled by @recurring_check():
            await self._task

async def _check_timeout(self):
        logger.info("Periodical check %s ", inactivity.track)
        if inactivity.track.is_timeout():
            logger.warning("Shutting down due to inactivity.")
            fail_agent_service()

[-] event_monitor.py [edit]
[-] lve_utils_install.py [edit]
[-] checkpoint.py [edit]
[-] icontact_sender.py [edit]
[-] event_hook_executor.py [edit]
[+] ..
[-] accumulate.py [edit]
[-] idle_time_out.py [edit]
[-] client.py [edit]
[-] ping.py [edit]
[+] __pycache__
[-] config_watcher.py [edit]
[-] cagefs.py [edit]
[-] analyst_cleanup_update.py [edit]
[-] myimunify.py [edit]
[-] send_server_config.py [edit]
[-] backup_info_sender.py [edit]
[-] __init__.py [edit]
[-] send_domain_list.py [edit]
[-] event_monitor_message_processor.py [edit]
[-] config_merger.py [edit]
[-] files_recurring_update.py [edit]