Admin

name	type	size	perms	modified	actions
[parent directory]
__pycache__/	dir	-	0755	2026-08-19 08:39:33
__init__.py	file	1,520 B	0644	2026-07-20 00:00:00	edit
helpers.py	file	4,953 B	0644	2026-07-20 00:00:00	edit
modify_admin_hook.py	file	598 B	0644	2026-07-20 00:00:00	edit
modify_domain_hook.py	file	1,873 B	0644	2026-07-20 00:00:00	edit
modify_package_hook.py	file	508 B	0644	2026-07-20 00:00:00	edit
modify_user_hook.py	file	2,269 B	0644	2026-07-20 00:00:00	edit

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 argparse
import inspect
import logging
import os
import re
import sys

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

logger = logging.getLogger(__name__)

def _sanitize_log_value(value):
    """Strip control characters from a value before logging."""
    return str(value).replace('\n', '\\n').replace('\r', '\\r')

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

# Linux + cPanel/Plesk-style usernames: must start with [a-z_], 1-32 chars,
# allow lowercase letters, digits, underscore, hyphen, dot. Dot is included
# because cPanel and some hosting environments accept dotted account names
# (bugbot finding ae6f9312 on !30).
_VALID_NAME_RE = re.compile(r'^[a-z_][a-z0-9._-]{0,31}$')

def valid_name(value):
    """argparse type-callback that validates a hosting account name.

Used by post_modify_user.py / pre_modify_user.py to reject names with
    shell metacharacters, path separators, or trailing newlines before
    dispatching to listener plugins.
    """
    if not isinstance(value, str) or not _VALID_NAME_RE.fullmatch(value):
        raise argparse.ArgumentTypeError(
            f'invalid name {value!r}: must match {_VALID_NAME_RE.pattern}')
    return value

@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__,
                    _sanitize_log_value(args), _sanitize_log_value(kwargs))

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

cancel