/opt/imunify360/venv/lib/python3.11/site-packages/clcommon
Edit: /opt/imunify360/venv/lib/python3.11/site-packages/clcommon/sysctl.py (7792B)
# -*- 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
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'
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',
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
"""
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_section(self.MAIN_SECTION)
self._migrate_config.set(self.MAIN_SECTION, param_name, 'true')
with open(self.MIGRATE_CONFIG_TMP_PATH, 'w', encoding='utf-8') as config_tmp:
self._migrate_config.write(config_tmp)
config_tmp.flush()
os.fsync(config_tmp.fileno())
os.rename(self.MIGRATE_CONFIG_TMP_PATH, self.MIGRATE_CONFIG_PATH)
def migrate(self, param_name, default_value):
# type: (AnyStr, AnyStr) -> None
"""
Migrate sysctl parameter from one config to another
in conformity with presence of parameter in source config and default value.
All cases of using you can see in doc:
https://docs.google.com/spreadsheets/d/1H_q3TA3CMFCj1YwAOn7G1LZgxcS1F4h90B6jCgiTUpo
"""
# We won't do anything if paramater already was migrated.
if self._is_migration_done(param_name=param_name):
return
# We use value from src file if parameter is present in it
if self._src_conf.has_parameter(param_name):
value = self._src_conf.get(param_name)
# otherwise, we use default value.
else:
value = default_value
# Remove parameter from src file.
self._src_conf.remove(param_name)
if value is not None:
# Write the migrated parameter to dst file
# We shouldn't overwrite existing value in 90-cloudlinux.conf by default value,
# but we should migrate value from src cfg to dst cfg
overwrite = True if default_value is None else False
self._dst_conf.set(param_name, value, overwrite=overwrite)
# Set the migrate state to True.
self._set_migration_state_to_done(param_name=param_name)