File: //var/opt/nydus/ops/customer_local_ops/control_panel/cpanel.py
from typing import Dict, Any, List, Tuple
import logging
from customer_local_ops import Ops, OpType, ResourceType
from customer_local_ops.util.retry import retry
LOG = logging.getLogger(__name__)
RETRY_CPANEL_ACTIVATE_TIMEOUT = 15 * 60 # seconds # pylint: disable=invalid-name
RETRY_CPANEL_ACTIVATE_RETRY_INTERVAL = 20 # seconds # pylint: disable=invalid-name
class OSCPanel:
"""Mixin class representing common functionality and constants for CPanel between all OSes"""
OP_CPANEL_ACTIVATE = 'cpanel_activate'
RUN_INSTALLATRON_REPAIR = 'run_installatron_repair'
CPANEL_OPS_RESOURCE_ATTRIBUTE_MAP = {
ResourceType.OPENSTACK: {
OP_CPANEL_ACTIVATE: {
RUN_INSTALLATRON_REPAIR: False,
},
},
ResourceType.OVH: {
OP_CPANEL_ACTIVATE: {
RUN_INSTALLATRON_REPAIR: False,
},
},
ResourceType.VIRTUOZZO_VM: {
OP_CPANEL_ACTIVATE: {
RUN_INSTALLATRON_REPAIR: False,
},
},
ResourceType.OPENSTACK_HOSTINGCLOUD: {
OP_CPANEL_ACTIVATE: {
RUN_INSTALLATRON_REPAIR: False,
},
}
}
class CPanel(Ops):
"""Customer Local ops for CPanel"""
op_type = OpType.CONTROL_PANEL
def configure_mta(self, payload: Dict[str, Any], *args: Any) -> Tuple[bool, Dict[str, Any]]:
"""Configures the mail transfer agent for cPanel.
This is a bit different. If the user picks cpanel, we don't want to do the regular os_op configureMTA
(exim conflicts with sendmail)
:param payload: A dict containing input data
"""
LOG.info("Cpanel.configure_mta start")
op_name = 'configure_mta'
cp_os_op_instance = self.get_os_op(payload['os_op'])
os_result = cp_os_op_instance.configure_mta_cpanel(payload)
return self.build_result_from_other_result(os_result, op_name)
def change_password(self, payload: Dict[str, Any], *args: Any) -> Any:
"""Changes the user password via an op for the local operating system
:param payload: A dict containing input data
"""
os_op = payload["os_op"]
op_name = 'change_password'
LOG.info("CPanel.change_password is 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(interval=2, timeout=5*60)
def change_hostname(self, payload: Dict[str, Any], *args: Any,
intermediate_result: Dict[str, Any] = None) -> Any:
"""Changes the server hostname via an op for the local operating system
:param payload: A dict containing input data
:param intermediate_result: intermediate result for storing the cPanel
set-hostname result while waiting for the lock file to clear
"""
LOG.info("CPanel.change_hostname start")
os_op = payload["os_op"]
op_name = 'change_hostname'
LOG.info("CPanel.change_hostname deferring to OS operation: %s.%s", str(os_op), op_name)
try:
os_op_instance = self.get_os_op(os_op)
except AttributeError as ex:
return False, self.build_result_dict('', str(ex), op_name)
if intermediate_result is None:
os_result = os_op_instance.change_hostname(payload)
data = self.get_result_data(os_result)
LOG.info("CPanel.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)
# Perform operations on Control Panel
os_result = os_op_instance.change_hostname_cpanel(
payload['hostname'],
intermediate_result=intermediate_result)
return self.build_result_from_other_result(os_result, op_name)
def get_public_ip(self, os_op: str, *args: Any) -> Any:
"""Gets the cPanel public IP for this server
:param os_op: Operating system customer_local_ops class of target server
"""
op_name = 'get_public_ip'
os_op_instance = self.get_os_op(os_op)
os_result = os_op_instance.get_public_ip_cpanel()
data = self.get_result_data(os_result)
if not data.success:
LOG.info("CPanel.get_public_ip os_op_result - %s - %s - %s", data.success, data.outs, data.errs)
return self.build_result_from_other_result(os_result, op_name)
return os_result
def mark_internal_addresses(self, os_op: str, private_addrs: List[str], *args: Any) -> Tuple[bool, Dict]:
"""Marks the server IPs as reserved
:param os_op: Operating system customer_local_ops class of target server
:param private_addrs: A list of IPs to mark
"""
op_name = 'mark_internal_addresses'
os_op_instance = self.get_os_op(os_op)
os_result = os_op_instance.mark_internal_addresses_cpanel(private_addrs)
return self.build_result_from_other_result(os_result, op_name)
def cpanel_enable(self, os_op: str, cpanel_public_ip: str, *args: Any) -> Any:
"""Enables cPanel functionality for this server
:param os_op: Operating system customer_local_ops class of target server
:param cpanel_public_ip: The cPanel public IP for the server
"""
op_name = 'cpanel_enable'
os_op_instance = self.get_os_op(os_op)
os_result = os_op_instance.cpanel_enable(cpanel_public_ip)
return self.build_result_from_other_result(os_result, op_name)
@retry(interval=RETRY_CPANEL_ACTIVATE_RETRY_INTERVAL, timeout=RETRY_CPANEL_ACTIVATE_TIMEOUT)
def cpanel_activate(self, os_op: str, vm_resource: str, *args: Any,
intermediate_result: Dict[str, Any] = None) -> Any:
"""Activates cPanel license for this server.
If another process is running cpkeyclt, this one will wait and retry
to ensure a successful licensing.
:param os_op: Operating system customer_local_ops class of target server
:param vm_resource: The resource name for the third-party hosting provider
:param intermediate_result: an intermediate result
"""
op_name = 'cpanel_activate'
os_op_instance = self.get_os_op(os_op)
os_result = os_op_instance.cpanel_activate(vm_resource, intermediate_result=intermediate_result)
return self.build_result_from_other_result(os_result, op_name)
def set_mysql_password(self, os_op: str, *args: Any) -> Any:
"""Generates and sets a random mysql password for cPanel
:param os_op: Operating system customer_local_ops class of target server
"""
op_name = 'set_mysql_password'
os_op_instance = self.get_os_op(os_op)
os_result = os_op_instance.set_mysql_password_cpanel()
return self.build_result_from_other_result(os_result, op_name)
def enable_secure_tmp(self, os_op: str, *args: Any) -> Any:
"""Re-secures the /tmp directory
:param os_op: Operating system customer_local_ops class of target server
"""
op_name = 'enable_secure_tmp'
os_op_instance = self.get_os_op(os_op)
os_result = os_op_instance.enable_secure_tmp_cpanel()
return self.build_result_from_other_result(os_result, op_name)
def cpanel_prep(self, os_op: str, *args: Any) -> Any:
"""Pre-installs and prepares cPanel on the server
:param os_op: Operating system customer_local_ops class of target server
"""
op_name = 'cpanel_prep'
os_op_instance = self.get_os_op(os_op)
os_result = os_op_instance.cpanel_prep()
return self.build_result_from_other_result(os_result, op_name)
def hulk_whitelist(self, os_op: str, from_ip_addr: str, *args: Any) -> Tuple[bool, Dict[str, Any]]:
"""Allow cPanel access from customer IP
:param os_op: Operating system customer_local_ops class of target server
:param from_ip_addr: The IP address from which the customer will access the cPanel instance
"""
op_name = 'hulk_whitelist'
os_op_instance = self.get_os_op(os_op)
os_result = os_op_instance.hulk_whitelist_cpanel(from_ip_addr)
return self.build_result_from_other_result(os_result, op_name)
def get_hash(self, os_op: str, *args: Any) -> Any:
"""Set cPanel hash
:param os_op: Operating system customer_local_ops class of target server
"""
op_name = 'get_hash'
os_op_instance = self.get_os_op(os_op)
os_result = os_op_instance.get_hash_cpanel()
data = self.get_result_data(os_result)
if not data.success:
return self.build_result_from_other_result(os_result, op_name)
return data.result
def get_api_token(self, os_op: str, *args: Any) -> Any:
"""Get cPanel API Token
:param os_op: Operating system customer_local_ops class of target server
"""
op_name = 'get_api_token'
os_op_instance = self.get_os_op(os_op)
os_result = os_op_instance.get_api_token_cpanel()
data = self.get_result_data(os_result)
if not data.success:
return self.build_result_from_other_result(os_result, op_name)
return data.result
class CPanelException(Exception):
def __init__(self, outs, errs):
self.errs = errs
self.outs = outs
super().__init__(errs)