import logging import os import shutil from defence360agent.contracts.config import SystemConfig from defence360agent.contracts.messages import MessageType from defence360agent.contracts.plugins import MessageSink, expect from im360.contracts.config import ProactiveDefence from im360.subsys import proactive logger = logging.getLogger(__name__) class PhpImmunityPlugin(MessageSink): def __init__(self): self._loop = None self.config_php_immunity = ProactiveDefence.PHP_IMMUNITY async def create_sink(self, loop) -> None: self._loop = loop @expect(MessageType.ConfigUpdate) async def on_config_update(self, message: MessageType.ConfigUpdate): if ( isinstance(message["conf"], SystemConfig) and self.config_php_immunity != ProactiveDefence.PHP_IMMUNITY ): self.config_php_immunity = ProactiveDefence.PHP_IMMUNITY if not ProactiveDefence.PHP_IMMUNITY: if os.path.isfile(proactive.PHP_IMMUNITY_COMPILED_RULES): os.unlink(proactive.PHP_IMMUNITY_COMPILED_RULES) else: shutil.rmtree( proactive.PHP_IMMUNITY_COMPILED_RULES, ignore_errors=True, ) else: if os.path.exists(proactive.PHP_IMMUNITY_FILENAME): await proactive.apply_php_immunity_rules()