1#@!#!123s

: / opt / imunify360 / venv / lib / python3.11 / site-packages / im360 / application /
Filename : settings.py
back
"""Additional changes, which need to use RPC/CLI schemas. """

import sqlite3
from pathlib import Path

from im360.contracts.config import IPSET_LISTS_PATH
from imav.application.settings import (
    configure as base_configure,
    get_schema_paths as base_get_schema_paths,
)
from imav.malwarelib.subsys import aibolit
import im360.model.update_hooks
import im360.subsys.ossec
import im360.subsys.panels.update_hooks
import im360.subsys.proactive
import im360.subsys.webshield
from defence360agent import files as base_files
from defence360agent.contracts.config import Model
from defence360agent.model import infected_domain
from im360 import files
from im360.application import tags
from im360.model import (
    firewall,
    incident,
    messages_to_send,
    proactive,
)
from im360.simple_rpc.schema import init_validator
from im360.simple_rpc.validate import SchemaValidator, validate_middleware


def get_schema_paths():
    path = Path(__file__).parent.parent.absolute()
    schema_paths = base_get_schema_paths(full=True) + (path / "simple_rpc",)
    return schema_paths


def setup_file_hooks(resident=False) -> None:
    if not resident:
        files.Index.add_hook(files.GEO, im360.model.update_hooks.update_geodb)
        files.Index.add_hook(
            files.MODSEC, im360.subsys.panels.update_hooks.update_vendors
        )
        files.Index.add_hook(files.OSSEC, im360.subsys.ossec.on_files_update)
        files.Index.add_hook(
            files.PHP_IMMUNITY, im360.subsys.proactive.update_hook
        )
        files.Index.add_hook(
            files.PHP_IMMUNITY_V2,
            im360.subsys.proactive.update_hook_immunity_v2,
        )
        files.Index.add_hook(
            files.PROACTIVE,
            im360.subsys.proactive.recreate_signatures_on_update,
        )
        files.Index.add_hook(
            files.SIGS,
            aibolit.restart_on_sigs_or_config_update,
        )
        files.Index.add_hook(
            files.IP_RECORD,
            im360.subsys.panels.update_hooks.update_iprecord,
        )
        files.DEFAULT_HOOKS[files.AUDITD_CONF] = Path(
            "/usr/sbin/imunify-auditd-log-reader-cfg-reload"
        )
        files.DEFAULT_HOOKS[files.EMAIL_RULES_V1] = Path(
            "/usr/sbin/ie_ruleupdate"
        )
        files.DEFAULT_HOOKS[files.PROACTIVE_CONFIG] = Path(
            "/usr/bin/i360-sync-config.sh"
        )


def configure(set_sentry_tags=tags.fill, resident=False):
    im360_path = Path(__file__).resolve().parent.parent
    base_configure(
        init_validator=init_validator,
        validator_cls=SchemaValidator,
        validate_middleware_wrap=validate_middleware,
        schema_paths=get_schema_paths(),
        models_modules=[
            firewall,
            incident,
            infected_domain,
            messages_to_send,
            proactive,
        ],
        set_sentry_tags=set_sentry_tags,
        migration_dirs=[im360_path / "migrations"],
        migrations_attached_dbs=[
            (Model.RESIDENT_PATH, "resident"),
            (IPSET_LISTS_PATH, "ipsetlists"),
        ],
        resident=resident,
    )
    if not resident:
        files.Index.add_hook(
            base_files.REALTIME_AV_CONF,
            aibolit.restart_on_sigs_or_config_update,
        )
    files.configure()
    setup_file_hooks(resident=resident)


def before_migrations():
    """
    Ensure `resident` db and `ipsetlists` DB have WAL in `journal_mode`.
    """
    with sqlite3.connect(Model.RESIDENT_PATH) as conn:
        conn.execute("PRAGMA journal_mode=WAL")
    with sqlite3.connect(IPSET_LISTS_PATH) as conn:
        conn.execute("PRAGMA journal_mode=WAL")