1#@!#!123s

: / opt / imunify360 / venv / lib / python3.11 / site-packages / im360 / subsys /
Filename : modsec_cache_dir.py
back
"""
https://github.com/SpiderLabs/ModSecurity/wiki/Reference-Manual-(v2.x)#SecDataDir
"""
import asyncio
import os
import logging

from defence360agent.subsys.panels import hosting_panel
from defence360agent.utils import retry_on
from defence360agent.subsys import web_server

logger = logging.getLogger(__name__)


def get_sec_data_dir(modsec_config_path):
    try:
        with open(modsec_config_path) as f:
            data = f.read()
        start = "<IfModule security2_module>"
        end = "</IfModule>"
        param_name = "SecDataDir"
        data_between = data.split(start)[1].split(end)[0]
        param = next(
            iter(
                [
                    line
                    for line in data_between.splitlines()
                    if param_name in line
                ]
            ),
            None,
        )
        return param.split(param_name)[1].strip()
    except (IndexError, AttributeError, OSError) as e:
        logger.error("Incorrect modsec config %s", e)


async def create_modsec_cache_directory():
    """
    Create modsec cache directory because sometimes a directory may not exist
    :return:
    """
    try:
        modsec_config_path = (
            hosting_panel.HostingPanel().get_modsec_config_path()
        )
    except NotImplementedError as e:
        logger.debug(
            "get_modsec_config_path is not implemented for current "
            "hosting_panel %s",
            e,
        )
        return
    if not modsec_config_path or not os.path.exists(modsec_config_path):
        return

    sec_data_dir = get_sec_data_dir(modsec_config_path)
    if sec_data_dir and not os.path.exists(sec_data_dir):
        try:
            os.makedirs(sec_data_dir)

            # The directory to which the directive points must be
            # writable by the web server user.

            async def pause(*_):
                await asyncio.sleep(60)  # wait a minute between attempts

            @retry_on(web_server.NotRunningError, on_error=pause, max_tries=5)
            async def coro():
                web_server.chown(sec_data_dir)

            await coro()

            logger.info("Successfully created sec_data_dir %s", sec_data_dir)
            await web_server.graceful_restart()
        except OSError as e:
            logger.error("Error when creating sec_data_dir %s", e)