SWAMPy

✎ CARVE FLESH

EDITING: queue.py

"""
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 collections
import logging
import time
from abc import ABC, abstractmethod
from functools import cached_property
from glob import iglob
from os import fsencode
from pathlib import Path
from typing import ClassVar, Deque, List, Optional
from uuid import uuid4

from defence360agent.api import inactivity
from defence360agent.contracts.config import (
    Malware as Config,
    MalwareScanIntensity,
)
from defence360agent.internals.global_scope import g
from defence360agent.subsys.panels.hosting_panel import HostingPanel
from defence360agent.utils import (
    antivirus_mode,
    create_task_and_log_exceptions,
)
from defence360agent.utils.common import HOUR, rate_limit
from defence360agent.utils.serialization import serialize_attr, unserialize
from imav.malwarelib.config import (
    MalwareScanResourceType,
    MalwareScanType,
    QueuedScanState,
)
from imav.malwarelib.scan.ai_bolit.detached import (
    AiBolitDetachedScan,
)
from imav.malwarelib.scan.app_version_detector import (
    AVDExecutionError,
    AppVersionDetector,
)
from imav.malwarelib.scan.detached import DetachedScan
from imav.malwarelib.scan.mds.detached import MDSDetachedScan
from imav.malwarelib.scan.mds.scanner import MalwareDatabaseScanner
from imav.malwarelib.utils.user_list import panel_users

logger = logging.getLogger(__name__)
throttled_log_error = rate_limit(period=HOUR, on_drop=logger.warning)(
    logger.error
)

INTENSITY_FALLBACK = {
    "low": {
        "intensity_cpu": 1,
        "intensity_io": 1,
        "intensity_ram": 1024,
    },
    "moderate": {
        "intensity_cpu": 4,
        "intensity_io": 4,
        "intensity_ram": 2048,
    },
    "high": {
        "intensity_cpu": 7,
        "intensity_io": 7,
        "intensity_ram": 4096,
    },
}

class QueuedScanBase(ABC):
    resource_type: ClassVar[MalwareScanResourceType]
    detached_scan: DetachedScan
    state = None

if intensity:
            intensity_fallback = INTENSITY_FALLBACK[intensity]
        else:
            intensity_fallback = {
                "intensity_cpu": MalwareScanIntensity.CPU,
                "intensity_io": MalwareScanIntensity.IO,
                "intensity_ram": MalwareScanIntensity.RAM,
            }

self.args = {
            "intensity_cpu": intensity_cpu
            or intensity_fallback["intensity_cpu"],
            "intensity_io": intensity_io or intensity_fallback["intensity_io"],
            "intensity_ram": intensity_ram
            or intensity_fallback["intensity_ram"],
            "initiator": initiator,
        }

home_dirs = home_dirs or []
        if scan_type == MalwareScanType.ON_DEMAND and Path(path) in home_dirs:
            scan_type = MalwareScanType.USER

self.scanid = scanid or uuid4().hex
        self.scan_type = scan_type
        self.created = created or int(time.time())
        self.state = (
            QueuedScanState.queued if state is None else QueuedScanState(state)
        )
        self.started = started
        self.scanner_task = None

@abstractmethod
    async def start(self):
        pass

@property
    @abstractmethod
    def total_resources(self):
        pass

def stop(self):
        if self.scanner_task:
            self.scanner_task.cancel()

def status(self):
        result = {
            "status": self.state.value,
            "path": self.path,
            "scanid": self.scanid,
            "started": self.started,
            "created": self.created,
            "scan_type": self.scan_type,
            "resource_type": self.resource_type.value,
            **self.args,
        }
        if self.state == QueuedScanState.running:
            result["phase"] = self.detached_scan.phase
            result["progress"] = self.detached_scan.progress
        return result

def __getstate__(self):
        state = self.__dict__.copy()
        del state["detached_scan"]
        del state["scanner_task"]
        return state

def __setstate__(self, state):
        self.__dict__.update(state)
        self.scanner_task = None

def __eq__(self, other):
        return (
            self.resource_type == other.resource_type
            and self.path == other.path
        )

def __repr__(self):
        return "<{}({!r}, scanid={})>".format(
            self.__class__.__qualname__, self.path, self.scanid
        )

# We don't need to hash queued scans
    __hash__ = None  # type: ignore

def to_dict(self):
        return {
            "path": self.path,
            "scanid": self.scanid,
            "scan_type": self.scan_type,
            "created": self.created,
            "started": self.started,
            "args": self.args,
            "resource_type": self.resource_type.value,
            "state": self.state.value,
        }

@classmethod
    def from_dict(cls, kwargs):
        for arg, value in kwargs.pop("args", {}).items():
            kwargs[arg] = value
        return cls(**kwargs)

class QueuedFileScan(QueuedScanBase):
    resource_type = MalwareScanResourceType.FILE

def __init__(
        self,
        path,
        *,
        scanid=None,
        scan_type: str = MalwareScanType.ON_DEMAND,
        created: int = None,
        started: Optional[float] = None,
        intensity=None,
        home_dirs=None,
        intensity_cpu=None,
        intensity_io=None,
        intensity_ram=None,
        file_patterns=None,
        exclude_patterns=None,
        follow_symlinks=None,
        detect_elf=None,
        initiator=None,
        state=None,
        **_,
    ):
        super().__init__(
            path,
            scanid=scanid,
            scan_type=scan_type,
            created=created,
            started=started,
            intensity=intensity,
            home_dirs=home_dirs,
            intensity_cpu=intensity_cpu,
            intensity_io=intensity_io,
            intensity_ram=intensity_ram,
            initiator=initiator,
            state=state,
        )
        self.args.update(
            file_patterns=file_patterns or None,
            exclude_patterns=exclude_patterns or None,
            follow_symlinks=follow_symlinks or False,
            detect_elf=detect_elf,
        )
        self.detached_scan = AiBolitDetachedScan(self.scanid)

def __setstate__(self, state):
        # WARNING: Avoid adding a new attribute to a serializable class.
        # If an object deserializes after a package upgrade, it will lack it
        super().__setstate__(state)
        self.detached_scan = AiBolitDetachedScan(self.scanid)

async def start(self):
        self.started = time.time()

from imav.malwarelib.scan.scanner import MalwareScanner

scanner = MalwareScanner(sink=g.sink, hooks=True)
        self.scanner_task = scanner.start(
            self.path,
            scan_id=self.scanid,
            scan_type=self.scan_type,
            started=self.started,
            **self.args,
        )
        scan_data = await scanner.async_wait()

if scan_data is None:
            logger.info("Scan cancelled for %s", self.path)
            self.state = QueuedScanState.stopped
        else:
            self.state = QueuedScanState.running
            scan_data["initiator"] = (
                # using `get` because there is no initiator before version 6.8
                self.args.get("initiator")
                # for compatibility reason: when `self.initiator` is available
                or getattr(self, "initiator", "undefined")
            )

return scan_data

def stop(self):
        if self.scanner_task:
            self.scanner_task.cancel()

@property
    def total_resources(self):
        return self.detached_scan.total_resources

class QueuedDbScan(QueuedScanBase):
    resource_type = MalwareScanResourceType.DB

def __init__(
        self,
        path: str,
        scanid: Optional[str] = None,
        # FIXME: Use Enum instead of a class with str attributes.
        scan_type: str = MalwareScanType.ON_DEMAND,
        created: int = None,
        started: Optional[float] = None,
        intensity: Optional[str] = None,
        home_dirs=None,
        intensity_cpu=None,
        intensity_io=None,
        intensity_ram=None,
        state=None,
        **_,
    ):
        super().__init__(
            path=path,
            scanid=scanid,
            scan_type=scan_type,
            created=created,
            started=started,
            intensity=intensity,
            home_dirs=home_dirs,
            intensity_cpu=intensity_cpu,
            intensity_io=intensity_io,
            intensity_ram=intensity_ram,
            state=state,
        )
        self.detached_scan = MDSDetachedScan(self.scanid)

def __setstate__(self, state):
        super().__setstate__(state)
        self.detached_scan = MDSDetachedScan(self.scanid)

@property
    def total_resources(self) -> int:
        return self.detached_scan.total_resources

async def _scan(self):
        # app-version-detector should recursive check all directories,
        # no need to extract them explicitly.
        # Used to make files and db scans idempotent (DEF-19264)
        # MDS scanner (php) should alerady handle /path/* as /path/ (DEF-19096)
        apps_path = (
            self.path.rstrip("*") if self.path.endswith("/*") else self.path
        )
        doc_roots = {}
        try:
            doc_roots = await HostingPanel().list_docroots()
        except Exception as exc:
            # ignore an error here to keep backward compatibility
            logger.error("Can't get docroots, run AVD anyway. Error: %s", exc)
        unglobbed_paths = [fsencode(d) for d in iglob(apps_path)]
        try:
            with inactivity.track.task("AVD_scan"):
                await AppVersionDetector().start(unglobbed_paths, doc_roots)
        except AVDExecutionError as exc:  # Exited with non-zero return code
            await self.detached_scan.handle_aborted_process(
                sink=g.sink,
                scan_path=self.path,
                scan_type=self.scan_type,
                scan_started=self.started,
                cmd=exc.command,
                out=exc.out,
                err=exc.err,
            )
        else:
            await MalwareDatabaseScanner(
                self.path,
                [self.path],  # FIXME: pass unglobbed_paths here
                **self.args,
                scan_type=self.scan_type,
                scan_id=self.scanid,
            ).scan(self.started)

async def start(self) -> None:
        self.started = time.time()
        self.scanner_task = create_task_and_log_exceptions(
            asyncio.get_event_loop(), self._scan
        )

# For backward compatibility to deserialize an old queue
State = QueuedScanState
QueuedScan = QueuedFileScan

serialize_scans = serialize_attr(path=Config.SCANS_PATH, attr="_scans_info")
SCAN_TYPE_CLASSES = {
    MalwareScanResourceType.FILE.value: QueuedFileScan,
    MalwareScanResourceType.DB.value: QueuedDbScan,
}

class ScanQueue:
    @property
    def _scans_info(self):
        return collections.deque(item.to_dict() for item in self._scans)

@cached_property
    def _scans(self) -> Deque[QueuedScanBase]:
        # it should be loaded once per instance
        scans = collections.deque()
        for scan_info in unserialize(
            path=Config.SCANS_PATH, fallback=collections.deque
        ):
            try:
                cls = SCAN_TYPE_CLASSES[scan_info["resource_type"]]
                scans.append(cls.from_dict(scan_info))
            except Exception as exc:
                # don't flood Sentry, send one error message
                throttled_log_error(
                    "Can't get scan class for %s due to %s", scan_info, exc
                )
        return scans

@property
    def current_scan(self):
        return self.peek(0)

@serialize_scans
    async def put(
        self,
        paths,
        resource_type: MalwareScanResourceType,
        prioritize=False,
        **scan_args,
    ):
        home_dirs = [Path(user["home"]) for user in await panel_users()]

if resource_type == MalwareScanResourceType.FILE:
            scans_to_add: List[QueuedScanBase] = [
                QueuedFileScan(path, home_dirs=home_dirs, **scan_args)
                for path in paths
            ]
        elif (
            antivirus_mode.disabled
            and resource_type == MalwareScanResourceType.DB
        ):
            scans_to_add = db_scans(paths, home_dirs, scan_args)
        else:
            raise ValueError("Unknown resource_type: {}".format(resource_type))

if prioritize and self._scans:
            running = self._scans.popleft()
            self._scans.extendleft(reversed(scans_to_add))
            self._scans.appendleft(running)
        else:
            self._scans.extend(scans_to_add)

@serialize_scans
    def remove(self, scan=None):
        if len(self) == 0:
            return
        scan = scan or self.current_scan
        self._scans.remove(scan)
        scan.stop()

logger.info("Scans pending: %d", len(self))

def peek(self, priority):
        if -1 < priority < len(self):
            return self._scans[priority]

def find_all(self, scan_ids):
        return [scan for scan in self._scans if scan.scanid in scan_ids]

def find(self, **kwargs) -> Optional[QueuedScanBase]:
        for scan in self._scans:
            if all([getattr(scan, k) == v for k, v in kwargs.items()]):
                return scan
        return None

def update(self, scan_ids, status) -> None:
        for scan in self._scans:
            if scan.scanid in scan_ids:
                scan.state = status

def get_scans_from_paths(self, paths):
        for scan in self.scans:
            if scan.path in paths:
                yield scan, scan.state.value

def scan_summaries(self, scans=None):
        scans = scans or self._scans
        return collections.OrderedDict(
            (
                scan.scanid,
                {
                    "path": scan.path,
                    "scan_status": scan.state.value,
                    "scan_type": scan.scan_type,
                    "started": scan.started,
                    "created": scan.created,
                    "error": None,
                    "total_resources": scan.total_resources,
                    "total_malicious": 0,
                    "resource_type": scan.resource_type.value,
                },
            )
            for scan in scans
        )

@property
    def scans(self):
        return list(self._scans)

def __bool__(self):
        return len(self._scans) > 0

def __contains__(self, scan):
        return scan in self._scans

def __len__(self):
        return len(self._scans)

def db_scans(paths, home_dirs, scan_args):
    return [
        QueuedDbScan(path, home_dirs=home_dirs, **scan_args) for path in paths
    ]

CANCEL

Name	Type	Size	Modified	Actions
↩ ..	DIR	—	—
📁 __pycache__/	DIR	—	2026-08-14 05:35
📁 ai_bolit/	DIR	—	2026-08-14 05:35
📁 cleaners/	DIR	—	2026-08-14 05:35
📁 mds/	DIR	—	2026-08-14 05:35
📄 __init__.py	PY	2.4 KB	2026-08-05 11:17	EDIT
📄 app_version_detector.py	PY	3.8 KB	2026-08-05 11:17	EDIT
📄 crontab.py	PY	1.4 KB	2026-08-05 11:17	EDIT
📄 detached.py	PY	6.8 KB	2026-08-05 11:17	EDIT
📄 queue.py	PY	15.8 KB	2026-08-05 11:17	EDIT
📄 queue_supervisor_sync.py	PY	5.1 KB	2026-08-05 11:17	EDIT
📄 scan_result.py	PY	9.4 KB	2026-08-05 11:17	EDIT
📄 scanner.py	PY	6.2 KB	2026-08-05 11:17	EDIT
📄 utils.py	PY	1.8 KB	2026-08-05 11:17	EDIT