?HAMA

Navigation

Breadcrumbs

Create

Edit File Size: 1.82 KB

/opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/subsys/notifier.py

Text Base64

"""Send events via Notification service"""

import asyncio
import base64
import json

SOCKET_PATH = "/opt/imunify360/lib/event.sock"
SOCKET_TIMEOUT = 10.0  # seconds
_LEN_BYTES = 4
_MAX_SIZE = 1024 * 1024

CONFIG_UPDATED_EVENT_ID = "CONFIG_UPDATED"
USER_SCAN_STARTED_EVENT_ID = "USER_SCAN_STARTED"
USER_SCAN_FINISHED_EVENT_ID = "USER_SCAN_FINISHED"
USER_SCAN_MALWARE_FOUND_EVENT_ID = "USER_SCAN_MALWARE_FOUND"
CUSTOM_SCAN_STARTED_EVENT_ID = "CUSTOM_SCAN_STARTED"
CUSTOM_SCAN_FINISHED_EVENT_ID = "CUSTOM_SCAN_FINISHED"
CUSTOM_SCAN_MALWARE_FOUND_EVENT_ID = "CUSTOM_SCAN_MALWARE_FOUND"
SCRIPT_BLOCKED_EVENT_ID = "SCRIPT_BLOCKED"

def _prepare_event(event_id: str, user: str, body: dict) -> bytes:
    event = json.dumps(
        {
            "event_id": event_id,
            "user": user,
            "body": base64.b64encode(json.dumps(body).encode("utf-8")).decode(
                "utf-8"
            ),
        }
    )
    binary = event.encode("utf-8")
    if len(binary) > _MAX_SIZE:
        raise Exception(
            "message size {} exceeds limit of {}".format(
                len(binary), _MAX_SIZE
            )
        )
    return len(binary).to_bytes(_LEN_BYTES, byteorder="big") + binary

async def _send_event(event: bytes) -> None:
    _, writer = await asyncio.open_unix_connection(SOCKET_PATH)
    try:
        writer.write(event)
        await writer.drain()
    finally:
        writer.close()

async def trigger_event(event_id: str, user: str, body: dict) -> None:
    """Send an event with given event_id and user, having given body."""
    event = _prepare_event(event_id, user, body)
    await asyncio.wait_for(_send_event(event), SOCKET_TIMEOUT)

async def config_updated() -> None:
    """Send CONFIG_UPDATED event.

This forces imunify-notifier to reread its config."""
    await trigger_event(CONFIG_UPDATED_EVENT_ID, "", {})

Exit & Reset Text mode: syntax highlighting auto-detects file type.

Name	Size	Perms	Modified	Actions
features DIR	-	drwxr-xr-x	2026-06-08 20:24:30	Edit Download recursive recursive
panels DIR	-	drwxr-xr-x	2026-06-08 20:24:30	Edit Download recursive recursive
__pycache__ DIR	-	drwxr-xr-x	2026-06-08 20:24:30	Edit Download recursive recursive
backup_systems.py	11.19 KB	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
clcagefs.py	7.79 KB	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
notifier.py	1.82 KB	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
persistent_state.py	1.88 KB	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
svcctl.py	5.76 KB	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
sysctl.py	379 B	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
systemd_notifier.py	1.58 KB	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
web_server.py	21.21 KB	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive
__init__.py	0 B	lrw-r--r--	2026-05-26 21:20:44	Edit Download recursive recursive

Directory Contents