Da

✏️ Editing: clcontrollib.py

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

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

import grp
import os
import re
import sys
import syslog
import functools
from typing import Tuple, Optional, List  # NOQA

import cldetectlib as detect
from clcommon import ClPwd, mysql_lib
from clcommon.clproc import ProcLve
from clcommon.utils import grep
from clcommon.cpapi.plugins import cpanel, directadmin, ispmanager, plesk

BAD_CODING_ERROR_CODE = 48

class ConfigDataError(Exception):
    """
    Should be raised when something went wrong during user's config data
    loading/parsing.
    Supposed to be used when we want to skip some user with broken
    configuration files
    """
    pass

def detect_panelclass():
    detect.getCPName()
    if detect.CP_NAME == 'Plesk':
        return Plesk()
    if detect.CP_NAME == 'cPanel':
        return Cpanel()
    if detect.CP_NAME == 'InterWorx':
        return InterWorx()
    if detect.CP_NAME == 'ISPManager':
        return ISPManager()
    if detect.CP_NAME == 'DirectAdmin':
        return DirectAdmin()
    if detect.CP_NAME == 'HostingNG':
        return HostingNG()
    return Unknown()

class GeneralPanel:

def __init__(self):
        # verifying proper use panel class
        self.users_data = ClPwd()
        self.class_name = self.__class__.__name__
        self.current_panel_name = detect.getCPName()
        if self.class_name.lower() != self.current_panel_name.lower():
            raise UserWarning(
                f'Class "{self.class_name}" not should be used on panel "{self.current_panel_name}"'
            )
        self.cpapi_plugin = None

def list_admins(self, debug=False):
        """
        List all admins names in given control panel
        :param bool debug: Do produce debug output or don't
        :return: list of strings
        """
        return self.cpapi_plugin.admins()

def is_admin(self, username):
        """
        Return True if username is in admin names
        :param str username: user to check
        :return: bool
        """
        return self.cpapi_plugin.is_admin(username)

def _warning(self, debug, method_name):
        if debug:
            sys.stderr.write(f"{method_name} is not implemented for {self.__class__.__name__}\n")

@staticmethod
    def _slog_warning(msg):
        syslog.syslog(syslog.LOG_WARNING, f'CL_CONTROLLIB: {msg}')

def _check_and_get_user(self, uid=None, username=None, debug=False,
                            syslog_=True):
        """
        Check that provided user is really exists in system and return it as:
        - uid if username was provided
        - username if uid was provided
        - None if user doesn't exists
        Produce syslog/debug output depending on flags if user isn't exists.

This helper function is mostly needed because it's possible that
        user is already deleted in system but his entity is still present in
        control panel files (like DA_USERS_DIR) so we should do additional
        checks to skip such users.

:return: (uid | username) or None
        """
        if uid is not None and username is not None:
            raise ValueError('This function accepts either uid or username, '
                             'but not both simultaneously')

if uid is None and username is None:
            raise ValueError('Incorrect call: uid or username is not specified')

if username is not None:
            try:
                # will return uid
                return self.users_data.get_uid(username)
            except ClPwd.NoSuchUserException as e:
                if debug:
                    print(e)
                if syslog_:
                    self._slog_warning(
                        f'User with username="{username}" is not present in the system'
                    )
                return None

try:
            # will return username
            return self.users_data.get_names(int(uid))[0]
        except ClPwd.NoSuchUserException as e:
            if debug:
                print(e)
            if syslog_:
                self._slog_warning(
                    f'User with uid="{uid}" is not present in the system'
                )
            return None
        except ValueError:
            print("Incorrect user id")
            return None

@staticmethod
    def get_file_lines(file_name):
        """
        Safely read file content and close it before return the content string.
        No exceptions will have caught!

:param string file_name: Absolute or relative path to file
        :rtype: string
        :return: Content of given file path
        """
        with open(file_name, 'r', encoding='utf-8') as f:
            content = f.readlines()
        return content

@staticmethod
    def write_file_lines(file_name, content, open_method='w'):
        """
        Safely write to file and close it before the function end.
        No exceptions will have caught!

:param string file_name: Absolute or relative path to file
        :param string content: What to write there
        :param string open_method: Either "w" or "a" or "b"
        :return: None
        """
        with open(file_name, open_method, encoding='utf-8') as f:
            f.writelines(content)

def list_all(self, debug=False):
        """
        Implements /usr/bin/getcontrolpaneluserspackages and
        /usr/bin/getcontrolpaneluserspackages --list-all commands
        Result is stdout output:
            uid package name

:param bool debug: Do produce debug output or don't
        :return: None
        """
        self._warning(debug, "List all")

def list_users_in_package(self, package, debug=False):
        """
        Implements /usr/bin/getcontrolpaneluserspackages --package=package_name
        Result is stdout output:
            uid

:param string package: Package name
        :param bool debug: Do produce debug output or don't
        :return: None
        """
        self._warning(debug, "List users in given package")

def list_users_in_reseller_package(self, package, reseller, debug=False):
        """
        Implements
        /usr/bin/getcontrolpaneluserspackages --package=package_name --reseller=reseller_name
        Result is stdout output:
            uid

:param string package: Package name
        :param string reseller: Reseller name
        :param bool debug: Do produce debug output or don't
        :return: None
        """
        self._warning(debug, "List users in given reseller's package")

def list_user_package(self, uid, debug=False):
        """
        Implements /usr/bin/getcontrolpaneluserspackages --userid=id
        Result is stdout output:
            uid package name

:param int uid: User ID
        :param bool debug: Do produce debug output or don't
        :return: None
        """
        self._warning(debug, "List given user's package")

def list_packages(self, debug=False):
        """
        Implements /usr/bin/getcontrolpaneluserspackages --list-packages
        Result is stdout output:
            package name

:param bool debug: Do produce debug output or don't
        :return: None
        """
        self._warning(debug, "List all packages")

def list_resellers_packages(self, debug=False):
        """
        Implements /usr/bin/getcontrolpaneluserspackages --list-resellers-packages
        Result is stdout output:
            reseller_name package name

:param bool debug: Do produce debug output or don't
        :return: None
        """
        self._warning(debug, "List packages belong to resellers")

def list_reseller_packages(self, reseller, debug=False):
        """
        Implements
        /usr/bin/getcontrolpaneluserspackages --list-reseller-packages=reseller
        Result is stdout output:
            package name

:param string reseller: Reseller name
        :param bool debug: Do produce debug output or don't
        :return: None
        """
        self._warning(debug, "List packages belong to given reseller")

def list_users(self, debug=False):
        """
        Implements /usr/bin/getcontrolpaneluserspackages --list-users
        Result is stdout output:
            uid,package name,reseller_name

:param bool debug: Do produce debug output or don't
        :return: None
        """
        self._warning(debug, "List users")

def _get_user_packages(self, debug=False, reseller_name=None):
        """
        Generate triple with uid, it's package, it's reseller

def list_reseller_users(self, debug=False, reseller=None):
        """
        Print uid and package for defined reseller
        :param bool debug: Do produce debug output or don't
        :param reseller: username of reseller
        :return: None
        """
        if self.class_name != 'GeneralPanel':
            if reseller is None:
                if debug:
                    print('Reseller is not defined')
                return
            for uid, package, _ in self._get_user_packages(debug=debug, reseller_name=reseller):
                print(f'{uid},{package}')

def get_reseller(self, uid, debug=False):
        """
        Determine user's reseller
        :param string uid: uid of user
        :param bool debug: Do produce debug output or don't
        :rtype: string
        :return: reseller name
        """
        self._warning(debug, "Get user's reseller")

class Cpanel(GeneralPanel):
    USERDIR = '/var/cpanel/users'
    PACKAGEDIR = '/var/cpanel/packages'
    RESELLERS_PATH = '/var/cpanel/resellers'
    RESELLERS_AND_USERS = '/etc/trueuserowners'

users_data = None

def __init__(self):
        super().__init__()
        self.min_uid = self.users_data.get_sys_min_uid()
        self.cpapi_plugin = cpanel.PanelPlugin()

def get_file_lines(self, file_name):
        """
        Safely read file content and close it before return the content string.
        No exceptions will have caught!

:param string file_name: Absolute or relative path to file
        :rtype: string
        :return: Content of given file path
        """
        ENCODING_LINK = "http://kb.cloudlinux.com/2017/08/how-to-fix-the-issue-" \
                        "with-non-utf-symbols-in-packages-names/"
        try:
            with open(file_name, 'r', encoding='utf-8') as f:
                content = f.readlines()
        except UnicodeDecodeError as e:
            message = f"Unable to read file '{file_name}', error is: {e}. " \
                      f"It looks like you use not an UTF-8 charset."
            if file_name.startswith(self.USERDIR):
                message += f"\nPlease read the instruction: {ENCODING_LINK}"
            print(message, file=sys.stderr)
            self._slog_warning(message)
            sys.exit(BAD_CODING_ERROR_CODE)
        return content

def _get_users_resellers_dict(self):
        """
        Get dict, where keys - user's names, values - reseller's names
        :return: dict {user1: reseller1, ...}
        """
        result = {}
        # The file has a following structure:
        # ------------
        # #userowners v1
        # asdf123: root
        # t2est2: root
        # ------------
        # We used to just skip the first line, but for some reason it wasn't
        # there on my machine, so we would skip an actual reseller info.
        # Now we're using a regular expression.
        # In cPanel a username can only contain letters and digits.
        pattern = re.compile(r"^(?P<name>\w+):\s*(?P<owner>\w+)$")
        if os.path.isfile(self.RESELLERS_AND_USERS):
            file_content = self.get_file_lines(self.RESELLERS_AND_USERS)
            for line in file_content:
                stripped = line.strip()
                match = re.match(pattern, stripped)
                if match is None:
                    continue
                result[match.group('name')] = match.group('owner')
        return result

def _get_resellers(self):
        """
        Get resellers list from RESELLERS_PATH

:rtype: list
        :return: list of resellers names
        """
        resellers = []
        if os.path.isfile(self.RESELLERS_PATH):
            file_content = self.get_file_lines(self.RESELLERS_PATH)
            resellers = [line.strip().split(':', 1)[0] for line in file_content]
        return resellers

def get_package(self, path):
        """
        Get package name from user's config file

:param string path: Path to file contains info about cPanel user
        :rtype: string
        :return: Package name
        """
        package = ''
        if os.path.isfile(path):
            file_content = self.get_file_lines(path)
            result = list(grep(
                r'PLAN\=.*$',
                fixed_string=False,
                match_any_position=False,
                multiple_search=False,
                data_from_file=file_content
            ))
            if result:
                package = result[0].strip().split('=')[1]
        return package

def _get_reseller(self, uid, debug=False):
        """
        Determine user's reseller based on `owner` field from user's config

:param string uid: uid of user
        :param bool debug: Do produce debug output or don't
        :rtype: string
        :return: reseller name
        """
        reseller = ''
        username = self._check_and_get_user(uid=uid, debug=debug)
        if username:
            path = os.path.join(self.USERDIR, username)
            if os.path.isfile(path):
                file_content = self.get_file_lines(path)
                result = list(grep(
                    r'OWNER\=.*$',
                    fixed_string=False,
                    match_any_position=False,
                    multiple_search=False,
                    data_from_file=file_content
                ))
                if result:
                    reseller = result[0].strip().split('=')[1]
        return reseller if reseller and reseller in self._get_resellers() else ''

def get_reseller(self, uid, debug=False):
        """
        Determine user's reseller based on `owner` field from user's config

:param string uid: uid of user
        :param bool debug: Do produce debug output or don't
        :rtype: string
        :return: reseller name
        """

print(self._get_reseller(uid, debug=debug))

def _get_packages(self):
        """
        Generate list of packages in current panel

:rtype: generator of (string, bool)
        :return: (name of package, True) for normal files and (/path/to/wrang/file, False) otherwise
        """
        if os.path.isdir(self.PACKAGEDIR):
            for filename in os.listdir(self.PACKAGEDIR):
                if os.path.isfile(os.path.join(self.PACKAGEDIR, filename)):
                    yield (filename, True)
                else:
                    yield (os.path.join(self.PACKAGEDIR, filename), False)

def list_packages(self, debug=False):  # --list-packages
        resellers = self._get_resellers()
        for filename, is_file in self._get_packages():
            if is_file:
                if '_' in filename and filename.split('_', 1)[0] in resellers:
                    continue
                print(filename)
            elif debug:
                print(f"Warning: {filename} is not a file.")

def _get_user_packages(self, debug=False, reseller_name=None):
        """
        Generate triple with uid, it's package and it's reseller for every user in USERDIR
        which uid is greater than MIN_UID

:param bool debug: Do produce debug output or don't
        :param string reseller_name: filter by this reseller
        :rtype: generator of (int, string, string)
        :return: (uid, package name, reseller name)
        """
        users_resellers_dict = self._get_users_resellers_dict()
        # users_resellers_dict example:
        #  {'r1': 'root', 'user1': 'res1res1', 'res1res1': 'res1res1',
        #   'r': 'root', 'res1root': 'root', 'cltest1': 'root'}
        if not users_resellers_dict:
            return
        if os.path.isdir(self.USERDIR):
            for filename in os.listdir(self.USERDIR):
                try:
                    uid = self.users_data.get_uid(filename)
                    if uid < self.min_uid:  # skip all systemusers with uid < MIN_UID
                        continue
                except ClPwd.NoSuchUserException as e:
                    if debug:
                        print(e)
                else:
                    path = os.path.join(self.USERDIR, filename)
                    reseller = users_resellers_dict.get(filename, self._get_reseller(uid, debug))
                    reseller = '' if self.is_admin(reseller) else reseller
                    if reseller_name is not None and reseller != reseller_name:
                        continue
                    yield (uid, self.get_package(path), reseller)

def list_all(self, package=None, debug=False, reseller=None):  # pylint: disable=arguments-renamed
        for uid, self_package, self_reseller in self._get_user_packages(debug, reseller_name=reseller):
            if self_package != '':
                if package is None:
                    print(uid, self_package)
                elif package == self_package:
                    print(uid)

def list_user_package(self, uid, debug=False):  # --userid
        try:
            users = self.users_data.get_names(int(uid))
        except ClPwd.NoSuchUserException:
            if debug:
                print("getcontrolpaneluserspackages: User not found")
        except ValueError:
            print("Incorrect user id")
        else:
            for user in users:
                path = self.USERDIR + '/' + user
                if os.path.isfile(path):
                    package = self.get_package(path)
                    if package != '':
                        print(package)

def list_resellers_packages(self, debug=False, reseller=None):
        resellers = self._get_resellers()
        for filename, is_file in self._get_packages():
            if is_file and '_' in filename:
                res_name = filename.split('_', 1)[0]
                if res_name not in resellers:
                    continue
                if reseller is not None and reseller != res_name:
                    continue
                print(f"{res_name} {filename}")
            elif not is_file and debug:
                print(f"Warning: {filename} is not a file.")

def list_reseller_packages(self, reseller, debug=False):
        resellers = self._get_resellers()
        if reseller not in resellers:
            if debug:
                print(f"Error: {reseller} is not reseller")
            return
        self.list_resellers_packages(debug=debug, reseller=reseller)

def list_users_in_package(self, package, debug=False):  # --package
        resellers = self._get_resellers()
        if "_" in package:
            parts = package.split('_')
            if any('_'.join(parts[:i]) in resellers for i in range(1, len(parts))):
                return
        if package is not None:
            self.list_all(package)
        elif debug:
            print("getcontrolpaneluserspackages: Undefined package")

def list_users_in_reseller_package(self, package, reseller, debug=False):
        resellers = self._get_resellers()
        if reseller in resellers:  # and package.startswith("%s_" % reseller):
            self.list_all(package, debug=debug, reseller=reseller)
        elif debug:
            print("getcontrolpaneluserspackages: Undefined reseller package")

def list_users(self, debug=False):
        for uid, self_package, reseller in self._get_user_packages(debug):
            print(f"{uid},{self_package},{reseller}")

class DirectAdmin(GeneralPanel):
    DA_USERS_DIR = '/usr/local/directadmin/data/users/'
    DA_CONF = '/usr/local/directadmin/conf/directadmin.conf'
    DA_ADMIN_LIST = '/usr/local/directadmin/data/admin/admin.list'
    DA_RESELLERS_LIST = '/usr/local/directadmin/data/admin/reseller.list'
    DA_ADMINS_PACKAGES = '/usr/local/directadmin/data/admin/packages.list'
    DA_ADMIN = ""
    ENCODING = ""
    users_data = None
    proc_lve = ProcLve()

def __init__(self):
        super().__init__()
        self.DA_ADMIN = detect.detect_DA_admin()
        self.cpapi_plugin = directadmin.PanelPlugin()
        # Detect DA native encoding (see LU-1334)
        self.ENCODING = self.cpapi_plugin.get_encoding_name()

def get_file_lines(self, file_name):
        try:
            with open(file_name, 'r', encoding=self.ENCODING) as f:
                content = f.readlines()
            return content
        except UnicodeDecodeError as e:
            message = f"Unable to read file '{file_name}'. " \
                      f"It looks like you use not an {self.ENCODING} charset."
            self._slog_warning(message + f" Error is: {e}.")
            raise RuntimeError(message) from e

def _get_user_info(self, username):
        # type: (str) -> Tuple[int, str, str]
        """
        Just a wrapper around _get_user_info_inner to send errors to syslog
        """
        try:
            return self._get_user_info_inner(username)
        # We catch all possible errors because anything can happen during
        # user's data loading and/or parsing/processing and all that we can do
        # about this is just log error and skip this user on upper level:
        except Exception as e:
            tpl = "Skipping user '{}' with bad configuration files due to '{}'"
            self._slog_warning(tpl.format(username, e))
            # Should be used on higher levels to skip user if it's applicable:
            raise ConfigDataError() from e

def _get_user_info_inner(self, username):
        # type: (str) -> Tuple[int, str, str]
        """
        Return uid, package name and owner of user (reseller)
        """
        userprofile = os.path.join(self.DA_USERS_DIR, username, 'user.conf')
        try:
            userprofile_content = self.get_file_lines(userprofile)
        except Exception:
            userprofile_content = None
        if not userprofile_content:     # not exists or empty
            raise ValueError('Unable to read any data from user.conf')

# TODO: maybe it is better to parse config file instead of grep?
        result_usertype = list(grep(
            r'usertype\=.*$',
            fixed_string=False,
            match_any_position=False,
            multiple_search=False,
            data_from_file=userprofile_content,
        ))
        result_creator = list(grep(
            r'creator\=.*$',
            fixed_string=False,
            match_any_position=False,
            multiple_search=False,
            data_from_file=userprofile_content,
        ))
        try:
            usertype = result_usertype[0].strip().split('=')[1]
        except IndexError as e:
            raise ValueError('No "usertype" record in user config') from e

try:
            reseller = result_creator[0].strip().split('=')[1]
        except IndexError as e:
            raise ValueError('No "creator" record in user config') from e

package = self._get_user_package(username, userprofile_content)

# DA stores 'creator' in config but we need
        # 'owner' or 'vendor', so additional checks needed
        if usertype == 'reseller':
            # reseller has another kind of package
            # so we mark owner as 'root'
            reseller = ''
            # when reseller has active limits his user ignores package limits
            # otherwise we have problems with limits inheritance
            # (yay, backwards compatibility!)
            lvp_id = self.users_data.get_uid(username)
            if self._is_reseller_limits_enabled(lvp_id):
                package = 'DEFAULT'
                reseller = username

# and finally get userid from passwd file
        uid = self.users_data.get_uid(username)
        return uid, package, reseller

def _is_reseller_limits_enabled(self, lvp_id):
        return self.proc_lve.exist_lvp(lvp_id)

def _get_user_package(self, username: str, userprofile_content: Optional[List[str]] = None) -> str:
        """
        Reads user's package just as in the `_get_user_info_inner` function, but without masking to DEFAULT package
        Be aware that it may return different package than the `_get_user_info_inner` function
        """
        if not userprofile_content:
            userprofile = os.path.join(self.DA_USERS_DIR, username, 'user.conf')
            try:
                userprofile_content = self.get_file_lines(userprofile)
            except Exception:
                userprofile_content = None
            if not userprofile_content:     # not exists or empty
                raise ValueError('Unable to read any data from user.conf')
        result_package = list(grep(
            r'package\=.*$',
            fixed_string=False,
            match_any_position=False,
            multiple_search=False,
            data_from_file=userprofile_content,
        ))
        result_original_package = list(grep(
            r'original_package\=.*$',
            fixed_string=False,
            match_any_position=False,
            multiple_search=False,
            data_from_file=userprofile_content,
        ))

# yay, sometimes directadmin stores package using other option name
        # https://www.directadmin.com/features.php?id=1180
        try:
            package = result_package[0].strip().split('=')[1]
        except IndexError as e:
            raise ValueError('No "package" record in user config') from e
        if package == 'custom' and result_original_package:
            try:
                package = result_original_package[0].strip().split('=')[1]
            except IndexError as e:
                raise ValueError('No "original_package" record in user config') from e
        return package

def _get_user_packages(self, debug=False, reseller_name=None):
        """
        Generate triple with uid, it's package, it's reseller
        :param bool debug: Do produce debug output or don't
        :param string reseller_name: filter by this reseller
        :rtype: generator of (int, string, string)
        :return: (uid, package name, reseller name)
        """
        list_admins = set(self.list_admins())
        for username in os.listdir(self.DA_USERS_DIR):
            # skip all DA admins, check added long-long time ago
            # you can find some details in LU-290
            if username in list_admins:
                continue

if self._check_and_get_user(username=username, debug=debug) is None:
                continue

try:
                uid, package, reseller = self._get_user_info(username)
            except ConfigDataError:
                continue

# filter users by reseller
            if reseller_name is not None and reseller != reseller_name:
                continue
            yield uid, package, reseller

def list_reseller_users(self, debug=False, reseller=None):
        # type: (bool, str) -> None
        """
        Print uid and package for users of a specified reseller
        :param debug: whether to produce debug output
        :param reseller: username of a reseller
        :return: None
        """
        path = os.path.join(self.DA_USERS_DIR, reseller, 'users.list')
        if not os.path.exists(path):
            return

reseller_users = self.get_file_lines(path)
        for username in reseller_users:
            try:
                uid, package, _ = self._get_user_info(username.rstrip('\n'))
            except ConfigDataError:
                continue
            print(f'{uid},{package}')

# do not forget about reseller's own user
        if not self.is_admin(reseller):
            try:
                uid, package, _ = self._get_user_info(reseller)
            except ConfigDataError:
                return
            print(f'{uid},{package}')

def get_reseller(self, uid, debug=False):
        """
        Determine user's reseller based on package from user's config
        :param string uid: uid of user
        :param bool debug: Do produce debug output or don't
        :rtype: string
        :return: reseller name
        """
        username = self._check_and_get_user(uid=uid, debug=debug)
        if username:
            try:
                _, _, reseller = self._get_user_info(username)
            except ConfigDataError:
                return
            print(reseller)

def is_da_admin(self, login_name):
        """
        Check login_name is da admin. Backward compatibility wrapper
        :param login_name: Admin name
        :return:
        """
        return self.is_admin(login_name)

def is_da_admin_for_hook(self, login_name):
        """
        Check login_name is da admin for using in hook

return True - if login_name is admin
               False - otherwise
        """
        # Attention!!!
        # In hook we can't search login_name in list, which returns by self.admins(),
        # because DA calls hook before (!) updating /usr/local/directadmin/data/admin/admin.list
        # So in hook we never find login_name in this file. And we should to determine
        # user type from his user.conf file. This is a universal method.
        result = False
        # TODO this functionality partially duplicates self._get_user_info
        path = os.path.join(self.DA_USERS_DIR, login_name, 'user.conf')
        if os.path.isfile(path):
            file_content = self.get_file_lines(path)
            result_usertype = list(grep(
                r'usertype\=.*$',
                fixed_string=False,
                match_any_position=False,
                multiple_search=False,
                data_from_file=file_content,
            ))
            if result_usertype:
                usertype = result_usertype[0].strip().split('=')[1]
                if usertype == 'admin':
                    result = True
        return result

def add_all_da_admins_to_sudoers(self, s_cmd):
        """
        Add all DA admins to sudoers. Now calls only from CageFS plugin installer
        :param s_cmd: String such as "Clsudo.add_cagefs_user(user)"
        :return:
        """
        try:
            # Do not remove Clsudo from import list
            # NOQA - local disable pyflakes unused import checks
            from clsudo import (Clsudo, NoSuchUser, UnableToReadFile, UnableToWriteFile)  # NOQA
        except ImportError:
            print("Cannot import Clsudo. Check if alt-python27-cllib package installed\n")
            sys.exit(1)
        da_admins = self.list_admins()
        for user in da_admins:
            try:
                eval(s_cmd)
            except (NoSuchUser, UnableToReadFile, UnableToWriteFile) as e:
                print(f'{e}\n')

def get_all_da_admins(self, debug=False):
        """
           Wrapper for compatibility with old code
        """
        return self.list_admins(debug)

def list_admins(self, debug=False):
        """
        Get all DA admins list

:rtype: list of strings
        :return: List of DA admins names
        """
        da_admins = []
        if os.path.isfile(self.DA_ADMIN_LIST):
            file_content = self.get_file_lines(self.DA_ADMIN_LIST)
            da_admins = [line.rstrip('\n') for line in file_content]
        return da_admins

def _get_packages(self):
        file_content = self.get_file_lines(self.DA_ADMINS_PACKAGES)
        for line in file_content:
            yield line.rstrip('\n')

def list_packages(self, debug=False):
        for line in self._get_packages():
            print(line)

def list_all(self, debug=False):
        for uid, package, _ in self._get_user_packages():
            print(uid, package)

def list_users(self, debug=False):
        for uid, package, reseller in self._get_user_packages():
            print(f"{uid},{package},{reseller}")

def list_user_package(self, uid, debug=False):
        username = self._check_and_get_user(uid=uid, debug=debug)
        if username:
            try:
                _, package, _ = self._get_user_info(username)
            except ConfigDataError:
                return
            print(package)

def list_users_in_package(self, package, debug=False):  # --package
        # here we work only with root packages
        for uid, package_, reseller in \
                self._get_user_packages(reseller_name=''):
            if package_ == package:
                print(uid)

def list_users_in_reseller_package(self, package, reseller, debug=False):  # --package --reseller
        # here we work only with end user's packages
        for uid, package_, _ in \
                self._get_user_packages(reseller_name=reseller):
            if package_ == package:
                print(uid)

def _get_resellers(self):
        """
        Return list of resellers names
        """
        resellers_list = []
        if os.path.isfile(self.DA_RESELLERS_LIST):
            file_content = self.get_file_lines(self.DA_RESELLERS_LIST)
            resellers_list = [line.rstrip() for line in file_content if line.rstrip()]
        resellers_list.extend(self.list_admins())
        return list(set(resellers_list))

def _get_reseller_packages(self, reseller):
        """
        Get all packages for giver reseller

:param string reseller: Reseller name
        :rtype: list of strings
        :return: List of packages names that belongs to given reseller
        """
        packages_list = []
        package_path = os.path.join(self.DA_USERS_DIR, reseller, "packages.list")
        if os.path.isfile(package_path):
            file_content = self.get_file_lines(package_path)
            packages_list = [line.rstrip() for line in file_content if line.rstrip()]
        return packages_list

def list_resellers_packages(self, debug=False):  # --list-resellers-packages
        first_exception = None
        for reseller in self._get_resellers():
            try:
                packages = self._get_reseller_packages(reseller)
            except Exception as e:
                if first_exception is None:
                    first_exception = e
                continue
            # it's work faster when cycle in cycle or previous variant
            # out_string = 'res1 pkg1\n res1 pkg2\n res1 pkg3'
            out_string = functools.reduce(
                lambda result, pkg, res=reseller: f'{result}{res} {pkg}\n',  # processing function
                packages,  # list for processing
                ''  # accumulator for result
            ).strip()  # remove last \n
            print(out_string)
        if first_exception is not None:
            raise first_exception

def list_reseller_packages(self, reseller, debug=False):  # --list-reseller-packages
        print('\n'.join(self._get_reseller_packages(reseller)))

class ISPManagerGetPackagesException(Exception):
    def __init__(self, message):
        Exception.__init__(self, message)

class ISPManager(GeneralPanel):
    ISP4_CONF = '/usr/local/ispmgr/etc/ispmgr.conf'
    users_data = None
    ISP5_GRP = 'mgrsecure'

def __init__(self):
        super().__init__()
        detect.getCP()
        ver_parts = detect.CP_VERSION.split('.')
        self.isp_major_ver = int(ver_parts[0])
        # ISP5 type - master/slave
        self.isp5_is_master = detect.ispmanager5_is_master()
        self.cpapi_plugin = ispmanager.PanelPlugin()

def _get_user_packages_dict(self):
        """
        Retrives list of pairs "uid - package"
        :return: Dictionary: uid -> package_name
            Example: {512: 'custom', 513: 'test1'}
         Case 2: (Error): (1, "error message")
        """
        dict_uid_package = {}
        if self.isp_major_ver == 4:
            # ISPManager 4
            if not os.path.isfile(self.ISP4_CONF):
                raise ISPManagerGetPackagesException(f"ERROR: Can't read {self.ISP4_CONF}")
            try:
                with open(self.ISP4_CONF, 'r', encoding='utf-8') as f:
                    lines = f.read()
                lines = lines.split('Account')
                lines.pop(0)
                for line in lines:
                    uid = line.split()[0]
                    uid = int(self.users_data.get_uid(uid.replace("\"", "")))
                    line = line.split('\n')
                    for line1 in line:
                        if line1.strip().startswith("Preset"):
                            pkg_name = ' '.join(line1.strip().split()[1:])
                            if uid in dict_uid_package:
                                dict_uid_package[uid] = dict_uid_package[uid] + ' ' + str(pkg_name)
                            else:
                                dict_uid_package[uid] = str(pkg_name)
            except (OSError, IOError) as e:
                # raise ISPManagerGetPackagesException("ERROR: Can't read %s - %s" % (self.ISP4_CONF, str(e)))
                raise ISPManagerGetPackagesException(
                    f"ERROR: Can't read {self.ISP4_CONF} - {str(e)}"
                ) from e
        else:
            # ISP Manager 5
            # We emmulate don`t retrive packages from isp 5. All users have default package
            # see more https://cloudlinux.atlassian.net/browse/LU-256
            # Isp5 use lvectl and THIS script when DB is locked. So we can`t get packages from isp 5 db.
            try:
                panel_users = grp.getgrnam(self.ISP5_GRP).gr_mem
                for user in panel_users:
                    try:
                        uid = self.users_data.get_uid(user)
                        dict_uid_package[uid] = 'default'
                    except ClPwd.NoSuchUserException:  # skip user without UID
                        pass
            except KeyError:  # group self.ISP5_GRP not found
                pass
        return dict_uid_package

@classmethod
    def get_reseller(cls, uid, debug=False):
        """
        Get reseller name from user's config file

:param string uid: uid of user
        :param bool debug: Do produce debug output or don't
        :rtype: string
        :return: reseller name
        """
        reseller = 'admin'
        print(reseller)

def list_packages(self, debug=False):  # --list-packages
        try:
            dict_uid_package = self._get_user_packages_dict()
            for package in set(dict_uid_package.values()):
                print(package)
        except ISPManagerGetPackagesException as e:
            print(e)
            sys.exit(1)

def list_all(self, debug=False):  # list all
        try:
            dict_uid_package = self._get_user_packages_dict()
            for uid, package in dict_uid_package.items():
                print(str(uid) + ' ' + package)
        except ISPManagerGetPackagesException as e:
            print(e)
            sys.exit(1)

def list_user_package(self, uid, debug=False):  # --userid
        try:
            uid = int(uid)
            dict_uid_package = self._get_user_packages_dict()
            if uid in dict_uid_package:
                print(dict_uid_package[uid])
        except ISPManagerGetPackagesException as e:
            print(e)
            sys.exit(1)
        except ValueError:
            print("Incorrect user id")

def list_users_in_package(self, package, debug=False):  # --package
        try:
            dict_uid_package = self._get_user_packages_dict()
            for uid, package_ in dict_uid_package.items():
                if package == package_:
                    print(str(uid))
        except ISPManagerGetPackagesException as e:
            print(e)
            sys.exit(1)

def list_users(self, debug=False):  # --list-users
        try:
            dict_uid_package = self._get_user_packages_dict()
            for uid, package in dict_uid_package.items():
                print(f"{uid},{package},admin")
        except ISPManagerGetPackagesException as e:
            print(e)
            sys.exit(1)

class Plesk(GeneralPanel):
    """
    See following link for information about database:
    https://github.com/plesk/db-schemas/blob/master/psadb.xml
    """
    PSA_SHADOW = '/etc/psa/.psa.shadow'
    ADMIN_ID = '1'
    NO_PACKAGE = 'None'
    users_data = None

def __init__(self):
        super().__init__()
        self.cpapi_plugin = plesk.PanelPlugin()

def fetch_data_from_db(self, sql, data=None):
        if not os.path.isfile(self.PSA_SHADOW):
            return False
        with open(self.PSA_SHADOW, 'r', encoding='utf-8') as f:
            passwd = f.read().strip()
        connector = mysql_lib.MySQLConnector(host='localhost', user='admin',
                                             passwd=str(passwd), db='psa',
                                             use_unicode=True, charset='utf8')
        with connector.connect() as db:
            return db.execute_query(sql, args=data)

def _get_user_packages(self, debug=False, reseller_name=None, username=None):
        """
        Generate triple with uid, it's package, it's reseller

:param bool debug: Do produce debug output or don't
        :param string reseller_name: filter by this reseller
        :param string username: filter by this username (ignored if reseller_name set)
        :rtype: generator of (int, string, string)
        :return: (uid, package name, reseller name)
        """
        # users in plesk can have no package!
        # please, be careful when you change this SQL query
        # always run tests from QA repository!
        query = """
            SELECT
                sys_user.login,
                template.name,
                reseller.login
            FROM
                sys_users AS sys_user
            JOIN
                hosting AS hosting ON hosting.sys_user_id = sys_user.id
            JOIN
                domains AS domain ON hosting.dom_id = domain.id AND domain.webspace_id = 0
            JOIN
                clients AS reseller ON domain.vendor_id = reseller.id
            LEFT JOIN
                Subscriptions AS subscription ON subscription.object_id = domain.id
                AND subscription.object_type = 'domain'
            LEFT JOIN
                PlansSubscriptions AS plan ON plan.subscription_id = subscription.id
            LEFT JOIN
                Templates AS template ON plan.plan_id = template.id
            WHERE
                sys_user.mapped_to IS NULL
                AND (template.type = 'domain' OR template.type IS NULL)
        """
        if reseller_name is not None:
            query = f"{query} AND reseller.login = %s"
            result = self.fetch_data_from_db(query, [reseller_name])
        elif username is not None:
            query = f"{query} AND sys_user.login = %s"
            result = self.fetch_data_from_db(query, [username])
        else:
            result = self.fetch_data_from_db(query)

for username_, package, reseller in result:
            package = package or self.NO_PACKAGE
            reseller = '' if reseller == 'admin' else reseller
            uid = self._check_and_get_user(username=username_, debug=debug)
            if username_ and uid is not None:
                yield uid, package, reseller

def list_packages(self, debug=False):  # --list-packages
        """
        Print packages (exclude reseller's)

:param bool debug: Do produce debug output or don't
        :return: packages names

"""

query = f"""
            SELECT name
            FROM psa.Templates WHERE owner_id = {self.ADMIN_ID} AND type = 'domain';
        """

for line in self.fetch_data_from_db(query):
            print(line[0])

def get_reseller(self, uid: str, debug: bool = False):  # --get-user-reseller
        """
        Get reseller name from DB
        :param string uid: uid of user
        :param bool debug: Do produce debug output or don't
        :rtype: string
        :return: reseller name
        """
        reseller = ''
        usernames = self.users_data.get_names(int(uid))
        if usernames:
            format_strings = ','.join(['%s'] * len(usernames))
            query = f"""SELECT reseller.login
                FROM sys_users AS sys_user
                JOIN hosting AS hosting ON hosting.sys_user_id = sys_user.id
                JOIN domains AS domain ON hosting.dom_id = domain.id
                JOIN clients AS reseller ON domain.vendor_id = reseller.id
                WHERE sys_user.login IN ({format_strings})"""
            result = self.fetch_data_from_db(query, tuple(usernames))
            if result:
                reseller = result[0][0]
        print('' if reseller == 'admin' else reseller)

def list_all(self, debug=False):  # --list-all
        """
        Print info about user's packages (include resellers)

:param bool debug: Do produce debug output or don't
        :return: pairs "uid package", where
        uid - unique id of user
        package - name of user's package

"""
        users = self._get_user_packages(debug=debug)
        for (uid, package, _) in users:
            print(f'{uid} {package}')

def list_users(self, debug=False):  # --list-users
        """
        Print info about user's packages (include resellers) and thouse resellers

:param bool debug: Do produce debug output or don't
        :return: triples "uid,package,reseller", where
        uid - unique id of user
        package - name of user's package
        reseller - name of package's reseller
        """
        users = self._get_user_packages(debug=debug)
        for (uid, package, reseller) in users:
            print(f'{uid},{package},{reseller}')

def list_user_package(self, uid, debug=False):  # --userid
        """
        Print all packages for user's uid

:param uid: user's unique id
        :param bool debug: Do produce debug output or don't
        :return: package name for user's uid

"""

packages = []
        try:
            names = self.users_data.get_names(int(uid))
        except ClPwd.NoSuchUserException as e:
            if debug:
                print(e)
        except ValueError:
            print("Incorrect user id")
        else:
            for name in names:
                result = self._get_user_packages(debug=debug, username=name)
                packages.extend([line[1] for line in result])
        for package in packages:
            print(package)

def list_users_in_package(self, package, debug=False):  # --package
        """
        Print all users in package

:param package: name of package
        :param bool debug: Do produce debug output or don't
        :return: user's uid
        """

query = f"""
        SELECT t5.login
        FROM psa.Templates AS t1
        JOIN psa.PlansSubscriptions AS t2 ON t2.plan_id = t1.id
        JOIN psa.Subscriptions AS t3 ON t3.id = t2.subscription_id
        JOIN psa.domains AS t4 ON t4.id = t3.object_id
        JOIN psa.hosting AS t6 ON t6.dom_id = t4.id
        RIGHT JOIN psa.sys_users AS t5 ON t6.sys_user_id = t5.id
        WHERE t1.name = '{package}' AND t1.owner_id = {self.ADMIN_ID} AND t1.type = 'domain';
        """

for login in self.fetch_data_from_db(query):
            print(self.users_data.get_uid(login[0]))

def list_users_in_reseller_package(self, package, reseller, debug=False):  # --package --reseller
        """
        Print info about users in resellers package

:param package: package name
        :param reseller: name of package's reseller
        :param bool debug: Do produce debug output or don't
        :return: users uids

"""
        reseller_query = f"""SELECT id FROM psa.clients WHERE type = 'reseller' AND login = '{reseller}'"""
        reseller_id = self.fetch_data_from_db(reseller_query)
        if len(reseller_id) != 1:
            return
        reseller_id = int(reseller_id[0][0])
        query = f"""
            SELECT t5.login
            FROM psa.Templates AS t1
            JOIN psa.PlansSubscriptions AS t2 ON t2.plan_id = t1.id
            JOIN psa.Subscriptions AS t3 ON t3.id = t2.subscription_id
            JOIN psa.domains AS t4 ON t4.id = t3.object_id
            JOIN psa.hosting AS t6 ON t6.dom_id = t4.id
            RIGHT JOIN psa.sys_users AS t5 ON t6.sys_user_id = t5.id
            WHERE t1.type = 'domain' AND t1.name = '{package}' AND t1.owner_id = '{reseller_id}';
        """
        result = self.fetch_data_from_db(query)
        for line in result:
            try:
                luid = self.users_data.get_uid(line[0])
            except ClPwd.NoSuchUserException as e:
                if debug:
                    print(e)
            else:
                print(str(luid))

def list_resellers_packages(self, debug=False):  # --list-resellers-packages
        """
        Print list of resellers packages

:param bool debug: Do produce debug output or don't
        :return: pairs "reseller package", where
        package - package name
        reseller - package's reseller

"""
        query = f"""SELECT t2.login, t1.name FROM
        (SELECT name, owner_id, type FROM psa.Templates) AS t1
        JOIN psa.clients AS t2 ON t1.owner_id = t2.id
        WHERE t1.owner_id != {self.ADMIN_ID} AND t1.type = 'domain';"""
        for line in self.fetch_data_from_db(query):
            package = line[1]
            reseller = 'root' if line[0] == 'admin' else line[0]
            print(f'{reseller} {package}')

def list_domain_packages_with_id(self) -> List[Tuple[Optional[str], str, int]]:
        """
        Return list of non-reseller packages with the plesk DB id
            [(reseller, package, plesk_id), (None, package, plesk_id)]
        """
        res = []
        query = """SELECT t2.login, t1.name, t1.id FROM
        (SELECT id, name, owner_id, type FROM psa.Templates) AS t1
        JOIN psa.clients AS t2 ON t1.owner_id = t2.id
        WHERE t1.type = 'domain';"""
        for line in self.fetch_data_from_db(query):
            package = line[1]
            reseller = 'root' if line[0] == 'admin' else line[0]
            _id = line[2]
            res.append((reseller, package, int(_id)))
        return res

def get_package_name_by_id_from_plesk_db(self, package_id: int) -> str:
        """
        Return package name by the plesk DB id
        """
        query = "SELECT id, name, type FROM psa.Templates WHERE id = %s;"
        data = self.fetch_data_from_db(query, (package_id,))
        return data[0][1]

def get_package_id_by_name_from_plesk_db(self, package_name: str) -> int:
        """
        Return package id by the plesk DB name
        """
        query = "SELECT id, name, type FROM psa.Templates WHERE name = %s;"
        data = self.fetch_data_from_db(query, (package_name,))
        return data[0][0]

def list_reseller_packages(self, reseller, debug=False):  # --list-reseller-packages
        """
        Print list reseller's packages

:param reseller: name of reseller
        :param bool debug: Do produce debug output or don't
        :return: packages names

"""
        if reseller == "root":
            reseller = "admin"
        if not re.match(r"^[\w_]*$", reseller):
            return None
        query = f"""SELECT t1.name FROM
        (SELECT name, owner_id, type FROM psa.Templates) as t1
        JOIN psa.clients as t2 ON t1.owner_id = t2.id AND t2.login = '{reseller}'
        WHERE t1.type = 'domain' AND t1.owner_id != {self.ADMIN_ID};"""
        for line in self.fetch_data_from_db(query):
            print(line[0])

def list_admins(self, debug=False):
        return ["admin"]

class InterWorx(GeneralPanel):
    def _warning(self, debug, method_name):
        if debug:
            sys.stderr.write("Doesn't support InterWorx anymore\n")
        sys.exit(0)

class Unknown(GeneralPanel):
    def _warning(self, debug, method_name):
        raise NotImplementedError(f"{method_name} wasn't implemented for Unknown panel")

class HostingNG(GeneralPanel):
    def _warning(self, debug, method_name):
        raise NotImplementedError(f"{method_name} wasn't implemented for HostingNG panel")

📁 Sid Gifari File Manager