OwlCyberSecurity - MANAGER

Edit File: pylve_wrapper.py

# Copyright © Cloud Linux GmbH & Co. KG, Cloud Linux Software, Inc 2010-2026 # All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT """pylve C-binding wrapper and related types. This module wraps the pylve C extension (PyLve, PyLveError) and lives outside lveapi deliberately: websiteisolation.* is imported by lveapi at the top level, so any websiteisolation sub-module that needs PyLve or PyLveError must not import from lveapi — that would form a cycle. Placing these types here breaks the cycle; lveapi re-exports them for backward compatibility. """ import contextlib import errno import time from clcommon.clproc import ProcLve from clcommon.clfunc import uid_max try: import pylve as _pylve_module except ImportError: _pylve_module = None UID_MAX = uid_max() # kernel devs say that lve_id type is uint32_t # so max id should be 0xFFFF_FFFF, in fact we have this: MAX_LVE_ID = 0x7FFFFFFF - 1 class PyLveError(Exception): pass class PyLve: """ Wrapper for generate traceback with pretty descriptions """ @staticmethod def _code_is_error(code): return isinstance(code, int) and code != -errno.ENOSYS and code != 0 def _arg_to_str(self, arg_var): if isinstance(arg_var, self._pylve.liblve_settings): # for pretty print liblve_settings object liblve_settings_attr = ', '.join( [f"{attr}={getattr(arg_var, attr)}" for attr in dir(arg_var) if not attr.startswith('_')] ) arg_var_str = f'<liblve_settings object {liblve_settings_attr}>' else: arg_var_str = str(arg_var) return arg_var_str def _wrapped_fun(self, call, *args, **kwargs): msg_template = kwargs.pop('err_msg', self.default_msg_template) ignore_error = kwargs.pop('ignore_error', self.ignore_error) code = call(*args, **kwargs) is_error = self._code_is_error(code) if is_error and self._retry: # wait and try again run function time.sleep(self._retry_time) code = call(*args, **kwargs) is_error = self._code_is_error(code) format_args = { 'code': code, 'fun_name': call.__name__, 'module': call.__module__, 'args_': ', '.join(list(map(self._arg_to_str, args)) + [f"{k}={self._arg_to_str(v)}" for k, v in kwargs.items()]) } if self.debug >= 1: print(self.debug_msg_template.format(**format_args)) if self.debug >= 2: self.traceback.print_stack() if not ignore_error and is_error: msg = msg_template.format(**format_args) raise PyLveError(msg) return code def _wrap_code(self, call): def fun(*args, **kwargs): return self._wrapped_fun(call, *args, **kwargs) return fun def __init__(self, pylve=_pylve_module, retry=True, retry_tyme=0.1, debug=0): self.debug = debug if self.debug >= 2: self.traceback = __import__('traceback') self.default_msg_template = 'Error code {code}; {module}.{fun_name}({args_})' self.debug_msg_template = "DEBUG [lvectl]: call {module}.{fun_name}({args_}) with code {code}" self.ignore_error = False self._pylve = pylve self._retry = retry self._proc = ProcLve() self._retry_time = retry_tyme self.api_version = self._pylve.lve_get_api_version() self.initialize = self._pylve.initialize self.lve_start = self._wrap_code(self._pylve.lve_start) self.liblve_settings = self._pylve.liblve_settings self.lve_create = self._wrap_code(self._pylve.lve_create) self.lve_destroy = self._wrap_code(self._pylve.lve_destroy) self.lve_info = self._pylve.lve_info self.lve_set_default = self._wrap_code(self._pylve.lve_set_default) self.lve_setup = self._wrap_code(self._pylve.lve_setup) self.lve_enter_pid = self._wrap_code(self._pylve.lve_enter_pid) self.lve_enter_pid_flags = self._wrap_code(self._pylve.lve_enter_pid_flags) self.lve_leave_pid = self._wrap_code(self._pylve.lve_leave_pid) if hasattr(pylve, 'lve_lvp_create'): self.lve_lvp_create = self._wrap_code(self._pylve.lve_lvp_create) self.lve_lvp_destroy = self._wrap_code(self._pylve.lve_lvp_destroy) self.lve_lvp_map = self._wrap_code(self._pylve.lve_lvp_map) self.lve_lvp_move = self._wrap_code(self._pylve.lve_lvp_move) # mocked functions. Not implement in pylve self.lve_lvp_setup = self._wrap_code(self.lve_lvp_setup) if hasattr(pylve, 'lve_lvp_create2'): self.lve_lvp_create2 = self._wrap_code(self._pylve.lve_lvp_create2) def resellers_supported(self): """ Check in pylve binding reseller limits supported """ return hasattr(self._pylve, 'lve_lvp_create') def domains_supported(self): """ Check if per-domain limits via hierarchical LVP (lve_lvp_create2) are supported by both the installed library and the running kernel module. liblve and lve-kmod may be at different versions when one is updated without rebooting. lve_lvp_create2 requires kernel module API >= 1.7. After pylve.initialize(), lve_kapi_ver() returns the API version that kapi_init() negotiated with the running kernel. """ if not hasattr(self._pylve, 'lve_lvp_create2'): return False kapi_major, kapi_minor = self._pylve.lve_kapi_ver() return (kapi_major, kapi_minor) >= (1, 7) def lve_exists(self, lve_id): """ Check if lve exists in kernel :rtype: bool """ try: self.lve_info(lve_id) return True except OSError: return False # TODO: remove this wrapper when kernel logic is ready # upd: kernel logic is still not ready yet (KMODLVE-79) def lve_lvp_setup(self, lvp_id, settings): # pylint: disable=method-hidden """ Wrapper for lve_lvp_setup. When reseller's limits changed, we should iterate over his user's limits and set them again; This behaviour is needed cause kernel does not update users limits after changing reseller's one Adjust parameters for top level container. :param int lvp_id: top level container ID, 0 by default; :param settings: liblve_settings instance. :return: 0 or errno value """ # is it real situation when lvp_id is 0? if lvp_id == 0: return self._pylve.lve_lvp_setup(lvp_id, settings) # reduce lve limits real_lve_settings: dict = {} for lve_id in self._proc.lve_id_list(lvp_id): # save real settings to restore them later try: real_settings = self.lve_info(lve_id) real_lve_settings[lve_id] = real_settings if real_settings.ls_cpu > settings.ls_cpu: # get current settings and reduce cpu temp_settings = self.lve_info(lve_id) temp_settings.ls_cpu = min(temp_settings.ls_cpu, settings.ls_cpu) self.lve_setup(lve_id, temp_settings) except OSError: # lve was destroyed, ignore that pass # set new reseller settings _lve_lvp_setup = self._wrap_code(self._pylve.lve_lvp_setup) # ignore errors here and raise errors after operations result = _lve_lvp_setup(lvp_id, settings, ignore_error=True) # pass lve's limits again, so kernel will apply them for lve_id in self._proc.lve_id_list(lvp_id): if lve_id in real_lve_settings: self.lve_setup(lve_id, real_lve_settings[lve_id]) else: # some lve was created during operations above # nothing bad, but we can do nothing with that # until we have no lock's for this method self.lve_setup(lve_id, self.lve_info(lve_id)) return result def get_available_lve_id(self, start=UID_MAX, stop=MAX_LVE_ID): """ Iter over lves and find available one. :param int start: value to start search from; UID_MAX by default :param int stop: max value when we will stop search :return int: lve_id """ for lve_id in range(start + 1, stop): try: self.lve_info(lve_id) except OSError: return lve_id raise PyLveError(f"Unable to find free lve in range ({start}, {stop})") @contextlib.contextmanager def context_ignore_error(self, ignore_error): self.ignore_error, saved_ignore_error = ignore_error, self.ignore_error try: yield finally: self.ignore_error = saved_ignore_error