22

✏️ Editing: ossec_rules_checker.py

import os

from defence360agent.contracts.plugins import MessageSink, Scope
from defence360agent.subsys.persistent_state import register_lock_file
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 defence360agent.utils.check_lock import check_lock

LOCK_FILE = register_lock_file("ossec-rules-check", Scope.IM360)

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)

MR SKB