File: //var/opt/nydus/ops/customer_local_ops/control_panel/plesk.py
# -*- 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)