MAINHACK

Viewing / Editing File: waf_rules_configurator.py

File is not writable. Editing disabled.
"""
A plugin responsible for periodically launching AppVersionDetector and
optionally limiting ModSecurity rulesets for the sites that use various
CMS.

Currently it sets up and maintains a cron job to achieve this.
"""
import os
from contextlib import suppress
from logging import getLogger
from pathlib import Path
from random import randint
from typing import Optional

from defence360agent.contracts.config import SystemConfig
from defence360agent.contracts.messages import MessageType
from defence360agent.contracts.plugins import MessageSink, expect
from defence360agent.utils.cronjob import CronJob
from im360.contracts.config import Modsec as Config
from defence360agent.subsys import web_server
from im360.subsys.panels.hosting_panel import HostingPanel
from im360.subsys.waf_rules_configurator import (
    is_webserver_supported,
    try_restore_config_from_backup,
)

logger = getLogger(__name__)

UPDATE_COMPONENTS_SCRIPT = (
    "/opt/imunify360/venv/share/imunify360/scripts/"
    "update_components_versions.py"
)
WAF_CONFIGURATOR_CRON_PATH = "/etc/cron.d/waf_configurator"


class WAFRuleSetConfigurator(MessageSink):
    def __init__(self):
        self._cron_job = self._read_cron()

    async def create_sink(self, loop):
        await self._update_cron_job(Config.APP_SPECIFIC_RULESET)
        await try_restore_config_from_backup()

    @staticmethod
    def _read_cron() -> Optional[str]:
        with suppress(FileNotFoundError):
            return CronJob.from_str(
                Path(WAF_CONFIGURATOR_CRON_PATH).read_text()
            )
        return None

    @staticmethod
    def _write_cron(cron_job: CronJob):
        path = Path(WAF_CONFIGURATOR_CRON_PATH)
        path.touch(mode=0o600)
        path.write_text(str(cron_job))

    @staticmethod
    def _get_cron_job(update_modsec_rulesets) -> CronJob:
        shell_cmd = (
            "/usr/libexec/report-command-error "
            + UPDATE_COMPONENTS_SCRIPT
            + (" --update-modsec-rulesets" if update_modsec_rulesets else "")
            + " > /dev/null 2>&1"
        )
        hour, minute = 4, randint(0, 59)
        return CronJob(minute=minute, hour=hour, cmd=shell_cmd)

    async def _update_cron_job(self, enabled):
        cron_job = self._get_cron_job(
            update_modsec_rulesets=enabled and await is_webserver_supported()
        )
        if self._cron_job and cron_job.cmd == self._cron_job.cmd:
            return
        logger.info("Updating AppVersionDetector version cron")
        self._write_cron(cron_job)
        self._cron_job = cron_job

    async def _truncate_conf(self):
        """
        If app-specific httpd config exists and is not empty,
        truncate it
        """
        try:
            config_path = HostingPanel().get_app_specific_waf_config()
            st = os.stat(config_path)
        except (FileNotFoundError, NotImplementedError):
            pass
        else:
            if st.st_size:
                open(config_path, "w").close()
                await web_server.graceful_restart()
                logger.info("App specific ruleset config truncated")

    @expect(MessageType.ConfigUpdate)
    async def update_cron_job(self, message):
        if isinstance(message["conf"], SystemConfig):
            enabled = Config.APP_SPECIFIC_RULESET
            await self._update_cron_job(enabled)
            if not enabled:
                await self._truncate_conf()
Cancel / Back
[ MAINHACK ]

Viewing / Editing File: waf_rules_configurator.py

Server Info

System Features

User Info
