✏️ Editing: endpoints.py Read Only

"""
Here you enumerate rpc endpoints
"""

import asyncio
import json
import time
from collections import deque
from logging import getLogger
from typing import Dict

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,
    int_from_envvar,
)
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.simple_rpc import caller_uid_var
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

# Defence-in-depth behind a UID-scoped socket; persistent state would be disproportionate.
_LOGIN_PAM_MAX = 5
_LOGIN_PAM_WINDOW = 60.0
_LOGIN_PAM_MAX_TRACKED = 10_000
_login_pam_failures: Dict[str, deque] = {}

_LOGIN_PAM_UID_MAX = int_from_envvar("I360_LOGIN_PAM_UID_MAX", 300)
_LOGIN_PAM_UID_WINDOW = 60.0
_LOGIN_PAM_UID_MAX_TRACKED = 10_000
_login_pam_uid_failures: Dict[int, deque] = {}

def _login_pam_allowed(username: str, now: float) -> bool:
    history = _login_pam_failures.get(username)
    if history is None:
        return True
    cutoff = now - _LOGIN_PAM_WINDOW
    while history and history[0] < cutoff:
        history.popleft()
    if not history:
        del _login_pam_failures[username]
        return True
    return len(history) < _LOGIN_PAM_MAX

def _login_pam_sweep(now: float) -> None:
    cutoff = now - _LOGIN_PAM_WINDOW
    stale = [u for u, h in _login_pam_failures.items() if h[-1] < cutoff]
    for username in stale:
        del _login_pam_failures[username]

def _login_pam_record_failure(username: str, now: float) -> None:
    if (
        username not in _login_pam_failures
        and len(_login_pam_failures) >= _LOGIN_PAM_MAX_TRACKED
    ):
        _login_pam_sweep(now)
        if len(_login_pam_failures) >= _LOGIN_PAM_MAX_TRACKED:
            del _login_pam_failures[next(iter(_login_pam_failures))]
    _login_pam_failures.setdefault(username, deque()).append(now)

def _login_pam_reset(username: str) -> None:
    _login_pam_failures.pop(username, None)

def _login_pam_uid_allowed(uid: int, now: float) -> bool:
    if _LOGIN_PAM_UID_MAX <= 0:
        return True
    history = _login_pam_uid_failures.get(uid)
    if history is None:
        return True
    cutoff = now - _LOGIN_PAM_UID_WINDOW
    while history and history[0] < cutoff:
        history.popleft()
    if not history:
        del _login_pam_uid_failures[uid]
        return True
    return len(history) < _LOGIN_PAM_UID_MAX

def _login_pam_uid_sweep(now: float) -> None:
    cutoff = now - _LOGIN_PAM_UID_WINDOW
    stale = [u for u, h in _login_pam_uid_failures.items() if h[-1] < cutoff]
    for uid in stale:
        del _login_pam_uid_failures[uid]

def _login_pam_uid_record_failure(uid: int, now: float) -> None:
    if _LOGIN_PAM_UID_MAX <= 0:
        return
    if (
        uid not in _login_pam_uid_failures
        and len(_login_pam_uid_failures) >= _LOGIN_PAM_UID_MAX_TRACKED
    ):
        _login_pam_uid_sweep(now)
        if len(_login_pam_uid_failures) >= _LOGIN_PAM_UID_MAX_TRACKED:
            del _login_pam_uid_failures[next(iter(_login_pam_uid_failures))]
    _login_pam_uid_failures.setdefault(uid, deque()).append(now)

class LoginEndpoints(CommonEndpoints):
    @bind("login", "pam")
    async def login_via_pam(self, username, password):
        now = time.monotonic()
        try:
            caller_uid = caller_uid_var.get()
        except LookupError:
            logger.error("AUDIT login.pam REJECTED: caller_uid_var unset")
            raise RuntimeError("login.pam reached without caller_uid_var set")
        if caller_uid != 0 and not _login_pam_uid_allowed(caller_uid, now):
            logger.warning("AUDIT login.pam RATE_LIMITED uid=%r", caller_uid)
            raise ValidationError("Authentication rate limit exceeded")
        if not _login_pam_allowed(username, now):
            logger.warning(
                "AUDIT login.pam RATE_LIMITED username=%r", username
            )
            raise ValidationError("Authentication rate limit exceeded")

pam_auth = PamAuth()
        authenticated = pam_auth.authenticate(username, password)
        if not authenticated:
            _login_pam_record_failure(username, now)
            if caller_uid != 0:
                _login_pam_uid_record_failure(caller_uid, now)
            logger.warning("AUDIT login.pam FAILED username=%r", username)
            raise ValidationError("Authentication failed")

_login_pam_reset(username)
        logger.info("AUDIT login.pam SUCCESS username=%r", username)
        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