HEX
Server: LiteSpeed
System: Linux php-prod-3.spaceapp.ru 5.15.0-151-generic #161-Ubuntu SMP Tue Jul 22 14:25:40 UTC 2025 x86_64
User: sarli3128 (1010)
PHP: 7.4.33
Disabled: pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare,
Upload Files
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)