Admin

name	type	size	perms	modified	actions
[parent directory]
__pycache__/	dir	-	0755	2026-09-01 07:52:27
cpanel/	dir	-	0755	2026-09-01 07:52:27
directadmin/	dir	-	0755	2026-09-01 07:52:27
generic/	dir	-	0755	2026-09-01 07:52:27
no_cp/	dir	-	0755	2026-09-01 07:52:27
plesk/	dir	-	0755	2026-09-01 07:52:27
__init__.py	file	0 B	0644	2026-09-01 07:52:27	edit
base.py	file	9,299 B	0644	2026-09-01 07:52:27	edit
hosting_panel.py	file	1,049 B	0644	2026-09-01 07:52:27	edit

editing: base.py

import pwd
from abc import ABC, abstractmethod
from collections import defaultdict
from dataclasses import dataclass
from enum import IntEnum
from pathlib import Path
from typing import Dict, List, Optional, Set

from defence360agent.utils.ipecho import APIError, IPEchoAPI

TCP_PORTS_COMMON = [
    "20",
    "21",
    "22",
    "25",
    "53",
    "80",
    "110",
    "443",
    "587",
    "993",
    "995",
]

GENERIC_PANEL_NAME = "generic panel"

class UserLevel(IntEnum):
    ADMIN = 1
    RESSELER = 2
    REGULAR_USER = 3

class PanelException(Exception):
    pass

class InvalidTokenException(Exception):
    pass

@dataclass
class DomainData:
    docroot: str
    domain: str
    type: str
    username: str

class AbstractPanel(ABC):
    """Abstract class that provides only basic hosting panel integration
    functionality."""

NAME = "MINIMAL"
    OPEN_PORTS = {
        "tcp": {
            "in": ["465"] + TCP_PORTS_COMMON,
            "out": ["113"] + TCP_PORTS_COMMON,
        },
        "udp": {
            "in": ["20", "21", "53", "443"],
            "out": ["20", "21", "53", "113", "123"],
        },
    }
    exception = PanelException
    smtp_allow_users = []  # type: List[str]

@classmethod
    @abstractmethod
    def is_installed(cls):
        """
        Checks if hosting panel installed on the known path
        :return: bool:
        """
        pass

@classmethod
    def get_server_ip(cls):
        """
        Stub with external IP as currently
        only implementation for cPanel needed
        """
        try:
            return IPEchoAPI.server_ip()
        except APIError:
            return ""

@classmethod
    async def version(cls):
        return None

@classmethod
    async def name(cls):
        return cls.NAME

@abstractmethod
    async def enable_imunify_plugin(self, name=None):
        """
        Registers and enables Imunify360 UI plugin in hosting panel
        """
        pass

@abstractmethod
    async def disable_imunify_plugin(self, name=None):
        """
        UnRegisters Imunify360 UI plugin in hosting panel
        """
        pass

@abstractmethod
    async def get_user_domains(self):
        """
        Returns domains hosted via control panel
        :return: list
        """
        pass

@abstractmethod
    async def get_users(self) -> List[str]:
        """
        Returns system users from hosting panel
        :return: list
        """
        pass

@abstractmethod
    async def get_domain_to_owner(self) -> Dict[str, List[str]]:
        """
        Returns dict with domain to list of users pairs
        """
        pass

@abstractmethod
    async def get_domains_per_user(self) -> Dict[str, List[str]]:
        """
        Returns dict with user to list of domains pairs
        """
        pass

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

async def get_user_details(self) -> Dict[str, Dict[str, str]]:
        """
        Returns dict with user to email pairs
        """

return {
            user: {"email": "", "locale": ""}
            for user in await self.get_users()
        }

async def users_count(self) -> int:
        return len(list(await self.get_users()))

def authenticate(self, protocol, data: dict):
        """
        Performs actions to distinguish endusers from admins
        :param protocol: _RpcServerProtocol
        :param data: parsed params
        :returns (user_type, user_name)
        """
        name = None
        if protocol._uid != 0:
            # we can get here if a non-root web panel user visits i360 UI
            # To emulate it:
            # su -s /bin/bash -c
            #  $'echo \'{"command":["config", "show"],"params":{}}\'
            #    | nc -U -w1 \
            #    /var/run/defence360agent/non_root_simple_rpc.sock'
            #  fakeuser
            pw = pwd.getpwuid(protocol._uid)
            name = pw.pw_name
        return protocol.user, name

@classmethod
    def get_modsec_config_path(cls):
        raise NotImplementedError

def get_SMTP_conflict_status(self) -> bool:
        """
        Return Conflict status
        """
        return False

@abstractmethod
    def basedirs(self) -> Set[str]:
        pass

@classmethod
    def base_home_dir(cls, home_dir: str) -> Path:
        base_dir = Path(home_dir).resolve().parent
        return base_dir

@classmethod
    def get_rapid_scan_db_dir(cls, home_dir: str) -> Optional[str]:
        try:
            base_dir = cls.base_home_dir(home_dir)
            resolved_home = Path(home_dir).resolve()
            tail = resolved_home.relative_to(base_dir)
        # Symbolic link loop could cause runtime error
        except (ValueError, RuntimeError):
            return None

from defence360agent.contracts.config import MalwareTune

if rapid_scan_basedir_override := getattr(
            MalwareTune, "RAPID_SCAN_BASEDIR_OVERRIDE", None
        ):
            base_dir = rapid_scan_basedir_override

return str(base_dir / ".rapid-scan-db" / tail)

@classmethod
    async def retrieve_key(cls) -> str:
        """
        Returns registration key from panel, if possible, raise
        PanelException if not successful (or wrong panel key provided),
        or NoImplemented if method not supported
        by the panel.
        """
        raise NotImplementedError

@classmethod
    async def notify(cls, *, message_type, params, user=None):
        """
        Notify a customer using the panel internal tooling
        """
        return None

@abstractmethod
    async def list_docroots(self) -> Dict[str, str]:
        """
        :return dict with docroot to domain
        """
        pass

@classmethod
    async def switch_ui_config(cls, myimunify_enabled: bool) -> None:
        """
        Switch UI panel configuration between Im360 and MyImunify
        """
        return None

async def get_domain_paths(self) -> Dict[str, List[str]]:
        """
        Domain to docroot list mapping
        Patchman expects a subdomain to be listed separately from main domain

:return: dict with domain to list of docroots
        """
        doc_roots = await self.list_docroots()
        domain_paths = defaultdict(list)
        for doc_root, domain_name in doc_roots.items():
            domain_paths[domain_name].append(doc_root)

return domain_paths

async def patchman_users(self):
        panel_users = await self.get_users()
        domain_paths = await self.get_domain_paths()
        user_domains = await self.get_domains_per_user()
        user_details = await self.get_user_details()

users = []
        for user_name in panel_users:
            details = user_details[user_name]
            email = details["email"]
            locale = details["locale"]
            parent = details.get("parent", "")
            level = details.get("level", int(UserLevel.REGULAR_USER))
            suspended = details.get("suspended", False)

domains = user_domains.get(user_name, [])
            user_domain_paths = []
            for domain_name in domains:
                paths = domain_paths.get(domain_name, [])
                user_domain_paths.append(
                    {
                        "domain": domain_name,
                        "paths": paths,
                    }
                )

users.append(
                {
                    "username": user_name,
                    "email": email,
                    "language": locale,
                    "parent": parent,
                    "level": level,
                    "suspended": suspended,
                    "domains": user_domain_paths,
                }
            )

return users

@classmethod
    async def get_user_domains_details(cls, username: str) -> list[DomainData]:
        raise NotImplementedError()

def ensure_valid_panel(**dec_kwargs):
    """
    Run function only if hosting panel is installed,
    elsewhere raise PanelException

This method is intended to be used as a decorator on AbstractPanel instance
    methods.

:raise PanelException:
    :param dec_kwargs: kwargs passed to is_installed function
    :return:
    """

def real_decorator(fn):
        """
        :param fn: coroutine
        """

async def wrapper(self, *args, **kwargs):
            if not self.is_installed(**dec_kwargs):
                raise self.exception(
                    "%s is not valid!" % self.__class__.__name__
                )
            return await fn(self, *args, **kwargs)

return wrapper

return real_decorator

class ModsecVendorsError(Exception):
    """
    Raises when its impossible to get modsec vendor
    """

pass

def forbid_dns_only(fn):
    """Decorator for functions on cPanel instance methods.

Calls original function if _is_dns_only() returns False, otherwise
    throws cPanelException."""

async def wrapper(self, *args, **kwargs):
        if self._is_dns_only():
            raise self.exception("Method is not allowed for dnsonly panel")
        return await fn(self, *args, **kwargs)

return wrapper

cancel