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