update_hooks.py

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 generic_panel_uses_coraza
from im360.subsys.panels.base import use_modsec_lock
from imav.api.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()
            # Skip the inline graceful_restart()  the unconditional
            # restart below already reloads.
            await _update_account_compromise_prevention_rule_state(
                skip_restart=True
            )
            await _update_app_based_rules()
            # A restart may already have been scheduled before modsec-rules
            # were downloaded, consuming the coalesce throttle window and
            # getting later requests dropped, so use the throttle-bypassing
            # restart. graceful_restart_confirmed() confirms the reload and
            # recovers a failed one; fall back to the legacy sync restart when
            # running against an older core that lacks it.
            confirmed_restart = getattr(
                web_server, "graceful_restart_confirmed", None
            )
            if confirmed_restart is not None:
                await confirmed_restart()
            else:
                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 (
                    generic_panel_uses_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)
        # When called from update_vendors, the caller issues its own
        # restart afterwards  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  the httpd/nginx restart
        # does not touch it  so reload_wafd() must run independently of
        # skip_restart.
        # (Sort of a workaround to avoid redundant imports which can
        # cause circular dependencies.)
        if (
            generic_panel_uses_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)

docroot: /home/induvedam/public_html

curdir: /opt/imunify360/venv/lib/python3.11/site-packages/im360/subsys/panels

	.	4096
	..	4096
X	__init__.py	0
	__pycache__	4096
X	base.py	20074
X	coraza_modsecurity.py	12409
	cpanel	138
	directadmin	124
	generic	83
X	hosting_panel.py	155
	no_cp	60
	plesk	102
X	update_hooks.py	6679