MAINHACK

Viewing / Editing File: smtp_blocking.py

File is not writable. Editing disabled.
import logging
from defence360agent.contracts.messages import MessageType
from defence360agent.contracts.plugins import MessageSink, expect
from defence360agent.utils import recurring_check
from im360.subsys.smtp_blocking import (
    reset_rules_for_all_versions,
    sync_rules_for_all_versions,
    is_SMTP_blocking_supported,
    read_SMTP_settings,
    get_active_settings_list,
    conflicts_exist,
)

logger = logging.getLogger(__name__)


class SMTPBlocker(MessageSink):
    """
    Makes sure SMTP blocking rules exist and synchronises them with
    SMTP_BLOCKING section of Imunify config when necessary.
    """

    RULE_CHECK_INTERVAL = 300  # delay in secs between two checks
    CONSECUTIVE_ERROR_LIMIT = 10  # retries until giving up

    def __init__(self):
        self._tasks = []
        self._loop = None
        self._task = None

    async def create_sink(self, loop):
        self._loop = loop
        if await is_SMTP_blocking_supported():
            self._task = self._loop.create_task(self._run())
        else:
            self._task = None
            logger.info("SMTP blocking is not supported by current kernel")

    async def shutdown(self):
        if self._task is not None:
            self._task.cancel()
            await self._task
        await reset_rules_for_all_versions()

    async def _check_state_and_reset(self, check_settings=False):
        if await is_SMTP_blocking_supported():
            if not await conflicts_exist():
                new_settings = read_SMTP_settings()
                if not check_settings or self._is_smtp_settings_changed(
                    new_settings
                ):
                    await sync_rules_for_all_versions(new_settings)
            else:
                await reset_rules_for_all_versions()

    @recurring_check(RULE_CHECK_INTERVAL, CONSECUTIVE_ERROR_LIMIT)
    async def _run(self):
        await self._check_state_and_reset()

    @expect(MessageType.ConfigUpdate)
    async def on_config_update(self, _):
        await self._check_state_and_reset(check_settings=True)

    @staticmethod
    def _is_smtp_settings_changed(new_settings):
        return any(
            active_settings != new_settings
            for active_settings in get_active_settings_list()
        )
Cancel / Back
[ MAINHACK ]

Viewing / Editing File: smtp_blocking.py

Server Info

System Features

User Info
