1#@!#!123s

: / opt / imunify360 / venv / lib / python3.11 / site-packages / im360 / subsys / panels / directadmin /
Filename : panel.py
back
import logging
from functools import lru_cache, partial
from typing import Set
from pathlib import Path

from defence360agent.subsys.panels.directadmin import DirectAdmin as Base
from defence360agent.subsys.panels.directadmin.config import ConfigOptions
from imav.malwarelib.subsys.pure_ftpd import PureFTPBaseConfig
from im360.subsys.panels.base import PanelInterface
from .mod_security import DirectAdminModSecurity, get_custombuild_webserver
from .remoteip import RemoteIP

PURE_FTP_CONFIG = (
    "/usr/local/directadmin/custombuild/configure/pureftpd/pure-ftpd.conf"
)
SETUP_TXT_OLD = Path("/usr/local/directadmin/scripts/setup.txt")
SETUP_TXT = Path("/usr/local/directadmin/conf/setup.txt")
VERSIONS = "/usr/local/directadmin/custombuild/versions.txt"

logger = logging.getLogger(__name__)


class DirectAdmin(Base, PanelInterface, DirectAdminModSecurity, RemoteIP):
    pure_ftp_conf_cls = partial(PureFTPBaseConfig, filename=PURE_FTP_CONFIG)

    @lru_cache(maxsize=1)
    def _get_config_options(self):
        config = {"is_SSL_on": False, "SSL_port": False, "port": "2222"}
        try:
            config["is_SSL_on"] = (ConfigOptions("SSL").get() or "0") == "1"
            config["SSL_port"] = ConfigOptions("ssl_port").get() or False
            config["port"] = ConfigOptions("port").get() or "2222"
        except (OSError, KeyError):
            logger.exception("Failed to obtain options from DA config")
        return config

    @lru_cache(maxsize=1)
    def _get_setup_file(self):
        if SETUP_TXT.exists():
            return SETUP_TXT
        elif SETUP_TXT_OLD.exists():
            return SETUP_TXT_OLD
        else:
            raise FileNotFoundError("setup.txt file not found")

    async def _get_all_admin_emails(self) -> list:
        emails = ConfigOptions("email", filename=self._get_setup_file()).get()
        if emails:
            return [email.strip() for email in emails.split(",") if email]
        return []

    def http_ports(self) -> Set[int]:
        config = self._get_config_options()
        if config["is_SSL_on"]:
            return set()
        try:
            return {int(config["port"])}
        except ValueError:
            return set()

    def https_ports(self) -> Set[int]:
        config = self._get_config_options()
        ports = set()
        try:
            if config["SSL_port"]:
                ports.add(int(config["SSL_port"]))
        except ValueError:
            logger.warning(
                "Bad value for `SSL_port` in DirectAdmin configuration: %s",
                config["SSL_port"],
            )
        try:
            if config["is_SSL_on"]:
                ports.add(int(config["port"]))
        except ValueError:
            logger.warning(
                "Bad value for `port` in DirectAdmin configuration: %s",
                config["port"],
            )
        return ports

    def remoteip_supported(self) -> bool:
        return True

    async def get_web_server(self):
        return await get_custombuild_webserver()