0x1998 - MANAGER

Düzenlenen Dosya: send_server_config.py

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 register_lock_file
from defence360agent.utils import Scope, recurring_check
from defence360agent.utils.check_lock import check_lock
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.subsys.shared_disabled_rules import get_shared_disabled_rules_list

logger = getLogger(__name__)

NON_IMUNIFY_VENDOR = "non-imunify-vendor"
NO_VENDORS_INSTALLED = "no-vendors-installed"
LOCK_FILE = register_lock_file("send-server-config", Scope.IM360)

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()
        for shared_disabled_rule in get_shared_disabled_rules_list():
            # Ensuring shared disabled rules compatible with current DisabledRule
            #  keys format.
            msg["disabled_rules"].append(
                {
                    DisabledRule.plugin.name: shared_disabled_rule["plugin"],
                    DisabledRule.rule_id.name: shared_disabled_rule["rule_id"],
                    DisabledRule.name.name: "shared-disabled-rule",
                },
            )

msg["agent_global_config"][
            "CORE.is_go_agent"
        ] = Core.GO_FLAG_FILE.exists()

return msg

geri dön