Admin Control Panel v2.0

✎ Editing: manager.py ← Back

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

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

import logging
import os
import pwd
import shutil
import subprocess
from collections import namedtuple

from packaging.version import Version

from clcommon.utils import get_file_lines, write_file_lines, mod_makedirs
from clcommon.public_hooks import CLOUDLINUX_HOOKS, CONTACT_SUPPORT_MESSAGE_FOOTER
from clcommon.cpapi import get_cp_description

BIN_DIR = os.path.join(CLOUDLINUX_HOOKS, 'directadmin/')

Hook = namedtuple('Hook', ['path', 'hook'])
HookPath = namedtuple('HookPath', ['min_version', 'max_version', 'path'])

HOOKS = { # None means +inf or -int depending on positions
    Hook(BIN_DIR + 'user_create_post', (HookPath(None, '1.60', 'user_create_post.sh'),
                                        HookPath('1.60', None, 'user_create_post/CL_user_create_post.sh'))),
    Hook(BIN_DIR + 'user_destroy_post', (HookPath(None, '1.60', 'user_destroy_post.sh'),
                                         HookPath('1.60', None, 'user_destroy_post/CL_user_destroy_post.sh'))),
    Hook(BIN_DIR + 'user_destroy_pre', (HookPath(None, '1.60', 'user_destroy_pre.sh'),
                                        HookPath('1.60', None, 'user_destroy_pre/CL_user_destroy_pre.sh'))),
    Hook(BIN_DIR + 'user_restore_post', (HookPath(None, '1.60', 'user_restore_post.sh'),
                                         HookPath('1.60', None, 'user_restore_post/CL_user_restore_post.sh'))),
    Hook(BIN_DIR + 'domain_change_post', (HookPath(None, '1.60', 'domain_change_post.sh'),
                                          HookPath('1.60', None, 'domain_change_post/CL_domain_change_post.sh'))),
}
DA_HOOK_DEST_DIR = '/usr/local/directadmin/scripts/custom'
if get_cp_description() is None:
    DA_VERSION = '1.0'  # in case we can't determine version we set default old
else:
    DA_VERSION = get_cp_description()["version"]

logger = logging.getLogger(__name__)

def _folder_hooks_compatibility():
    """
    Check that DA has compatibility with folder-based hooks
    :return: Bool
    """
    return Version(DA_VERSION) >= Version('1.60')

def _get_hook_from_structure(hook):
    """
    hook = HookPath stucture
    return suitable hook name depending on version
    if return None - suitable hook not found
    """
    for version_hook in hook:
        if version_hook.min_version is None:
            min_version = True
        else: # if DA bigger than the min val
            min_version = Version(DA_VERSION) >= Version(version_hook.min_version)
        if version_hook.max_version is None:
            max_version = True
        else: # if DA lesser that the max val
            max_version = Version(DA_VERSION) < Version(version_hook.max_version)
        if min_version and max_version:
            return version_hook.path
    return None

def create_da_hook(da_hook_filename, command, da_hook_default_dir=DA_HOOK_DEST_DIR):
    """
    Creates DA hook
    Example args:
    da_hook_filename = user_create_post.sh
    da_hook_src = /usr/share/cagefs-plugins/hooks/directadmin/user_create_post.sh - command

:param string da_hook_filename: How to name that hook in DA panel
    :param command: what we should run on hook
    :param string da_hook_default_dir:
    :return: None
    """

hook_fullname = os.path.join(da_hook_default_dir, da_hook_filename)

logger.debug('Registering %s action hook', hook_fullname)
    try:
        da_user = pwd.getpwnam('diradmin')
    except (KeyError,) as e:
        logger.error("failed to find 'diradmin' user: %s", str(e))
        return

da_user_uid = da_user.pw_uid
    da_user_gid = da_user.pw_gid

try:
        # if hook file already in system
        if os.path.isfile(hook_fullname):
            # get hook content
            content = get_file_lines(hook_fullname)
            content = [line for line in content if line != '\n']
            # Flags for check if hook installed and if hook on bash
            hook_installed = False
            hook_on_bash = False

for line in content:
                # check if hook installed
                if line.find(command) != -1:
                    hook_installed = True
                    break
                # check if hook on bash
                if line.startswith('#!/') and (line.find('/sh') != -1 or line.find('/bash') != -1):
                    hook_on_bash = True

# if hook not installed
            if not hook_installed:
                # if hook on bash
                if hook_on_bash:
                    # add command for cagefs hook
                    content.append('\n' + command + '\n')
                    write_file_lines(hook_fullname, content, 'w')
                else:  # if hook not on bash
                    try:
                        # backup old hook filename
                        dirname, fname = os.path.split(hook_fullname)
                        old_hook_backup = os.path.join(dirname, 'old_'+fname)

# copy backup old hook
                        shutil.copyfile(hook_fullname, old_hook_backup)
                        os.chmod(old_hook_backup, 0o700)
                        os.chown(old_hook_backup, da_user_uid, da_user_gid)

# replace old hook by new hook on bash
                        write_file_lines(hook_fullname,
                                         '#!/bin/bash\n' + command + '\n' + old_hook_backup + '\n', 'w')
                        os.chmod(hook_fullname, 0o700)
                        os.chown(hook_fullname, da_user_uid, da_user_gid)
                    except (OSError, shutil.Error) as e:
                        logger.error('Failed to create hook for DirectAdmin: %s: %s. %s',
                                     hook_fullname, str(e), CONTACT_SUPPORT_MESSAGE_FOOTER)
    except (OSError, IOError) as e:
        logger.error('Failed to install hook for DirectAdmin: %s. %s',
                     str(e), CONTACT_SUPPORT_MESSAGE_FOOTER)

# Install hook if it's not installed
    if not os.path.isfile(hook_fullname):
        try:
            if not os.path.isdir(os.path.dirname(hook_fullname)):
                mod_makedirs(os.path.dirname(hook_fullname), 0o700)
                os.chmod(os.path.dirname(hook_fullname), 0o700)
                os.chown(os.path.dirname(hook_fullname), da_user_uid, da_user_gid)
            write_file_lines(hook_fullname, '#!/bin/bash\n' + command + '\n', 'w')
            os.chmod(hook_fullname, 0o700)
            os.chown(hook_fullname, da_user_uid, da_user_gid)
        except (OSError, IOError) as e:
            logger.error('Failed to install hook for DirectAdmin: %s. %s',
                         str(e), CONTACT_SUPPORT_MESSAGE_FOOTER)

# DO NOT CHANGE THIS METHOD
# STILL USED IN LVEMANAGER & CAGEFS
def remove_da_hook(da_hook_filename, command, da_hook_default_dir=DA_HOOK_DEST_DIR):
    """
    Removes DA hook
    da_hook_default_dir = DA_HOOK_DEST_DIR = /usr/local/directadmin/scripts/custom
    da_hook_filename = user_create_post.sh
    command = /usr/share/cagefs-plugins/hooks/directadmin/user_create_post.sh

:param string da_hook_filename: How to name that hook in DA panel
    :param command: what we should run on hook
    :param da_hook_default_dir: default dir for hooks
    :return: None
    """
    logger.debug('Unregistering %s action hook', da_hook_filename)

hook_fullname = os.path.join(da_hook_default_dir, da_hook_filename)

# check if hook exist in system
    if not os.path.isfile(hook_fullname):
        logger.info('Hook %s is not installed; skip', hook_fullname)
        return

try:
        content = get_file_lines(hook_fullname)
        new_content = []
        for line in content:
            # check for hook execution command in hook
            if line != '\n' and line.find(command) == -1:
                new_content.append(line)
        # write changes to hook
        write_file_lines(hook_fullname, new_content, 'w')
    except IOError as e:
        logger.error('Failed to remove hook for DirectAdmin: %s. %s',
                     str(e), CONTACT_SUPPORT_MESSAGE_FOOTER)

def install_hooks():
    if not _folder_hooks_compatibility():
        subprocess.run(
            f"touch {DA_HOOK_DEST_DIR}/.old_hooks_present",
            shell=True,
            executable="/bin/bash",
            check=False,
        )
    else:
        if os.path.isfile(f"{DA_HOOK_DEST_DIR}/.old_hooks_present"):
            remove_hooks()

for hook_structure in HOOKS:
        hook_name = _get_hook_from_structure(hook_structure.hook)
        if hook_name is None:
            logger.error('Failed to install hook for DirectAdmin: %s. %s',
                         "Can't find suitable version", CONTACT_SUPPORT_MESSAGE_FOOTER)
            continue
        create_da_hook(hook_name, hook_structure.path)

def remove_hooks():
    if _folder_hooks_compatibility and os.path.isfile(f"{DA_HOOK_DEST_DIR}/.old_hooks_present"):
        subprocess.run(
            f"rm -f {DA_HOOK_DEST_DIR}/.old_hooks_present",
            shell=True,
            executable="/bin/bash",
            check=False,
        )
    for hook_structure in HOOKS:
        hook_names = [version_hook.path for version_hook in hook_structure.hook]
        for name in hook_names:
            remove_da_hook(name, hook_structure.path)

Cancel