File Manager

✏️ ooks/bundle/cpanel/manager.py

# -*- 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 subprocess
import os
import json
import logging

from typing import NamedTuple
from clcommon.public_hooks import CLOUDLINUX_HOOKS, CONTACT_SUPPORT_MESSAGE_FOOTER

BIN_DIR = os.path.join(CLOUDLINUX_HOOKS, 'cpanel/')
MANAGE_HOOK = '/usr/local/cpanel/bin/manage_hooks'

# this class is directly used in other projects, e.g lvemanager:
# usr/share/l.v.e-manager/cpanel/extension/install_lve_extension.py
class Hook(NamedTuple):
    path: str
    category: str
    event: str
    stage: str
    escalateprivs: bool = False

HOOKS = (
    Hook(BIN_DIR + 'postwwwacct', 'Whostmgr', 'Accounts::Create', 'post'),
    Hook(BIN_DIR + 'postkillacct', 'Whostmgr', 'Accounts::Remove', 'post'),
    Hook(BIN_DIR + 'prekillacct', 'Whostmgr', 'Accounts::Remove', 'pre'),
    Hook(BIN_DIR + 'postmodifyacct', 'Whostmgr', 'Accounts::Modify', 'post'),
    Hook(BIN_DIR + 'premodifyacct', 'Whostmgr', 'Accounts::Modify', 'pre'),
    Hook(BIN_DIR + 'postrestoreacct', 'PkgAcct', 'Restore', 'post'),
    Hook(BIN_DIR + 'postsuspendacct', 'Whostmgr', 'Accounts::suspendacct', 'post'),
    Hook(BIN_DIR + 'postunsuspendacct', 'Whostmgr', 'Accounts::unsuspendacct', 'post'),
    Hook(BIN_DIR + 'postaddondom', 'Cpanel', 'Api2::AddonDomain::addaddondomain', 'post', escalateprivs=True),
    Hook(BIN_DIR + 'postkilladdondom', 'Cpanel', 'Api2::AddonDomain::deladdondomain', 'post', escalateprivs=True),
    Hook(BIN_DIR + 'postaddondom', 'Cpanel', 'Api2::SubDomain::addsubdomain', 'post', escalateprivs=True),
    Hook(BIN_DIR + 'postkilladdondom', 'Cpanel', 'Api2::SubDomain::delsubdomain', 'post', escalateprivs=True),
    Hook(BIN_DIR + 'prechangesubdomaindocroot', 'Cpanel', 'Api2::SubDomain::changedocroot',
         'pre', escalateprivs=True),
    Hook(BIN_DIR + 'postchangesubdomaindocroot', 'Cpanel', 'Api2::SubDomain::changedocroot',
         'post', escalateprivs=True),
)

logger = logging.getLogger(__name__)

def _install_hook(hook_path, category, event, stage, escalateprivs):
    is_installed = _is_hook_installed(event, category, hook_path, stage)
    if is_installed:
        logger.info('Hook for %s:%s:%s action '
                    'is already installed; skip installing',
                    category, event, stage)
        return True

logger.debug('Registering %s:%s:%s action hook', category, event, stage)
    command = [
            MANAGE_HOOK, 'add', 'script', hook_path,
            '--category', str(category),
            '--event', str(event),
            '--stage', str(stage), '--manual'
        ]
    if escalateprivs:
        command.append('--escalateprivs')
    try:
        output = subprocess.check_output(command, stderr=subprocess.STDOUT, text=True)
    except (OSError, subprocess.CalledProcessError) as e:
        if isinstance(e, subprocess.CalledProcessError):
            message = e.output.rstrip('\n')
        else:
            message = str(e)
        logger.error('Can\'t install hook `%s` to category: `%s` event: `%s`; message: `%s`',
                     os.path.basename(hook_path), category, str(event), message)
        return False
    else:
        logger.info('Register hook ended successfully; tool output: `%s`', output.rstrip())
        return True

def _is_hook_installed(event, category, hook_script_path, stage):
    """
    Check if hook with given parameters
    is installed in control panel
    :return: boolean
    """
    all_hooks = _get_hooks_config()
    if all_hooks is None:
        return None
    try:
        for i in all_hooks[category][event]:
            if i['hook'] == hook_script_path and i['stage'] == stage:
                return True
    except KeyError:
        return False
    return False

def _get_hooks_config():
    """
    Reads main hooks config file
    """
    try:
        stream = subprocess.check_output(
            ['/usr/local/cpanel/bin/manage_hooks', 'list', '--output=JSON'],
            text=True
        )
    except (OSError, subprocess.CalledProcessError) as e:
        logger.error('Unable to load list of '
                     'already installed hooks. Reason is `%s`', str(e))
        return None

try:
        all_hooks = json.loads(stream)
    except (ValueError, TypeError):
        logger.exception('Received list of installed hooks '
                         'is malformed and not valid JSON. %s',
                         CONTACT_SUPPORT_MESSAGE_FOOTER)
        return None
    return all_hooks

def _delete_hook_script(hook_script_path, category, event, stage):
    """
    Delete hook script and unregister in cPanel
    """
    # if something bad happened and _is_hook_installed is None, we must try
    # to uninstall hooks in order not to leave garbage
    if _is_hook_installed(event, category, hook_script_path, stage) is False:
        logger.info('Hook for %s:%s:%s action '
                    'is not installed; skip removing',
                    category, event, stage)
        return True

logger.debug('Unregistering %s:%s:%s action hook', category, event, stage)
    try:
        output = subprocess.check_output([
            MANAGE_HOOK, 'delete', 'script', hook_script_path,
            '--category', str(category),
            '--event', str(event),
            '--stage', str(stage), '--manual'
        ], stderr=subprocess.STDOUT, text=True)
    except (OSError, subprocess.CalledProcessError) as e:
        if isinstance(e, subprocess.CalledProcessError):
            message = e.output.rstrip('\n')
        else:
            message = str(e)
        logger.error('Can\'t delete hook with category:`%s` event:`%s`; message: `%s`. %s',
                     category, str(event), message, CONTACT_SUPPORT_MESSAGE_FOOTER)
        return False
    else:
        logger.info('Unregister hook ended successfully; tool output: `%s`', output.rstrip())
        return True

def install_hooks(hooks=HOOKS):
    success = True
    for hook in hooks:
        success = _install_hook(
            hook.path, hook.category, hook.event, hook.stage, hook.escalateprivs) and success
    return success

def remove_hooks(hooks=HOOKS):
    success = True
    for hook in hooks:
        success = _delete_hook_script(
            hook.path, hook.category, hook.event, hook.stage) and success
    return success

Batal

⬅ bundle 5 item

Nama	Tipe	Ukuran	Diubah	Aksi
📁__pycache__	dir	—	2026-03-11 18:53	📂 Buka
📁bin	dir	—	2026-03-11 18:53	📂 Buka
🐍__init__.py	py	345 B	2026-02-10 15:10	✏️ 👁️
🐍lib.py	py	12.9 KB	2026-02-10 15:10	✏️ 👁️
🐍manager.py	py	6.3 KB	2026-02-10 15:10	✏️ 👁️

✏️ Rename