ꦫꦣꦺꦤ꧀ꦄꦤ꧀ꦠꦱꦺꦤ

PATH: opt / imunify360 / venv / lib / python3.11 / site-packages / im360 / simple_rpc

from defence360agent.rpc_tools import ValidationError
from defence360agent.rpc_tools.lookup import RootEndpoints, bind
from defence360agent.utils import Scope
from im360.subsys.features import get_applicable_features
from defence360agent.subsys.features.abstract_feature import (
    FeatureError,
    FeatureStatus,
)

class FeaturesEndpoints(RootEndpoints):
    SCOPE = Scope.IM360

@bind("features", "list")
    async def feature_list(self):
        return {"items": list(get_applicable_features().keys())}

@bind("features", "install")
    async def feature_install(self, name):
        return await self.call("install", name)

@bind("features", "remove")
    async def feature_remove(self, name):
        return await self.call("remove", name)

@bind("features", "status")
    async def feature_status(self, name):
        return await self.call("status", name)

async def call(self, action, name):
        try:
            _cls = get_applicable_features()[name]
        except KeyError:
            raise ValidationError("Feature is not available: {}".format(name))
        else:
            if _cls is None:
                raise ValidationError(f"No implementation for feature: {name}")

try:
            feature = await _cls(sink=self._sink).init()
            return await getattr(feature, action)()
        except FeatureError as e:
            # implemented new format for status, but not for install/remove yet
            # also will be nice to move log file path to status response
            # because now, if status returned "installing", UI have to call
            # "install" (!) to get log file path
            if action == "status":
                return {
                    "items": {
                        "status": FeatureStatus.ERROR,
                        "message": str(e),
                    }
                }
            else:
                raise ValidationError(str(e))

[-] unavailable_on_freemium.py [edit]
[-] configuration_management.py [edit]
[-] whitelisted_domains.py [edit]
[-] smtp_blocking.py [edit]
[-] control_panel.py [edit]
[-] schema.py [edit]
[-] proactive.py [edit]
[-] whitelist_rbl.py [edit]
[-] countries.py [edit]
[+] schema
[-] feature.py [edit]
[+] __pycache__
[-] middleware.py [edit]
[-] whitelisted_crawlers.py [edit]
[-] hosting_panel.py [edit]
[-] validate.py [edit]
[-] uninstall_cleanup.py [edit]
[-] conflicts.py [edit]
[-] resident_socket.py [edit]
[-] remote_proxy.py [edit]
[-] health.py [edit]
[-] smart_advice.py [edit]
[-] malware.py [edit]
[-] __init__.py [edit]
[-] lists.py [edit]
[-] csf_imports.py [edit]
[+] schema_responses
[+] ..
[-] incidents.py [edit]
[-] kcarectl.py [edit]
[-] disabled_rules.py [edit]
[-] custom_lists.py [edit]