0x1998 - MANAGER

Düzenlenen Dosya: 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,
)
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"
    PER_DOMAIN_DISABLED_RULES_CONFIG_FILENAME = (
        "/etc/imunify360-wafd/disabled-ids-rules.json"
    )
    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) -> bool:
        """
        :param list rule_list: rules to sync
        :return: True if config was changed, False otherwise
        """
        changed = cls.write_global_disabled_rules(rule_list)
        if changed:
            await cls.reload_modsec()
        return changed

@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) -> bool:
        """
        :param list rule_list: rules to sync
        :return: True if config was changed, False otherwise
        """
        os.makedirs(cls.DISABLED_RULES_CONFIG_DIR, exist_ok=True)
        return 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]
    ):
        logger.info("Sync disabled rules for [%s]", ",".join(domain_rules_map))

config_map = {}
        try:
            with open(cls.PER_DOMAIN_DISABLED_RULES_CONFIG_FILENAME, "r") as f:
                old_config_list = json.load(f)
                config_map = {
                    item["domain"]: item["disabled_ids"]
                    for item in old_config_list
                }
        except (json.JSONDecodeError, FileNotFoundError):
            logger.warning(
                "Could not load or parse existing config file. A new one will"
                " be created."
            )

config_map.update(domain_rules_map)

final_config = [
            {"domain": domain, "disabled_ids": [int(_id) for _id in ids]}
            for domain, ids in config_map.items()
            if ids
        ]

if atomic_rewrite(
            cls.PER_DOMAIN_DISABLED_RULES_CONFIG_FILENAME,
            json.dumps(final_config),
            backup=False,
        ):
            logger.info(
                "Disabled rules for domains were successfully updated on"
                " path %s.",
                cls.PER_DOMAIN_DISABLED_RULES_CONFIG_FILENAME,
            )
        else:
            logger.error(
                "Failed to update disabled rules for domains on path %s.",
                cls.PER_DOMAIN_DISABLED_RULES_CONFIG_FILENAME,
            )

class CorazaFilesVendor(FilesVendor):
    modsec_interface = CorazaModSecurity

async def _remove_vendor(self):
        """
        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):
        return f"imunify360-{cls.get_ruleset_suffix()}-coraza"

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

@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
    # WARN: Resolution in seconds is not enough to make path unique, append nanoseconds.
    # Monotonic clock is used in addition to system clock to improve name uniqueness.
    curr_time = f"{time.strftime('%Y-%m-%dT%H%M%S')}.{time.monotonic_ns()}"
    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)

geri dön