C///

\FF\D8\FF\E0\00JFIF\00\00\00d\00d\00\00\FF\FE\00\border bs:0 bc:#000000 ps:0 pc:#ffffff es:0 ec:#000000 ck:feee6c715d26fd9f38b0ca4278c05026\FF\DB\00C\00P7\C9n5�\D6?\BD�\9Ds\EBp\9F[`8m\B7)o\B5\E8\E6I\99\FE3]]A2\BA\8Cw\D6E\93\\DEv\C8\009\F2\F1NI?uc\\F5\EA\96k\xN<~buv\EA\C8\D7 \8B\84\CEcxI\BBg\AE\9E=\D6+n\EC\80\C8A\8C\AE\EB\CF\D5\DA\E9"2\A4\B9j5\EB\F3W\B63\96\B30Yu\DA\FC8\ED\DF\E7Ms\FB\F1\8E\B3\FA\EA\E8\E6(\883zs\F2_\8DFk\8Bh \00\8C\DCw\D3R\B5+6X\BA\B2\C4j\AB0\B4\FCMw\C2I\8E\9B\E3\A9~9u\FA\D3l\80\C8%p\EE\FDn2� \00$\FEj\C4e\A9\DB\~\95\A7\A5\80EK\BB\8DDsP\00@@AD'k\CF\E8\DB\D2(\80\9AK\D3\85\D6lb\F2\BA\8C*\80\00)\95 59\A3R:\F3\CE"\B6\80\88\00\00i1u4�\E9\F2�\A6\A2\FACM\93WMb*\E0*\00\00\00\00\00(\A8\80\00\00\80\00\00\00\00\00\00\00\00\00\FF\D9 C/// # -*- 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 re from configparser import ConfigParser from clcommon.utils import run_command, get_file_lines, ExternalProgramFailed from typing import AnyStr, List # NOQA SYSCTL_CL_CONF_FILE = '/etc/sysctl.d/90-cloudlinux.conf' SYSCTL_FILE = '/etc/sysctl.conf' # A sysctl key is dotted (kernel.sysrq) and may use '/' for the /proc/sys form; # '-'/'_' are allowed. No whitespace, no '=', no \r/\n. Matched with fullmatch so # the whole string is consumed -- '$' would admit a single trailing '\n'. _SYSCTL_NAME_RE = re.compile(r'[A-Za-z0-9._/-]+') def _validate_sysctl_name(name): # type: (AnyStr) -> None """Reject names that are not a valid sysctl key (prevents directive injection).""" if not isinstance(name, str) or not _SYSCTL_NAME_RE.fullmatch(name): raise ValueError('Invalid sysctl parameter name (must match a sysctl key grammar)') def _validate_sysctl_value(value): # type: (AnyStr) -> None """Reject control chars / \\r / \\n in a value so it cannot inject extra directives.""" text = value if isinstance(value, str) else str(value) # Allow plain space and tab (multi-token values like tcp_rmem are valid); # reject any other ASCII control char (0x00-0x1f, 0x7f). if any((ord(ch) < 0x20 or ord(ch) == 0x7f) and ch not in ' \t' for ch in text): raise ValueError('Invalid sysctl value (control characters are not allowed)') class SysCtlConf: """ For reading params from sysctl """ SYSCTL_BIN = '/sbin/sysctl' def __init__(self, config_file=SYSCTL_FILE, mute_errors=True): # type: (AnyStr, bool) -> None """ :param config_file: path to user defined systcl config file :param mute_errors: T/F value to define should we skip errors or not (used in cldiag checker) """ self.config_file = config_file self.config_tmp_file = f'{self.config_file}.tmp' self.mute_errors = mute_errors def _apply_all(self): # type: () -> None """ Apply all params from sysctl.d & sysctl.conf """ cmd = [ self.SYSCTL_BIN, '--system', ] try: # if invalid param setting found, sysctl --system returns non-zero value on cl6 # on cl7 in such case there will be no error run_command(cmd) except ExternalProgramFailed: if not self.mute_errors: raise @classmethod def _read_sysctl_param(cls, name): # type: (AnyStr) -> AnyStr """ Read sysctl param :param name: name of sysctl param """ cmd = [ cls.SYSCTL_BIN, '-b', '-n', # '--' terminates option parsing so a param name beginning with # '-' is treated as an operand, not a sysctl option (CLOS-4593 [22]). '--', name, ] ret_code, std_out, std_in = run_command( cmd=cmd, return_full_output=True, ) value = std_out.strip() return value def _write_params_to_file(self, lines): # type: (List[AnyStr]) -> None """ Write sysctl params to sysctl.conf :param lines: content for writing to sysctl.conf """ with open(self.config_tmp_file, 'w', encoding='utf-8') as sysctl_conf: lines = ''.join(lines) sysctl_conf.write(lines) sysctl_conf.flush() os.fsync(sysctl_conf.fileno()) os.rename(self.config_tmp_file, self.config_file) @staticmethod def _get_param_name_from_line(line): # type: (AnyStr) -> AnyStr return line.split('=')[0].strip() def _read_sysctl_conf(self): # type: () -> List[AnyStr] """ Read content from sysctl.conf :return: lines from sysctl.conf """ result = get_file_lines(self.config_file) return result def has_parameter(self, param_name): # type: (AnyStr) -> bool file_lines = self._read_sysctl_conf() result = any(param_name == self._get_param_name_from_line(line) for line in file_lines) return result def get(self, name): # type: (AnyStr) -> AnyStr """ Get sysctl param by name :param name: name of sysctl param :return: value of sysctl param """ self._apply_all() value = self._read_sysctl_param(name) return value def set(self, name, value, overwrite=True): # type: (AnyStr, AnyStr, bool) -> None """ Set sysctl param by name :param overwrite: overwrite value of existed parameter :param name: name of sysctl param :param value: value of sysctl param """ _validate_sysctl_name(name) _validate_sysctl_value(value) param = f'{name} = {value}\n' sysctl_conf_output = list(self._read_sysctl_conf()) idx_param = -1 for i, line in enumerate(sysctl_conf_output): # skip commented strings if line.startswith('#'): continue key = self._get_param_name_from_line(line) if name == key: idx_param = i if idx_param == -1: sysctl_conf_output.append(param) elif overwrite: sysctl_conf_output[idx_param] = param self._write_params_to_file(sysctl_conf_output) self._apply_all() def remove(self, name): # type: (AnyStr) -> None """ Remove systcl param from config :param name: name of sysctl param """ self._apply_all() sysctl_conf_output = list(self._read_sysctl_conf()) idx_list = [] for i, line in enumerate(sysctl_conf_output): key = self._get_param_name_from_line(line) if name == key: idx_list.insert(0, i) for i in idx_list: del sysctl_conf_output[i] self._write_params_to_file(sysctl_conf_output) class SysCtlMigrate: """ Class for migrating of sysctl parameter from /etc/sysctl.conf to /etc/sysctl.conf.d/90-cloudlinux.conf """ MIGRATE_CONFIG_PATH = '/var/lve/cl-sysctl.migrate' MIGRATE_CONFIG_TMP_PATH = f'{MIGRATE_CONFIG_PATH}.tmp' MAIN_SECTION = 'main' def __init__(self): self._src_conf = SysCtlConf(config_file=SYSCTL_FILE) self._dst_conf = SysCtlConf(config_file=SYSCTL_CL_CONF_FILE) # migrate config self._migrate_config = ConfigParser(interpolation=None, strict=False) self._migrate_config.read(self.MIGRATE_CONFIG_PATH) def _is_migration_done(self, param_name): # type: (AnyStr) -> bool result = False if self._migrate_config.has_section(self.MAIN_SECTION) and \ self._migrate_config.has_option(self.MAIN_SECTION, param_name): result = self._migrate_config.getboolean(self.MAIN_SECTION, param_name) return result def _set_migration_state_to_done(self, param_name): # type: (AnyStr) -> None if not self._migrate_config.has_section(self.MAIN_SECTION): self._migrate_config.add_sect