MAINHACK

Viewing / Editing File: myimunify_modsec_checker.py

File is not writable. Editing disabled.
"""
Plugin is used to turn on/off optional modsec rules
provided by "MyImunify" feature
"""
from abc import abstractmethod
import asyncio
from logging import getLogger

import blinker

from defence360agent.contracts.config import MyImunifyConfig
from defence360agent.contracts.messages import MessageType
from defence360agent.contracts.permissions import myimunify_protection_enabled
from defence360agent.contracts.plugins import MessageSink, expect
from defence360agent.subsys import web_server
from defence360agent.utils import nice_iterator, recurring_check
from im360.subsys.panels.base import use_modsec_lock
from im360.subsys.panels.hosting_panel import HostingPanel

logger = getLogger(__name__)


class Signals:
    updated = blinker.Signal()


# This plugin is disabled for now, because on first stage
# we do not manage modsec for MyImunify.protection=enabled users
class ModsecMyImunifyChecker(MessageSink):
    """
    MyImunify modsec rules are checked in the following cases:
      - on agent startup
      - on config update (MYIMUNIFY.enable setting change)
      - on user permissions update
    """
    # NOTE: This method makes inspect.isabstract() thinks, that
    #       this plugin is abstract and therefore shouldn't be
    #       started during plugin instantiation
    #       (check BasePlugin.get_active_plugins())
    #
    #       To enable plugin, simply remove this method
    @abstractmethod
    def abstract(self):
        pass

    def __init__(self):
        self._loop = None
        self._task = None
        self._previous_myimunify_status = None
        self._updating_modsec_rules_required = asyncio.Event()
        Signals.updated.connect(self._on_myimunify_update)

    def _on_myimunify_update(self, user):
        self._updating_modsec_rules_required.set()

    async def create_sink(self, loop):
        self._loop = loop
        # Note: modsec rules should be checked on agent startup
        # (first ConfigUpdate message)
        self._task = self._loop.create_task(
            recurring_check(30)(
                self.apply_myimunify_modsec_rules_for_domains
            )()
        )

    async def shutdown(self):
        if self._task is not None:
            self._task.cancel()
            await self._task

    async def apply_myimunify_modsec_rules_for_domains(self):
        await self._updating_modsec_rules_required.wait()
        self._updating_modsec_rules_required.clear()
        await self._apply_myimunify_modsec_rules_for_domains()

    @use_modsec_lock
    async def _apply_myimunify_modsec_rules_for_domains(self):
        # To avoid possible loss of an user during check
        # try to apply MyImunify settings for all domains on each attempt
        panel = HostingPanel()
        users_domains = await panel.get_domains_per_user()
        enabled_users_domains = {}
        disabled_users_domains = {}
        if MyImunifyConfig.ENABLED:
            async for user, domains in nice_iterator(
                users_domains.items(), chunk_size=1000
            ):
                if myimunify_protection_enabled(user):
                    enabled_users_domains[user] = domains
                else:
                    disabled_users_domains[user] = domains
        else:
            logger.info("Disable MyImunify settings for all domains")
            disabled_users_domains = users_domains
        try:
            logger.info("Apply MyImunify settings")
            updated_domains = (
                await panel.apply_myimunify_modsec_rules_for_domains(
                    enabled_users_domains=enabled_users_domains,
                    disabled_users_domains=disabled_users_domains
                )
            )
            if updated_domains:
                logger.info(
                    "MyImunify settings were updated for domains: %s",
                    updated_domains
                )
                await web_server.graceful_restart()
        except Exception as exc:
            logger.exception(
                "Something went wrong during applying "
                "modsec MyImunify rules: %s", exc
            )

    @expect(MessageType.ConfigUpdate)
    async def on_config_update(self, message: MessageType.ConfigUpdate):
        myimunify_enabled = MyImunifyConfig.ENABLED
        if self._previous_myimunify_status != myimunify_enabled:
            self._updating_modsec_rules_required.set()
            self._previous_myimunify_status = myimunify_enabled
Cancel / Back
[ MAINHACK ]

Viewing / Editing File: myimunify_modsec_checker.py

Server Info

System Features

User Info
