File manager

# 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