✏️ Editing: lfd_block.py Read Only

#!/opt/imunify360/venv/bin/python3 -u
"""BLOCK_REPORT script invoked by Login Failure Daemon (CSF) for a blocked ip.

- report the incident to imunify360
- run the replaced user BLOCK_REPORT script (block_report_user)
"""
import json
import logging
import os
import socket
import subprocess
import sys
from collections import namedtuple

import defence360agent.internals.logger

BLOCK_REPORT_TIMEOUT = 10  # seconds
SOCKET = "/var/run/defence360agent/generic_sensor.sock.2"

Event = namedtuple(
    "Event",
    (
        "ip",
        "ports",
        "permanent",
        "inout",
        "timeout",
        "message",
        "logs",
        "trigger",
    ),
)

def run_user_script(
    args,
    *,
    logger=None,
    timeout=None,
    # see defence360/src/asyncclient/defence360agent/plugins/sensor/lfd.py
    script=os.path.join(os.path.dirname(__file__), "block_report_user"),
):
    if os.path.isfile(script):
        if os.path.realpath(script) == os.path.abspath(__file__):
            # If for whatever reason script tries to call itself, ignore it
            logger.error("Not running %s since it is a loop", script)
            return
        try:  # NOTE: ignore user script errors
            subprocess.run([script] + args, timeout=timeout)
        except subprocess.TimeoutExpired:
            raise TimeoutError("imunify lfd_block user script timeout")

def main(logger):
    if len(sys.argv) != (len(Event._fields) + 1):
        # logger.warning is to find evidence of call without arguments
        # in logs (to find a possible automation call mistake)
        logger.warning(
            "This script is intended to be used as "
            "BLOCK_REPORT script for CSF"
        )
        sys.exit(1)

e = Event(*sys.argv[1:])
    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
        sock.settimeout(BLOCK_REPORT_TIMEOUT)
        try:
            sock.connect(SOCKET)
            msg = {
                "method": "INCIDENT",
                "attackers_ip": e.ip,
                "plugin_id": "lfd",
                "ttl": e.timeout,
                "rule": e.trigger,
                "name": e.trigger,
                "message": e.message,
            }
            sock.sendall(json.dumps(msg).encode() + b"\n")
        except (
            ConnectionRefusedError,
            FileNotFoundError,
        ):  # allow other errors to propagate
            # agent appears to be turned off or hanged
            pass  # do nothing
        except socket.timeout:  # also do nothing
            logger.debug("failed to send incident report in time")
        finally:
            run_user_script(
                sys.argv[1:], timeout=BLOCK_REPORT_TIMEOUT, logger=logger
            )

if __name__ == "__main__":
    defence360agent.internals.logger.reconfigure()
    logger = logging.getLogger(sys.argv[0])
    try:
        main(logger)
    except Exception:  # <-- ignore SystemExit
        # do not left unreported
        logger.exception("imunify lfd_block script error")

🔐 Sid Gifari File Manager Pro