��ѧۧݧ�ӧ�� ާ֧ߧ֧էا֧�

��ѧۧݧ�ӧ�� ާ֧ߧ֧էا֧� - ��֧էѧܧ�ڧ��ӧѧ�� - /opt/cloudlinux/venv/lib/python3.11/site-packages/clcommon/public_hooks/bundle/cpanel/lib.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 json import os import subprocess import logging import urllib.parse import raven from clcommon.public_hooks import ( POST_MODIFY_DOMAIN, PRE_MODIFY_USER, POST_MODIFY_USER ) from clcommon.utils import run_command, ExternalProgramFailed, is_user_present from clcommon import cpapi from secureio import makedirs_secure, write_file_secure, read_file_secure EXIT_NO_USER_FOUND = 1 ERR_NO_USER_FOUND = 'Unable to find username in hook cmdline' WHMAPI1 = '/usr/sbin/whmapi1' LVE_DIR = '/var/lve' TMP_DIR = os.path.join(LVE_DIR, 'tmp') logger = logging.getLogger(__name__) def print_response(hook_name, success=True): """ cPanel expects that each custom hook prints two values in the end of the execution: - status, where 1 means success - message, which explains non-1 statuses otherwise nothing really breaks, but logs are full of "script returned invalid response" msgs :param hook_name: name, path or anything else to fill message with in order to understand what exactly failed :param success: is it everything ended successfully? :return: Nothing """ if not success: print(0, f"Failed to execute hook {hook_name}; you can find logs in " "/var/log/cloudlinux/hooks/info.log " "and contact CloudLinux Support if you need " "help with the issue.") else: print(1, "Ok") # ATTENTION: we use this function in # processpaneluserspackages and lvemanager def call_sync_map(): """ Run lvectl sync-map and log possible stdout\|err in case of errors. :return: None """ with subprocess.Popen( ['lvectl', 'sync-map'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) as proc: stdout, stderr = proc.communicate() if proc.returncode != 0: logger.error('Error during "lvectl sync-map", code: %s, ' 'stderr: `%s`, stdout: `%s`. Reseller limits ' 'kernel mapping might be not synchronized.' 'Contact CloudLinux Support for help.' '', proc.returncode, stdout, stderr) def cpanel_postwwwacct_main(data): """ Post create account hook of cPanel :return: None """ user = data.get('user', None) owner = data.get('owner', None) if not user: logger.warning(ERR_NO_USER_FOUND) return EXIT_NO_USER_FOUND return subprocess.call([ POST_MODIFY_USER, 'create', '--username', user, '--owner', owner], env={'CPANEL_RESTORE': str(data.get('is_restore', 0))}) def cpanel_prekillacct_main(data): """ Pre kill account hook of cPanel :return: None """ # It's necessary destroy lve before remove user home directory, # otherwise will be error due busy mount points of cagefs user = data.get('user', None) if not user: logger.warning(ERR_NO_USER_FOUND) return EXIT_NO_USER_FOUND if not is_user_present(user): logger.warning('User %s does not present in the system, skip hook', user) return 0 return subprocess.call([ PRE_MODIFY_USER, 'delete', '--username', user]) def cpanel_postkillacct_main(data): """ Post kill account hook of cPanel :return: None """ user = data.get('user', None) if not user: logger.warning(ERR_NO_USER_FOUND) return EXIT_NO_USER_FOUND return subprocess.call([ POST_MODIFY_USER, 'delete', '--username', user]) def cpanel_postunsuspendacct_main(data): """ Post unsuspend account hook of cPanel data: {'result': 1, 'args': {'user': 'susp2'}, 'user': 'root'} """ user = data.get('args', {}).get('user', None) if not user: logger.warning(ERR_NO_USER_FOUND) return EXIT_NO_USER_FOUND if not is_user_present(user): logger.warning('User %s does not present in the system, skip hook', user) return 0 return subprocess.call([ POST_MODIFY_USER, 'unsuspend', '--username', user]) def cpanel_postsuspendacct_main(data): """ Post unsuspend account hook of cPanel data: {'result': 1, 'args': {'user': 'susp2'}, 'user': 'root'} """ user = data.get('args', {}).get('user', None) if not user: logger.warning(ERR_NO_USER_FOUND) return EXIT_NO_USER_FOUND if not is_user_present(user): logger.warning('User %s does not present in the system, skip hook', user) return 0 return subprocess.call([ POST_MODIFY_USER, 'suspend', '--username', user]) def _read_old_domain(user): """ Read old domain for modified account :param user: name of user :return: old domain :rtype: str """ domain = None filename = os.path.join(TMP_DIR, user) try: content = read_file_secure(filename, uid=0, gid=0, exit_on_error=False, # This log is intended to be used only by # cagefs update command write_log=False) domain = content[0] except (IndexError, OSError, IOError): # use Raven carefully and only in places where # you sure that sentry is already initialized raven.base.Raven.captureException( message='failed to read old domain for user (pre hook no called?)') return domain def cpanel_postmodifyacct_main(data): """ Post modify account hook of cPanel :return: None """ user = data.get('user', None) new_user = data.get('newuser', None) domain = data.get('domain', None) exit_code = 0 # changing owner of user # FIXME: this check does not work because cpanel sends `owner` always new_owner = data.get('owner', data.get('OWNER')) args = [POST_MODIFY_USER, 'modify', '-u', user] if new_owner: args += ['--new-owner', new_owner] if all((user, new_user,)) and user != new_user: args += ['--new-username', new_user] exit_code += subprocess.call(args) old_domain = _read_old_domain(user) if domain is not None and domain != old_domain: # looks like domain is renamed exit_code += subprocess.call([ POST_MODIFY_DOMAIN, 'modify', '--username', user if new_user is None else new_user, '--domain', old_domain, '--new-domain', domain, '--include-subdomains']) return exit_code def cpanel_postrestoreacct_main(data): """ Post restore account hook of cPanel :return: None """ user = data.get('user', None) if not user: logger.warning(ERR_NO_USER_FOUND) return EXIT_NO_USER_FOUND return subprocess.call([ POST_MODIFY_USER, 'restore', '--username', user]) def _get_old_domain(user): """ Get old domain for modified account :param user: name of user :return: old domain :rtype: str """ domain = None try: cmd = [ WHMAPI1, 'listaccts', f'search={user}', 'searchtype=user', 'searchmethod=exact', 'want=domain', '--output=json' ] std_out = run_command(cmd, return_full_output=True)[1] # take only std_out, ignore std_err data = json.loads(std_out) domain = data['data']['acct'][0]['domain'] except (ExternalProgramFailed, IndexError, KeyError): # use Raven carefully and only in places where # you sure that sentry is already initialized raven.base.Raven.captureException(message='failed to get old domain for user from cpanel') return domain def cpanel_premodifyacct_main(data): """ Pre modify account hook of cPanel :return: None """ user = data.get('user') # getting old domain # TODO: why not cpapi? domain = _get_old_domain(user) if domain is None: return 0 # save old domain to file filename = os.path.join(TMP_DIR, user) makedirs_secure(TMP_DIR, perm=0o750, uid=0, gid=0, parent_path=LVE_DIR) write_file_secure([domain], filename, uid=0, gid=0, perm=0o700) return 0 def cpanel_postcreateaddondom_main(data): args = data.get('args', {}) user = data.get('user', None) newdomain = args.get('newdomain', None) or args.get('domain', None) if not all([user, newdomain]): logger.warning("Missing required arguments: user=%s, newdomain=%s", str(user), str(newdomain)) return 0 return subprocess.call( [ POST_MODIFY_DOMAIN, 'create', '--username', user, '--domain', newdomain] ) def cpanel_postkilladdondom_main(data): args = data.get('args', {}) user = data.get('user', None) domain = args.get('domain', None) if not all([user, domain]): logger.warning("Missing required arguments: user=%s, domain=%s", str(user), str(domain)) return 0 return subprocess.call( [ POST_MODIFY_DOMAIN, 'delete', '--username', user, '--domain', domain] ) def cpanel_prechangesubdomaindocroot_main(data): """ Pre change subdomain document root hook of cPanel Called when subdomain document root is being changed via Api2::SubDomain::changedocroot Saves old docroot to file for post hook to use :param data: hook data from cPanel :return: exit code """ args = data.get('args', {}) user = data.get('user', None) subdomain = args.get('subdomain', None) rootdomain = args.get('rootdomain', None) new_dir = args.get('dir', None) if not all([user, subdomain, rootdomain, new_dir]): logger.error("Missing required arguments: user=%s, subdomain=%s, rootdomain=%s, new_dir=%s", str(user), str(subdomain), str(rootdomain), str(new_dir)) return 0 # Construct full domain name (subdomain.rootdomain) domain = f"{subdomain}.{rootdomain}" # Get old docroot using cpapi try: old_docroot, _ = cpapi.docroot(domain) except Exception: raven.base.Raven.captureException( message=f'failed to get old docroot for subdomain {domain}') logger.warning("Could not get old docroot for domain %s", domain) return 0 # Save old docroot and new dir to file for post hook docroot_tmp_dir = os.path.join(TMP_DIR, 'prechangesubdomaindoc') filename = os.path.join(docroot_tmp_dir, domain) makedirs_secure(TMP_DIR, perm=0o750, uid=0, gid=0, parent_path=LVE_DIR) makedirs_secure(docroot_tmp_dir, perm=0o750, uid=0, gid=0, parent_path=TMP_DIR) write_file_secure([old_docroot], filename, uid=0, gid=0, perm=0o700) return 0 def _read_old_docroot(domain): """ Read saved docroot data for a subdomain """ docroot_tmp_dir = os.path.join(TMP_DIR, 'prechangesubdomaindoc') filename = os.path.join(docroot_tmp_dir, domain) try: content = read_file_secure(filename, uid=0, gid=0, exit_on_error=False, write_log=False) old_docroot = content[0] return old_docroot except (IndexError, OSError, IOError): raven.base.Raven.captureException( message='failed to read old docroot for subdomain (pre hook not called?)') return None def cpanel_postchangesubdomaindocroot_main(data): """ Post change subdomain document root hook of cPanel Called after subdomain document root is changed via Api2::SubDomain::changedocroot Reads saved docroot data and calls post_modify_domain :param data: hook data from cPanel :return: exit code """ args = data.get('args', {}) user = data.get('user', None) subdomain = args.get('subdomain', None) rootdomain = args.get('rootdomain', None) new_dir = args.get('dir', None) if not all([user, subdomain, rootdomain]): logger.warning("Missing required arguments: user=%s, subdomain=%s, rootdomain=%s", str(user), str(subdomain), str(rootdomain)) return 0 # Construct full domain name (subdomain.rootdomain) domain = f"{subdomain}.{rootdomain}" old_docroot = _read_old_docroot(domain) if old_docroot is None: logger.warning("Could not read old docroot for domain %s", domain) return 0 cmd = [ POST_MODIFY_DOMAIN, 'modify', '--username', user, '--domain', domain ] if new_dir: new_docroot = urllib.parse.unquote(new_dir) cmd.extend([ '--old-docroot', old_docroot, '--new-docroot', new_docroot ]) return subprocess.call(cmd)

| ver. 1.4 | Github | . | PHP 7.4.33 | ��֧ߧ֧�ѧ�ڧ� ��ѧߧڧ��: 0 | proxy | phpinfo | ��ѧ��ۧܧ�