""" This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this program.  If not, see <https://www.gnu.org/licenses/>. Copyright © 2019 Cloud Linux Software Inc. This software is also available under ImunifyAV commercial license, see <https://www.imunify360.com/legal/eula> """ import logging from defence360agent.contracts.config import SystemConfig from defence360agent.contracts.messages import ConfigUpdate from defence360agent.contracts.plugins import MessageSink, expect from defence360agent.utils import Scope from imav.contracts.config import Malware as MalwareConfig from imav.malwarelib.subsys import aibolit logger = logging.getLogger(__name__) class DetectAdminToolsWatcher(MessageSink): SCOPE = Scope.IM360 def __init__(self): self._loop = None self._previous_detect_admin_tools = MalwareConfig.DETECT_ADMIN_TOOLS async def create_sink(self, loop) -> None: self._loop = loop @expect(ConfigUpdate) async def on_config_update(self, message: ConfigUpdate): if isinstance(message["conf"], SystemConfig): current_value = MalwareConfig.DETECT_ADMIN_TOOLS if self._previous_detect_admin_tools != current_value: logger.info( "DETECT_ADMIN_TOOLS changed from %s to %s, restarting" " ai-bolit", self._previous_detect_admin_tools, current_value, ) self._previous_detect_admin_tools = current_value await aibolit.restart_on_detect_admin_tools_update()