MAINHACK

Viewing / Editing File: modsec_ruleset_checker.py

File is not writable. Editing disabled.
import asyncio
from logging import getLogger

from defence360agent import utils
from defence360agent.contracts.messages import MessageType
from defence360agent.contracts.plugins import MessageSink, expect
from defence360agent.files import FILES_DIR
from defence360agent.subsys.panels.base import (
    ModsecVendorsError,
    PanelException,
)
from defence360agent.utils import recurring_check
from defence360agent.utils.common import DAY
from im360.contracts.config import Modsec
from im360.subsys.panels.base import (
    ModsecImunifyVendorNotInstalled,
    ModsecNotInstalledVendors,
    is_modsec_locked,
    use_modsec_lock,
)
from im360.subsys.panels.hosting_panel import HostingPanel
from im360.subsys.panels.update_hooks import (
    update_account_compromise_prevention_rule_state,
    update_vendors,
)

logger = getLogger(__name__)


class ModsecRulesetChecker(MessageSink):
    VERSION_FILE = FILES_DIR / "modsec/v2/VERSION"

    def __init__(self):
        self._loop = None
        self._task = None
        self._install_rules_task = None
        self.panel = None

    async def create_sink(self, loop):
        self.panel = HostingPanel()
        self._loop = loop
        self._task = self._loop.create_task(self.reinstall_vendor_if_needed())
        self._install_rules_task = self._loop.create_task(self.update_rules())

    async def shutdown(self):
        self._task.cancel()
        self._install_rules_task.cancel()
        await self._task
        await self._install_rules_task

    @use_modsec_lock
    async def reinstall_vendor_if_needed(self):
        panel = self.panel or HostingPanel()
        token = panel.installing_settings_var.set(True)
        try:
            try:
                # We can't do anything if Panel doesn't support modsec
                # or generate specific panel error
                installed_im360_vendor = await panel.get_i360_vendor_name()
            except (ModsecNotInstalledVendors,
                    ModsecImunifyVendorNotInstalled):
                # this means that we can't find any modsec vendors or
                # there is no im360 vendor
                installed_im360_vendor = None
            if installed_im360_vendor is None or \
                    Modsec.RULESET.lower() not in installed_im360_vendor:
                logger.info(
                    'Installed i360 vendor {} does not match expected type of '
                    'ruleset: {}\n Trying to reinstall modsec ruleset'.format(
                        str(installed_im360_vendor), str(Modsec.RULESET)))
                await panel.apply_modsec_files_update()
        except asyncio.CancelledError:
            raise
        except Exception as e:
            logger.error(
                'Something went wrong during reinstalling modsec ruleset: {}'.
                format(e))
        finally:
            panel.installing_settings_var.reset(token)

    @recurring_check(DAY / 2)
    async def update_rules(self):
        """
        Reinstall ModSec rules if installed version is not the same as in files
        """
        await self._update_rules()

    async def _update_rules(self):
        if is_modsec_locked():
            # Already being updated via imunify files
            return

        try:
            installed_vendor_version = (
                await self.panel.get_i360_vendor_version()
            )
        except (ModsecVendorsError, PanelException):
            return
        available_vendor_version = self.get_vendor_version_from_files()
        if installed_vendor_version and available_vendor_version:
            if installed_vendor_version != available_vendor_version:
                logger.info(
                    "Reinstalling ModSec rules."
                    " Installed: %s. Available: %s",
                    installed_vendor_version,
                    available_vendor_version,
                )
                await update_vendors(None, is_updated=True)

    @classmethod
    def get_vendor_version_from_files(cls) -> str:
        try:
            with open(cls.VERSION_FILE) as f:
                return f.read().strip()
        except FileNotFoundError:
            return ""

    @expect(MessageType.ConfigUpdate)
    @utils.log_error_and_ignore()
    async def on_config_update(self, _):
        await self.reinstall_vendor_if_needed()

    @expect(MessageType.ConfigUpdate)
    async def check_cms_account_compromise_prevention(self, _):
        """
        Update *WP_REDIRECT_CONF* to apply changes
        to *cms_account_compromise_prevention* config setting.
        """
        await update_account_compromise_prevention_rule_state()
Cancel / Back
[ MAINHACK ]

Viewing / Editing File: modsec_ruleset_checker.py

Server Info

System Features

User Info
