1#@!#!123s

: / opt / imunify360 / venv / lib / python3.11 / site-packages / im360 / plugins /
Filename : ossec_rules_checker.py
back
import asyncio
import os

from pathlib import Path

from defence360agent.contracts.plugins import MessageSink
from defence360agent.subsys.persistent_state import PERSISTENT_STATE_DIR
from defence360agent.utils import recurring_check
from defence360agent.utils.common import LooseVersion
from im360.subsys import ossec
from im360.subsys.ossec import rules_update_lock
from im360.utils.check_lock import check_lock

LOCK_FILE = PERSISTENT_STATE_DIR / ".ossec-rules-check.lock"


class OssecRulesChecker(MessageSink):
    """Plugin that reinstall ossec rules if a newer version is available
    or there are no rules installed.
    """

    def __init__(self):
        self.loop = None
        self.task = None

    async def create_sink(self, loop):
        self.loop = loop
        self.task = loop.create_task(self.check_rules_recurrent())

    async def shutdown(self):
        if self.task:
            self.task.cancel()
            await self.task

    @recurring_check(
        check_lock,
        check_period_first=True,
        check_lock_period=3600,
        lock_file=LOCK_FILE,
    )
    async def check_rules_recurrent(self):
        await self.check_rules()

    async def check_rules(self):
        if rules_update_lock.locked():
            # Already being updated via imunify files
            return

        if not os.path.isdir(ossec.ETC_DIR):
            # ossec is not ready yet
            return

        installed_version = ossec.get_rules_installed_version()
        available_version = ossec.get_rules_version()
        if available_version is None:
            return
        available_version = LooseVersion(available_version)

        if (installed_version is None) or (
            available_version > installed_version
        ):
            await ossec.on_files_update(None, is_updated=True)