submit.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 logging

from defence360agent.rpc_tools.validate import ValidationError
from defence360agent.utils import is_cluster
from imav.malwarelib.config import PENDING
from imav.malwarelib.utils import malware_response

logger = logging.getLogger(__name__)

FALSE_POSITIVE = "fp"
FALSE_NEGATIVE = "fn"

SUBMIT_SUCCESS = "OK"
SUBMIT_PENDING = PENDING
SUBMIT_FAILED = "failed"

async def submit_malware(filename, type, reason=None):
    """
    Submit file to MRS for analysis
    :param filename: str -> path to file to submit
    :param type: str -> report type (fp or fn)
    :param scanner: int -> scanner id
    :return: one of SUBMIT_SUCCESS, SUBMIT_PENDING, SUBMIT_FAILED
    :raises LicenseError:
    """

# submit to MRS
    # Convert CLI format to short format
    upload_reason = (
        malware_response.FALSE_POSITIVE
        if type == FALSE_POSITIVE
        else malware_response.FALSE_NEGATIVE
    )

if is_cluster():
        from imav.malwarelib.cleanup.storage import (
            resolve_file_for_upload_k8s,
        )
        from imav.malwarelib.model import MalwareHit
        from imav.malwarelib.tenant_path import split_prefixed, to_prefixed
        from imav.malwarelib.utils.user_list import registered_apps

# The cluster branch works with a plain str: downstream sqlite
        # binds and os.fspath() recursion on a nested MalwareHitPath
        # wrapper blow up. The non-cluster branch below must NOT coerce 
        # the uploader reads content_path from the wrapper, and str() would
        # collapse it to the live real_path.
        filename_str = str(filename)

# On k8s the tenant is encoded in the path itself:
        # /<app_id>/var/www/... The UI malicious list displays bare
        # paths, so a bare path is also accepted iff it resolves to
        # exactly ONE tenant's stored hits  deterministic, unlike the
        # removed newest-hit heuristic. Ambiguous or unknown paths are
        # rejected.
        registered = await registered_apps()
        app_id, _ = split_prefixed(filename_str)
        if app_id not in registered:
            candidates = [to_prefixed(filename_str, u) for u in registered]
            owners = {
                hit.user
                for hit in MalwareHit.select(MalwareHit.user).where(
                    MalwareHit.orig_file.in_(candidates)
                )
                if hit.user
            }
            if len(owners) == 1:
                app_id = owners.pop()
                filename_str = to_prefixed(filename_str, app_id)
            else:
                raise ValidationError(
                    "On Kubernetes the path must carry the application id"
                    " as its first component (/<app_id>/var/www/...);"
                    " %r does not resolve to exactly one registered"
                    " application (matches: %d)" % (filename_str, len(owners))
                )

async with resolve_file_for_upload_k8s(
            filename_str, app_id=app_id
        ) as upload_paths:
            if not upload_paths:
                logger.error(
                    f"File {filename_str} not available for upload in K8s mode"
                )
                return SUBMIT_FAILED
            files_to_upload = [
                malware_response.MalwareHitPath(path, filename_str)
                for path in upload_paths
            ]
            try:
                for file_to_upload in files_to_upload:
                    await malware_response.upload_with_retries(
                        file_to_upload,
                        upload_reason=upload_reason,
                        notify_timeout=malware_response.SUBMIT_TIMEOUT,
                    )
            except malware_response.TimeoutError as e:
                logger.warning(
                    f"File {filename_str} uploading timed out."
                    f" Marking as pending. {e}"
                )
                return SUBMIT_PENDING
            except malware_response.ClientUploadError as e:
                logger.error(f"File {filename_str} uploading failed. {e}")
                return SUBMIT_FAILED

return SUBMIT_SUCCESS

try:
        await malware_response.upload_with_retries(
            filename,
            upload_reason=upload_reason,
            notify_timeout=malware_response.SUBMIT_TIMEOUT,
        )
        return SUBMIT_SUCCESS
    except malware_response.TimeoutError as e:
        logger.warning(
            f"File {filename} uploading timed out. Marking as pending. {e}"
        )
        return SUBMIT_PENDING
    except malware_response.ClientUploadError as e:
        logger.error(f"File {filename} uploading failed. {e}")
        return SUBMIT_FAILED

docroot: /home/induvedam/public_html

curdir: /opt/imunify360/venv/lib/python3.11/site-packages/imav/malwarelib/utils

	.	4096
	..	4096
X	__init__.py	8202
	__pycache__	4096
X	chattr.py	2446
X	check_file.py	2486
X	cloudways.py	2565
X	crontab.py	20802
X	endpoints.py	1464
X	malware_response.py	18772
X	persistent_mrs_queue.py	12596
X	revisium.py	5752
X	sqlite_limits.py	1689
X	submit.py	5516
X	user_list.py	22567