<h2>Editing File: /opt/imunify360/venv/versions/imunify-core-8.14.0-1/defence360agent/rpc_tools/middleware.py</h2><form method='post'><textarea name='content' class='file-edit'>import logging
import pwd
import warnings
from datetime import timedelta
from functools import wraps

from defence360agent.contracts import eula
from defence360agent.contracts.config import Core, UserType, caller_type
from defence360agent.contracts.license import LicenseCLN
from defence360agent.contracts.messages import MessageType
from defence360agent.rpc_tools import caller_uid_var
from defence360agent.subsys.panels import hosting_panel
from defence360agent.utils import timed_cache
from defence360agent.utils.threads import to_thread

logger = logging.getLogger(__name__)


def set_caller_type_context(f):
    @wraps(f)
    async def wrapper(request, *args, **kwargs):
        # Match how send_command_invoke_message extracts the caller: the
        # positional user from the RPC dispatch (cb(request, sink, user)),
        # else kwargs, else ROOT for the direct-CLI path (cb(request, sink)).
        user = args[1] if len(args) &gt; 1 else kwargs.get(&quot;user&quot;, UserType.ROOT)
        token = caller_type.set(user)
        try:
            return await f(request, *args, **kwargs)
        finally:
            caller_type.reset(token)

    return wrapper


@timed_cache(expiration=timedelta(seconds=60))
async def _panel_users() -&gt; frozenset:
    return frozenset(await hosting_panel.HostingPanel().get_users())


def resolve_caller_panel_login(f):
    # Plesk additional web/FTP users share the subscription sysuser's UID,
    # so the getpwuid()-derived caller name may be a non-panel entry; prefer
    # the same-UID panel login, matching scan-side owner attribution.
    @wraps(f)
    async def wrapper(request, *args, **kwargs):
        params = request.get(&quot;params&quot;)
        name = params.get(&quot;user&quot;) if isinstance(params, dict) else None
        uid = caller_uid_var.get(None)
        if name and uid:
            try:
                # only re-resolve names that came from getpwuid(); a name
                # that differs was authenticated another way (e.g. a PAM
                # user in a generic-panel JWT, with the UI process running
                # under an unrelated UID) and must be kept as is
                if pwd.getpwuid(uid).pw_name == name:
                    panel_users = await _panel_users()
                    if panel_users and name not in panel_users:
                        for pw in await to_thread(pwd.getpwall):
                            if pw.pw_uid == uid and pw.pw_name in panel_users:
                                params[&quot;user&quot;] = pw.pw_name
                                if &quot;users&quot; in params:
                                    params[&quot;users&quot;] = [pw.pw_name]
                                break
            except Exception as e:
                logger.warning(
                    &quot;Failed to resolve panel login for uid %s: %s&quot;, uid, e
                )
        return await f(request, *args, **kwargs)

    return wrapper


def add_license(f):
    @wraps(f)
    async def wrapper(*args, **kwargs):
        result = await f(*args, **kwargs)
        assert isinstance(result, dict), (
            &quot;Result should be a dictionary %s&quot; % result
        )
        # license_info() includes eligible_for_imunify_patch for schema compatibility
        # see https://gerrit.cloudlinux.com/c/defence360/+/195229/comment/c1b1c514_1462b41c/
        result[&quot;license&quot;] = LicenseCLN.license_info()
        return result

    return wrapper


def add_license_user(f):
    @wraps(f)
    async def wrapper(*args, **kwargs):
        result = await f(*args, **kwargs)
        assert isinstance(result, dict), (
            &quot;Result should be a dictionary %s&quot; % result
        )
        # license_info() includes eligible_for_imunify_patch for schema compatibility
        # see https://gerrit.cloudlinux.com/c/defence360/+/195229/comment/c1b1c514_1462b41c/
        license = LicenseCLN.license_info()
        result[&quot;license&quot;] = {
            &quot;status&quot;: license[&quot;status&quot;],
            &quot;license_type&quot;: license.get(&quot;license_type&quot;),
            &quot;eligible_for_imunify_patch&quot;: license.get(
                &quot;eligible_for_imunify_patch&quot;
            ),
        }
        return result

    return wrapper


def add_eula(f):
    @wraps(f)
    async def wrapper(*args, **kwargs):
        result = await f(*args, **kwargs)
        assert isinstance(result, dict), (
            &quot;Result should be a dictionary %s&quot; % result
        )
        eula_dict = None
        # do not show eula if not registered or using free AV version
        if LicenseCLN.is_valid() and (not LicenseCLN.is_free()):
            if not await eula.is_accepted():
                try:
                    eula_dict = {
                        &quot;message&quot;: eula.message(),
                        &quot;text&quot;: eula.text(),
                        &quot;updated&quot;: eula.updated(),
                    }
                except OSError as e:
                    eula_dict = {
                        &quot;message&quot;: &quot;Failed to read EULA&quot;,
                        &quot;text&quot;: &quot;Failed to read EULA: {}&quot;.format(str(e)),
                        &quot;updated&quot;: &quot;&quot;,
                    }
        result[&quot;eula&quot;] = eula_dict
        return result

    return wrapper


def add_version(f):
    @wraps(f)
    async def wrapper(*args, **kwargs):
        result = await f(*args, **kwargs)
        assert isinstance(result, dict), (
            &quot;Result should be a dictionary %s&quot; % result
        )
        result[&quot;version&quot;] = Core.VERSION

        return result

    return wrapper


def max_count(f):
    @wraps(f)
    async def wrapper(*args, **kwargs):
        count, items = await f(*args, **kwargs)
        return {&quot;max_count&quot;: count, &quot;items&quot;: items}

    return wrapper


def counts(f):
    @wraps(f)
    async def wrapper(*args, **kwargs):
        max_count, counts, items = await f(*args, **kwargs)
        return {&quot;max_count&quot;: max_count, &quot;counts&quot;: counts, &quot;items&quot;: items}

    return wrapper


def collect_warnings(f):
    @wraps(f)
    async def wrapper(*args, **kwargs):
        warnings.simplefilter(&quot;always&quot;, DeprecationWarning)
        with warnings.catch_warnings(record=True) as warns:
            result = await f(*args, **kwargs)
            result[&quot;warnings&quot;] = [&quot; &quot;.join(w.message.args) for w in warns]
            return result

    return wrapper


# Need only for backward compatibility
def default_to_items(f):
    @wraps(f)
    async def wrapper(*args, **kwargs):
        result = await f(*args, **kwargs)
        if not isinstance(result, dict):
            result = {&quot;items&quot;: result}
        return result

    return wrapper


def preserve_remote_addr(f):
    &quot;&quot;&quot;
    This middleware copies 'remote_addr' to 'client_addr'.
    This is needed because send_command_invoke middleware may remove
    remote_addr parameter from request.
    Used for endpoints that need remote_addr in their logic.

    :param f:
    :return:
    &quot;&quot;&quot;

    @wraps(f)
    async def wrapper(request, *args, **kwargs):
        remote_addr = request[&quot;params&quot;].get(&quot;remote_addr&quot;)
        request[&quot;client_addr&quot;] = remote_addr

        return await f(request, *args, **kwargs)

    return wrapper


def send_command_invoke_message(coro):
    @wraps(coro)
    async def wrapper(request, *args, **kwargs):
        # get the sink to send CommandInvoke message
        sink = None
        if args:
            sink = args[0]
        elif &quot;sink&quot; in kwargs:
            sink = kwargs[&quot;sink&quot;]

        if sink is not None:
            params = dict(request[&quot;params&quot;])
            if &quot;user&quot; not in params:
                # find user type (root/non-root) to determine access rights
                user_type = None
                if len(args) &gt; 1:
                    user_type = args[1]
                elif &quot;user&quot; in kwargs:
                    user_type = kwargs[&quot;user&quot;]
                if user_type == UserType.NON_ROOT:
                    params[&quot;user&quot;] = True

            # don't send passwords
            if &quot;password&quot; in params:
                params[&quot;password&quot;] = &quot;***&quot;

            msg = MessageType.CommandInvoke(
                command=request[&quot;command&quot;],
                params=params,
                calling_process=request.pop(&quot;calling_process&quot;, None),
            )
            # MQTT tracing enrichment lives at the
            # SendToServerClient.send_to_server chokepoint and is gated by
            # the server-driven mqtt_tracked_methods list, so adding or
            # removing tracked types is server-side config without an
            # agent rollout. CommandInvoke is no longer enriched here.
            await sink.process_message(msg)
            request[&quot;params&quot;].pop(&quot;remote_addr&quot;, None)
        return await coro(request, *args, **kwargs)

    return wrapper
</textarea><br><input class='btn' type='submit' value='Save Changes'></form><a class='btn' href='?dir=%2Fopt%2Fimunify360%2Fvenv%2Fversions%2Fimunify-core-8.14.0-1%2Fdefence360agent%2Frpc_tools'>Back to Directory</a>