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()