Edit File - File Manager

Edit File: coraza_modsecurity.py

"""CoraZa ModSecurity interface for Imunify360"""

import json
import logging
import os
import shutil
import time
import zipfile
from itertools import chain
from pathlib import Path
from typing import Dict, List, Optional
from urllib.parse import urlparse

from defence360agent.subsys import svcctl
from defence360agent.utils import atomic_rewrite, async_lru_cache
from im360.subsys.panels.base import (
    FilesVendor,
    FilesVendorList,
    ModSecurityInterface,
    OPENLITESPEED,
    MODSEC_NAME_TEMPLATE,
)
from defence360agent.subsys.panels.base import forbid_dns_only
from defence360agent.subsys.panels.cpanel.whm import catch_exception
from im360.subsys.int_config import is_force_use_coraza
from im360.subsys.shared_disabled_rules import (
    get_shared_disabled_modsec_rules_ids,
)
from im360.utils import is_apache2nginx_enabled

logger = logging.getLogger(__name__)

CORAZA_RULES_DIR = "/var/imunify360/modsec/coraza"

class CorazaModSecurity(ModSecurityInterface):
    """Coraza ModSecurity interface for Imunify360"""

APP_BASED_EXCLUDE_CONF_NAME = "../app-specific-rules.json"
    GLOBAL_DISABLED_RULES_CONFIG_FILENAME = "coraza.i360_disabled_rules.conf"
    DISABLED_RULES_CONFIG_DIR = "/etc/imunify360-wafd/modsecurity.d"
    WAFD_CORAZA_AUDIT_DIR = "/var/log/imunify360/modsec_audit"
    WAFD_CORAZA_AUDIT_LOG_FILE = "/var/log/imunify360/modsec_audit.log"

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

@classmethod
    async def sync_global_disabled_rules(cls, rule_list):
        """
        :param list rule_list: rules to sync
        :return:
        """
        cls.write_global_disabled_rules(rule_list)
        await cls.reload_modsec()

@classmethod
    def get_audit_logdir_path(cls):
        return cls.WAFD_CORAZA_AUDIT_DIR

@classmethod
    def generate_disabled_rules_config(cls, rule_list):
        tpl = """SecRuleRemoveById {rules_list}"""
        content = ""
        rules_ids = {
            str(id_)
            for id_ in chain(
                rule_list,
                get_shared_disabled_modsec_rules_ids(),
            )
        }
        if rules_ids:
            content = tpl.format(rules_list=" ".join(sorted(rules_ids)))
        return content

@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,
        )

@forbid_dns_only
    @catch_exception
    async def _install_settings(self, reload_wafd=True):
        await CorazaFilesVendorList.apply()
        await self.reload_modsec(reload_wafd=reload_wafd)

@classmethod
    async def installed_modsec(cls):
        # For Coraza scenario, installed modsec can be detected by presence of
        # enabled apache2nginx
        return is_force_use_coraza() or is_apache2nginx_enabled()

async def revert_settings(self, reload_wafd=True):
        # Potentially revert to an empty state or remove Coraza rules if needed
        # Reverting means removing Coraza support:
        await CorazaFilesVendorList.revert()
        await self.reload_modsec(reload_wafd=reload_wafd)

@classmethod
    async def modsec_vendor_list(cls) -> List[str]:
        """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_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 enabled_modsec_vendor_list(cls) -> List[str]:
        """
        Return a list of enabled modsec vendors
        Checked by existence of CORAZA_RULES_DIR
        """
        return await cls.modsec_vendor_list()

REBUILD_HTTPDCONF_CMD = None

@classmethod
    def _get_conf_dir(cls):
        return cls.DISABLED_RULES_CONFIG_DIR

@classmethod
    async def modsec_get_directive(cls, directive_name, default=None):
        """
        N/A for Coraza
        """
        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 Coraza
        """
        raise NotImplementedError

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

@classmethod
    async def reload_modsec(cls, reload_wafd=True):
        """Reload the wafd service to apply updated rules"""
        if not reload_wafd:
            logger.info("Skipping reload wafd.")
            return
        unitctl = svcctl.imunify360_wafd_service()
        try:
            await unitctl.reload()
        except Exception:
            logger.warning("Failed to reload 'imunify360-wafd'")

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

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

@classmethod
    async def _apply_modsec_files_update(cls):
        if await CorazaFilesVendorList.install_or_update():
            await cls.reload_modsec()

@classmethod
    async def sync_disabled_rules_for_domains(
        cls, domain_rules_map: Dict[str, list]
    ):
        # Unused.
        pass

class CorazaFilesVendor(FilesVendor):
    modsec_interface = CorazaModSecurity

async def _add_vendor(self, url, name, *args, **kwargs):
        """Only one vendor is supported"""
        await CorazaFilesVendorList.install_or_update()

async def _remove_vendor(self, vendor, *args, **kwargs):
        """
        Clear vendor data
        """
        atomically_swap_folders()  # swap existing folder with empty one

async def apply(self):
        """
        Atomically swap vendor folders. We can swap them atomically
        only by swapping symlinks to them
        """
        atomically_swap_folders(self._item["local_path"])

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 CorazaFilesVendorList(FilesVendorList):
    files_vendor = CorazaFilesVendor
    modsec_interface = CorazaModSecurity

@classmethod
    async def _get_compatible_name(cls, installed_vendors):
        web_server = OPENLITESPEED

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

@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 revert(cls, *_):
        if os.path.islink(CORAZA_RULES_DIR):
            dst_name = os.readlink(CORAZA_RULES_DIR)
            dst = os.path.join(os.path.dirname(CORAZA_RULES_DIR), dst_name)
            shutil.rmtree(dst, ignore_errors=True)
            os.remove(CORAZA_RULES_DIR)
        else:
            shutil.rmtree(CORAZA_RULES_DIR, ignore_errors=True)

def atomically_swap_folders(zip_path=None):
    """
    Tries to swap folders atomically. The idea is to atomically swap symlynks
    to the folders while folders themselves can be removed afterwards.
    If path to zip file is passed, current folder is replaced with contents of
    the archive. Otherwise it's replaced with empty folder
    """
    # FIXME: Wrap this sync code into async wrappers
    curr_time = time.strftime("%Y-%m-%dT%H%M%S")
    dest_dirpath = f"{CORAZA_RULES_DIR}_{curr_time}"
    dest_symlink = f"{CORAZA_RULES_DIR}_{curr_time}_sym"
    os.makedirs(dest_dirpath, mode=0o700)

if zip_path:
        with zipfile.ZipFile(zip_path) as zf:
            for member in zf.namelist():
                filename = os.path.basename(member)
                if not filename:
                    continue
                target = os.path.join(dest_dirpath, filename)
                with zf.open(member) as src, open(target, "wb") as dst:
                    shutil.copyfileobj(src, dst)
    os.symlink(os.path.basename(dest_dirpath), dest_symlink)

if os.path.islink(CORAZA_RULES_DIR):
        dst_name = os.readlink(CORAZA_RULES_DIR)
        dst = os.path.join(os.path.dirname(CORAZA_RULES_DIR), dst_name)
        os.rename(dest_symlink, CORAZA_RULES_DIR)
        shutil.rmtree(dst, ignore_errors=True)
    else:
        shutil.rmtree(CORAZA_RULES_DIR, ignore_errors=True)
        os.rename(dest_symlink, CORAZA_RULES_DIR)

Back to File Manager