MOONy

✎ EDIT FILE

EDITING: wpos_modify_user_hook.py

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

from __future__ import absolute_import, print_function

import os
import re
import subprocess
import sys
import shutil

from clcommon.clpwd import drop_privileges
from clcommon.public_hooks.lib import ModifyUserHook, ModifyDomainHook
from clwpos.cl_wpos_exceptions import WposError
from clwpos.optimization_features import OBJECT_CACHE_FEATURE
from clwpos.feature_suites import get_admin_config_directory, ALL_SUITES, UNSUPPORTED_SUITES_FOR_RESELLER
from clwpos.user.config import UserConfig
from clwpos.utils import (
    WposUser,
    update_redis_conf,
    update_wp_config,
    user_uid,
    is_user_owned_by_reseller,
)
from clwpos.constants import SUITES_MARKERS, DISABLE_FEATURE_IN_HOOK_FLAG
from clwpos.object_cache.redis_utils import reload_redis
from cldetectlib import is_plesk

# Hosting account names use the useradd NAME_REGEX charset; reject anything with
# a list delimiter / whitespace / shell metacharacter so the value cannot expand
# the `set-suite --users` (comma-split) target set to unrelated accounts.
VALID_USERNAME_RE = re.compile(r'^[a-zA-Z0-9._-]+$')

class WposModifyUserHook(ModifyUserHook, ModifyDomainHook):
    """
    Update user's data in Redis and WP configs.
    """

def _post_reload_redis(self, username):
        """
        Reload Redis if the user has a Redis configuration file.
        We should use it because after modification of the domain the cagefsctl --force-update kill redis
        :param username: username of the domain owner
        """
        user = WposUser(username)
        with drop_privileges(username):
            if not os.path.exists(user.redis_conf):
                return
            try:
                reload_redis()
            except Exception as e:
                print(f'Error occurred during daemon reload: {e}', file=sys.stderr)

def post_modify_user(self, username: str, new_name: str = None, **kwargs):
        """
        Update the name of the user's home directory in:
        - ~/.clwpos/redis.conf,
        - wp-config.php for Wordpresses with enabled object caching.
        For Plesk also perform checks for incompatibilities after
        PHP version changes
        (they fall into the same Plesk event 'Update physical hosting')
        """
        if is_plesk():
            subprocess.run(["/usr/share/cloudlinux/wpos/plesk/admin_disable_caching.py"])

if new_name is None:
            return

new_user = WposUser(new_name)
        if not new_user.home_dir.endswith(new_name):
            raise WposError('Internal Error. Contact CloudLinux support')
        old_user_homedir = os.path.join(new_user.home_dir[: -len(new_name)], username)
        old_user = WposUser(username, homedir=old_user_homedir)

with drop_privileges(new_name):
            # Update ~/.clwpos/redis.conf
            if not os.path.exists(new_user.redis_conf):
                # we don't continue if redis.conf is missing
                # because it means that user has never enabled the module
                return

update_redis_conf(new_user, old_user)

# Update wp-config.php files
            # use user's WPOS config to get paths to Wordpresses with WPOS object caching enabled
            user_config = UserConfig(new_name)
            for abs_wp_path in user_config.wp_paths_with_enabled_module(OBJECT_CACHE_FEATURE):
                update_wp_config(abs_wp_path, new_user, old_user)

try:
                reload_redis()
            except Exception as e:
                print(f'Error occurred during daemon reload: {e}', file=sys.stderr)

def post_create_user(self, username, owner, **kwargs):
        if not username or not VALID_USERNAME_RE.match(username):
            print(f'Refusing to set suites for invalid username {username!r}', file=sys.stderr)
            return

is_owned_by_admin = not is_user_owned_by_reseller(username)

suites_to_allow = []
        suites_to_disallow = []

# Iterate over all suites and determine if they should be allowed or disallowed
        for suite_name in ALL_SUITES.keys():
            if (is_owned_by_admin or suite_name not in UNSUPPORTED_SUITES_FOR_RESELLER) and os.path.isfile(
                SUITES_MARKERS[suite_name]
            ):
                suites_to_allow.append(suite_name)
            elif not is_owned_by_admin:
                suites_to_disallow.append(suite_name)

# Run the collected suites in one command
        if suites_to_allow:
            print(f'Allowing {", ".join(suites_to_allow)} for user {username}')
            subprocess.run(
                [
                    '/usr/bin/cloudlinux-awp-admin',
                    'set-suite',
                    '--suites',
                    ','.join(suites_to_allow),
                    '--users',
                    username,
                    '--allowed',
                ]
            )
        if suites_to_disallow:
            print(f'Disallowing {", ".join(suites_to_disallow)} for user {username} owned by reseller')
            subprocess.run(
                [
                    '/usr/bin/cloudlinux-awp-admin',
                    'set-suite',
                    '--suites',
                    ','.join(suites_to_disallow),
                    '--users',
                    username,
                    '--disallowed',
                ]
            )

def pre_delete_user(self, username, **kwargs):
        """
        Triggered before deleting user.
        NOT TRIGGERED ON PLESK since there is no pre hooks
        Cleans up garbage left in /var/clwpos/uids/{user_uid}
        Not run on Solo edition
        :param username: account name
        """
        cfg_dir = get_admin_config_directory(user_uid(username=username))
        print(f'Checking {cfg_dir}')
        if os.path.isdir(cfg_dir):
            print(f'{cfg_dir} exists, must be cleaned')
            shutil.rmtree(cfg_dir)
        else:
            print(f'{cfg_dir} does not exist, nothing to be done')

if os.path.exists(DISABLE_FEATURE_IN_HOOK_FLAG):
            command = ['/usr/bin/cloudlinux-awp-user', '--user', username, 'disable-all']
            print(f'{DISABLE_FEATURE_IN_HOOK_FLAG} exists, calling command={command} to disable features for user')
            result = subprocess.run(command, capture_output=True, text=True)
            print(f'Command={command} stdout={result.stdout}, stderr={result.stderr}, return_code={result.returncode}')

def post_create_domain(self, username, domain):
        """Hook to be called after domain creation."""
        self._post_reload_redis(username)

def post_modify_domain(self, username, domain, new_domain=None, include_subdomains=None, **kwargs):
        """Hook to be called after domain modification."""
        self._post_reload_redis(username)

def post_delete_domain(self, username, domain):
        """Hook to be called after domain deletion."""
        self._post_reload_redis(username)

CANCEL

Name	Type	Size	Modified	Actions
↩ ..	DIR	—	—
📁 __pycache__/	DIR	—	2026-08-25 23:50
📁 cpanel/	DIR	—	2026-08-25 23:50
📁 lib/	DIR	—	2026-08-25 23:50
📁 plesk/	DIR	—	2026-08-25 23:50
📄 additional_hooks.py	PY	1.7 KB	2026-07-31 10:24	EDIT
📄 wpos_modify_user_hook.py	PY	7 KB	2026-07-31 10:24	EDIT
📄 wpos_user_dirs_hook.py	PY	791 B	2026-07-31 10:24	EDIT