JFIF $
d d 7
Viewing File: /opt/imunify360/venv/lib/python3.11/site-packages/imav/patchman/rpc/endpoints.py
"""
This program is free software: you can redistribute it and/or modify it under
the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License,
or (at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
See the GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
Copyright © 2019 Cloud Linux Software Inc.
This software is also available under ImunifyAV commercial license,
see <https://www.imunify360.com/legal/eula>
"""
import logging
import pathlib
from imav.patchman.config import PatchmanConfig
from imav.patchman.constants import PATCHMAN_PACKAGE, PATCHMAN_SERVICE_NAME
from imav.patchman.custom_integration import PatchmanCustomIntegration
from imav.patchman.exceptions import PatchmanError
from imav.patchman.feature import (
FeatureStatus,
PatchmanFeature,
RealtimeFeature,
)
from imav.patchman.model import Domain, Path, User, patchman_db
from imav.patchman.license import License
from imav.patchman.utils import get_current_ip, is_ipv4, is_private_ip
from imav.subsys.realtime_av import REALTIME_SERVICE_NAME
from defence360agent.contracts.config import Core
from defence360agent.contracts.messages import MessageType
from defence360agent.rpc_tools.lookup import RootEndpoints, bind
from defence360agent.subsys.panels import hosting_panel
from defence360agent.utils import (
check_run,
finally_happened,
OsReleaseInfo,
os_version,
)
logger = logging.getLogger(__name__)
OLD_PATCHMAN_PACKAGE = "patchman-client"
OLD_PATCHMAN_BIN = "/usr/local/patchman/patchmand"
def is_debian() -> bool:
return OsReleaseInfo.id_like() & OsReleaseInfo.DEBIAN
class PatchmanEndpoints(RootEndpoints):
KEYFILE = "/etc/patchman/license/key"
@bind("patchman", "users")
async def users(
self,
custom_integration=False,
integration_type=None,
metadata_path=None,
):
"""Writes the users and their domains to database.
Patchman go agent executes this endpoint to fill the database. In case
of custom integration it retrieves the user-defined parameters from
Patchman Portal and passes it here.
:param custom_integration: true if custom integration, false by default
:param integration_type: only for custom integration, file or script
:param metadata_path: only for custom integration, path to file or
script that represent metadata (File type 1 or Script 1 from
https://docs.imunify360.com/Patchman_custom_integration.pdf)
"""
if custom_integration:
hp = PatchmanCustomIntegration(integration_type, metadata_path)
else:
hp = hosting_panel.HostingPanel()
patchman_users = await hp.patchman_users()
with patchman_db(
[User, Domain, Path], target_dir=Core.TMPDIR, prefix="users"
) as db:
for user_data in patchman_users:
user_name = user_data["username"]
email = user_data["email"] or ""
language = user_data["language"]
parent = user_data["parent"] or ""
level = user_data["level"]
suspended = user_data["suspended"]
user = User.create(
name=user_name,
email=email,
language=language,
parent=parent,
level=level,
suspended=suspended,
)
domains = user_data["domains"]
for domain_data in domains:
domain_name = domain_data["domain"]
domain = Domain.create(name=domain_name, user=user)
paths = domain_data["paths"]
for path_name in paths:
Path.create(name=path_name, domain=domain)
return {"path": db.database}
@bind("patchman", "register")
async def register(self, regkey=None):
"""
Writes a key to Patchman's key file.
Patchman automatically checks for registration every 5 seconds.
"""
keyfile_path = pathlib.Path(self.KEYFILE)
keyfile_path.parent.mkdir(parents=True, exist_ok=True)
with keyfile_path.open("w") as f:
f.write(regkey)
@property
def patchman_feature(self):
return PatchmanFeature()
@property
def realtime_feature(self):
return RealtimeFeature()
async def is_installed(self):
status = (await self.patchman_feature.status())["items"]["status"]
return status in [FeatureStatus.INSTALLING, FeatureStatus.INSTALLED]
async def _migrate_patchman(self) -> None:
if await self.is_installed():
logger.info("%s package already installed", PATCHMAN_PACKAGE)
return
package_manager = "apt-get" if is_debian() else "yum"
cmd = [package_manager, "-y", "install", PATCHMAN_PACKAGE]
if not is_debian():
# use swap to replace old patchman package(s)
cmd = [
package_manager,
"-y",
"swap",
OLD_PATCHMAN_PACKAGE,
PATCHMAN_PACKAGE,
]
if not os_version().startswith("7"): # CentOS/CL > 7
# it was added only here
# https://github.com/rpm-software-management/dnf/pull/61
cmd += ["--allowerasing"]
await check_run(cmd)
await self._sink.process_message(
MessageType.EnsureServiceState(service=PATCHMAN_SERVICE_NAME)
)
@bind("patchman", "install")
async def install(self, ip_address=None):
"""Install imunify-patchman package"""
if ip_address and not is_ipv4(ip_address):
raise ValueError("Only ipv4 address is supported")
if not ip_address:
current_ip = get_current_ip()
if is_private_ip(current_ip):
raise ValueError(
"Can't automatically detect the outbound IP "
"on this machine, since you're using a NAT. "
"Please manually provide the outbound IPv4 address "
"that we should use."
)
if ip_address: # custom IP
PatchmanConfig.set("network", "ip", ip_address)
await self.patchman_feature.install()
if await finally_happened(
self.patchman_feature.has_status,
FeatureStatus.INSTALLED,
max_tries=10,
delay=6,
):
await self._sink.process_message(
MessageType.EnsureServiceState(service=PATCHMAN_SERVICE_NAME)
)
@bind("patchman", "migrate")
async def migrate(self):
"""Migrate old patchman agent to imunify-patchman"""
if not pathlib.Path(OLD_PATCHMAN_BIN).exists():
logger.info("There is nothing to migrate")
return
if not License.has_clean_mode():
logger.warning("Unsupported license type to migrate")
return
await self._migrate_patchman()
@bind("patchman", "uninstall")
async def uninstall(self):
"""Uninstall imunify-patchman package"""
await self.patchman_feature.remove()
@bind("patchman", "status")
async def get_status(self):
"""Get patchman status"""
return {
"status": (await self.patchman_feature.status())["items"][
"status"
],
"realtime": (await self.realtime_feature.status())["items"][
"status"
],
}
@bind("patchman", "install", "realtime")
async def install_realtime(self):
"""Install imunify-realtime-av package"""
status = (await self.patchman_feature.status())["items"]["status"]
if status != FeatureStatus.INSTALLED:
# don't install realtime package without patchman package
raise PatchmanError(
"Can't install realtime package, install patchman first"
)
if not License.has_realtime_enabled():
raise PatchmanError(
"Realtime is not enabled for the current Patchman license"
)
await self.realtime_feature.install()
if await finally_happened(
self.realtime_feature.has_status,
FeatureStatus.INSTALLED,
max_tries=10,
delay=6,
):
await self._sink.process_message(
MessageType.EnsureServiceState(service=REALTIME_SERVICE_NAME)
)
@bind("patchman", "uninstall", "realtime")
async def uninstall_realtime(self):
"""Uninstall imunify-realtime-av package"""
await self.realtime_feature.remove()
Back to Directory
nL+D550H?Mx ,D"v]qv;6*Zqn)ZP 0 !1 A "#a$2Qr D8 a Ri[f\mIykIw0 cuFcRı?lO7к_f˓[C$殷WF<_W ԣs KcëIzyQy /_LKℂ;C",pFA: /]=H ~,ls/9ć:[=/#f;)x {ٛEQ )~ =𘙲r*2~ a _V=' kumF D}KYYC)({ *g&f`툪ry`=^cJ.I](*`wq 1dđ#̩͑0;H]u搂@:~וKL Nsh}OIR*8:2 !lDJVo( 3=M(zȰ+i*NA r6KnSl )!JJӁ* %݉?|D}d5:eP0R;{$X'xF@.ÊB {,WJuQɲRI;9QE琯62fT.DUJ;*cP A\ILNj!J۱+O\ ͔]ޒ SJȧc%ANolՎprULZԛerE2=XDXgVQeӓk yP7U*omQIs,K`)6\G3t?pgjrmۛجwluGtf h9uyP0D;Uڽ"OXlif$)&|ML0Zrm1[HXPlPR0'G=i2N +0e2]] 9VTPO7h(F*癈 '=QVZDF,d߬~TX
G[`le69CR(!S2!P
<0x < !1AQ "Raq02Br#SCTb
?Ζ"]mH5WR7k.ۛ!}Q~+yԏz|@T20S~Kek *zFf^2X*(@8r?CIuI|֓>^ExLgNUY+{.RѪ
τVYTD I62'8Y2 7'\TP.6d&˦@Vqi|8- OΕ]ʔ U=TL8=;6c| !qfF3aů&~$l}'NWUs$Uk^SV:U# 6w++s&r+nڐ{@29 gL
u"TÙM=6 (^"7r} =6YݾlCuhquympǦ
G jhsǜNlɻ}o7 #S6aw4!OSrD57%|?x>L
|/nD6?/8w#[)L7+6〼T ATg ! %5MmZ/c-{1_Je"|^$'O&ޱմ Trb$ w)R$&
N1EtdU3Uȉ1pM" N*(DN yd96.(jQ)X
5cQɎMyW? Q*!R>6=7)Xj5`J]e 8%t!+'!1Q 5 !1 AQaqё#2"0BRb ? Gt^## .llQT $v,,m 㵜5 ubV =sY+@d{N! dnO< .-B; _wJt6;QJd.Qc%p{ 1,sNDdFHI0ГoXшe黅XۢF:)[FGXƹ/w_cMeD,ʡcc. WDtA$j@:) -# u
c1<@ۗ9F)KJ-hpP]_x[qBlbp ʖw q"LFGd ƶ*s+ډ_Zc"? %t[IP 6J]#=ɺVvvCGsGh1 >)6|ey?Lӣm,4GWUi` ]uJVoV DG< SB 6ϏQ@ TiUly OU0kfV~~}SZ@*WUUi##; s/[ =!7}" WN]'(L! ~y5g9T̅JkbM' +s:S + B)v@Mj
e Cf jE 0Y\QnzG1д~Wo{T 9?`Rmyhsy3!HAD]mc1~2L Su7xT;j$`}4->L#vzŏ ILS ֭ T { rjGKC; bpU= -`BsK.SFw4Mq]ZdHS0)tLg