# -*- 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 os import shutil import subprocess import simplejson as json import configparser CL_CONFIG = '/etc/sysconfig/cloudlinux' HOOK_SECTION = 'hooks' HOOK_SEPARATOR = ';' DEBUG_STRING = 'python-cllib:clhook:' def get_config(file_name): """ read config from file_name return config """ config = configparser.ConfigParser(interpolation=None, strict=False) config.optionxform = str # make config case sensitive config.read(file_name) return config def write_config(config, file_name, debug_local_name): """ write config to file_name return [BOOLEAN_STATE, MESSAGE] """ try: tmp_cfg_path = f'{file_name}.{os.getpid()}' with open(tmp_cfg_path, 'w', encoding='utf-8') as tmpconfig: config.write(tmpconfig) shutil.move(tmp_cfg_path, file_name) except IOError as e: return [False, f'{debug_local_name}{file_name} write error\n{str(e)}'] return [True, 'OK\n'] def get_hook_list_from_file(hook_prefix, config_file = CL_CONFIG): """ return [BOOLEAN_STATE, hook_list] """ hook_list = [] config = get_config(config_file) try: hook_list = config.get(HOOK_SECTION, hook_prefix).split(HOOK_SEPARATOR) except (configparser.NoOptionError, configparser.NoSectionError): return [False, hook_list] return [True, hook_list] def register_hook(hook_path, hook_prefix, config_file = CL_CONFIG): """ add hook to config file return [BOOLEAN_STATE, MESSAGE] """ debug_local_name = DEBUG_STRING + 'register_hook:' config = get_config(config_file) try: hook_list = config.get(HOOK_SECTION, hook_prefix).split(HOOK_SEPARATOR) if hook_path in hook_list: return [True, debug_local_name + 'Hook ' + str(hook_path) + ' already registred\n' ] hook_list.append(hook_path) config.set(HOOK_SECTION, hook_prefix, HOOK_SEPARATOR.join(hook_list)) except configparser.NoSectionError: config.add_section(HOOK_SECTION) config.set(HOOK_SECTION, hook_prefix, hook_path) except configparser.NoOptionError: config.set(HOOK_SECTION, hook_prefix, hook_path) return write_config(config, config_file, debug_local_name) def unregister_hook(hook_path, hook_prefix, config_file = CL_CONFIG): """ remove hook from config_file return [BOOLEAN_STATE, MESSAGE] """ debug_local_name = DEBUG_STRING + 'unregister_hook:' config = get_config(config_file) try: hook_list = config.get(HOOK_SECTION, hook_prefix).split(HOOK_SEPARATOR) hook_list.remove(hook_path) config.set(HOOK_SECTION, hook_prefix, HOOK_SEPARATOR.join(hook_list)) return write_config(config, config_file, debug_local_name) except (configparser.NoSectionError, configparser.NoOptionError, ValueError): return [True, 'OK\n'] def call_hook_list(hook_list, data): """ call hooks and send data as json to stdin return [BOOLEAN_STATUS, message] """ debug_local_name = DEBUG_STRING + 'call_hook:' json_encode = json.dumps(data) for hook in hook_list: if os.path.isfile(hook) and os.access(hook, os.X_OK): try: with subprocess.Popen( hook, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) as proc: proc.stdin.write(json_encode) proc.communicate() except OSError: return [False, debug_local_name + 'Hook call error\n'] else: return [False, debug_local_name + str(hook) + ' : isn`t file or isn`t runnable\n'] return [True, 'OK\n']