?HAMA

Edit File Size: 5.94 KB

/opt/imunify360/venv/lib/python3.11/site-packages/imav/malwarelib/scan/scanner.py

Text Base64

"""
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/>.

This software is also available under ImunifyAV commercial license,
see <https://www.imunify360.com/legal/eula>
"""
import asyncio
import itertools
import logging
import time
from typing import Dict, Iterator, List, Optional, Tuple

from imav.malwarelib.scan import ScanFailedError
from defence360agent.contracts.config import Core as Config
from defence360agent.contracts.messages import MessageType

from .ai_bolit.scanner import AiBolit
from .scan_result import DirectAiBolit, PrepareFileList, ScanResult, event_hook

logger = logging.getLogger(__name__)

class MalwareScanner:
    def __init__(self, tmpdir=None, sink=None, hooks=False):
        self._tmpdir = tmpdir or Config.TMPDIR
        self._sink = sink
        self._hooks = hooks

self.aibolit = None
        self._task = None
        self._loop = asyncio.get_event_loop()
        self._tmpdir = tmpdir
        self._current_task = None
        self._future_result = None
        self._is_running = asyncio.Lock()

def start(self, path, *args, **kwargs):
        self._future_result = asyncio.Future()
        self._task = self._loop.create_task(self._start(path, *args, **kwargs))
        return self._task

def wait(self):
        if self._future_result is None:
            raise asyncio.exceptions.InvalidStateError("Need to start action")

self._loop.run_until_complete(self._future_result)
        return self._future_result.result()

async def async_wait(self):
        if self._future_result is None:
            raise asyncio.exceptions.InvalidStateError("Need to start action")

await self._future_result
        return self._future_result.result()

def stop(self):
        if self.status():
            self._task.cancel()

def status(self):
        return self._is_running.locked()

async def _start(self, path, *args, **kwargs):
        async with self._is_running:
            try:
                execute_scan = self._execute_scan
                if self._hooks:
                    execute_scan = event_hook(self._sink)(execute_scan)
                result = await execute_scan(path, *args, **kwargs)
            except asyncio.CancelledError:
                self._future_result.set_result(None)
            except Exception as e:
                self._future_result.set_exception(e)
            else:
                self._future_result.set_result(result)

async def _send_scan_failed_message(self, e):
        if self._sink:
            try:
                msg = MessageType.ScanFailed()

msg["exception"] = e.__class__.__name__
                if isinstance(e, ScanFailedError):
                    msg["return_code"] = e.return_code
                    msg["command"] = e.command
                    msg["out"] = e.out
                    msg["err"] = e.err
                    msg["message"] = e.message
                    msg["path"] = e.path
                    msg["scan_id"] = e.scan_id
                else:
                    msg["message"] = str(e)

await self._sink.process_message(msg)
            except asyncio.CancelledError:
                raise
            except Exception:
                logger.exception("Exception while sending ScanFailed message")

async def _execute_scan(self, path, *args, **kwargs):
        logger.info("Scan started")

files_provider = (
            DirectAiBolit if isinstance(path, str) else PrepareFileList
        )

prepare_file_list = files_provider(self._tmpdir)
        kwargs["begin_time"] = time.time()

execute_scan = prepare_file_list(self._run_aibolit)
        result = await execute_scan(path, *args, **kwargs)

assert isinstance(result, ScanResult)
        end_time = None if result.is_detached() else time.time()
        result.set_start_stop(end_time=end_time)

if result.is_detached():
            return result.to_dict_initial()
        else:
            if self.aibolit and self.aibolit.cmd:
                result.args = self.aibolit.cmd
            return result.to_dict()

async def _run_aibolit(
        self, file: Optional[str], **scan_options
    ) -> Tuple[List[Iterator[Dict]], Optional[str]]:
        """
        :param file: file with list of files, that should be scanned
        :param scan_options: some options for the scan
        :return: parse report iterator and error
        """
        iterators = []
        error = None
        self.aibolit = AiBolit()
        scan = self.aibolit.scan

try:
            self._current_task = scan
            try:
                iterator = await scan(file, **scan_options)
            except asyncio.CancelledError:
                # all over here, task is cancelled, passing exc up
                raise
            except ScanFailedError as e:
                await self._send_scan_failed_message(e)
                logger.exception("Scan failed with ScanFailed exception")
                iterator = []
                error = e.message

except Exception as e:
                await self._send_scan_failed_message(e)
                logger.exception("Scan failed with ordinary exception")
                iterator = []
                error = repr(e)

# copy iterator
            result, _ = itertools.tee(iterator)
            iterators.append(result)
        finally:
            self._current_task = None

return iterators, error

Exit & Reset Text mode: syntax highlighting auto-detects file type.

Name	Size	Perms	Modified	Actions
ai_bolit DIR	-	drwxr-xr-x	2026-03-03 09:00:23	Edit Download recursive recursive
cleaners DIR	-	drwxr-xr-x	2026-03-03 09:00:23	Edit Download recursive recursive
mds DIR	-	drwxr-xr-x	2026-03-03 09:00:23	Edit Download recursive recursive
__pycache__ DIR	-	drwxr-xr-x	2026-03-03 09:00:23	Edit Download recursive recursive
app_version_detector.py	3.82 KB	lrw-r--r--	2026-01-23 13:55:22	Edit Download recursive recursive
crontab.py	1.35 KB	lrw-r--r--	2026-01-23 13:55:22	Edit Download recursive recursive
detached.py	6.77 KB	lrw-r--r--	2026-01-23 13:55:22	Edit Download recursive recursive
queue.py	15.83 KB	lrw-r--r--	2026-01-23 13:55:22	Edit Download recursive recursive
queue_supervisor_sync.py	5.11 KB	lrw-r--r--	2026-01-23 13:55:22	Edit Download recursive recursive
scanner.py	5.94 KB	lrw-r--r--	2026-01-23 13:55:22	Edit Download recursive recursive
scan_result.py	9.55 KB	lrw-r--r--	2026-01-23 13:55:22	Edit Download recursive recursive
utils.py	1.79 KB	lrw-r--r--	2026-01-23 13:55:22	Edit Download recursive recursive
__init__.py	2.36 KB	lrw-r--r--	2026-01-23 13:55:22	Edit Download recursive recursive

Directory Contents