import base64 import json import logging import os from binascii import Error as base64Error from collections import defaultdict from pathlib import Path from typing import Dict, List, Set, Tuple, Union from xml.etree import ElementTree from defence360agent.application.determine_hosting_panel import ( is_plesk_installed, ) from defence360agent.contracts import config from defence360agent.utils import OsReleaseInfo, check_run from defence360agent.utils.common import get_hostname from .. import base from . import api from .utils import PleskConfig PLESK_KEY_REGISTRY = "/etc/sw/keys/" PLESK_IMUNIFY360_PRODUCT_NAME = "ext-imunify360" TCP_PORTS_PLESK = base.TCP_PORTS_COMMON + ["953", "990", "8443", "8447"] PLESK_NOTIFICATION_SCRIPT_PATH = "/usr/local/psa/admin/plib/modules/imunify360/scripts/send-notifications.php" PLESK_NOTIFICATION_HOOK_PATH = ( "/opt/imunify360/venv/share/imunify360/scripts/send-notifications" ) logger = logging.getLogger(__name__) def _safe_get_text(node, tag) -> str: """Avoid AttributeError if tag not found""" _node = node.find(tag) if _node is not None: return _node.text return "" def _get_key_data(key) -> Tuple[str, str]: """Return product name and filename from key data""" filename = "" key_product_name = "" for data in key.findall("value/struct/member"): if _safe_get_text(data, "name") == "filename": filename = _safe_get_text(data, "value/string") if _safe_get_text(data, "name") == "key_product_name": key_product_name = _safe_get_text(data, "value/string") return key_product_name, filename class PleskException(base.PanelException): pass class Plesk(base.AbstractPanel): NAME = "Plesk" OPEN_PORTS = { "tcp": { "in": ["143", "465", "8880", "49152-65535"] + TCP_PORTS_PLESK, "out": ["113", "5224"] + TCP_PORTS_PLESK, }, "udp": { "in": ["20", "21", "53", "443"], "out": ["20", "21", "53", "113", "123"], }, } exception = PleskException @classmethod def is_installed(cls): return is_plesk_installed() @staticmethod async def version(): with open("/usr/local/psa/version", "r") as f: return f.read().split()[0] @base.ensure_valid_panel() async def enable_imunify_plugin(self, name=None): pass @base.ensure_valid_panel() async def disable_imunify_plugin(self, plugin_name=None): pass async def get_users(self) -> List[str]: """Returns a list of Plesk system users""" try: return await api.get_users() except base.PanelException as e: logger.error("Failed to get users: %s", e) return [] async def patchman_users(self): tuples = await api.get_users_for_patchman() res = defaultdict(dict) # https://cloudlinux.slite.com/app/docs/nrQKL-Raf_3ps4#e0bf3d51 client_type_to_level = defaultdict(lambda: base.UserLevel.REGULAR_USER) client_type_to_level.update( { "admin": base.UserLevel.ADMIN, "reseller": base.UserLevel.RESSELER, "client": base.UserLevel.REGULAR_USER, } ) for ( username, email, parent, locale, client_type, domain, homedir, suspended, ) in tuples: res[username]["email"] = "" if email == "NULL" else email res[username]["language"] = "" if locale == "NULL" else locale res[username]["username"] = username res[username]["parent"] = "" if parent == "NULL" else parent res[username]["level"] = int(client_type_to_level[client_type]) res[username]["suspended"] = bool(int(suspended)) if res[username].get("domains") is None: res[username]["domains"] = [] if domain != "NULL": res[username]["domains"].append( { "domain": domain, "paths": [homedir], } ) return list(res.values()) async def get_user_domains(self): """ :return: list: domains hosted on server via plesk """ return await api.get_domains() async def get_domain_to_owner(self): """ :return: domain to list of users pairs """ return await api.get_domain_to_user() async def get_user_to_email(self) -> Dict[str, str]: """ Returns dict with user to email pairs """ return await api.get_user_to_email() async def get_domains_per_user(self): """ :return: user to list of domains pairs """ return await api.get_user_to_domain() async def users_count(self) -> int: return await api.count_customers_with_subscriptions() @classmethod def get_modsec_config_path(cls): if OsReleaseInfo.id_like() & OsReleaseInfo.DEBIAN: return "/etc/apache2/mods-available/security2.conf" else: return "/etc/httpd/conf.d/security2.conf" def basedirs(self) -> Set[str]: basedir = PleskConfig("HTTPD_VHOSTS_D").get() return {basedir} if basedir else set() @classmethod def base_home_dir(cls, _) -> Path: # Local import to save memory on other panels from configparser import ConfigParser with open("/etc/psa/psa.conf") as c: text = "[dummy section]\n" + c.read() config = ConfigParser(delimiters=[" ", "\t"]) config.read_string(text) base_dir = Path( config["dummy section"].get("HTTPD_VHOSTS_D", "/var/www/vhosts") ) return base_dir @classmethod def _retrieve_key(cls) -> Union[str, None]: """Parse xml of registry and corresponding key file to retrive product key. return: str key or None if not found. """ registry = ElementTree.parse( os.path.join(PLESK_KEY_REGISTRY, "registry.xml") ) for member in registry.getroot().findall("struct/member"): if _safe_get_text(member, "name") == "active": for key in member.findall("value/struct/member"): key_product_name, filename = _get_key_data(key) if key_product_name == PLESK_IMUNIFY360_PRODUCT_NAME: key_value = _safe_get_text( ElementTree.parse( os.path.join( PLESK_KEY_REGISTRY, "keys", filename ) ), "{http://parallels.com/schemas/keys/aps/3}" "key-body", ) return base64.b64decode(key_value.encode()).decode() return None @classmethod async def retrieve_key(cls) -> str: """Returns registration key from registered keys, if possible, raise PleskException if not successful.""" try: result = cls._retrieve_key() except (ElementTree.ParseError, base64Error, FileNotFoundError) as e: raise PleskException("failed to retrieve key with error %s" % e) if result: logger.info("key retrieved %s", result) return result raise PleskException("The key not found") async def list_docroots(self): """ :return: dict docroot to domain """ docroot_domains_users = await api.list_docroots_domains_users() return { docroot: domain for docroot, domain, _ in docroot_domains_users } async def panel_user_link(self, username) -> str: """ Returns panel url :return: str """ return "" @classmethod async def notify(cls, *, message_type, params, user=None): """ Notify a customer using Plesk Notifications Hook """ if not Path(PLESK_NOTIFICATION_SCRIPT_PATH).exists(): return False if not config.AdminContacts.ENABLE_ICONTACT_NOTIFICATIONS: return False if not config.should_send_user_notifications(username=user): return False data = {"message_type": message_type, "params": params, "user": user} logger.info(f"{cls.__name__}.notify(%s)", data) stdin = json.dumps(data) await check_run([PLESK_NOTIFICATION_HOOK_PATH], input=stdin.encode()) return { "message_type": message_type, "mainip": cls.get_server_ip(), "base_url": "", "host_server": get_hostname(), "sent_to_root": user is None, "params": params, } @classmethod async def get_user_domains_details( cls, username: str ) -> list[base.DomainData]: all_domains = await api.get_user_domains_details() return [ domain for domain in all_domains if domain.username == username ]