ÿØÿàJFIFÿþ ÿÛC       ÿÛC ÿÀÿÄÿÄ"#QrÿÄÿÄ&1!A"2qQaáÿÚ ?Øy,æ/3JæݹÈ߲؋5êXw²±ÉyˆR”¾I0ó2—PI¾IÌÚiMö¯–þrìN&"KgX:Šíµ•nTJnLK„…@!‰-ý ùúmë;ºgµŒ&ó±hw’¯Õ@”Ü— 9ñ-ë.²1<yà‚¹ïQÐU„ہ?.’¦èûbß±©Ö«Âw*VŒ) `$‰bØԟ’ëXÖ-ËTÜíGÚ3ð«g Ÿ§¯—Jx„–’U/ÂÅv_s(Hÿ@TñJÑãõçn­‚!ÈgfbÓc­:él[ðQe 9ÀPLbÃãCµm[5¿ç'ªjglå‡Ûí_§Úõl-;"PkÞÞÁQâ¼_Ñ^¢SŸx?"¸¦ùY騐ÒOÈ q’`~~ÚtËU¹CڒêV  I1Áß_ÿÙimport logging import json from defence360agent.contracts.config import HookEvents from defence360agent.model.event_hook import EventHook from defence360agent.rpc_tools import ValidationError from defence360agent.rpc_tools.lookup import RootEndpoints, bind from defence360agent.contracts.hooks import HooksConfig from defence360agent.subsys import notifier from defence360agent.contracts.license import LicenseCLN 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()