1#@!#!123s

: / opt / imunify360 / venv / lib / python3.11 / site-packages / im360 / plugins /
Filename : send_server_config.py
back
from logging import getLogger

from defence360agent.contracts.config import Core
from defence360agent.plugins.send_server_config import (
    SendServerConfig as SendServerConfigAV,
)
from defence360agent.subsys.panels.base import (
    ModsecVendorsError,
    PanelException,
)
from defence360agent.subsys.persistent_state import PERSISTENT_STATE_DIR
from defence360agent.utils import Scope, recurring_check
from im360.model.incident import DisabledRule
from im360.subsys import ossec
from im360.subsys.panels import hosting_panel
from im360.subsys.panels.base import (
    ModsecImunifyVendorNotInstalled,
    ModsecNotInstalledVendors,
)
from im360.utils.check_lock import check_lock


logger = getLogger(__name__)


NON_IMUNIFY_VENDOR = "non-imunify-vendor"
NO_VENDORS_INSTALLED = "no-vendors-installed"
LOCK_FILE = PERSISTENT_STATE_DIR / ".send-server-config.lock"


class SendServerConfig(SendServerConfigAV):
    SCOPE = Scope.IM360

    async def create_source(self, loop, sink):
        self._loop = loop
        self._sink = sink

        self._task = self._loop.create_task(
            recurring_check(
                check_lock,
                check_period_first=True,
                check_lock_period=self._period,
                lock_file=LOCK_FILE,
            )(self._send_server_config)()
        )

    async def _create_server_config_msg(self):
        msg = await super()._create_server_config_msg()
        hp = hosting_panel.HostingPanel()

        modsec_installed = await hp.installed_modsec()
        vendor_name = None
        vendor_version = None
        vendor_enabled = None
        active_vendors = None

        if modsec_installed:
            try:
                vendor_name = await hp.get_modsec_vendor_from_release_file()
            except ModsecImunifyVendorNotInstalled as e:
                logger.warning(str(e))
                vendor_name = NON_IMUNIFY_VENDOR
            except ModsecNotInstalledVendors as e:
                logger.warning(str(e))
                vendor_name = NO_VENDORS_INSTALLED
            except ModsecVendorsError as e:
                logger.warning(str(e))

            else:
                try:
                    # Do not forget to modify getting vendor_enabled flag
                    # when we will do DEF-9878
                    vendor_enabled = bool(
                        vendor_name in await hp.enabled_modsec_vendor_list()
                    )
                except PanelException as e:
                    logger.warning(
                        "Cannot determine if vendor is enabled "
                        "because we can't get enabled vendors "
                        "list: %s",
                        str(e),
                    )

                try:
                    vendor_version = await hp.get_i360_vendor_version()
                except (ModsecVendorsError, PanelException) as e:
                    logger.warning(
                        "Cannot determine Imunify360 vendor version: %s",
                        str(e),
                    )

        try:
            active_vendors = await hp.enabled_modsec_vendor_list()
        except Exception as e:
            logger.warning("Cannot determine active vendors: %s", str(e))

        modsec_msg = {
            "vendor": vendor_name,
            "enabled": vendor_enabled,
            "version": vendor_version,
            "modsec_installed": modsec_installed,
            "engine_mode": hp.get_modsec_engine_mode(),
            "active_vendors": active_vendors,
            "conf_files": hp.get_modsec_active_conf_files(),
            "vendor_updates": hp.get_modsec_vendor_updates(),
        }
        msg["modsec"] = modsec_msg
        msg["ossec"] = {"version": ossec.get_rules_version()}
        msg["disabled_rules"] = DisabledRule.as_list()
        msg["agent_global_config"][
            "CORE.is_go_agent"
        ] = Core.GO_FLAG_FILE.exists()

        return msg