D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
opt
/
imunify360
/
venv
/
lib
/
python3.11
/
site-packages
/
defence360agent
/
subsys
/
Filename :
notifier.py
back
Copy
"""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, "", {})