D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
opt
/
imunify360
/
venv
/
lib
/
python3.11
/
site-packages
/
imav
/
plugins
/
Filename :
event_hook_executor.py
back
Copy
""" 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> """ from defence360agent.contracts.hook_events import HookEvent from defence360agent.contracts.messages import MessageType from defence360agent.contracts.plugins import ( expect, ) from defence360agent.hooks.execute import execute_hooks from defence360agent.plugins.event_hook_executor import EventHookExecutor from imav.malwarelib.config import MalwareHitStatus from imav.malwarelib.model import MalwareHit EVENTS = ( HookEvent.MalwareScanningStarted, HookEvent.MalwareScanningFinished, HookEvent.MalwareCleanupStarted, HookEvent.MalwareCleanupFinished, HookEvent.MalwareDetectedCritical, ) async def detected_hook( sink, scan_id, scan_type, started, path, total, error=None ): total_malicious, hits = MalwareHit.malicious_list( by_scan_id=scan_id, ignore_cleaned=True ) if hits: malware_detected = HookEvent.MalwareDetectedCritical( scan_id=scan_id, scan_type=scan_type, error=error, started=started, path=path, total_files=total, total_malicious=total_malicious, DUMP=hits, ) await sink.process_message(malware_detected) class ImavEventHookExecutor(EventHookExecutor): @expect(*EVENTS) async def receive_event(self, event): self._loop.create_task(execute_hooks(event)) @expect(MessageType.MalwareCleanup) async def process_malware_cleanup(self, message): error = message["error"] status = "failed" if error else "ok" hits = message["hits"] cleaned = [h for h in hits if h.status in MalwareHitStatus.RESTORABLE] dump = [hit.as_dict() for hit in hits] malware_cleanup = HookEvent.MalwareCleanupFinished( cleanup_id=message["cleanup_id"], started=message["started"], error=error, total_files=len(hits), total_cleaned=len(cleaned), status=status, DUMP=dump, ) await self._sink.process_message(malware_cleanup) @expect(MessageType.MalwareScan) async def process_malware_scan(self, message): summary = message["summary"] # Message without summary means that detached scan is finished # and summary will arrive along with results in another message - this # is when the hooks should be called. if not summary.get("path"): return await detected_hook( self._sink, summary["scanid"], summary["type"], summary["started"], summary["path"], summary["total_files"], summary["error"], ) @expect(MessageType.MalwareDatabaseScan) async def process_malware_db_scan(self, message): await detected_hook( self._sink, message.scan_id, message.type, message.started, message.path, message.total_resources, message.error, )