1#@!#!123s

: / opt / imunify360 / venv / lib / python3.11 / site-packages / im360 / subsys / panels / generic /
Filename : mod_security.py
back
import asyncio
import json
import logging
import os
import shutil
import zipfile
from contextlib import suppress
from pathlib import Path
from typing import Optional, List, Dict
from urllib.parse import urlparse

from defence360agent.api.integration_conf import IntegrationConfig
from defence360agent.subsys.panels.base import PanelException
from defence360agent.utils import (
    atomic_rewrite,
    async_lru_cache,
)
from im360.subsys.panels.base import (
    FilesVendor,
    FilesVendorList,
    ModSecSettingInterface,
    ModSecurityInterface,
    APACHE,
    LITESPEED,
    NGINX,
    OPENLITESPEED,
)
from defence360agent.subsys.web_server import graceful_restart_sync

logger = logging.getLogger(__name__)

GENERIC_DIR = "/etc/sysconfig/imunify360/generic/"
MODSEC_CONF = os.path.join(GENERIC_DIR, "modsec.conf")
MODSEC_CONF_DIR = os.path.join(GENERIC_DIR, "modsec.conf.d")
RULES_DIR = "/var/imunify360/modsec/generic/rules"
GLOBAL_DISABLED_RULES_CONFIG = os.path.join(
    GENERIC_DIR, "global_disabled_rules.conf"
)
MODSEC_SETTINGS_CONF = os.path.join(GENERIC_DIR, "modsec2.imunify.conf")
MODSEC_SETTINGS_CONF_NGINX = os.path.join(
    GENERIC_DIR, "nginx.modsec3.imunify.conf"
)
_APACHE_COMPATIBLE_WEBSERVERS = (APACHE, LITESPEED, OPENLITESPEED)
_NGINX_COMPATIBLE_WEBSERVERS = (NGINX,)


class GenericPanelModSecException(PanelException):
    pass


class _ModSecSettings(ModSecSettingInterface):
    #: apache-compatible modsec rules
    I360_RULES_INCLUDE_APACHE = (
        "<IfModule security2_module>"
        '\n\tIncludeOptional "{}/*.conf"'
        # Include without asterisk will error out on Apache < 2.4.30.
        '\n\tIncludeOptional "{}"'
        '\n\tIncludeOptional "{}/*"'
        "\n</IfModule>\n"
        "Include {}\n"
    ).format(
        RULES_DIR,
        GLOBAL_DISABLED_RULES_CONFIG,
        MODSEC_CONF_DIR,
        MODSEC_SETTINGS_CONF,
    )

    I360_RULES_INCLUDE_NGINX = (
        "include {}/*.conf\ninclude {}\ninclude {}/*\ninclude {}\n"
    ).format(
        RULES_DIR,
        GLOBAL_DISABLED_RULES_CONFIG,
        MODSEC_CONF_DIR,
        MODSEC_SETTINGS_CONF_NGINX,
    )

    @classmethod
    def get_i360_rules(cls):
        """Return modsec rules for the web server."""
        return (
            cls.I360_RULES_INCLUDE_NGINX
            if _get_web_server_type() == NGINX
            else cls.I360_RULES_INCLUDE_APACHE
        )

    @classmethod
    async def revert(cls, **kwargs):
        try:
            with open(MODSEC_CONF, "w"):
                # Opening in write mode is enough to truncate the file.
                pass
        except (IOError, OSError):
            logger.warning("%s does not exist")

    @classmethod
    async def apply(cls):
        """Populate modsec.conf"""
        # Note: customer is responsible for including it in the server config
        # https://docs.imunify360.com/stand_alone/#interaction-with-modsecurity
        rules = cls.get_i360_rules()
        try:
            atomic_rewrite(MODSEC_CONF, rules, backup=False)
        except (IOError, OSError) as err:
            raise GenericPanelModSecException(
                "Could not access ModSec config for generic panel"
            ) from err


class GenericPanelModSecurity(ModSecurityInterface):
    REBUILD_HTTPDCONF_CMD = None

    @classmethod
    def _get_conf_dir(cls) -> str:
        return MODSEC_CONF_DIR

    @classmethod
    async def installed_modsec(cls):
        return True

    async def _install_settings(self):
        # configure modsec on a compatible web server
        try:
            await _ModSecSettings.apply()
        except GenericFilesVendorList.CompatiblityCheckFailed as e:
            logger.error(str(e))  # send to sentry, ignore otherwise
        else:
            await GenericFilesVendorList.apply()
            graceful_restart_sync()

    async def modsec_get_directive(self, directive_name, default=None):
        """
        Used for `imunify360-agent check modsec directives` which
        allows to check whether the global ModSecurity directives have values
        recommended by Imunify360.

        N/A for GenericPanel
        """
        raise NotImplementedError

    async def reset_modsec_directives(self):
        """
        Used for `imunify360-agent fix modsec directives` to
        reset ModSecurity settings to values chosen by Imunify360

        N/A for GenericPanel
        """
        raise NotImplementedError

    async def reset_modsec_rulesets(self):
        # Unused.
        raise NotImplementedError

    async def revert_settings(self):
        await _ModSecSettings.revert()
        await GenericFilesVendorList.revert()
        graceful_restart_sync()

    @classmethod
    def detect_cwaf(cls):
        # Unused.
        return False

    @classmethod
    @async_lru_cache(maxsize=1)
    async def _get_release_info_from_file(cls) -> Optional[dict]:
        modsec_release_file = await cls.build_vendor_file_path(
            vendor="Not used", filename="RELEASE"
        )
        try:
            with modsec_release_file.open() as release_f:
                json_data = json.load(release_f)
            return json_data
        except (OSError, json.JSONDecodeError):
            return None

    @classmethod
    async def modsec_vendor_list(cls) -> list:
        """Return a list of installed ModSecurity vendors."""
        vendor_list = []
        vendor = await cls.get_modsec_vendor_from_release_file()
        if vendor:
            vendor_list.append(vendor)
        return vendor_list

    @classmethod
    async def enabled_modsec_vendor_list(cls) -> list:
        """Return a list of enabled ModSecurity vendors."""
        try:
            with open(MODSEC_CONF, "r") as f:
                if not any(
                    line.strip() == _ModSecSettings.get_i360_rules()
                    for line in f
                ):
                    return []
        except FileNotFoundError as e:
            raise GenericPanelModSecException(
                "Rules not configured yet"
            ) from e
        return await cls.modsec_vendor_list()

    @classmethod
    async def build_vendor_file_path(cls, vendor: str, filename: str) -> Path:
        return Path(RULES_DIR) / filename

    @classmethod
    async def _apply_modsec_files_update(cls):
        await GenericFilesVendorList.install_or_update()

    @classmethod
    def get_audit_log_path(cls):
        try:
            return IntegrationConfig.to_dict()["web_server"][
                "modsec_audit_log"
            ]
        except KeyError as err:
            raise GenericPanelModSecException(
                "Integration config is missing modsec_audit_log field"
            ) from err

    @classmethod
    def get_audit_logdir_path(cls):
        try:
            return IntegrationConfig.to_dict()["web_server"][
                "modsec_audit_logdir"
            ]
        except KeyError as err:
            raise GenericPanelModSecException(
                "Integration config is missing modsec_audit_logdir field"
            ) from err

    @classmethod
    def _generate_disabled_rules_apache_config(cls, rule_list: list) -> str:
        # *super* uses Apache compatible config syntax
        return super().generate_disabled_rules_config(rule_list)

    @classmethod
    def _generate_disabled_rules_nginx_config(cls, rule_list: list) -> str:
        text = ""
        if rule_list:
            text = "SecRuleRemoveById {}\n".format(
                " ".join(map(str, rule_list))
            )
        return text

    @classmethod
    def generate_disabled_rules_config(cls, rule_list: list) -> str:
        """
        Returns config text with disabled rules according to used webserver.
        """
        web_server = _get_web_server_type()
        if web_server in _APACHE_COMPATIBLE_WEBSERVERS:
            return cls._generate_disabled_rules_apache_config(rule_list)
        if web_server in _NGINX_COMPATIBLE_WEBSERVERS:
            return cls._generate_disabled_rules_nginx_config(rule_list)

    @classmethod
    def write_global_disabled_rules(cls, rule_list: List[str]):
        atomic_rewrite(
            GLOBAL_DISABLED_RULES_CONFIG,
            cls.generate_disabled_rules_config(rule_list),
            backup=False,
        )

    @classmethod
    async def sync_global_disabled_rules(cls, rule_list: List[str]):
        cls.write_global_disabled_rules(rule_list)

    @classmethod
    async def sync_disabled_rules_for_domains(
        cls, domain_rules_map: Dict[str, list]
    ):
        """
        Disable mod_security rules on domain level for each domain
        specified in a map.
        """

        def to_byte_row(domain: str, config: str) -> bytes:
            return json.dumps({"domain": domain, "content": config}).encode(
                "utf-8"
            )

        try:
            domain_config_script = IntegrationConfig.to_dict()[
                "integration_scripts"
            ]["modsec_domain_config_script"]
        except KeyError as err:
            raise GenericPanelModSecException(
                "Integration config is missing"
                " modsec_domain_config_script field"
            ) from err

        if not domain_config_script:
            raise GenericPanelModSecException(
                "modsec_domain_config_script specified in integration.conf"
                " is empty"
            )

        lines = [
            to_byte_row(domain, cls.generate_disabled_rules_config(rule_list))
            for domain, rule_list in domain_rules_map.items()
        ]

        # don't send empty line to the customer's script
        # the script may not expect this, according to documentation
        # https://docs.imunify360.com/stand_alone/#interaction-with-modsecurity
        if lines:
            try:
                proc = await asyncio.create_subprocess_exec(
                    *domain_config_script.split(),
                    stdin=asyncio.subprocess.PIPE,
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                )
            except FileNotFoundError as err:
                raise GenericPanelModSecException(
                    "Could not run modsec_domain_config_script: {}."
                    "\nTry specifying a full path.".format(err)
                )
            # add '\n' at the end to avoid possible problems with
            # custom script (see DEF-12901 for details)
            data = b"\n".join(lines) + b"\n"
            stdout, stderr = await proc.communicate(input=data)
            if proc.returncode:
                raise GenericPanelModSecException(
                    "Failed to update domains config using %s\n"
                    "STDOUT:\n%r\nSTDERR:\n%r"
                    % (
                        domain_config_script,
                        stdout,
                        stderr,
                    )
                )
            else:
                logger.info("Successfully updated ModSec domain configs")


class GenericFilesVendor(FilesVendor):
    modsec_interface = GenericPanelModSecurity

    def _add_vendor(self, url, name, *args, **kwargs):
        pass

    async def _remove_vendor(self, vendor, *args, **kwargs):
        with suppress(FileNotFoundError):
            shutil.rmtree(RULES_DIR)
        os.makedirs(RULES_DIR, exist_ok=True, mode=0o700)

    async def apply(self):
        shutil.rmtree(RULES_DIR, ignore_errors=True)
        os.makedirs(RULES_DIR, exist_ok=True, mode=0o700)
        with zipfile.ZipFile(self._item["local_path"]) as zf:
            for member in zf.namelist():
                filename = os.path.basename(member)
                if not filename:
                    continue
                target = os.path.join(RULES_DIR, filename)
                with zf.open(member) as src, open(target, "wb") as dst:
                    shutil.copyfileobj(src, dst)

    def _vendor_id(self):
        basename = os.path.basename(urlparse(self._item["url"]).path)
        basename_no_zip, _ = os.path.splitext(basename)
        return basename_no_zip


class GenericFilesVendorList(FilesVendorList):
    files_vendor = GenericFilesVendor
    modsec_interface = GenericPanelModSecurity

    @classmethod
    def vendor_fit_panel(cls, item):
        # use plesk item from description.json cause plesk,
        # da and generic panel share the same zip
        return item["name"].endswith("plesk")

    @classmethod
    async def _get_compatible_name(cls, installed_vendors):
        web_server = _get_web_server_type()
        basename = "imunify360-{modsec3}{ruleset_suffix}-{webserver}-{panel}"
        return basename.format(
            modsec3="modsec3-" * (web_server == NGINX),
            ruleset_suffix=cls.get_ruleset_suffix(),
            webserver=web_server,
            panel="plesk",
        )


def _get_web_server_type() -> Optional[str]:
    """Return web_server.server_type or None."""
    try:
        web_server = IntegrationConfig.to_dict()["web_server"]["server_type"]
    except KeyError as err:
        raise GenericPanelModSecException(
            "Integration config is missing server_type field"
        ) from err
    if web_server not in (APACHE, LITESPEED, NGINX, OPENLITESPEED):
        raise GenericFilesVendorList.CompatiblityCheckFailed(
            "Imunify360 mod_security vendor does not support '{}' "
            "webserver".format(web_server)
        )
    return web_server