Last Piece Hacktivist Crew

Editing: conflicts.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>
"""
from defence360agent.contracts.config import ANTIVIRUS_MODE
from imav.malwarelib.subsys.cxs import cxswatch_running
from imav.malwarelib.subsys.pure_ftpd import thirdparty_uploadscript
from defence360agent.subsys.panels.hosting_panel import HostingPanel
from defence360agent.rpc_tools.lookup import RootEndpoints, bind
from defence360agent.utils import Scope
from defence360agent.utils.hyperscan import is_ssse3_supported
from defence360agent.utils.resource_limits import (
    LimitsMethod,
    get_current_method,
)

class ConflictsEndpoints(RootEndpoints):
    SCOPE = Scope.AV

@bind("3rdparty", "conflicts")
    async def thirdparty_conflicts(self):
        results = {}
        for func in self._funcs:
            await func(results)
        return {"items": results}

@property
    def _funcs(self):
        """List of functions for "3rdparty conflicts" endpoint"""

common_conflict_methods = [
            self._intensity_conflicts,
            self._hyperscan_conflicts,
        ]
        extended_conflict_methods = [
            self._malware_inotify_scan_conflicts,
            self._malware_pureftp_scan_conflicts,
        ] * (not ANTIVIRUS_MODE)
        return common_conflict_methods + extended_conflict_methods

@staticmethod
    async def _intensity_conflicts(results_dict):
        if await get_current_method() == LimitsMethod.NICE:
            results_dict["scan_intensity_io_cfq"] = True
        return results_dict

@classmethod
    async def _hyperscan_conflicts(cls, results_dict):
        if not is_ssse3_supported():
            # SSSE3 instruction set is required to run HyperScan
            # We may change it to SSE2 in the future
            results_dict["scan_hyperscan_no_ssse3"] = True

return results_dict

@staticmethod
    async def _malware_inotify_scan_conflicts(results_dict):
        if not HostingPanel().is_installed():
            results_dict["scan_inotify_control_panel"] = True
        elif cxswatch_running():
            results_dict["scan_inotify_cxswatch_daemon"] = True
        return results_dict

@staticmethod
    async def _malware_pureftp_scan_conflicts(results_dict):
        _thirdparty_uploadscript = thirdparty_uploadscript()
        if _thirdparty_uploadscript is not None:
            results_dict["scan_pure_ftpd"] = True
        return results_dict

Cancel