Admin

name	type	size	perms	modified	actions
[parent directory]
__pycache__/	dir	-	0755	2026-09-01 07:52:27
__init__.py	file	46 B	0644	2026-09-01 07:52:27	edit
api.py	file	7,460 B	0644	2026-09-01 07:52:27	edit
panel.py	file	9,224 B	0644	2026-09-01 07:52:27	edit
upgrade_urls.py	file	2,943 B	0644	2026-09-01 07:52:27	edit
utils.py	file	191 B	0644	2026-09-01 07:52:27	edit

editing: panel.py

import base64
import json
import logging
import os
from binascii import Error as base64Error
from collections import defaultdict
from pathlib import Path
from typing import Dict, List, Set, Tuple, Union
from xml.etree import ElementTree

from defence360agent.application.determine_hosting_panel import (
    is_plesk_installed,
)
from defence360agent.contracts import config
from defence360agent.utils import OsReleaseInfo, check_run
from defence360agent.utils.common import get_hostname

from .. import base
from . import api
from .utils import PleskConfig

PLESK_KEY_REGISTRY = "/etc/sw/keys/"
PLESK_IMUNIFY360_PRODUCT_NAME = "ext-imunify360"
TCP_PORTS_PLESK = base.TCP_PORTS_COMMON + ["953", "990", "8443", "8447"]

PLESK_NOTIFICATION_SCRIPT_PATH = "/usr/local/psa/admin/plib/modules/imunify360/scripts/send-notifications.php"
PLESK_NOTIFICATION_HOOK_PATH = (
    "/opt/imunify360/venv/share/imunify360/scripts/send-notifications"
)

logger = logging.getLogger(__name__)

def _safe_get_text(node, tag) -> str:
    """Avoid AttributeError if tag not found"""

_node = node.find(tag)
    if _node is not None:
        return _node.text
    return ""

def _get_key_data(key) -> Tuple[str, str]:
    """Return product name and filename from key data"""

filename = ""
    key_product_name = ""
    for data in key.findall("value/struct/member"):
        if _safe_get_text(data, "name") == "filename":
            filename = _safe_get_text(data, "value/string")
        if _safe_get_text(data, "name") == "key_product_name":
            key_product_name = _safe_get_text(data, "value/string")
    return key_product_name, filename

class PleskException(base.PanelException):
    pass

class Plesk(base.AbstractPanel):
    NAME = "Plesk"
    OPEN_PORTS = {
        "tcp": {
            "in": ["143", "465", "8880", "49152-65535"] + TCP_PORTS_PLESK,
            "out": ["113", "5224"] + TCP_PORTS_PLESK,
        },
        "udp": {
            "in": ["20", "21", "53", "443"],
            "out": ["20", "21", "53", "113", "123"],
        },
    }
    exception = PleskException

@classmethod
    def is_installed(cls):
        return is_plesk_installed()

@staticmethod
    async def version():
        with open("/usr/local/psa/version", "r") as f:
            return f.read().split()[0]

@base.ensure_valid_panel()
    async def enable_imunify_plugin(self, name=None):
        pass

@base.ensure_valid_panel()
    async def disable_imunify_plugin(self, plugin_name=None):
        pass

async def get_users(self) -> List[str]:
        """Returns a list of Plesk system users"""
        try:
            return await api.get_users()
        except base.PanelException as e:
            logger.error("Failed to get users: %s", e)
            return []

async def patchman_users(self):
        tuples = await api.get_users_for_patchman()
        res = defaultdict(dict)
        # https://cloudlinux.slite.com/app/docs/nrQKL-Raf_3ps4#e0bf3d51
        client_type_to_level = defaultdict(lambda: base.UserLevel.REGULAR_USER)
        client_type_to_level.update(
            {
                "admin": base.UserLevel.ADMIN,
                "reseller": base.UserLevel.RESSELER,
                "client": base.UserLevel.REGULAR_USER,
            }
        )
        for (
            username,
            email,
            parent,
            locale,
            client_type,
            domain,
            homedir,
            suspended,
        ) in tuples:
            res[username]["email"] = "" if email == "NULL" else email
            res[username]["language"] = "" if locale == "NULL" else locale
            res[username]["username"] = username
            res[username]["parent"] = "" if parent == "NULL" else parent
            res[username]["level"] = int(client_type_to_level[client_type])
            res[username]["suspended"] = bool(int(suspended))
            if res[username].get("domains") is None:
                res[username]["domains"] = []
            if domain != "NULL":
                res[username]["domains"].append(
                    {
                        "domain": domain,
                        "paths": [homedir],
                    }
                )

return list(res.values())

async def get_user_domains(self):
        """
        :return: list: domains hosted on server via plesk
        """
        return await api.get_domains()

async def get_domain_to_owner(self):
        """
        :return: domain to list of users pairs
        """
        return await api.get_domain_to_user()

async def get_user_to_email(self) -> Dict[str, str]:
        """
        Returns dict with user to email pairs
        """
        return await api.get_user_to_email()

async def get_domains_per_user(self):
        """
        :return: user to list of domains pairs
        """
        return await api.get_user_to_domain()

async def users_count(self) -> int:
        return await api.count_customers_with_subscriptions()

@classmethod
    def get_modsec_config_path(cls):
        if OsReleaseInfo.id_like() & OsReleaseInfo.DEBIAN:
            return "/etc/apache2/mods-available/security2.conf"
        else:
            return "/etc/httpd/conf.d/security2.conf"

def basedirs(self) -> Set[str]:
        basedir = PleskConfig("HTTPD_VHOSTS_D").get()
        return {basedir} if basedir else set()

@classmethod
    def base_home_dir(cls, _) -> Path:
        # Local import to save memory on other panels
        from configparser import ConfigParser

with open("/etc/psa/psa.conf") as c:
            text = "[dummy section]\n" + c.read()
        config = ConfigParser(delimiters=[" ", "\t"])
        config.read_string(text)
        base_dir = Path(
            config["dummy section"].get("HTTPD_VHOSTS_D", "/var/www/vhosts")
        )
        return base_dir

@classmethod
    def _retrieve_key(cls) -> Union[str, None]:
        """Parse xml of registry and corresponding key file to retrive
        product key.

return: str key or None if not found.
        """

registry = ElementTree.parse(
            os.path.join(PLESK_KEY_REGISTRY, "registry.xml")
        )
        for member in registry.getroot().findall("struct/member"):
            if _safe_get_text(member, "name") == "active":
                for key in member.findall("value/struct/member"):
                    key_product_name, filename = _get_key_data(key)
                    if key_product_name == PLESK_IMUNIFY360_PRODUCT_NAME:
                        key_value = _safe_get_text(
                            ElementTree.parse(
                                os.path.join(
                                    PLESK_KEY_REGISTRY, "keys", filename
                                )
                            ),
                            "{http://parallels.com/schemas/keys/aps/3}"
                            "key-body",
                        )
                        return base64.b64decode(key_value.encode()).decode()
        return None

@classmethod
    async def retrieve_key(cls) -> str:
        """Returns registration key from registered keys, if possible, raise
        PleskException if not successful."""

try:
            result = cls._retrieve_key()
        except (ElementTree.ParseError, base64Error, FileNotFoundError) as e:
            raise PleskException("failed to retrieve key with error %s" % e)
        if result:
            logger.info("key retrieved %s", result)
            return result
        raise PleskException("The key not found")

async def list_docroots(self):
        """
        :return: dict docroot to domain
        """
        docroot_domains_users = await api.list_docroots_domains_users()
        return {
            docroot: domain for docroot, domain, _ in docroot_domains_users
        }

async def panel_user_link(self, username) -> str:
        """
        Returns panel url
        :return: str
        """
        return ""

@classmethod
    async def notify(cls, *, message_type, params, user=None):
        """
        Notify a customer using Plesk Notifications Hook
        """
        if not Path(PLESK_NOTIFICATION_SCRIPT_PATH).exists():
            return False
        if not config.AdminContacts.ENABLE_ICONTACT_NOTIFICATIONS:
            return False
        if not config.should_send_user_notifications(username=user):
            return False

data = {"message_type": message_type, "params": params, "user": user}

logger.info(f"{cls.__name__}.notify(%s)", data)

stdin = json.dumps(data)
        await check_run([PLESK_NOTIFICATION_HOOK_PATH], input=stdin.encode())

return {
            "message_type": message_type,
            "mainip": cls.get_server_ip(),
            "base_url": "",
            "host_server": get_hostname(),
            "sent_to_root": user is None,
            "params": params,
        }

@classmethod
    async def get_user_domains_details(
        cls, username: str
    ) -> list[base.DomainData]:
        all_domains = await api.get_user_domains_details()
        return [
            domain for domain in all_domains if domain.username == username
        ]

cancel