✏️ Editing: endpoints.py Read Only

"""
Here you enumerate rpc endpoints
"""

import asyncio
import json
from logging import getLogger

from defence360agent import files
from defence360agent.api.jwt_issuer import JWTIssuer
from defence360agent.api.newsfeed import NewsFeed
from defence360agent.api.pam_auth import PamAuth
from defence360agent.contracts import config, eula
from defence360agent.contracts.config import (
    ANTIVIRUS_MODE,
    Core as CoreConfig,
    ImmutableMerger,
    LocalConfig,
    MutableMerger,
    effective_user_config,
)
from defence360agent.contracts.license import LicenseCLN
from defence360agent.internals.cln import CLN, CLNError, InvalidLicenseError
from defence360agent.myimunify.billing import (
    collect_billing_incompatibilities,
    get_license_type,
)
from defence360agent.rpc_tools import ValidationError
from defence360agent.rpc_tools.lookup import (
    CommonEndpoints,
    RootEndpoints,
    bind,
)
from defence360agent.subsys.panels.base import PanelException
from defence360agent.utils import (
    IMUNIFY_PACKAGE_NAMES,
    CheckRunError,
    antivirus_mode,
    check_db,
    getpwnam,
    system_packages_info,
)
from defence360agent.utils.config import update_config
from defence360agent.utils.support import ZendeskAPIError, send_request
from defence360agent.utils.whmcs import sync_billing_data

from defence360agent.utils.doctor import get_doctor_key

if antivirus_mode.disabled:
    from im360.subsys.panels import hosting_panel
else:
    from defence360agent.subsys.panels import hosting_panel

logger = getLogger(__name__)

class ConfigEndpoints(CommonEndpoints):
    @bind("config", "show")
    async def config_show(self, user=None):
        full_conf = config.ConfigFile()
        if user:
            user_conf_dict = effective_user_config(
                full_conf, config.ConfigFile(user)
            )
            return {"items": user_conf_dict}
        else:
            return {"items": full_conf.config_to_dict()}

@bind("config", "show", "defaults")
    async def config_show_defaults(self):
        layer_paths = MutableMerger.get_layer_names()
        return {
            "items": {
                "mutable_config": MutableMerger(layer_paths).configs_to_dict(),
                "local_config": LocalConfig().config_to_dict(normalize=False),
                "immutable_config": ImmutableMerger(
                    layer_paths
                ).configs_to_dict(),
            }
        }

@bind("config", "update")
    async def config_update(self, items=None, data=None, user=None):
        # workaround for https://cloudlinux.atlassian.net/browse/DEF-3902
        # TODO: remove items from method parameters
        if items:
            data = items[0]
        new_data = json.loads(data)
        logger.warning("AUDIT config.update user=%r data=%r", user, new_data)
        await update_config(
            self._sink,
            new_data,
            user,
        )
        return await self.config_show(user)

@bind("config", "patch")
    async def config_update_ui(self, data=None, user=None):
        logger.warning("AUDIT config.patch user=%r data=%r", user, data)
        await update_config(self._sink, data, user)
        return await self.config_show(user)

@bind("config", "patch-many")
    async def config_update_many_ui(self, data=None, users=None):
        if users is None:
            users = []
        logger.warning("AUDIT config.patch-many users=%r data=%r", users, data)
        for user in users:
            await update_config(self._sink, data, user)
        return {}

@bind("config", "get-many")
    async def config_get_many_ui(self, users=None):
        if users is None:
            return {}
        result = {"items": {}}
        full_conf = config.ConfigFile()
        for user in users:
            user_conf_dict = effective_user_config(
                full_conf, config.ConfigFile(user)
            )
            result["items"][user] = user_conf_dict
        return result

class LoginEndpoints(CommonEndpoints):
    @bind("login", "pam")
    async def login_via_pam(self, username, password):
        pam_auth = PamAuth()
        authenticated = pam_auth.authenticate(username, password)
        if not authenticated:
            logger.warning("AUDIT login.pam FAILED username=%r", username)
            raise ValidationError("Authentication failed")

return {
            "items": JWTIssuer().get_token(
                username, await pam_auth.get_user_type(username)
            )
        }

class RootLoginEndpoints(RootEndpoints):
    @bind("login", "get")
    async def login_get(self, username):
        if not getpwnam(username):
            raise ValidationError("User name not found")

return {
            "items": JWTIssuer().get_token(
                username, await PamAuth().get_user_type(username)
            )
        }

class PackageVersionsEndpoints(CommonEndpoints):
    @bind("get-package-versions")
    async def get_package_versions(self, user=None):
        return {"items": await system_packages_info(IMUNIFY_PACKAGE_NAMES)}

class NewsEndpoints(RootEndpoints):
    @bind("get-news")
    async def get_news(self):
        return {"items": await NewsFeed.get()}

class Endpoints(RootEndpoints):
    license_info = LicenseCLN.license_info

@bind("register")
    async def register(self, regkey=None):
        LicenseCLN.get_token.cache_clear()
        if LicenseCLN.is_registered():
            if LicenseCLN.is_valid():
                if not ANTIVIRUS_MODE:
                    raise ValidationError("Agent is already registered")
            else:
                logger.info(
                    "Unregistering invalid license: %s"
                    % LicenseCLN.get_token()
                )
                await self.unregister()
        try:
            await CLN.register(regkey)
        except InvalidLicenseError as e:
            raise ValidationError(str(e))
        except CLNError as e:
            logger.warning(
                "Can't register %r as imunify360 key. Trying to "
                "register it as a web panel key instead",
                regkey,
            )
            try:
                await CLN.register(
                    await hosting_panel.HostingPanel().retrieve_key()
                )
            except NotImplementedError:
                logger.warning(
                    "Registration with web panel's key doesn't supported"
                )
                raise ValidationError(str(e))
            except PanelException as panel_e:
                raise ValidationError("{}, {}".format(str(e), str(panel_e)))
            except (CLNError, InvalidLicenseError) as e:
                raise ValidationError(str(e))
        return {}

@bind("unregister")
    async def unregister(self):
        if not LicenseCLN.is_registered():
            raise ValidationError("Agent is not registered yet")
        if LicenseCLN.is_free():
            raise ValidationError("Free license can not be unregistered")

await CLN.unregister()
        return {}

@bind("update-license")
    async def update_license(self):
        if not LicenseCLN.is_registered():
            raise ValidationError("Unregistered (server-id is not assigned)")
        token = LicenseCLN.get_token()
        LicenseCLN.users_count = (
            await hosting_panel.HostingPanel().users_count()
        )
        new_token = await CLN.refresh_token(token)
        if new_token is None:
            raise ValidationError("License does not exist. Agent unregistered")
        return {}

@bind("rstatus")
    async def rstatus(self, paid=False):
        LicenseCLN.get_token.cache_clear()
        if not LicenseCLN.is_valid():
            raise ValidationError("License is invalid for current server")
        if paid and LicenseCLN.is_free():
            raise ValidationError("Free license")
        return self.license_info()

@bind("version")
    async def version(self):
        return {"items": CoreConfig.VERSION}

@bind("wakeup")
    async def wakeup(self):
        """Wake up the agent, so it can process the request, if it's sleeping"""
        return {}

@bind("update")
    async def update_files(
        self, subj=None, force=False, list=False, version="latest"
    ):
        if subj and subj in config.FilesUpdate.DISABLED:
            if list:
                return files.Index(subj).get_list()
            if version:
                return await files.Index(subj).update_to(version, force)
        else:
            if list or version != "latest":
                raise ValidationError(
                    "Listing and version are not supported for this files type"
                )
        try:
            await files.update(subj, force)
        except (asyncio.TimeoutError, files.UpdateError):
            pass  # the error has been logged in files.update already

@bind("eula", "accept")
    async def eula_accept(self):
        await eula.accept()

@bind("eula", "show")
    async def eula_show(self):
        return eula.text()

@bind("checkdb")
    async def checkdb(self, recreate_schema=False):
        """Check DB consistency and repair if needed.
        If recreate_schema is set recreate schema for attached DB."""
        if recreate_schema:
            check_db.recreate_schema()
        else:
            check_db.check_and_repair()

@bind("doctor")
    async def doctor(self):
        key = await get_doctor_key()
        return (
            "Please, provide this key:\n%s\nto Imunify360 Support Team\n" % key
        )

@bind("support", "send")
    async def send_to_support(
        self, email, subject, description, cln=None, attachments=None
    ):
        # Generating doctor and extracting key from output
        try:
            doctor_key = await get_doctor_key()
        except CheckRunError:
            doctor_key = None

# Sending request via Zendesk API
        # https://developer.zendesk.com/rest_api/docs/core/requests#anonymous-requests
        try:
            ticket_url = await send_request(
                email, subject, description, doctor_key, cln, attachments
            )
        except ZendeskAPIError as e:
            logger.error(
                "Got error from Zendesk API. error=%s, description=%s,"
                " details=%s",
                e.error,
                e.description,
                e.details,
            )
            raise

return {"items": [ticket_url]}

class WhmcsEndpoint(RootEndpoints):
    """
    Describes all endpoints for interaction with WHMCS
    """

# needed by WHMCS to know whether it is compatible
    VERSION = "1"

@bind("billing", "sync")
    async def billing_sync(self, data):
        try:
            decoded_data = json.loads(data)
        except json.JSONDecodeError:
            raise ValueError("Invalid JSON")
        result = await sync_billing_data(self._sink, decoded_data)
        return {"result": "success", "data": result}

@bind("billing", "get-config")
    async def billing_get_config(self):
        result = dict(
            version=self.VERSION,
            billing_license=get_license_type(),
            issues=await collect_billing_incompatibilities(),
        )
        return {"result": "success", "data": result}

🔐 Sid Gifari File Manager Pro