1#@!#!123s

: / opt / imunify360 / venv / lib / python3.11 / site-packages / defence360agent / simple_rpc /
Filename : endpoints.py
back
"""
Here you enumerate rpc endpoints
"""

import asyncio
import json
import os
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 eula
from defence360agent.contracts import config
from defence360agent.contracts.config import ANTIVIRUS_MODE
from defence360agent.contracts.config import Core as CoreConfig
from defence360agent.contracts.config import (
    ImmutableMerger,
    LocalConfig,
    MutableMerger,
    Packaging,
    effective_user_config,
)
from defence360agent.contracts.license import LicenseCLN
from defence360agent.internals.cln import CLN, CLNError, InvalidLicenseError
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 (
    CheckRunError,
    antivirus_mode,
    check_db,
    check_run,
    getpwnam,
    system_packages_info,
)
from defence360agent.utils.config import update_config
from defence360agent.utils.whmcs import sync_billing_data
from defence360agent.myimunify.billing import (
    get_license_type,
    collect_billing_incompatibilities,
)
from defence360agent.utils.support import ZendeskAPIError, send_request

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

logger = getLogger(__name__)

DOCTOR_CMD = (
    "wget -qq -O - "
    "https://repo.imunify360.cloudlinux.com/defence360/imunify-doctor.sh "
    "| bash"
)


async def _package_get_doctor_key():
    dir_ = Packaging.DATADIR
    if not os.path.isdir(dir_):
        dir_ = ".."
    out = await check_run([os.path.join(dir_, "scripts", "imunify-doctor.sh")])
    key = out.decode().strip()
    return key


async def _repo_get_doctor_key():
    out = await check_run(DOCTOR_CMD, shell=True)
    key = out.decode().strip()
    if not key:
        raise ValueError("Doctor key is empty")
    return key


async def _get_doctor_key():
    try:
        key = await _repo_get_doctor_key()
    except (CheckRunError, ValueError):
        key = await _package_get_doctor_key()
    return key


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)
        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):
        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 = []
        for user in users:
            await update_config(self._sink, data, user)
        return {}


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:
            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):
        package_list = {
            "imunify-ui",
            "imunify360-firewall",
            "imunify-antivirus",
            "imunify-core",
        }

        return {"items": await system_packages_info(package_list)}


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):
        LicenseCLN.get_token.cache_clear()
        if not LicenseCLN.is_valid():
            raise ValidationError("License is invalid for current server")
        return self.license_info()

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

    @bind("update")
    async def update_files(self, subj=None, force=False):
        try:
            await files.update(subj, force)
        except (asyncio.TimeoutError, files.UpdateError) as err:
            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}