✏️ Editing: update_hooks.py Read Only

import logging
import shutil
from pathlib import Path

from defence360agent.subsys import web_server
from defence360agent.subsys.panels.base import (
    ModsecVendorsError,
    PanelException,
)
from defence360agent.utils import (
    atomic_rewrite,
    check_run,
    CheckRunError,
    file_hash,
    log_error_and_ignore,
)
from im360 import files
from im360.contracts.config import Modsec
from im360.subsys import modsec_app_version_detector, waf_rules_configurator
from im360.subsys.int_config import is_force_use_coraza
from im360.subsys.panels.base import use_modsec_lock
from imav.malwarelib.subsys.malware import HackerTrapHitsSaver

from .hosting_panel import HostingPanel

logger = logging.getLogger(__name__)

async def reload_wafd():
    if Path("/usr/bin/imunify360-wsctl").is_file():
        args = ["imunify360-wsctl", "reload"]
    else:
        args = ["systemctl", "reload", "imunify360-wafd"]
    try:
        await check_run(args)
    except CheckRunError:
        logger.warning("Failed to reload 'imunify360-wafd'")

WP_REDIRECT_CONF_PATTERN = "*_Disable_WP_Redirect.conf"
DISABLED_REDIRECT_CONF_CONTENT = "SecRuleRemoveById 33355 33357"

@use_modsec_lock
async def update_vendors(_, is_updated):
    if is_updated:
        hp = HostingPanel()
        if hp.is_installed():
            await hp.apply_modsec_files_update()
            # rules contains empty ip-record.db, so after
            # updating rules, we should also update ip-record.db
            await _update_iprecord(_, is_updated)
            await HackerTrapHitsSaver.init()
            # DEF-41801: skip the inline graceful_restart() — the
            # unconditional graceful_restart_sync() below already
            # reloads.
            await _update_account_compromise_prevention_rule_state(
                skip_restart=True
            )
            await _update_app_based_rules()
            # DEF-35552: during startup a webserver restart may be scheduled
            # before modsec-rules are downloaded, consuming the throttle
            # window and causing subsequent async restart requests to be
            # dropped.  Use the synchronous restart which bypasses the
            # coalesce throttle entirely (same pattern as _install_settings).
            web_server.graceful_restart_sync()

@use_modsec_lock
async def update_iprecord(_, is_updated):
    return await _update_iprecord(_, is_updated)

async def _update_iprecord(_, is_updated):
    if not is_updated:
        return

def _warn(e):
        logger.warning("Can't update ip-record.db, reason: %s" % str(e))

hp = HostingPanel()
    try:
        vendor = await hp.get_i360_vendor_name()
        ip_record = await hp.build_vendor_file_path(vendor, "ip-record.db")
    except (ModsecVendorsError, PanelException, ValueError) as e:
        _warn(e)
        return
    src = (
        Path(files.Index.files_path(files.IP_RECORD))
        / "ip-record"
        / "ip-record.db"
    )
    if src.exists():
        try:
            if not Path(ip_record).exists() or file_hash(src) != file_hash(
                ip_record
            ):
                shutil.copy(src, ip_record)
                await web_server.graceful_restart()
                # Sort of a workaround to avoid redundant imports which can
                # cause circular dependencies
                if (
                    is_force_use_coraza()
                    or hp.__class__.__name__ == "cPanelCoraza"
                ):
                    logger.info(
                        "Reloading 'imunify360-wafd' as coraza ruleset is in"
                        " action"
                    )
                    await reload_wafd()
        except Exception as e:
            _warn(e)
            return
    else:
        _warn("source file is missing")
        return

def _get_account_prevention_state(path: Path):
    content = path.read_text().strip()
    if not content:
        return True
    elif content == DISABLED_REDIRECT_CONF_CONTENT:
        return False
    else:
        logger.warning("Invalid content in %s: %s", path, content)
        return None

@use_modsec_lock
async def update_account_compromise_prevention_rule_state():
    return await _update_account_compromise_prevention_rule_state()

@log_error_and_ignore()
async def _update_account_compromise_prevention_rule_state(
    skip_restart: bool = False,
):
    is_prevention_enabled = Modsec.CMS_ACCOUNT_COMPROMISE_PREVENTION
    hp = HostingPanel()
    try:
        vendor = await hp.get_i360_vendor_name()
    except (ModsecVendorsError, PanelException) as e:
        logger.warning(str(e))
        return
    try:
        vendor_path = await hp.build_vendor_file_path(vendor, "")
        wp_redirect_conf_path = next(
            vendor_path.glob(WP_REDIRECT_CONF_PATTERN)
        )
    except (ModsecVendorsError, StopIteration):
        logger.exception("Can't get %s file", WP_REDIRECT_CONF_PATTERN)
        return
    current_state = _get_account_prevention_state(wp_redirect_conf_path)
    if current_state != is_prevention_enabled:
        content = (
            "" if is_prevention_enabled else DISABLED_REDIRECT_CONF_CONTENT
        )
        atomic_rewrite(str(wp_redirect_conf_path), content, backup=False)
        # DEF-41801: when called from update_vendors, the caller will
        # issue graceful_restart_sync() itself — skip the inline httpd
        # restart to avoid a double reload.
        if not skip_restart:
            await web_server.graceful_restart()
        # Coraza's wafd is a separate daemon — graceful_restart_sync()
        # reloads httpd/nginx only — so reload_wafd() must run
        # independently of skip_restart.
        # (Sort of a workaround to avoid redundant imports which can
        # cause circular dependencies.)
        if is_force_use_coraza() or hp.__class__.__name__ == "cPanelCoraza":
            logger.info(
                "Reloading 'imunify360-wafd' as coraza ruleset is in action"
            )
            await reload_wafd()

async def _update_app_based_rules():
    if Modsec.APP_SPECIFIC_RULESET:
        try:
            await waf_rules_configurator.update_waf_rules_config()
        except (
            waf_rules_configurator.NotSupportedWebserverError,
            modsec_app_version_detector.DatabaseNotFoundError,
        ) as e:
            logger.warning("App based rules not updated: %s", e)

🔐 Sid Gifari File Manager Pro