Current File : /opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/simple_rpc//hooks.py
import json
import logging
from defence360agent.contracts.config import HookEvents
from defence360agent.contracts.hooks import HooksConfig
from defence360agent.contracts.license import LicenseCLN
from defence360agent.model.event_hook import EventHook
from defence360agent.rpc_tools import ValidationError
from defence360agent.rpc_tools.lookup import RootEndpoints, bind
from defence360agent.subsys import notifier
logger = logging.getLogger(__name__)
class HooksEndpoints(RootEndpoints):
def _check_event(self, event, extra=None):
if event not in HookEvents.EVENTS and event != extra:
raise ValidationError(
'"{}" is not valid event for hook'.format(event)
)
@bind("hook", "add")
async def hook_add(self, event, path):
self._check_event(event)
result = EventHook.add_hook(event=event, path=path)
if not result:
raise ValidationError(
'Unable to add hook "{} {}"'.format(event, path)
)
result["status"] = "registered"
return {"items": result}
@bind("hook", "delete")
async def hook_delete(self, event, path):
self._check_event(event)
result = EventHook.delete_hook(event=event, path=path)
if not result:
raise ValidationError(
'Unable to delete hook "{} {}"'.format(event, path)
)
result["status"] = "unregistered"
return {"items": result}
@bind("hook", "list")
async def hook_list(self, event):
self._check_event(event, "all")
result = EventHook.list_events(event)
return {"items": result}
@bind("hook", "add-native")
async def hook_add_native(self, event, path):
self._check_event(event)
result = EventHook.add_hook(event=event, path=path, native=True)
if not result:
raise ValidationError(
'Unable to add native hook "{} {}"'.format(event, path)
)
result["status"] = "registered"
return {"items": result}
@bind("notifications-config", "show")
async def show(self):
return {"items": HooksConfig().get()}
@bind("notifications-config", "update")
async def update(self, items=None, data=None):
if LicenseCLN.is_demo():
raise ValidationError("This action is not allowed in demo version")
if items:
data = items[0]
new_data = json.loads(data)
HooksConfig().update(new_data)
await notifier.config_updated()
return await self.show()
@bind("notifications-config", "patch")
async def update_ui(self, data=None):
if LicenseCLN.is_demo():
raise ValidationError("This action is not allowed in demo version")
HooksConfig().update(data)
await notifier.config_updated()
return await self.show()
