FM / control

/var/opt/nydus/ops/customer_local_ops/control_panel/

/var/opt/nydus/ops/customer_local_ops/control_panel

Name	Size	Mode	Actions
powershell/	-	0755	rm
__pycache__/	-	0755	rm
cpanel.py	11751	0644	edit dl rm
ispconfig.py	3194	0644	edit dl rm
linux_cpanel.py	32228	0644	edit dl rm
linux_ispconfig.py	5342	0644	edit dl rm
linux_plesk.py	23852	0644	edit dl rm
plesk.py	13679	0644	edit dl rm
windows_plesk.py	15253	0644	edit dl rm
__init__.py	77	0644	edit dl rm

Edit: /var/opt/nydus/ops/customer_local_ops/control_panel/plesk.py (13679B)

# -*- coding: utf-8 -*-

import logging
import os

from typing import Any, Tuple, Dict, List

from customer_local_ops import NydusResult, Ops, ResourceType, OpType
from customer_local_ops.util.retry import retry, Retry

LOG = logging.getLogger(__name__)

# These two constants are used to define the interval and timeout for retrying a request when the Python/PHP
# DLL conflict is detected on Windows platforms. We want to retry the request twice, for a total of three attempts,
# including the original attempt. The timeout value is technically a little larger than it needs to be to allow for
# any slop in the timing.
CHECK_FOR_DLL_CONFLICT_RETRY_INTERVAL = 10
CHECK_FOR_DLL_CONFLICT_RETRY_TIMEOUT = 25

class OSPlesk:
    """Mixin class representing common functionality and constants for Plesk between all OSes"""
    OP_ENABLE_PLESK = 'enable_plesk'
    RUN_DISABLE_SESSION_IP_CHECK = 'run_disable_session_ip_check'
    RUN_HIDE_INTERNAL_IP = 'run_hide_internal_ip'
    SETUP_CMD = 'setup_cmd'
    PLESK_SETUP_CMD_FORMAT_STRING_LIST = ['{init_conf_cmd} {init_conf_setup_flag}',
                                          '-name admin', '-passwd {password}',
                                          '-license_agreed true', '-email noreply@secureserver.net',
                                          '-admin_info_not_required true']

PLESK_OPS_RESOURCE_ATTRIBUTE_MAP = {
        ResourceType.OPENSTACK: {
            OP_ENABLE_PLESK: {
                RUN_DISABLE_SESSION_IP_CHECK: False,
                RUN_HIDE_INTERNAL_IP: True,
                SETUP_CMD: ' '.join(PLESK_SETUP_CMD_FORMAT_STRING_LIST)
            },
        },
        ResourceType.OVH: {
            OP_ENABLE_PLESK: {
                RUN_DISABLE_SESSION_IP_CHECK: True,
                RUN_HIDE_INTERNAL_IP: False,
                SETUP_CMD: ' '.join(PLESK_SETUP_CMD_FORMAT_STRING_LIST)
            },
        },
        ResourceType.VIRTUOZZO_VM: {
            OP_ENABLE_PLESK: {
                RUN_DISABLE_SESSION_IP_CHECK: True,
                RUN_HIDE_INTERNAL_IP: False,
                SETUP_CMD: ' '.join(PLESK_SETUP_CMD_FORMAT_STRING_LIST)
            },
        },
        ResourceType.OPENSTACK_HOSTINGCLOUD: {
            OP_ENABLE_PLESK: {
                RUN_DISABLE_SESSION_IP_CHECK: True,
                RUN_HIDE_INTERNAL_IP: False,
                SETUP_CMD: ' '.join(PLESK_SETUP_CMD_FORMAT_STRING_LIST)
            },
        }
    }

def get_plesk_dir(self) -> str:
        """Find the installation directory for Plesk

:raises RuntimeError: If a Plesk path cannot be found
        :return: Full path to Plesk installation
        """
        # Implement this in OSPlesk child classes (e.g. WindowsPlesk, LinuxPlesk)
        raise NotImplementedError

def get_path_plesk(self, filename: str = None, dirs: List[str] = None) -> str:
        """Get the path for a Plesk file/command/directory

:param filename: Name of file or executable. If None, only directory is returned.
        :param dirs: List of directories under Plesk directory, in order of depth. These are joined to make the path.
        :raises RuntimeError: If a Plesk path cannot be found
        :return: Full path to Plesk file/command/directory
        """
        if dirs is None:
            dirs = ['bin']
        args = dirs
        if filename is not None:
            args.append(filename)
        return os.path.join(self.get_plesk_dir(), *args)

class Plesk(Ops):
    op_type = OpType.CONTROL_PANEL

RETRY_LICENSE_PLESK_TIMEOUT = 300
    RETRY_LICENSE_PLESK_INTERVAL = 20

@retry(RETRY_LICENSE_PLESK_INTERVAL, RETRY_LICENSE_PLESK_TIMEOUT)
    def license_plesk(self, os_op: str, activation_key: str, *args: Any,
                      intermediate_result: Dict[str, Any] = None) -> Tuple[bool, Dict[str, Any]]:
        """Run license utility on local vm with activation key

:param os_op: Operating system customer_local_ops class of target server
        :param activation_key: Key value passed back from license plesk op on hfs executor
        :param intermediate_result: Dict containing retry-related metadata
        :return: tuple with success, error data
        """
        LOG.info("license plesk")
        op_name = 'license_plesk'
        cp_os_op_instance = self.get_os_op(os_op)
        os_result = cp_os_op_instance.license_plesk(activation_key, intermediate_result=intermediate_result)
        return self.build_result_from_other_result(os_result, op_name)

RETRY_ENABLE_PLESK_TIMEOUT = 1800
    RETRY_ENABLE_PLESK_INTERVAL = 20

@retry(RETRY_ENABLE_PLESK_INTERVAL, RETRY_ENABLE_PLESK_TIMEOUT)
    def enable_plesk(self, os_op: str, vm_ip: str, vm_resource: str, plesk_user: str, plesk_pass: str, *args: Any,
                     intermediate_result: Dict[str, Any] = None) -> Tuple[bool, Dict[str, Any]]:
        """Enable plesk on local server

:param os_op: Operating system customer_local_ops class of target server
        :param vm_ip: External IP address of the server
        :param vm_resource: The resource name for the third-party hosting provider
        :param plesk_user: User name to be used on Plesk instance
        :param plesk_pass: Password to be used on Plesk instance
        :param intermediate_result: Dict containing retry-related metadata
        :return: tuple with success, error data
        """
        LOG.info("enable plesk")
        op_name = 'enable_plesk'
        cp_os_op_instance = self.get_os_op(os_op)
        os_result = cp_os_op_instance.enable_plesk(vm_ip, vm_resource, plesk_user, plesk_pass,
                                                   intermediate_result=intermediate_result)
        return self.build_result_from_other_result(os_result, op_name)

RETRY_SITE_LIST_TIMEOUT = 300
    RETRY_SITE_LIST_INTERVAL = 20

@retry(RETRY_SITE_LIST_INTERVAL, RETRY_SITE_LIST_TIMEOUT)
    def site_list(self, os_op: str, *args: Any,
                  intermediate_result: Dict[str, Any] = None) -> Any:
        """Retrieve a list of Plesk sites

:param os_op: Operating system customer_local_ops class of target server
        :param intermediate_result: Dict containing retry-related metadata
        :return: list of sites or tuple with error data
        """
        LOG.info("get site list")
        cp_os_op_instance = self.get_os_op(os_op)
        return cp_os_op_instance.site_list_plesk(intermediate_result=intermediate_result)

RETRY_SERVER_PREP_TIMEOUT = 3600
    RETRY_SERVER_PREP_INTERVAL = 20

@retry(RETRY_SERVER_PREP_INTERVAL, RETRY_SERVER_PREP_TIMEOUT)
    def server_prep(self, os_op: str, *args: Any,
                    intermediate_result: Dict[str, Any] = None) -> Tuple[bool, Dict[str, Any]]:
        """ Install Plesk on a server

:param os_op: Operating system customer_local_ops class of target server
        :param intermediate_result: Dict containing retry-related metadata
        :return: tuple with success, error data
        """
        LOG.info("install plesk on '%s'", str(self))
        LOG.info("server prep")
        op_name = 'server_prep'
        cp_os_op_instance = self.get_os_op(os_op)
        os_result = cp_os_op_instance.server_prep_plesk(intermediate_result=intermediate_result)
        return self.build_result_from_other_result(os_result, op_name)

RETRY_GET_CLIENT_TIMEOUT = 300
    RETRY_GET_CLIENT_INTERVAL = 20

@retry(RETRY_GET_CLIENT_INTERVAL, RETRY_GET_CLIENT_TIMEOUT)
    def get_client(self, os_op: str, *args: Any, intermediate_result: Dict[str, Any] = None) -> Any:
        """Get Plesk SSO URL

:param os_op: Operating system customer_local_ops class of target server
        :param intermediate_result: Dict containing retry-related metadata
        :return: SSO URL string or tuple with error data
        """
        LOG.info("get client")
        cp_os_op_instance = self.get_os_op(os_op)
        return cp_os_op_instance.get_client_plesk(intermediate_result=intermediate_result)

RETRY_CONFIGURE_MTA_TIMEOUT = 1200  # seconds  # pylint: disable=invalid-name
    RETRY_CONFIGURE_MTA_RETRY_INTERVAL = 30  # seconds  # pylint: disable=invalid-name

# the default steps already do what we want for plesk. stub put here in case we want to change that in the future.
    @retry(interval=RETRY_CONFIGURE_MTA_RETRY_INTERVAL, timeout=RETRY_CONFIGURE_MTA_TIMEOUT)
    def configure_mta(self, payload: Dict[str, Any],
                      *args: Any, intermediate_result: Any = None) -> Tuple[bool, Dict[str, Any]]:
        """Configure mail transfer agent on the Plesk server

:param payload: Dict containing op params
        :param intermediate_result: Dict containing result data and meta data for retries
        :return: tuple with success, retry or error data
        """
        os_op = payload["os_op"]
        op_name = 'configure_mta'
        LOG.info("Plesk.configure_mta start")
        LOG.info("Plesk.configure_mta deferring to OS operation: %s.%s", str(os_op), op_name)
        # Tell the OS Op do something now...
        # Assuming os_op has format module.os e.g. windows.Windows2016:
        try:
            cp_os_op_instance = self.get_os_op(os_op)
        except AttributeError as ex:
            return False, self.build_result_dict('', str(ex), op_name)

os_result = cp_os_op_instance.do_configure_mta(payload, op_name, intermediate_result=intermediate_result)
        # OS Op could return a retry
        if isinstance(os_result, Retry):
            return os_result
        data = self.get_result_data(os_result)
        LOG.info("Plesk.configure_mta cp_os_op result - %s  - %s - %s", data.success, data.outs, data.errs)
        return self.build_result_from_other_result(os_result, op_name)

def change_password(self, payload: Dict[str, Any], *args: Any) -> Tuple[bool, Dict[str, Any]]:
        """ Change password on the Plesk server

:param payload: Dict containing op params
        :return: tuple with success or error data
        """
        op_name = 'change_password'
        os_op = payload["os_op"]
        LOG.info("Plesk.change_password is a NOOP, deferring to OS operation: %s.%s", str(os_op), op_name)
        os_op_instance = self.get_os_op(os_op)
        return os_op_instance.change_password(payload)

RETRY_CHANGE_ADMIN_PASSWORD_TIMEOUT = 300
    RETRY_CHANGE_ADMIN_PASSWORD_INTERVAL = 20

@retry(RETRY_CHANGE_ADMIN_PASSWORD_INTERVAL, RETRY_CHANGE_ADMIN_PASSWORD_TIMEOUT)
    def change_admin_password(self, payload: Dict[str, Any], *args: Any,
                              intermediate_result: Dict[str, Any] = None) -> Tuple[bool, Dict[str, Any]]:
        """ Change admin password on the Plesk server

:param payload: Dict containing op params
        :param intermediate_result: Dict containing retry-related metadata
        :return: tuple with success or error data
        """
        os_op = payload["os_op"]
        op_name = 'change_admin_password'
        cp_os_op_instance = self.get_os_op(os_op)
        os_result = cp_os_op_instance.change_admin_password_plesk(payload['plesk_admin_pass'],
                                                                  intermediate_result=intermediate_result)
        return self.build_result_from_other_result(os_result, op_name)

RETRY_CHANGE_HOSTNAME_TIMEOUT = 120
    RETRY_CHANGE_HOSTNAME_INTERVAL = 20

@retry(RETRY_CHANGE_HOSTNAME_INTERVAL, RETRY_CHANGE_HOSTNAME_TIMEOUT)
    def change_hostname(self, payload: Dict[str, Any], *args: Any,
                        intermediate_result: Dict[str, Any] = None) -> Tuple[bool, Dict[str, Any]]:
        """ Change hostname on the Plesk server

:param payload: Dict containing op params
        :param intermediate_result: Dict containing retry-related metadata
        :return: tuple with success or error data
        """
        # Do some plesk specific work?
        LOG.debug("Plesk.change_hostname start")
        os_op = payload["os_op"]
        op_name = 'change_hostname'
        hostname = payload['hostname']
        cp_os_op_instance = self.get_os_op(os_op)

# Tell the OS Op do something now...
        LOG.info("Plesk.change_hostname deferring to OS operation: %s.%s", str(os_op), op_name)
        os_result = cp_os_op_instance.change_hostname(payload)

data = self.get_result_data(os_result)
        LOG.info("Plesk.change_hostname os_op_result - %s  - %s - %s", data.success, data.outs, data.errs)
        if not data.success:
            return self.build_result_from_other_result(os_result, op_name)

# Now change hostname for Plesk
        result = cp_os_op_instance.change_hostname_plesk(hostname, intermediate_result=intermediate_result)
        return self.build_result_from_other_result(result, op_name)

RETRY_SET_OUTGOING_MAIL_TIMEOUT = 300
    RETRY_SET_OUTGOING_MAIL_INTERVAL = 20

@retry(RETRY_SET_OUTGOING_MAIL_INTERVAL, RETRY_SET_OUTGOING_MAIL_TIMEOUT)
    def set_outgoing_email_ip(self, os_op: str, address: str, *args: Any,
                              intermediate_result: Dict[str, Any] = None) -> NydusResult:
        """Set Plesk's outgoing e-mail IP address.

See General #6 for more information:
        https://docs.plesk.com/en-US/obsidian/administrator-guide/mail/configuring-serverwide-mail-settings.59430/

:param os_op: Operating system customer_local_ops class of target server
        :param address: IP address from which to send e-mail
        :param intermediate_result: Dict containing retry-related metadata
        :return: Nydus operation result
        """
        return self.get_os_op(os_op).set_outgoing_email_ip(address, intermediate_result=intermediate_result)

cmd: