Server IP : 66.29.132.122 / Your IP : 3.144.101.139 Web Server : LiteSpeed System : Linux business142.web-hosting.com 4.18.0-553.lve.el8.x86_64 #1 SMP Mon May 27 15:27:34 UTC 2024 x86_64 User : admazpex ( 531) PHP Version : 7.2.34 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : OFF | Pkexec : OFF Directory : /proc/self/root/proc/self/root/proc/thread-self/root/proc/self/root/opt/imunify360/venv/lib64/python3.11/site-packages/defence360agent/simple_rpc/ |
Upload File : |
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()