1#@!#!123s

: / opt / imunify360 / venv / lib / python3.11 / site-packages / im360 / plugins / protector /
Filename : lfd_protector.py
back
import logging
import os

from defence360agent import utils
from defence360agent.contracts import config as common_config
from defence360agent.contracts.messages import MessageType, Reject
from defence360agent.contracts.plugins import (
    MessageSink,
    MessageSource,
    expect,
)
from im360.contracts import config
from im360.internals import strategy
from im360.subsys import csf

logger = logging.getLogger(__name__)


class LFDProtector(MessageSink, MessageSource):
    PROCESSING_ORDER = MessageSink.ProcessingOrder.LFD
    STRATEGY = strategy.Strategy.CSF_COOP_STRATEGY
    AVAILABLE_ON_FREEMIUM = False

    BLOCK_REPORT_SCRIPT = os.path.join(
        common_config.Packaging.DATADIR, "scripts", "lfd_block.py"
    )
    USER_SCRIPT_LINK = os.path.join(
        common_config.Packaging.DATADIR, "scripts", "block_report_user"
    )

    @property
    def _script_installed(self):
        try:
            current_script = csf.Config("BLOCK_REPORT").get()
        except (FileNotFoundError, NotADirectoryError):
            return False
        else:
            return os.path.realpath(current_script) == os.path.realpath(
                self.BLOCK_REPORT_SCRIPT
            )

    async def create_sink(self, loop):
        self._loop = loop

    async def create_source(self, loop, sink):
        self._sink = sink

    @expect(MessageType.SensorAlert, plugin_id=config.OssecSensor.PLUGIN_ID)
    async def ignore_ossec_alert(self, _):
        if (
            self._script_installed
            and strategy.Strategy.current == self.STRATEGY
        ):
            raise Reject("CSF is running")

    @expect(MessageType.SensorIncident, plugin_id="lfd")
    async def copy_lfd_incident_to_alert(self, message):
        """protector only"""
        alert = MessageType.SensorAlert.from_incident(message)
        await self._sink.process_message(alert)

    @expect(MessageType.SensorAlert, plugin_id="lfd")
    async def unblock_in_csf(self, message):
        logger.info(
            "Unblocking %s in CSF before adding to graylist",
            message["attackers_ip"],
        )
        await csf.unblock(message["attackers_ip"])