^^&&^^

Edit File: mod_security.py

import json
import logging
import os
from contextlib import suppress
from packaging.version import Version
from pathlib import Path
from typing import Dict, Optional
from urllib.parse import urlparse

from defence360agent.contracts.config import ConfigFile, Core
from defence360agent.subsys.panels import base
from defence360agent.utils import (
    OsReleaseInfo,
    async_lru_cache,
    atomic_rewrite,
    check_run,
)
from im360.subsys.panels.base import (
    MODSEC_NAME_TEMPLATE,
    FilesVendor,
    FilesVendorList,
    ModSecSettingInterface,
    ModSecurityInterface,
    skip_if_not_installed_modsec,
)
from defence360agent.subsys.web_server import graceful_restart

MODSEC_API = "/usr/local/psa/admin/sbin/modsecurity_ctl "
SERV_PERF = "/usr/local/psa/bin/server_pref "
#: full path to server_pref executable
SERVER_PREF_BIN = Path("/usr/local/psa/bin/server_pref")
HTTPDMNG = "/usr/local/psa/admin/bin/httpdmng "

CUSTOM_VENDOR_NAME = "custom"

logger = logging.getLogger(__name__)

class PleskModSecException(base.PanelException):
    pass

async def run_cmd(cmd):
    logger.debug("Running CMD: %s", cmd)
    data = await check_run(cmd.split(), raise_exc=PleskModSecException)
    return data.decode().strip()

async def plesk_supports_custom_vendors():
    # import used to avoid cyclic import
    from defence360agent.subsys.panels.plesk import Plesk

return Version(await Plesk.version()) >= Version("17.5")

class ModSecSettings(ModSecSettingInterface):
    # ModSec Custom directives file
    USER_MODSEC_CONF_REDHAT = "/etc/httpd/conf/plesk.conf.d/modsecurity.conf"
    I360_INCLUDE_REDHAT = 'Include "/etc/httpd/conf.d/modsec2.imunify.conf"'
    I360_MODSEC_CONF_DEBIAN = (
        "/etc/apache2/conf-available/modsec2.imunify.conf"
    )
    # Plesk adds its own config file (with mod_security settings)
    # as zz010_psa_httpd.conf. So to overwrite Plesk mod_security settings,
    # we choose this kind of name
    I360_MODSEC_CONF_SYMLINK_DEBIAN = (
        "/etc/apache2/conf-enabled/zz999_modsec2.imunify.conf"
    )
    config_key = "prev_settings"

@classmethod
    async def waf_rule_engine_mode(cls):
        data = await run_cmd(SERV_PERF + "--show-web-app-firewall")
        data = iter(data.splitlines())

rv = next((data for line in data if line == "[waf-rule-engine]"), None)
        status = rv and next(rv, None)
        return status

@classmethod
    def _read_custom_conf(cls):
        content = ""
        try:
            with open(cls.USER_MODSEC_CONF_REDHAT) as f:
                content = f.read()
        except OSError:
            pass
        return content

@classmethod
    def _include_modsec_conf_redhat(cls):
        content = cls._read_custom_conf()
        if cls.I360_INCLUDE_REDHAT not in content:
            content += "\n" + cls.I360_INCLUDE_REDHAT + "\n"
            atomic_rewrite(cls.USER_MODSEC_CONF_REDHAT, content, backup=False)

@classmethod
    def _include_modsec_conf_debian(cls):
        try:
            os.symlink(
                cls.I360_MODSEC_CONF_DEBIAN,
                cls.I360_MODSEC_CONF_SYMLINK_DEBIAN,
            )
        except FileExistsError:
            pass

@classmethod
    def include_modsec_conf(cls):
        if OsReleaseInfo.id_like() & OsReleaseInfo.DEBIAN:
            cls._include_modsec_conf_debian()
        else:
            cls._include_modsec_conf_redhat()

@classmethod
    def _revert_conf_include_redhat(cls):
        content = cls._read_custom_conf()
        if cls.I360_INCLUDE_REDHAT in content:
            content = content.replace(cls.I360_INCLUDE_REDHAT, "")
            atomic_rewrite(cls.USER_MODSEC_CONF_REDHAT, content, backup=False)

@classmethod
    def _revert_conf_include_debian(cls):
        for conf in (
            cls.I360_MODSEC_CONF_SYMLINK_DEBIAN,
            PleskModSecurity.conf_disable_global_symlink(),
        ):
            with suppress(FileNotFoundError):
                os.unlink(conf)

@classmethod
    def revert_conf_include(cls):
        if OsReleaseInfo.id_like() & OsReleaseInfo.DEBIAN:
            cls._revert_conf_include_debian()
        else:
            cls._revert_conf_include_redhat()

@classmethod
    async def apply(cls):
        cls.include_modsec_conf()
        prev_value = status = await cls.waf_rule_engine_mode()

if status != "off":
            return status

waf_option = " --update-web-app-firewall -waf-rule-engine on"

if await plesk_supports_custom_vendors():
            # New Plesk won't try to install tortix (see DEF-2102) so calling
            # without -waf-rule-set is fine.
            await run_cmd(SERV_PERF + waf_option)

return prev_value

# FIXME: installing CRS here in order to avoid yum circular call.
        # See DEF-2102 for more info.
        waf_option += " -waf-rule-set crs"
        await run_cmd(SERV_PERF + waf_option)
        await run_cmd(MODSEC_API + "--disable-all-rules --ruleset crs")
        await run_cmd(MODSEC_API + "--uninstall --ruleset crs")
        return prev_value

@classmethod
    async def revert(cls, prev_value: str):
        cls.revert_conf_include()
        if prev_value:
            await run_cmd(
                SERV_PERF
                + "--update-web-app-firewall -waf-rule-engine {}".format(
                    prev_value
                )
            )

@classmethod
    def is_enabled(cls):
        return True

class PleskModSecurity(ModSecurityInterface):
    AUDIT_LOG_FILE = "/var/log/modsec_audit.log"
    CWAF_INSTALLATION_DIR = "/usr/local/cwaf"
    DOMAIN_INCLUDE_DIR_TML = "/var/www/vhosts/system/{domain}/conf/siteapp.d/"
    APP_BASED_EXCLUDE_CONF_NAME = "zz999-i360-app-based-excludes.conf"

@classmethod
    def _get_conf_dir(cls) -> str:
        if OsReleaseInfo.id_like() & OsReleaseInfo.DEBIAN:
            return "/etc/apache2/conf-enabled"
        return "/etc/httpd/conf.d/"

@classmethod
    def _get_global_include_dir(cls):
        if OsReleaseInfo.id_like() & OsReleaseInfo.DEBIAN:
            return "/etc/apache2/plesk.conf.d"
        else:
            return "/etc/httpd/conf/plesk.conf.d"

@classmethod
    def conf_disable_global_symlink(cls):
        return os.path.join(
            cls._get_conf_dir(), "zz999_modsec2.imunify_disable.conf"
        )

@classmethod
    async def sync_disabled_rules_for_domains(
        cls, domain_rules_map: Dict[str, list]
    ):
        for domain, rule_list in domain_rules_map.items():
            conf_dir = cls.DOMAIN_INCLUDE_DIR_TML.format(domain=domain)
            os.makedirs(conf_dir, exist_ok=True)
            atomic_rewrite(
                os.path.join(conf_dir, "i360_modsec_disable.conf"),
                cls.generate_disabled_rules_config(rule_list),
                backup=False,
            )
            await run_cmd(
                HTTPDMNG
                + "--reconfigure-domain {} -skip-broken -no-restart".format(
                    domain
                )
            )

@classmethod
    def write_global_disabled_rules(cls, rule_list):
        """
        :param list rule_list: rule ids to sync
        :raise OSError: if httpd reconfigure returned not zero exit code
        :return:
        """
        os.makedirs(cls._get_global_include_dir(), exist_ok=True)
        atomic_rewrite(
            os.path.join(
                cls._get_global_include_dir(), "i360_modsec_disable.conf"
            ),
            cls.generate_disabled_rules_config(rule_list),
            backup=False,
        )
        # Symlink for
        try:
            os.symlink(
                os.path.join(
                    cls._get_global_include_dir(), "i360_modsec_disable.conf"
                ),
                cls.conf_disable_global_symlink(),
            )
        except FileExistsError:
            pass

@classmethod
    async def sync_global_disabled_rules(cls, rule_list):
        """
        just alias to write_global_disabled_rules()
        """
        cls.write_global_disabled_rules(rule_list)

@classmethod
    def get_audit_log_path(cls):
        return cls.AUDIT_LOG_FILE

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

@classmethod
    async def installed_modsec(cls):
        """Check if all the panel utilities we use are available."""
        if not os.path.isfile(MODSEC_API.rstrip()):
            return False
        try:
            await run_cmd(SERV_PERF + "--show-web-app-firewall")
        except PleskModSecException:
            return False
        else:
            return True

@base.ensure_valid_panel()
    async def _install_settings(self, reload_wafd=True):
        config = ConfigFile()
        for setting in self._get_avalible_settings():
            config.set("MOD_SEC", setting.config_key, await setting.apply())
        await graceful_restart()

async def modsec_get_directive(self, directive_name, default=None):
        raise NotImplementedError

async def reset_modsec_directives(self):
        raise NotImplementedError

async def reset_modsec_rulesets(self):
        raise NotImplementedError

@base.ensure_valid_panel()
    async def revert_settings(self, reload_wafd=True):
        if not await self.installed_modsec():
            logger.warning(
                "Skipping vendor removal, because ModSecurity isn't installed"
            )
            return

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)

await graceful_restart()

@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 modsec_vendor_list(cls):
        """Return a list of installed ModSecurity vendors."""
        if not await cls.installed_modsec():
            return []
        data = await run_cmd(MODSEC_API + "--list-rulesets")
        return await cls._vendor_list_with_custom_vendor_removed(data)

@classmethod
    async def _vendor_list_with_custom_vendor_removed(cls, data):
        """
        On new plesk panels we have convert all 'custom' vendors from plesk api
        calls to real imunify-vendor names(ex. imunify360-full-apache) as
        on this panel 'custom' - is a fixed vendor name for 3rd party
        installed vendor. If we will find RELEASE file with real vendor name
        in vendor directory, it means it is imunify vendor, we will return
        this real name
        :param data:
        :return: vendor list, example:
        ['imunify360-full-apache', 'other-vendor']
        """
        vendor_list = [vendor for vendor in data.splitlines() if vendor]

if not await plesk_supports_custom_vendors():
            return vendor_list

vendor = await cls.get_modsec_vendor_from_release_file()
        if vendor:
            try:
                vendor_list.remove(CUSTOM_VENDOR_NAME)
            except ValueError:
                pass
            vendor_list.append(vendor)
        return vendor_list

@classmethod
    @async_lru_cache(maxsize=1)
    async def _get_release_info_from_file(cls) -> Optional[dict]:
        if await plesk_supports_custom_vendors():
            # On new plesk we will look into custom vendor directory anyway.
            # We should not call get_i360_vendor_name() here because of
            # recursion
            modsec_release_file = await cls.build_vendor_file_path(
                None, "RELEASE"
            )
        else:
            modsec_release_file = await cls.build_vendor_file_path(
                await cls.get_i360_vendor_name(), "RELEASE"
            )
        try:
            with modsec_release_file.open() as release_f:
                json_data = json.load(release_f)
            return json_data
        except (OSError, IOError, json.JSONDecodeError):
            return None

@classmethod
    async def get_i360_vendor_version(cls) -> str:
        release_dict = await cls._get_release_info_from_file()
        if release_dict and release_dict.get("version"):
            return release_dict["version"]
        return await super().get_i360_vendor_version()

@classmethod
    @async_lru_cache(maxsize=1)
    async def enabled_modsec_vendor_list(cls):
        """Return a list of enabled ModSecurity vendors."""
        if not await cls.installed_modsec():
            return []
        data = await run_cmd(MODSEC_API + "--list-rulesets --enabled")
        return await cls._vendor_list_with_custom_vendor_removed(data)

@classmethod
    async def build_vendor_file_path(
        cls, vendor: Optional[str], filename: str
    ) -> Path:
        """
        :param vendor: vendor directory: old plesk panels - imunify360-*;
        new plesk panels - imunify360-* or None - we will look into custom
        directory anyway
        :param filename:
        :return:
        """
        rule_base_dir = await run_cmd(MODSEC_API + "--rules-base-dir")

if not await plesk_supports_custom_vendors():
            if not vendor:
                raise ValueError(
                    "Vendor directory can't be None on old plesk panels"
                )
            vendor_dir = vendor
        else:
            # On plesk 17.5+ ruleset is in the "custom" directory.
            if vendor and not vendor.startswith(Core.PRODUCT):
                raise ValueError(
                    "Vendor directory {} should be None or "
                    "starts with {} on new plesk "
                    "panels".format(vendor, Core.PRODUCT)
                )
            vendor_dir = CUSTOM_VENDOR_NAME
        return Path(rule_base_dir) / vendor_dir / filename

@classmethod
    def _get_avalible_settings(cls):
        return [ModSecSettings, PleskFilesVendorList]

@classmethod
    @skip_if_not_installed_modsec
    async def _apply_modsec_files_update(cls):
        await cls.invalidate_installed_vendors_cache()
        await PleskFilesVendorList.install_or_update()

class PleskFilesVendor(FilesVendor):
    modsec_interface = PleskModSecurity
    TMP_VENDOR_PATH = os.path.join(Core.TMPDIR, "i360_modsec_vendor.zip")

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

async def revert(self):
        """
        Removes and disables ModSecurity vendor + on new plesk also we will
        always delete "custom" ruleset in all cases
        """
        if await self._is_installed():
            await self._remove_vendor(self.vendor_id)
            logger.info("Successfully removed vendor %r.", self.vendor_id)

async def _remove_obsoleted(self):
        enabled_vendors = set(
            await self.modsec_interface.enabled_modsec_vendor_list()
        )
        obsoleted = set(self._item.get("obsoletes", []))
        for vendor in enabled_vendors & obsoleted:
            logger.info("Removing obsoleted vendor %r", vendor)
            await self._remove_vendor(vendor)

async def _add_or_update_vendor(self):
        # DEF-9877: use all installed vendors, not just enabled.
        # Or don't try to remove old vendor at all.
        installed_vendors = (
            await self.modsec_interface.enabled_modsec_vendor_list()
        )

await self._add_vendor(name=self.vendor_id)

if self.vendor_id in installed_vendors:
            logger.info("Successfully updated vendor %r.", self.vendor_id)
        else:
            logger.info("Successfully installed vendor %r.", self.vendor_id)

async def _remove_vendor(self, vendor: str, *args, **kwargs):
        if await plesk_supports_custom_vendors():
            logger.warning("Plesk doesn't support uninstalling rulesets.")
            try:
                await run_cmd(
                    SERV_PERF
                    + "--update-web-app-firewall -waf-rule-engine off"
                )
            except Exception as e:
                logger.error("Couldn't turn WAF rule engine off: %s ", e)
            return

await run_cmd(MODSEC_API + "--disable-all-rules --ruleset %s" % vendor)
        await run_cmd(MODSEC_API + "--uninstall --ruleset %s" % vendor)

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

async def _add_vendor(self, name, *args, **kwargs):
        if await plesk_supports_custom_vendors():
            await self._install_or_update_custom_ruleset()
        else:
            await run_cmd(
                (
                    MODSEC_API
                    + "--install --with-backup "
                    "--enable-ruleset --ruleset %s "
                    "--archive-path %s" % (name, self._item["local_path"])
                )
            )

async def _install_or_update_custom_ruleset(self):
        """
        Install or update custom ruleset from *TMP_VENDOR_PATH*.

Installation happens iff the "installing_settings" var is true.
        Installation turns on the waf rule engine.

The update command preserves waf rule engine mode.
        If the war rule engine is off, then the update command is skipped.

Unknown (None/empty) mode is considered to be "off."
        """
        old_mode = await ModSecSettings.waf_rule_engine_mode() or "off"
        installing = self.modsec_interface.installing_settings_var.get()
        if old_mode != "off" or installing:
            new_mode = "on" if installing else old_mode
            logger.info(
                "%s %s ruleset, waf-rule-engine mode: %s",
                "Installing" if installing else "Updating",
                CUSTOM_VENDOR_NAME,
                (
                    f"{old_mode=!r} {new_mode=!r}"
                    if old_mode != new_mode
                    else repr(new_mode)
                ),
            )
            await check_run(
                [SERVER_PREF_BIN, "--update-web-app-firewall"]
                + ["-waf-rule-engine", new_mode]
                + [
                    "-waf-rule-set",
                    CUSTOM_VENDOR_NAME,
                    "-waf-archive-path",
                    self._item["local_path"],
                ],
                raise_exc=PleskModSecException,
            )
        else:
            logger.info(
                "waf rule engine remains off."
                " Skip the update command as a workaround for DEF-15857."
            )

class PleskFilesVendorList(FilesVendorList):
    files_vendor = PleskFilesVendor
    modsec_interface = PleskModSecurity

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

@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="plesk",
        )