# -*- 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)