from defence360agent.rpc_tools import ValidationError from defence360agent.rpc_tools.lookup import RootEndpoints, bind from defence360agent.subsys import svcctl from defence360agent.subsys.panels.base import PanelException from defence360agent.utils import Scope from im360.contracts.config import Modsec from im360.simple_rpc.unavailable_on_freemium import unavailable_on_freemium from im360.subsys import modsec_app_version_detector, waf_rules_configurator from im360.subsys.panels import hosting_panel from im360.subsys.panels.base import use_modsec_lock from im360.subsys.panels.cpanel.panel import cPanel from im360.utils import is_apache2nginx_enabled class HostingPanelEndpoints(RootEndpoints): SCOPE = Scope.IM360 async def restart_agent(self): """Reload the wafd service to apply updated rules""" unitctl = svcctl.imunify360_agent_service() await unitctl.restart() @bind("switch-max-webserver") @unavailable_on_freemium @use_modsec_lock async def switch_max_webserver(self): try: apache2nginx_status = is_apache2nginx_enabled() if apache2nginx_status: cpanel_modsec = cPanel(force_cpanel=True) await cpanel_modsec.revert_settings() else: coraza_modsec = cPanel(force_coraza=True) await coraza_modsec.revert_settings(reload_wafd=False) await self.restart_agent() if Modsec.APP_SPECIFIC_RULESET: try: await waf_rules_configurator.update_waf_rules_config() except modsec_app_version_detector.DatabaseNotFoundError as e: raise ValidationError(str(e)) except waf_rules_configurator.NotSupportedWebserverError: pass return await self.hosting_panel.install_settings(reload_wafd=False) except PanelException as e: raise ValidationError(str(e)) @bind("install-vendors") @unavailable_on_freemium @use_modsec_lock async def install_vendors(self): try: return await self.hosting_panel.install_settings() except PanelException as e: raise ValidationError(str(e)) @bind("uninstall-vendors") @use_modsec_lock async def uninstall_vendors(self): try: return await self.hosting_panel.revert_settings() except PanelException as e: raise ValidationError(str(e)) @bind("admin-emails") async def get_admin_emails(self): return await self.hosting_panel.get_admin_emails() @property def hosting_panel(self): try: return hosting_panel.HostingPanel() except PanelException as e: raise ValidationError(str(e))