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) > 1 else kwargs.get("user", 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() -> 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("params") name = params.get("user") 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["user"] = pw.pw_name if "users" in params: params["users"] = [pw.pw_name] break except Exception as e: logger.warning( "Failed to resolve panel login for uid %s: %s", 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), ( "Result should be a dictionary %s" % result ) # license_info() includes eligible_for_imunify_patch for schema compatibility # see https://gerrit.cloudlinux.com/c/defence360/+/195229/comment/c1b1c514_1462b41c/ result["license"] = 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), ( "Result should be a dictionary %s" % 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["license"] = { "status": license["status"], "license_type": license.get("license_type"), "eligible_for_imunify_patch": license.get( "eligible_for_imunify_patch" ), } return result return wrapper def add_eula(f): @wraps(f) async def wrapper(*args, **kwargs): result = await f(*args, **kwargs) assert isinstance(result, dict), ( "Result should be a dictionary %s" % 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 = { "message": eula.message(), "text": eula.text(), "updated": eula.updated(), } except OSError as e: eula_dict = { "message": "Failed to read EULA", "text": "Failed to read EULA: {}".format(str(e)), "updated": "", } result["eula"] = 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), ( "Result should be a dictionary %s" % result ) result["version"] = Core.VERSION return result return wrapper def max_count(f): @wraps(f) async def wrapper(*args, **kwargs): count, items = await f(*args, **kwargs) return {"max_count": count, "items": items} return wrapper def counts(f): @wraps(f) async def wrapper(*args, **kwargs): max_count, counts, items = await f(*args, **kwargs) return {"max_count": max_count, "counts": counts, "items": items} return wrapper def collect_warnings(f): @wraps(f) async def wrapper(*args, **kwargs): warnings.simplefilter("always", DeprecationWarning) with warnings.catch_warnings(record=True) as warns: result = await f(*args, **kwargs) result["warnings"] = [" ".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 = {"items": result} return result return wrapper def preserve_remote_addr(f): """ 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: """ @wraps(f) async def wrapper(request, *args, **kwargs): remote_addr = request["params"].get("remote_addr") request["client_addr"] = 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 "sink" in kwargs: sink = kwargs["sink"] if sink is not None: params = dict(request["params"]) if "user" not in params: # find user type (root/non-root) to determine access rights user_type = None if len(args) > 1: user_type = args[1] elif "user" in kwargs: user_type = kwargs["user"] if user_type == UserType.NON_ROOT: params["user"] = True # don't send passwords if "password" in params: params["password"] = "***" msg = MessageType.CommandInvoke( command=request["command"], params=params, calling_process=request.pop("calling_process", 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["params"].pop("remote_addr", None) return await coro(request, *args, **kwargs) return wrapper