File Manager

Editing 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/>.

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 File Manager