|
Server : Apache System : Linux srv.rainic.com 4.18.0-553.47.1.el8_10.x86_64 #1 SMP Wed Apr 2 05:45:37 EDT 2025 x86_64 User : rainic ( 1014) PHP Version : 7.4.33 Disable Function : exec,passthru,shell_exec,system Directory : /opt/imunify360/venv/lib64/python3.11/site-packages/defence360agent/simple_rpc/ |
Upload File : |
"""
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,
)
from defence360agent.utils.config import update_config
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__)
async def _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
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", "update-ui")
async def config_update_ui(self, data=None, user=None):
await update_config(self._sink, data, user)
return await self.config_show(user)
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 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):
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):
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]}