File Manager

X7ROOT File Manager

Current Path: /opt/imunify360/venv/lib/python3.11/site-packages/clcommon/public_hooks/lib

opt / imunify360 / venv / lib / python3.11 / site-packages / clcommon / public_hooks / lib /

📁 ..
📄 __init__.py(1.48 KB)
📁 __pycache__
📄 helpers.py(3.8 KB)
📄 modify_admin_hook.py(598 B)
📄 modify_domain_hook.py(1.52 KB)
📄 modify_package_hook.py(508 B)
📄 modify_user_hook.py(1.85 KB)

Editing: helpers.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 inspect
import logging
import os
import sys

import raven
import time
from io import StringIO
from contextlib import contextmanager
from functools import wraps

logger = logging.getLogger(__name__)

LISTENERS_DIRECTORY = '/usr/share/cloudlinux/hooks/listeners/'

@contextmanager
def capture_output(stdo, stde):
    stdout = sys.stdout
    stderr = sys.stderr
    try:
        sys.stdout = stdo or StringIO()
        sys.stderr = stde or StringIO()
        yield
    finally:
        sys.stdout = stdout
        sys.stderr = stderr

def hook_method(func):
    """
    Magic decorator that calls all subclass methods
    that override base decorated one.
    Requirements:
    - subclass must be defined in .py file in LISTENERS_DIRECTORY
    - subclass must NOT start with '_' char
    - subclass must override base event method (the one with '@hook_method')
    """
    @wraps(func)
    def _wrapped(self, *args, **kwargs):
        # this only return direct subclasses, so we can't make `proxies` now
        for subclass in self.__class__.__subclasses__():
            listener_path = os.path.dirname(inspect.getmodule(subclass).__file__)
            # skip child if it is not in expected directory
            if os.path.normpath(LISTENERS_DIRECTORY) != os.path.normpath(listener_path):
                logger.warning('%s is not in %s directory; it is in %s,'
                               ' skip', subclass, LISTENERS_DIRECTORY, listener_path)
                continue
            # skip internal classes
            if subclass.__name__.startswith('_'):
                continue

# magic: get method only if it is defined in child (NOT in parent)
            listener = getattr(subclass(), func.__name__)
            if getattr(listener, 'is_magic_method', False):
                logger.debug('skip %s is not implemented in %s',
                             func.__name__, subclass.__name__)
                continue

logger.info('executing %s:%s', func.__name__, subclass.__name__)
            now = time.time()
            stdout, stderr = StringIO(), StringIO()
            try:
                with capture_output(stdout, stderr):
                    listener(*args, **kwargs)
            except Exception:
                # use Raven carefully and only in places where
                # you sure that sentry is already initialized
                raven.base.Raven.captureException(
                    fingerprint=['{{ default }}', subclass.__name__, func.__name__],
                    extra={'stdout': stdout.getvalue(), 'stderr': stderr.getvalue()}
                )
                logger.warning('listener %s:%s crashed', subclass.__name__, func.__name__, exc_info=1)
            finally:
                elapsed = time.time() - now
                stdout_str = stdout.getvalue()
                if stdout_str:
                    logger.info('captured stdout of %s:%s\n~BEGIN OUTPUT~\n%s\n~END OUTPUT~\n',
                                func.__name__, subclass.__name__, stdout_str)
                stderr_str = stderr.getvalue()
                if stderr_str:
                    logger.debug('captured stderr of %s:%s\n~BEGIN OUTPUT~\n%s\n~END OUTPUT~\n',
                                 func.__name__, subclass.__name__, stderr_str)
                logger.debug('running %s: %.4f elapsed', func.__name__, elapsed)
        logger.info('%s executed by the user with uid %s and gid %s',
                    func.__name__, os.geteuid(), os.getegid())
        logger.info('ended %s(%s, %s)', func.__name__, args, kwargs)

# special marker to determine overrided methods
    _wrapped.is_magic_method = True
    return _wrapped

X7ROOT File Manager

Editing: helpers.py

Upload File

Create Folder