X7ROOT File Manager
Current Path:
/opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/plugins
opt
/
imunify360
/
venv
/
lib
/
python3.11
/
site-packages
/
defence360agent
/
plugins
/
📁
..
📄
__init__.py
(0 B)
📁
__pycache__
📄
accumulate.py
(3.53 KB)
📄
analyst_cleanup_update.py
(5.53 KB)
📄
backup_info_sender.py
(3.06 KB)
📄
checkpoint.py
(1.23 KB)
📄
client.py
(9.7 KB)
📄
config_merger.py
(828 B)
📄
config_watcher.py
(1.89 KB)
📄
event_hook_executor.py
(777 B)
📄
event_monitor.py
(3.32 KB)
📄
event_monitor_message_processor.py
(6.33 KB)
📄
files_recurring_update.py
(1.09 KB)
📄
icontact_sender.py
(4.36 KB)
📄
idle_time_out.py
(1.21 KB)
📄
lve_utils_install.py
(1.58 KB)
📄
myimunify.py
(1.93 KB)
📄
ping.py
(536 B)
📄
send_domain_list.py
(2.79 KB)
📄
send_server_config.py
(10.37 KB)
Editing: idle_time_out.py
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()
Upload File
Create Folder