ACIL FM

Nama	Ukuran	Aksi
__pycache__	-	Open
accumulate.py	3.63 MB	View DL Edit
analyst_cleanup_update.py	5.53 MB	View DL Edit
backup_info_sender.py	3.06 MB	View DL Edit
cagefs.py	5.17 MB	View DL Edit
checkpoint.py	1.23 MB	View DL Edit
client.py	10.6 MB	View DL Edit
config_merger.py	828 B	View DL Edit
config_watcher.py	1.89 MB	View DL Edit
event_hook_executor.py	777 B	View DL Edit
event_monitor.py	3.32 MB	View DL Edit
event_monitor_message_processor.py	6.33 MB	View DL Edit
files_recurring_update.py	1.09 MB	View DL Edit
icontact_sender.py	4.42 MB	View DL Edit
idle_time_out.py	1.21 MB	View DL Edit
lve_utils_install.py	1.58 MB	View DL Edit
myimunify.py	1.98 MB	View DL Edit
ping.py	536 B	View DL Edit
send_domain_list.py	2.78 MB	View DL Edit
send_server_config.py	10.96 MB	View DL Edit
__init__.py	0 B	View DL Edit

Edit file: /opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/plugins/backup_info_sender.py

import asyncio
import time
from contextlib import suppress
from datetime import timedelta
from logging import getLogger
from typing import Union

from defence360agent.contracts.messages import MessageType
from defence360agent.contracts.plugins import MessageSource
from defence360agent.subsys.backup_systems import (
    get_current_backend,
    get_last_backup_timestamp,
)
from defence360agent.subsys.persistent_state import load_state, save_state
from defence360agent.utils import Scope, recurring_check

logger = getLogger(__name__)

SEND_INTERVAL = int(timedelta(hours=24).total_seconds())
RECURRING_CHECK_INTERVAL = 5

class BackupInfoSender(MessageSource):
    """Send user backup statistics to CH periodically"""

SCOPE = Scope.IM360

async def create_source(self, loop, sink):
        self._loop = loop
        self._sink = sink
        self._send_event = asyncio.Event()
        self._last_send_timestamp = self.load_last_send_timestamp()
        self._check_task = self._loop.create_task(
            self._recurring_check_data_to_send()
        )
        self._send_stat_task = self._loop.create_task(
            self._recurring_send_stat()
        )

async def shutdown(self):
        for task in [self._check_task, self._send_stat_task]:
            task.cancel()
            with suppress(asyncio.CancelledError):
                await task
        self.save_last_send_timestamp()

@staticmethod
    def is_valid_timestamp(timestamp: Union[int, float]) -> bool:
        return isinstance(timestamp, (int, float)) and timestamp > 0

def save_last_send_timestamp(self, ts: Union[int, float] = None):
        timestamp = self._last_send_timestamp if ts is None else ts
        if not self.is_valid_timestamp(timestamp):
            logger.warning("Invalid timestamp: %s", timestamp)
            return
        save_state("BackupInfoSender", {"last_send_timestamp": timestamp})

def load_last_send_timestamp(self):
        timestamp = load_state("BackupInfoSender").get("last_send_timestamp")
        if not self.is_valid_timestamp(timestamp):
            logger.warning("Invalid timestamp loaded, resetting to 0")
            timestamp = 0
        return timestamp

@recurring_check(RECURRING_CHECK_INTERVAL)
    async def _recurring_check_data_to_send(self):
        if time.time() - self._last_send_timestamp >= SEND_INTERVAL:
            self._send_event.set()

@recurring_check(0)
    async def _recurring_send_stat(self):
        await self._send_event.wait()
        try:
            await self._send_server_config()
        except Exception as e:
            logger.exception("Failed to collect backup info: %s", e)
        finally:
            # Ensure backup info is not sent too frequently, even after an error
            self._last_send_timestamp = time.time()
            self._send_event.clear()

async def _send_server_config(self):
        confg_msg = MessageType.BackupInfo(
            backup_provider_type=get_current_backend(),
            last_backup_timestamp=await get_last_backup_timestamp(),
        )
        await self._sink.process_message(confg_msg)

Batal