#!/opt/cloudlinux/venv/bin/python3 -bb # -*- coding: utf-8 -*- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT from __future__ import print_function from __future__ import division from __future__ import absolute_import import getopt import glob import os import re import subprocess import sys from typing import Dict # NOQA import simplejson as json from packaging.version import Version from clselect.utils import file_writelines, file_readlines, file_write, \ file_read, grep from clcommon.ui_config import UIConfig NAME = os.path.basename(sys.argv[0]) THEMES_LOCATION = '/usr/local/cpanel/base/frontend' sync_association = ( # None is given to hide old resource usage # cPane config dynamicui 0 - not inverse cPane config value # key name key name 1 - inverse cPane config value # 1 - show # 0 - hide ('hideRubyApp', 'lverubysel', 1), ('hidePythonApp', 'lvepythonsel', 1), ('hideNodeJsApp', 'lvenodejssel', 1), ('hideLVEUserStat', 'lveresusage', 1), ('hidePhpApp', 'lvephpsel', 1), ('hideXrayApp', 'lvexray', 1), ('hideAccelerateWPApp', 'lvewpos', 1), (None, 'enduserlve', 1), # for x3 theme (None, 'cpu_and_concurrent_connection_usage', 1) # for paper lantern theme ) def usage(): print('Usage: %s [-h] [--show dynamicui_key1[,dynamicui_key2..]] [--hide dynamicui_key1[,dynamicui_key2..]]' % NAME) print('[--sync-conf conf_key1[,conf_key2..]] [--no-rebuild-sprites]') print('Options:') print(' -h | --help :print this message') print(' -v | --verbose :detail output actions') print(' --fix-dynamicui :fix new json-style dynamicui.conf') print(' --sync-conf :sync with dynamicui configs') print(' supported keys: all, %s' % ' '.join([k_[0] for k_ in sync_association if isinstance(k_[0], str)])) print(' --show dynamicui_key1[,dynamicui_key2..] \n' ' :show app[s] in cPanel web-interface for user') print(' --hide dynamicui_key1[,dynamicui_key2..] \n' ' :hide app[s] in cPanel web-interface for user') print(' --path PATH :path to specific dynamicui config file') print(' --no-rebuild-sprites :don\'t rebuild sprites after patching dynamicui configs') print(' Association:') print(' conf_key\t\tdynamicui_key') for c1, c2, _ in sync_association: print(' %s\t%s' % (c1, c2)) print('') print(' Examples:') print(' %s --hide=lverubysel --hide=lvepythonsel --show=enduserlve' % NAME) print(' %s --hide=lverubysel,lvepythonsel --show=enduserlve' % NAME) print(' %s --sync-conf all --no-rebuild-sprites' % NAME) print(' %s --sync-conf hideRubyApp, hidePythonApp' % NAME) # add support function "any" for python 2.4 if 'any' not in globals(): def any(iterable): """ backported buildin function "any" for python 2.4 >>> any([False, False, False]) False >>> any([True, False, False]) True :param iterable: :return boll: """ for element in iterable: if element: return True return False def parse_json_with_comments(text, **kwargs): """ Parse json file with comments """ regex_str = r'\s*(#|\/{2}).*$' REGEX_COMMENT = re.compile(regex_str) REGEX_LINE_COMMENT = re.compile(r"^" + regex_str, re.IGNORECASE) REGEX_INLINE = re.compile(r'(:?(?:\s)*([A-Za-z\d\.{}]*)|((?<=\").*\"),?)(?:\s)*(((#|(\/{2})).*)|)$') lines = text.split('\n') for index, line in enumerate(lines): if REGEX_COMMENT.search(line): if REGEX_LINE_COMMENT.search(line): lines[index] = "" elif REGEX_INLINE.search(line): lines[index] = REGEX_INLINE.sub(r'\1', line) # remove trailing comma output = "\n".join(lines) output = re.sub(r",[ \t\r\n]*}", "}", output) output = re.sub(r",[ \t\r\n]*\]", "]", output) try: return json.loads(output, **kwargs) except ValueError: return None def fix_json_format(path): """ Fix json format conf file. Remove remove old-style lines :param: `str` path: path to config file """ if not os.path.isfile(path): return None # skip files with old-style config values if grep('^description=>', path): return False # clean old format lines from file try: f = open(path) output_lines = [] for line in f: line = line.rstrip("\n") directive = DynUiLine(line) if not directive.get_string(): # don`t add old-style line to file output_lines.append(line) except IOError: return None else: f.close() # update conf file try: f = open(path, "w") f.write("\n".join(output_lines)) except IOError: return None else: f.close() return 0 class DynUiLine(object): """ """ key_types = {'skipobj': int} def __init__(self, line): """ Constructor """ self.line = line.strip() self.line_parsed_ = self.pars_directives() def pars_directives(self): """ Parse line "self.line" and return parsed data :return list: return parsed structure; for future manipulation """ line_parsed_ = list() line = self.line for l_ in line.split(','): if l_.count('=>') != 1: continue key_, val_ = list(map(str.strip, l_.split('=>', 1))) # parse and clear line like "key=>val" type_converter = self.key_types.get(key_) if type_converter: try: val_ = type_converter(val_) except ValueError: pass line_parsed_.append([key_, val_]) return line_parsed_ def has_dir(self, key, val=None): """ Check if directive present :param str key: directive name :param str|None val: directive value; if value is None value not check :return bool: return True if directive present [and value] present """ for key_, val_ in self.line_parsed_: if key_ == key: if (val is None) or (val == val_): return True return False def del_dir(self, key): """ Delete directive :param str key: directive name need delete :return bool: return True if directive was deleted """ was_deleted = False for index_, (key_, _) in enumerate(self.line_parsed_): if key_ == key: del self.line_parsed_[index_] was_deleted = True return was_deleted def set_dir(self, key, val): """ Change or create directive :param str key: directive name :param str|int val: directive value :return bool: return True if change or create directive is success; False if not need change """ for index_, (key_, val_) in enumerate(self.line_parsed_): if key == key_: if val != val_: self.line_parsed_[index_][1] = val return True else: return False # add directive if not present self.line_parsed_.append([key, val]) return True def get_string(self): """ Generate and return raw string after manipulations :return str: return line after manipulations """ return ','.join(['=>'.join(map(str, key_val_)) for key_val_ in self.line_parsed_]) def hide_object(self, file_=None): """ Modify directives to hide object :param str|None file_: file name that need hide :return bool: return true if state was changed """ if self.has_dir('file', file_): # append directive "skipobj=>1" to line if len(self.line_parsed_) >= 2: return self.set_dir('skipobj', 1) return False def show_object(self, file_=None): """ Modify directives to show object :param str|None file_: file name that need hide :return bool: return true if state was changed """ if self.has_dir('file', file_): if len(self.line_parsed_) > 2: # detect directive "skipobj=>1" or "skipobj=>0" and delete them return self.del_dir('skipobj') elif len(self.line_parsed_) == 2: if self.has_dir('skipobj'): # detect line "file=>some_name,skipobj=>1" and delete all line self.line_parsed_ = list() return True return False def patch_dynamicui(path, app_show=None, app_hide=None, verbose=False): """ patching dynamicui.conf for add or delete direcitve 'skipobj=>1' in line :param str path: path to dynamicui config file :param list app_show: list apps to show in web-interface; delete directive 'skipobj=>1' in line :param list app_hide: list apps to hide in web-interface; add directive 'skipobj=>1' in line :return: `dict` {app: True,...} | None """ app_show = app_show or list() app_hide = app_hide or list() file_was_patched = False patched_apps = set() dynamicui_lines = file_readlines(path) for line_index, line in enumerate(dynamicui_lines): if line.lstrip('\n').startswith('#'): # ignoring line with comments continue line = line.rstrip('\n') directives = DynUiLine(line) if not directives.line: continue line_was_patched = False for app_name in app_show: if directives.has_dir('file', app_name): patched_apps.add(app_name) if directives.show_object(app_name): line_was_patched = True break for app_name in app_hide: if directives.has_dir('file', app_name): patched_apps.add(app_name) if directives.hide_object(app_name): line_was_patched = True break # line_was_patched = any(map(directives.show_object, app_show) + map(directives.hide_object, app_hide)) if line_was_patched: dynamicui_lines[line_index] = directives.get_string() + '\n' file_was_patched = True if file_was_patched: file_writelines(path, dynamicui_lines, 'w') if verbose: print('%s: file %s was patched' % (NAME, path)) else: if verbose: print('%s: file %s was skiped, not need to patch' % (NAME, path)) return patched_apps def patch_json_dynamicui(path, app_show, app_hide, verbose=False): """ Patch dynamicui.conf file in json format :param: `str` path: path to dynamicui config file :param: `list` app_show: list apps to show in web-interface; delete key 'skipobj' :param: `list` app_hide: list apps to hide in web-interface; add key 'skipobj: 1' :return: `dict` {app: True,...} | None """ # clean json format file for valid parse json conf = parse_json_with_comments(file_read(path)) if conf is None: if verbose: print('%s: file %s was skipped because contains invalid JSON format' % (NAME, path)) return None file_was_patched = False patched_apps = set() file_items = [] for item in conf: app_name = item.get("file") if app_name in app_show: item.pop("skipobj", None) file_was_patched = True patched_apps.add(app_name) elif app_name in app_hide: item["skipobj"] = 1 file_was_patched = True patched_apps.add(app_name) file_items.append(item) # add unexists items with skipobj key # non_patched_hide_app = set(app_hide).difference(patched_apps) # if non_patched_hide_app: # file_was_patched = True # for app_name in non_patched_hide_app: # file_items.append({"name": app_name, "file": app_name, "skipobj": 1}) # patched_apps.add(app_name) if file_was_patched: # https://stackoverflow.com/questions/18337407/saving-utf-8-texts-in-json-dumps-as-utf8-not-as-u-escape-sequence file_write(path, json.dumps(file_items, indent=4, ensure_ascii=False).encode('utf8'), mode='wb') if verbose: print('%s: file %s was patched' % (NAME, path)) else: if verbose: print('%s: file %s was skiped, not need to patch' % (NAME, path)) return patched_apps def patch_dynamicui_wrapper(path, app_show=None, app_hide=None, verbose=False, is_json_format=False): """ Wrapper for patch dynamicui config file. Catch base exceptions and print base messages """ path = os.path.abspath(path) if os.path.isfile(path) and not os.path.islink(path): try: # detect file format json_format = is_json_format and not grep('^description=>', path) if json_format: return patch_json_dynamicui(path, app_show, app_hide, verbose) else: return patch_dynamicui(path, app_show, app_hide, verbose) except IOError as e: if verbose: print('%s: %s' % (NAME, str(e))) else: if verbose: print('%s: file %s not present, skip' % (NAME, path)) return None def patch_themes(themes_dir_list, app_show, app_hide, verbose=False, is_json_format=False): """ Patch dynamicui configuration files for all themes """ for theme_path in themes_dir_list: # set of patched apps of current theme patched_set = set() # first change additional configs for app_name in (app_hide + app_show): dynamicui_conf_path = os.path.join(theme_path, 'dynamicui', 'dynamicui_%s.conf' % app_name) if app_name in app_hide: res = patch_dynamicui_wrapper(dynamicui_conf_path, app_hide=[app_name], verbose=verbose) patched_set |= (res or set()) elif app_name in app_show: res = patch_dynamicui_wrapper(dynamicui_conf_path, app_show=[app_name], verbose=verbose) patched_set |= (res or set()) # now filter all patched apps from start lists not_patched_app_show = list(sorted(set(app_show).difference(patched_set))) not_patched_app_hide = list(sorted(set(app_hide).difference(patched_set))) # check is all application was patched or no if not len(not_patched_app_show) and not len(not_patched_app_hide): continue # next step change main config if need patch anything dynamicui_conf_path = os.path.join(theme_path, 'dynamicui.conf') patch_dynamicui_wrapper(dynamicui_conf_path, not_patched_app_show, not_patched_app_hide, verbose, is_json_format) def load_json_conf(verbose=False, key_filter=None): """ Loading config from UI config file value must be only False or True :param verbose: detailed output enabled :param key_filter: keys to be filtered output dict :return dict: """ key_filter = key_filter or list() output_dict = dict() parsed_config = UIConfig().get_param('uiSettings') if parsed_config: for key, value in parsed_config.items(): if key not in key_filter: continue try: output_dict.update({key: value}) except ValueError: if verbose: print('WARNING: incorrect key "%s" in UI config file, skipped;' ' value must be True or False' % key) # add selector`s status config to result dict sync_selector_setting(output_dict, 'hideNodeJsApp') sync_selector_setting(output_dict, 'hidePythonApp') sync_selector_setting(output_dict, 'hidePhpApp') return output_dict def sync_selector_setting(external_conf, plugin_flag): # type: (Dict, str) -> () """ Adds value of selector status to cpanel config :param external_conf: dict with cpanel config values :param plugin_flag: plugin's flag name """ try: if plugin_flag == 'hideNodeJsApp': from clselect.clselectnodejs.node_manager import NodeManager m = NodeManager() elif plugin_flag == 'hidePythonApp': from clselect.clselectpython.python_manager import PythonManager m = PythonManager() elif plugin_flag == 'hidePhpApp': from clselect.clselectphp.php_manager import PhpManager m = PhpManager() else: raise NotImplementedError() external_conf[plugin_flag] = not m.selector_enabled except ImportError: external_conf[plugin_flag] = False except Exception as e: external_conf[plugin_flag] = False print(e) def sync_conf(themes_dir_list, sync_association=sync_association, key_filter=None, verbose=False, is_json_format=False): app_hide = list() app_show = list() conf = load_json_conf(verbose=verbose, key_filter=key_filter) # generate 'app_show' and 'app_hide' app list for key_, app_name, reverse_boolean in sync_association: val_ = conf.get(key_) if key_ is None: app_hide.append(app_name) # if key_ is None then hide the app if val_ is None: continue if reverse_boolean: val_ = not val_ if val_ is True: app_show.append(app_name) elif val_ is False: app_hide.append(app_name) patch_themes(themes_dir_list, app_show, app_hide, verbose, is_json_format) def main(args=None): try: opts, _ = getopt.getopt(args or sys.argv[1:], 'h:v', ['show=', 'hide=', 'path=', 'sync-conf=', 'no-rebuild-sprites', "fix-dynamicui", 'verbose', 'help']) except getopt.GetoptError as err: print(str(err)) usage() sys.exit(2) fix_dynamicui = False verbose = False app_show = list() app_hide = list() dynamicui_conf_path = None rebuild_sprites = True sync = list() for o, _ in opts: if o in ('-h', '--help'): usage() sys.exit() elif o == '--show': app_show.extend(_.split(',')) elif o == '--hide': app_hide.extend(_.split(',')) elif o == '--path': dynamicui_conf_path = _ elif o == '--fix-dynamicui': fix_dynamicui = True elif o == '--sync-conf': if _ == 'all': sync = [k_[0] for k_ in sync_association] else: sync.extend(_.split(',')) elif o == '--no-rebuild-sprites': rebuild_sprites = False elif o in ('--verbose', '-v'): verbose = True else: usage() sys.exit(2) if fix_dynamicui: for path in glob.glob(os.path.join(THEMES_LOCATION, '*')): if os.path.isdir(path) and not os.path.islink(path): fix_json_format(os.path.join(path, "dynamicui.conf")) return 0 if len(app_show + app_hide) == 0 and not sync: usage() sys.exit(2) # checking the uniqueness of values in the 'app_show' and 'app_hide' if len(set(app_show + app_hide)) != len(app_show + app_hide): print("ERROR: Application names in options '--show' and '--hide' should not be repeated") sys.exit(2) # get CP version import cldetectlib as detect detect.getCP() if detect.CP_NAME != 'cPanel': print('WARNING: Unsupported control panel, this script works only on cPanel. Exiting.') sys.exit(2) is_json_format = Version(detect.CP_VERSION) >= Version("54.0") if dynamicui_conf_path: patch_dynamicui_wrapper(dynamicui_conf_path, app_show, app_hide, verbose, is_json_format) return # list cPanel themes directories, without directories links themes_dir_list = [path for path in glob.glob(os.path.join(THEMES_LOCATION, '*')) if os.path.isdir(path) and not os.path.islink(path)] if sync: sync_conf(themes_dir_list, verbose=verbose, key_filter=sync, is_json_format=is_json_format) patch_themes(themes_dir_list, app_show, app_hide, verbose, is_json_format) if rebuild_sprites: if verbose: print('%s: rebuilding sprites in background' % NAME) if os.path.exists('/usr/local/cpanel/bin/sprite_generator'): subprocess.run("/usr/local/cpanel/bin/sprite_generator --all > /dev/null 2>&1 &", shell=True, executable='/bin/bash') else: subprocess.run("/usr/local/cpanel/bin/rebuild_sprites -cponly -quiet > /dev/null 2>&1 &", shell=True, executable='/bin/bash') if __name__ == '__main__': main()