Edit File - File Manager

Edit File: mod_security.py

import logging
import os
from contextlib import suppress
from functools import lru_cache
from pathlib import Path
from typing import Dict, List, Optional
from urllib.parse import urlparse

import yaml
from defence360agent.contracts.config import ConfigFile
from defence360agent.subsys.panels.base import (
    ModsecVendorsError,
    forbid_dns_only,
)
from defence360agent.subsys.panels.cpanel.whm import (
    WHMAPIException,
    catch_exception,
    whmapi1,
)
from defence360agent.utils import (
    CheckRunError,
    async_lru_cache,
    atomic_rewrite,
    check_run,
    nice_iterator,
)
from im360.contracts.config import ModSecurityDirectives
from im360.subsys.panels.base import (
    MODSEC_NAME_TEMPLATE,
    FilesVendor,
    FilesVendorList,
    ModSecSettingInterface,
    ModSecurityInterface,
    skip_if_not_installed_modsec,
)

#: path to paths.conf file from the ea-apache24-config-runtime package
EA4_PATHS_CONF_PATH = Path("/etc/cpanel/ea4/paths.conf")
#: presence of the file indicates it is EasyApache 4
IS_EA4_PATH = Path("/etc/cpanel/ea4/is_ea4")
#: path to modsec_vendor script
MODSEC_VENDOR_BIN = "/usr/local/cpanel/scripts/modsec_vendor"
NON_CONFLICTING_RULESETS = (
    "comodo_apache",
    "comodo_litespeed",
    "imunify360_rules",
    "configserver",
)

logger = logging.getLogger(__name__)

async def _modsec_vendor_cmd(cmd, param):
    """
    :raise subprocess.CalledProcessError:
    """
    await check_run([MODSEC_VENDOR_BIN, cmd, param], raise_exc=WHMAPIException)

class ModSecSetting(ModSecSettingInterface):
    config_key = "prev_settings"

directives_ids = {
        "SecAuditEngine": "0",
        "SecConnEngine": "1",
        "SecRuleEngine": "2",
    }
    OFF = "Off"

@classmethod
    async def apply(cls):
        """
        Reset ModSecurity settings to values chosen by Imunify360.
        :return str: previous settings
        """

# save previous settings
        previous_settings = await cPanelModSecurity.get_settings(
            *cls.directives_ids
        )

# set settings from the Imunify360 config module
        for directive_name, directive_id in cls.directives_ids.items():
            await whmapi1(
                "modsec_set_setting",
                setting_id=cls.directives_ids[directive_name],
                state=getattr(ModSecurityDirectives, directive_name),
            )

await whmapi1("modsec_deploy_settings_changes")
        return ",".join(
            "{id}:{state}".format(id=cls.directives_ids[name], state=state)
            for name, state in sorted(previous_settings.items())
            if state.strip()
        )

@classmethod
    async def revert(cls, prev_setting_value):
        """apply the setting previous value"""

if prev_setting_value:
            for item in prev_setting_value.split(","):
                setting_id, state = item.split(":")
                if state.strip():
                    await whmapi1(
                        "modsec_set_setting",
                        setting_id=setting_id,
                        state=state,
                    )

await whmapi1("modsec_deploy_settings_changes")

class cPanelModSecurity(ModSecurityInterface):
    CWAF_INSTALLATION_DIR = "/var/cpanel/cwaf"

AUDIT_LOG_FILE = "/usr/local/apache/logs/modsec_audit.log"
    DISABLED_RULES_CONFIG_DIR = "/etc/apache2/conf.d/"
    # to load after modsec2.conf
    GLOBAL_DISABLED_RULES_CONFIG_FILENAME = "modsec2.i360_disabled_rules.conf"

PER_DOMAIN_DISABLED_RULES_CONFIG_FILENAME = "i360_modsec_disable.conf"
    # it should be included before disabled rules config
    # to ensure that MyImunify rules can be disabled by a customer
    PER_DOMAIN_MYIMUNIFY_RULES_CONFIG_FILENAME = "i360_001_myimunify.conf"

REBUILD_HTTPDCONF_CMD = ("/usr/local/cpanel/scripts/rebuildhttpdconf",)

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

@classmethod
    @lru_cache(maxsize=1)
    def _get_ea4_paths(cls) -> Optional[Dict[str, str]]:
        """Return a dict with ea4 paths or None for non-ea4 case."""
        if IS_EA4_PATH.exists():
            # easy apache 4
            with EA4_PATHS_CONF_PATH.open() as f:
                paths_config = dict(
                    map(str.strip, line.partition("=")[::2])
                    for line in f
                    if "=" in line
                )
            return paths_config

@classmethod
    def _get_userdata_dir(cls):
        ea4_paths_config = cls._get_ea4_paths()
        if ea4_paths_config is not None:
            return ea4_paths_config.get(
                "dir_conf_userdata", "/etc/apache2/conf.d/userdata"
            )
        else:
            # easy apache 3
            return "/usr/local/apache/conf/userdata"

@classmethod
    def _get_domain_confs_paths(cls, user, domain, *, conf_name):
        # use hardcode version according to
        # https://docs.cpanel.net/ea4/apache/modify-apache-virtual-hosts-with-include-files/
        # now supported version apache >= 2.4
        # EA3 is EOL
        confs = []
        version_str = "2_4"
        for vhost_type in ("ssl", "std"):
            confs.append(
                Path(cls._get_userdata_dir())
                / vhost_type
                / version_str
                / user
                / domain
                / conf_name
            )
        return confs

@classmethod
    def _delete_domain_conf(cls, user, domain, *, conf_name) -> bool:
        deleted = False
        for conf_path in cls._get_domain_confs_paths(
            user, domain, conf_name=conf_name
        ):
            with suppress(FileNotFoundError):
                conf_path.unlink(missing_ok=False)
                deleted = True
        return deleted

@classmethod
    def _add_domain_conf(cls, user, domain, *, conf_name, conf_text) -> bool:
        updated = False
        for conf_path in cls._get_domain_confs_paths(
            user, domain, conf_name=conf_name
        ):
            logger.info("Adding domain configuration: %s", conf_path)
            conf_path.parent.mkdir(parents=True, exist_ok=True)
            updated |= atomic_rewrite(str(conf_path), conf_text, backup=False)
        return updated

@classmethod
    async def sync_disabled_rules_for_domains(
        cls, domain_rules_map: Dict[str, list]
    ):
        logger.info("Sync disabled rules for [%s]", ",".join(domain_rules_map))
        for domain, rule_list in domain_rules_map.items():
            data = await whmapi1("domainuserdata", domain=domain)
            user = data["userdata"]["user"]
            cls._add_domain_conf(
                user,
                domain,
                conf_name=cls.PER_DOMAIN_DISABLED_RULES_CONFIG_FILENAME,
                conf_text=cls.generate_disabled_rules_config(rule_list),
            )

await check_run(cls.REBUILD_HTTPDCONF_CMD)

@classmethod
    async def apply_myimunify_modsec_rules_for_domains(
        cls, *, enabled_users_domains: dict, disabled_users_domains: dict
    ) -> set:
        vendor = await cls.get_i360_vendor_name()
        myimunify_conf = await cls.build_vendor_file_path(vendor, "myimunify")
        conf_text = f"IncludeOptional {str(myimunify_conf)}\n"
        conf_name = cls.PER_DOMAIN_MYIMUNIFY_RULES_CONFIG_FILENAME
        updated_domains = set()

def gen_user_domain(users_domains):
            for user, domains in users_domains.items():
                for domain in domains:
                    yield user, domain

# add extended ruleset for enabled domains
        async for user, domain in nice_iterator(
            gen_user_domain(enabled_users_domains), chunk_size=1000
        ):
            if cls._add_domain_conf(
                user, domain, conf_name=conf_name, conf_text=conf_text
            ):
                updated_domains.add(domain)
        # delete extended ruleset for disabled domains
        async for user, domain in nice_iterator(
            gen_user_domain(disabled_users_domains), chunk_size=1000
        ):
            if cls._delete_domain_conf(user, domain, conf_name=conf_name):
                updated_domains.add(domain)
        if updated_domains:
            await check_run(cls.REBUILD_HTTPDCONF_CMD)
        return updated_domains

@classmethod
    def write_global_disabled_rules(cls, rule_list):
        """
        :param list rule_list: rules to sync
        :return:
        """
        os.makedirs(cls.DISABLED_RULES_CONFIG_DIR, exist_ok=True)
        atomic_rewrite(
            os.path.join(
                cls.DISABLED_RULES_CONFIG_DIR,
                cls.GLOBAL_DISABLED_RULES_CONFIG_FILENAME,
            ),
            cls.generate_disabled_rules_config(rule_list),
            backup=False,
        )

@classmethod
    async def sync_global_disabled_rules(cls, rule_list):
        """
        :param list rule_list: rules to sync
        :raise OSError: if rebuildhttpdconf returned not zero exit code
        :return:
        """
        cls.write_global_disabled_rules(rule_list)
        await check_run(cls.REBUILD_HTTPDCONF_CMD)

@classmethod
    def _get_avalible_settings(cls):
        return [ModSecSetting, cPanelFilesVendorList]

@classmethod
    def get_audit_log_path(cls):
        ea4_paths_config = cls._get_ea4_paths()
        if ea4_paths_config is not None:
            log_dir = ea4_paths_config.get("dir_logs", "/etc/apache2/logs")
            return os.path.join(log_dir, "modsec_audit.log")
        else:
            # easy apache 3
            return cls.AUDIT_LOG_FILE

@classmethod
    def get_audit_logdir_path(cls):
        return "/var/log/apache2/modsec_audit"

@classmethod
    async def installed_modsec(cls):
        try:
            rc = (await whmapi1("modsec_is_installed"))["data"]["installed"]
        except (WHMAPIException, OSError):
            return False  # can't get status, assume not installed
        else:
            return rc == 1  # rc==1 means modsec is installed

@forbid_dns_only
    @catch_exception
    async def _install_settings(self, reload_wafd=True):
        await self.reset_modsec_directives()
        await self.reset_modsec_rulesets()

async def reset_modsec_directives(self):
        # implement abstractmethod ModSecurityInterface.reset_modsec_directives
        await self._reset_modsec_setting(ModSecSetting)

async def reset_modsec_rulesets(self):
        # implement abstractmethod ModSecurityInterface.reset_modsec_rulesets
        await self._reset_modsec_setting(cPanelFilesVendorList)

@forbid_dns_only
    async def _reset_modsec_setting(self, setting):
        config = ConfigFile()
        config.set("MOD_SEC", setting.config_key, await setting.apply())

@forbid_dns_only
    @skip_if_not_installed_modsec
    @catch_exception
    async def revert_settings(self, reload_wafd=True):
        """Revert install_settings()"""
        config = ConfigFile()
        for setting in self._get_avalible_settings():
            await setting.revert(config.get("MOD_SEC", setting.config_key))
            config.set("MOD_SEC", setting.config_key, None)

@classmethod
    def detect_cwaf(cls):
        """
        Detects Comodo ModSecurity Rule Set
        :return: bool installed
        """
        return os.path.exists(cls.CWAF_INSTALLATION_DIR)

@classmethod
    @async_lru_cache(maxsize=1)
    async def installed_modsec_vendors_data(cls) -> List[Dict]:
        """Returns list of dicts that describes ModSecurity vendors."""
        vendors_dict = (await whmapi1("modsec_get_vendors")).get("vendors", [])
        return vendors_dict

@classmethod
    async def enabled_modsec_vendors_data(cls) -> List[Dict]:
        """Returns list of dicts that describes enabled ModSecurity vendors."""
        vendors_dict = await cls.installed_modsec_vendors_data()
        return [vendor for vendor in vendors_dict if vendor["enabled"]]

@classmethod
    async def invalidate_installed_vendors_cache(cls):
        cls.installed_modsec_vendors_data.cache_clear()  # NOSONAR Pylint:E1101

@classmethod
    async def modsec_vendor_list(cls) -> List[str]:
        """Return a list of installed ModSecurity vendors."""
        return [
            v["vendor_id"] for v in await cls.installed_modsec_vendors_data()
        ]

@classmethod
    async def enabled_modsec_vendor_list(cls) -> List[str]:
        return [
            v["vendor_id"] for v in await cls.enabled_modsec_vendors_data()
        ]

@classmethod
    async def modsec_get_directive(cls, directive_name, default=None):
        # implement abstractmethod ModSecurityInterface.modsec_get_directive
        try:
            return (await cls.get_settings(directive_name))[directive_name]
        except WHMAPIException:
            logger.exception("failed to get %s directive", directive_name)
            return default

@classmethod
    async def get_settings(cls, *directive_names):
        """Return a mapping ModSecurity directive -> its state."""
        settings = (await whmapi1("modsec_get_settings"))["settings"]
        try:
            return dict(
                (item["directive"], item["state"])
                for item in settings
                if item["directive"] in directive_names
            )
        except (KeyError, StopIteration):
            raise WHMAPIException("Could not parse whmapi1 output")

@classmethod
    async def build_vendor_file_path(cls, vendor: str, filename: str) -> Path:
        vendors_data = await cls.installed_modsec_vendors_data()
        vendor_path = next(
            (v["path"] for v in vendors_data if v["vendor_id"] == vendor), None
        )
        if vendor_path:
            return Path(vendor_path) / filename

raise ModsecVendorsError(
            "Can't get vendor record for vendor {}."
            " Installed vendors: {}".format(
                vendor, [v["vendor_id"] for v in vendors_data]
            )
        )

@classmethod
    def get_modsec_active_conf_files(cls) -> List[str]:
        return ModsecDatastore().get_as_list("active_configs")

@classmethod
    def get_modsec_engine_mode(cls) -> str:
        return ModsecDatastore().get("settings").get("SecRuleEngine")

@classmethod
    def get_modsec_vendor_updates(cls) -> List[str]:
        return ModsecDatastore().get_as_list("vendor_updates")

@classmethod
    @skip_if_not_installed_modsec
    async def _apply_modsec_files_update(cls):
        await cls.invalidate_installed_vendors_cache()
        await cPanelFilesVendorList.apply()

class ModsecDatastore:
    PATH = "/var/cpanel/modsec_cpanel_conf_datastore"

def get(self, section):
        try:
            with open(self.PATH) as f:
                return yaml.safe_load(f).get(section, {})
        except (yaml.YAMLError, FileNotFoundError):
            logger.error("Modsec datastore is not found: %s", self.PATH)
            return {}

def get_as_list(self, section):
        """Get values as a list from the following yaml structure:
        ```
        section:
          value1: 1
          value2: 1
          value3: 0
        ```
        """
        as_list = []
        for value, enabled in self.get(section).items():
            if enabled == 1:
                as_list.append(value)
        return as_list

class cPanelFilesVendor(FilesVendor):
    modsec_interface = cPanelModSecurity

async def apply(self):
        await self.modsec_interface.invalidate_installed_vendors_cache()
        await self._remove_obsoleted()
        await self._add_or_update_vendor()

async def _remove_obsoleted(self):
        logger_ = logging.getLogger("%s.%s" % (__name__, "_remove_obsoleted"))

installed_vendors = set(
            await self.modsec_interface.modsec_vendor_list()
        )

if installed_vendors:
            logger_.info(
                "Installed_vendors were detected: %r", installed_vendors
            )
        else:
            logger_.info("No installed_vendors were detected.")
            return

for to_be_removed in installed_vendors & set(
            self._item.get("obsoletes", [])
        ):
            logger_.info("Removing obsoleted vendor %r", to_be_removed)
            await self._remove_vendor(to_be_removed)
            installed_vendors.discard(to_be_removed)

# Here we are removing vendors that are no more appropriate for
        # the current setup (obsoleted ones (DEF-4434)) or those are
        # not appropriate for active webserver (apache, litespeed))
        for to_be_removed in installed_vendors:
            if (
                to_be_removed.startswith("imunify360")
                and to_be_removed != self.vendor_id
            ):
                logger_.info(
                    "Removing vendor %r which is inappropriate for this setup",
                    to_be_removed,
                )
                await self._remove_vendor(to_be_removed)

async def _add_or_update_vendor(self):
        installed_vendors = await self.modsec_interface.modsec_vendor_list()
        if self.vendor_id in installed_vendors:
            enabled_vendors = (
                await self.modsec_interface.enabled_modsec_vendor_list()
            )
            if self.vendor_id in enabled_vendors:
                try:
                    await check_run([MODSEC_VENDOR_BIN, "update", "--auto"])
                except CheckRunError as e:
                    logger.error(
                        "%r failed with error %r", MODSEC_VENDOR_BIN, e
                    )
                else:
                    logger.info(
                        "Successfully updated vendor %r.", self.vendor_id
                    )
        else:
            await self._add_vendor(url=self._item["url"], name=self.vendor_id)
            logger.info("Successfully installed vendor %r.", self.vendor_id)

@classmethod
    async def _add_vendor(cls, url, name, **kwargs):
        await _modsec_vendor_cmd("add", url)

@classmethod
    async def _remove_vendor(cls, vendor, **kwargs):
        return await _modsec_vendor_cmd("remove", vendor)

def _vendor_id(self):
        basename = os.path.basename(urlparse(self._item["url"]).path)
        basename_no_yaml, _ = os.path.splitext(basename)
        if basename_no_yaml.startswith("meta_"):
            return basename_no_yaml[len("meta_") :]
        else:
            return None

class cPanelFilesVendorList(FilesVendorList):
    files_vendor = cPanelFilesVendor
    modsec_interface = cPanelModSecurity

_FULLY_COMPATIBLE_VENDORS = {"configserver"}

@classmethod
    async def _get_compatible_name(cls, installed_vendors):
        web_server = await cls._get_web_server()
        if not web_server:
            raise cls.CompatiblityCheckFailed(
                "Web-server is not running, skipping "
                "imunify360 vendor installation",
                installed_vendors,
            )

return MODSEC_NAME_TEMPLATE.format(
            ruleset_suffix=cls.get_ruleset_suffix(),
            webserver=web_server,
            panel="cpanel",
        )

@classmethod
    def vendor_fit_panel(cls, item):
        return item["name"].endswith("cpanel")

Back to File Manager