1#@!#!123s

: / opt / imunify360 / venv / lib / python3.11 / site-packages / clcommon /
Filename : utils.py
back
# -*- coding: utf-8 -*-

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2022 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
#

import configparser
import datetime
import grp
import io
import os
import platform
import pwd
import re
import stat
import subprocess
import sys
from configparser import ConfigParser, Error
from functools import lru_cache
from typing import Any, AnyStr, Dict, Optional, Tuple, Union  # NOQA

import psutil
import secureio
from lxml import etree

# These commands used to be in this file, but is_ubuntu cannot be imported if they're here
# due to a circular import with clcommon.lib.__init__ -> clcommon.lib.mysql_governor_lib.
# To avoid that, we move them to a separate file and import into this namespace
# so that things like `from clcommon.utils import run_command` still work.
from clcommon.utils_cmd import (  # NOQA
    ExternalProgramFailed,
    check_command,
    exec_utility,
    run_command,
)

# Not all imports are used in this file - they are made so that other
# modules can import them from utils, not utils_cmd. Therefore, NOQA.

RHN_SYSTEMID_FILE = "/etc/sysconfig/rhn/systemid"
WEEK_DAYS = ("Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun")
LITESPEED_STATUS_FOR_USERS = "/opt/cloudlinux/litespeed_status"


def create_symlink(link_value, link_path):
    """
    Create symlink link_path -> link_value if it does not exist or
    points to different location
    :param link_value: path that symlink should point to (symlink value)
    :type link_value: str
    :param link_path: path where to create symlink
    :type link_path: str
    """
    link_to = None
    if os.path.islink(link_path):
        try:
            link_to = os.readlink(link_path)
        except OSError:
            pass
    if link_value != link_to:
        try:
            os.unlink(link_path)
        except OSError:
            pass
        os.symlink(link_value, link_path)


def get_file_lines(path, unicode_errors_handle=None):
    """
    Read file and return file's lines
    errors param may be passed to define how handle
    unicode errors, errors=None is default value of open()
    :param path: path to file
    :param unicode_errors_handle: how to handle unicode errors
    :return: list of file's lines
    """

    content = []
    if os.path.isfile(path):
        with open(path, "r", encoding="utf-8", errors=unicode_errors_handle) as f:
            content = f.readlines()
    return content


def write_file_lines(path, content, mode):
    """
    Write lines to file
    :param content: list of lines for writing to file
    :param path: path to file
    :param mode: open mode
    :return: None
    """

    with open(path, mode, encoding='utf-8') as f:
        f.writelines(content)


def delete_line_from_file(path, line):
    """
    Delete line from file. Return True when line(s) have been deleted, False otherwise (specified line is not found)
    :param path: path to file
    :type path: string
    :param line: line to delete without EOL ('\n')
    :type line: string
    :rtype bool
    """
    file_lines = get_file_lines(path)
    out_file_lines = [item for item in file_lines if line != item.rstrip("\n")]
    found = len(file_lines) != len(out_file_lines)
    write_file_lines(path, out_file_lines, "w+")

    return found


def is_root_or_exit():
    """
    Check whether current user is effectively root and exit if not
    """
    euid = os.geteuid()
    if euid != 0:
        try:
            # Additional admins placed in this special group
            # by lvemanager hooks to add root-like privileges to them
            gr_gid = grp.getgrnam("clsupergid").gr_gid
            if gr_gid in os.getgroups() or os.getegid() == gr_gid:
                return
        except KeyError:
            pass  # No group - no privileges
        print("Error: root privileges required. Abort.", file=sys.stderr)
        sys.exit(-1)


def is_ea4():
    """
    Detects is EA4 installed
    :return: True - EA4 present; False - EA4 absent
    """
    return os.path.isfile("/etc/cpanel/ea4/is_ea4")


def grep(
        pattern,
        path=None,
        fixed_string=False,
        match_any_position=True,
        multiple_search=False,
        data_from_file=None,
):
    """
    Grep pattern in file
    :param multiple_search: if True - search all match,
                            False - search first match
    :param pattern: pattern for search
    :param path: path to file
    :param data_from_file: read data from file for parsing
    :param fixed_string: if True - search only fixed string,
           False - search by regexp
    :param match_any_position: if True - search any match position,
           False - search only from string begin
    :return: Generator with matched strings
    """
    if data_from_file is None:
        data_from_file = get_file_lines(path)
    result = None
    if not fixed_string:
        # It's append the symbol ^ to the regexp
        # if we are searching from the begin of a string and by the regexp
        if not pattern.startswith("^") and not match_any_position:
            pattern = f"^{pattern}"
        pattern_comp = re.compile(pattern)
    else:
        pattern_comp = None
    for line in data_from_file:
        if fixed_string:
            if match_any_position and line.find(pattern) != -1:
                result = line
            elif line.startswith(pattern):
                result = line
        else:
            if pattern_comp.search(line):
                result = line
        if multiple_search and result is not None:
            yield result
        elif result is not None:
            break
        result = None
    if result is not None:
        yield result


def _parse_systemid_file():
    """
    :rtype: lxml.etree._ElementTree obj
    """
    return etree.parse(RHN_SYSTEMID_FILE)  # pylint: disable=c-extension-no-member


def get_rhn_systemid_value(name) -> Optional[str]:
    """
    find a member in xml by name and return value
    :type name: str
    :rtype: str|None
    """
    try:
        rhn_systemid_xml = _parse_systemid_file()
        for member in rhn_systemid_xml.iter("member"):
            if member.find("name").text == name:
                return member.find("value")[0].text
    except (IOError, IndexError, KeyError, etree.ParseError):  # pylint: disable=c-extension-no-member
        return None
    return None


def get_file_system_in_which_file_is_stored_on(file_path):
    # type: (str) -> Dict[str, Any]
    """
    This function is written for detect file system in which file is stored on.
    E.g., the file can be stored in NFS and this can affect the normal operation of the file.
    We want to receive information about FS in emergency situations during reading or writing
    :param file_path: path to file, for which we want to detect file system
    :return: dict, which contains two keys:
            key 'success' can be equals to False if we got error or True if we got normal result
            key 'details' can contais error string if key 'success' is False or result if key 'success' is True
    """
    result = {
        "success": False,
        "details": f'File "{file_path}" isn\'t exists',
    }
    if not os.path.exists(file_path):
        return result
    # Command:  mount | grep "on $(df <file_name> | tail -n 1 | awk '{print $NF}') type"
    # Result:   /usr/tmpDSK on /var/tmp type ext3 (rw,nosuid,noexec,relatime,data=ordered)
    try:
        mount_point = subprocess.check_output(
            [f"df {file_path} | tail -n 1 | awk '{{print $NF}}'"],
            shell=True,
            executable="/bin/bash",
            text=True,
        ).strip()
        data = subprocess.check_output(
            [f'mount | grep "on {mount_point} type"'],
            shell=True,
            executable="/bin/bash",
            text=True,
        ).strip()
        result["success"] = True
        result["details"] = data
    except (subprocess.CalledProcessError, OSError) as err:
        result["details"] = f'We can\'t get file system for file "{file_path}". Exception "{err}"'

    return result


def is_testing_enabled_repo() -> bool:
    """
    Checks if testing is enabled in /etc/yum.repos.d/cloudlinux.repo config
    :return: bool value if testing enabled or not
    """
    parser = ConfigParser(interpolation=None, strict=False)
    try:
        parser.read("/etc/yum.repos.d/cloudlinux.repo")
        res = parser.getboolean("cloudlinux-updates-testing", "enabled")
    # use base exception for config parser class
    except Error:
        res = False
    return res


def get_cl_version() -> Optional[str]:
    """
    Returns cl version taking into account release version
    E.g: release =
        2.6.32-896.16.1.lve1.4.54.el6.x86_64
        5.14.0-162.6.1.el9_1.x86_64
    el6 = cl6
    el8 = cl8
    ........
    ubuntu+cl_extesions = cl8
    :return appropriate version string
    """
    if is_ubuntu():
        return "cl8"
    check_vals_decoder = {
        "el6.": "cl6",
        "el6h.": "cl6h",
        "el7.": "cl7",
        "el7h.": "cl7h",
        "el8.": "cl8",
        ".el9": "cl9",  # note the dot at the beginning because of possible `el9_1` and so on
    }
    release: Optional[str] = platform.release()
    for check_val, cl_version in check_vals_decoder.items():
        if release and check_val in release:
            return cl_version
    release = get_rhn_systemid_value("os_release")
    ret_val = None

    if release is None:
        ret_val = None
    elif "6" in release:
        ret_val = "cl6"
    elif "7" in release:
        ret_val = "cl7"
    elif "8" in release:
        ret_val = "cl8"
    elif "9" in release:
        ret_val = "cl9"
    return ret_val


@lru_cache(maxsize=None)
def get_virt_type() -> Optional[str]:
    """
    Returns virtualization type on current system.
    It is reachable via virt-what utility.
    E.g.: 'kvm', 'bhyve', 'openvz', 'qemu'
    All acceptable outputs are listed here:
    https://people.redhat.com/~rjones/virt-what/virt-what.txt

    Output will be returned with at least two rows
    Sample:
        > kvm
        >
    Furthermore, there is a possibility for multiple text rows
    Sample:
        > xen
        > xen-domU
    That's why, the result will be taken from a first row.
    If the output is empty, and there were no errors, the machine
    is either non-virtual, or virt-what tool isn't familiar with it's
    hypervisor. But the list of supported hypervisors and containers
    covers all popular VMs types.

    :return: virt_type - Optional[AnyStr]
        - appropriate virtualization type string,
        - 'physical' if there is no virtualization,
        - None if there was an error
    """
    try:
        virt_what_output = run_command(["/usr/sbin/virt-what"]).strip()
    except (subprocess.CalledProcessError, FileNotFoundError):
        return None
    # Check for the non-empty output - virtualization exists
    if virt_what_output != "":
        return virt_what_output.split("\n", maxsplit=1)[0]
    else:
        return "physical"


def check_pid(pid: int):
    """
    Checks for a process existence by os.kill command
    If os.kill will be used as os.kill(pid, 0), it will
    just check for a presence of such PID

    And if such pid can't be reached with kill method,
    there will be raised OSError
    """
    try:
        os.kill(pid, 0)
    except OSError:
        return False
    else:
        return True


def is_process_running(pid_file_path: str) -> bool:
    """Check if process running using pid file
    Arguments:
        pid_file_path: path to the pid file of service

    Returns:
        bool: True or False
    """
    if os.path.isfile(pid_file_path):
        with open(pid_file_path, encoding="utf-8") as f:
            try:
                return check_pid(int(f.read().strip()))
            except ValueError:
                pass
    return False


def is_litespeed_running() -> bool:
    """
    Detects that server works under Litespeed.
    Note: be careful when modifying this method.
    It is used in X-Ray, ask @dkavchuk or someone else from C-Projects team
    for details.
    return: True - LS working; False - LS not running (stopped or absent)
    """
    if os.geteuid() == 0:
        pid_file_path = "/tmp/lshttpd/lshttpd.pid"
        return is_process_running(pid_file_path)
    else:
        if not os.path.exists(LITESPEED_STATUS_FOR_USERS):
            return False
        else:
            with open(LITESPEED_STATUS_FOR_USERS, encoding="utf-8") as f:
                status = f.read().strip()
            return status == "0"


def is_nginx_running() -> bool:
    """Check if nginx is running

    Returns:
        Bool: True or False
    """
    pid_file_path = "/run/nginx.pid"
    return is_process_running(pid_file_path)


def get_passenger_package_name():
    """
    Return proper passenger package according to apache version
    :rtype: str
    """
    if is_ea4():
        # Package that brings passenger on cPanel has different names on different systems:
        # CL6 - ea-ruby24-mod_passenger
        # CL7/8 - ea-ruby27-mod_passenger
        # CL9 - ea-apache24-mod-passenger
        # But they all provide "apache24-passenger" package
        return "apache24-passenger"
    return "alt-mod-passenger"


def is_package_installed(package_name):
    """
    Checks that package installed on server
    :param package_name: str
    :rtype: bool
    """
    try:
        if is_ubuntu():
            run_command(["dpkg", "-s", package_name])
        else:
            run_command(["rpm", "-q", package_name])
    except ExternalProgramFailed:
        return False
    return True


def get_package_db_errors():
    """
    Pick the package manager to check depending on the OS.
    If we're not on Ubuntu, assume a CL variant with RPM.

    :return: Error string, if any, None otherwise.
    """
    if is_ubuntu():
        return get_apt_db_errors()
    else:
        return get_rpm_db_errors()


def get_apt_db_errors():
    """
    Check the dpkg DB as described in https://man7.org/linux/man-pages/man1/dpkg.1.html
    See `--audit`.

    :return: Error string, if any, None otherwise.
    """
    try:
        with subprocess.Popen(
            ["dpkg", "--audit"],
            shell=False,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
        ) as proc:
            std_out, std_err = proc.communicate()
            # Should return 1 if the audit fails.
            if proc.returncode != 0:
                # Check error
                return f"dpkg audit error: {std_out}\n{std_err}."
    except (OSError, IOError,) as err:
        return str(err)
    # There is no dpkg DB errors
    return None


def get_rpm_db_errors():
    """
    Check RPM DB consistency.

    :return: None - No RPM DB errors
            string_message - Error description
    """
    doc_link = (
        "https://cloudlinux.zendesk.com/hc/en-us/articles/"
        "115004075294-Fix-rpmdb-Thread-died-in-Berkeley-DB-library"
    )
    try:
        with subprocess.Popen(
            ["/bin/rpm", "--dbpath", "/var/lib/rpm", "--verifydb"],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
        ) as proc:
            std_out, std_err = proc.communicate()
            if proc.returncode != 0:
                # Check error
                return f"RPM DB check error: {std_out}\n{std_err}.\nSee doc: {doc_link}"
    except (OSError, IOError,) as err:
        return str(err)
    # There is no RPM DB errors
    return None


def silence_stdout_until_process_exit():
    """
    Upon process exit, Sentry sometimes prints:

        Sentry is attempting to send 1 pending error messages
        Waiting up to 10 seconds
        Press Ctrl-C to quit

    This causes broken JSON in output.
    See also this issue: https://github.com/getsentry/raven-python/issues/904
    """
    sys.stdout = io.StringIO()
    sys.stderr = io.StringIO()


def mod_makedirs(path, mod):
    """
    Create directories with desired permissions

    Changed in version 3.7: The mode argument no longer affects
    the file permission bits of newly-created intermediate-level directories.
    Because it we use umask while creating dirs

    :param mod: desired permissions
    """
    inverted_mod = 0o777 - (mod & 0o777)
    with secureio.set_umask(inverted_mod):
        os.makedirs(path, mod)


def is_user_present(username: str) -> bool:
    """
    Check user existence in the system
    """
    try:
        pwd.getpwnam(username)
    except KeyError:
        return False
    return True


def is_uid_present(uid: int) -> bool:
    """
    Check uid existence in the system
    """
    try:
        pwd.getpwuid(uid)
    except KeyError:
        return False
    return True


def is_socket_file(path: AnyStr) -> Optional[bool]:
    """
    Check that file by path is socket
    """
    try:
        mode = os.lstat(path).st_mode
    except (FileNotFoundError, IOError, OSError):
        return None
    is_socket = stat.S_ISSOCK(mode)

    return is_socket


def get_system_runlevel() -> int:
    """
    Get number of system run level by command `runlevel`.
    """

    output = subprocess.check_output(
        "/sbin/runlevel",
        shell=True,
        executable="/bin/bash",
        text=True,
    )
    # output: N 5
    # there is N - previous value of runlevel, 5 is current runlevel
    result = output.strip().split()[1]
    # `S` means single-mode. Equals to level `1`
    level = 1 if result == "S" else int(result)
    return level


def _get_service_state_on_init_d_system(service_name: str) -> Tuple[bool, bool]:
    """
    Returns state of a service (present and enabled) for init.d system.
    Returns False, False if a service doesn't exist
    Returns True, False if a service exists and it's not enabled
    Returns True, True if a service exists and it's enabled
    """

    runlevel = get_system_runlevel()
    try:
        # LANG=C parameter allows to use C  programming language (en-US)
        # locale instead of current, since there can be non-English
        # results in the chkconfig output, while we search for the `on`
        output = subprocess.check_output(
            # the command return non-zero code if a service doesn't exist
            f"LANG=C /sbin/chkconfig --list {service_name}",
            shell=True,
            executable="/bin/bash",
            text=True,
        )
    except (subprocess.CalledProcessError, FileNotFoundError,):
        return False, False
    # split output:
    # `cl_plus_sender  0:off   1:off   2:on    3:on    4:on    5:on    6:off`
    output_list = output.strip().split()
    for state_info in output_list[1:]:
        state_list = state_info.strip().split(":")
        is_active = state_list[1] == "on"
        state_runlevel = int(state_list[0])
        if runlevel == state_runlevel:
            return True, is_active
    return True, False


def _get_service_state_on_systemd_system(service_name: str) -> Tuple[bool, bool]:
    """
    Returns state of service (present and enabled) for systemd system
    Returns False, False if a service doesn't exist
    Returns True, False if a service exists and it's not enabled
    Returns True, True if a service exists and it's enabled
    """
    try:
        subprocess.check_call(
            # the command return non-zero code if a service isn't enabled or
            # it's not present
            f"/usr/bin/systemctl is-enabled {service_name} &> /dev/null",
            shell=True,
            executable="/bin/bash",
        )
        return True, True
    except (subprocess.CalledProcessError, FileNotFoundError,):
        try:
            with subprocess.Popen(
                [
                    "/usr/bin/systemctl",
                    "status",
                    service_name,
                ],
                stderr=subprocess.DEVNULL,
                stdout=subprocess.DEVNULL,
            ) as proc:
                proc.communicate()
                # 0 - service is run
                # 3 - service is stopped
                # 4 - service doesn't exist
                if proc.returncode in (0, 3):
                    return True, False
                else:
                    return False, False
        except (subprocess.CalledProcessError, FileNotFoundError,):
            return False, False


def service_is_enabled_and_present(service_name: str) -> Tuple[bool, bool]:
    """
    Returns state of service (present and enabled)
    :param service_name: name of a service
    """
    cl_ver = get_cl_version()
    if cl_ver and "cl6" in cl_ver:
        is_present, is_enabled = _get_service_state_on_init_d_system(
            service_name=service_name,
        )
    else:
        is_present, is_enabled = _get_service_state_on_systemd_system(
            service_name=service_name,
        )
    return is_present, is_enabled


def process_is_running(process_file_path: str, strict_match: bool,) -> bool:
    """
    Check that a file in path is running.
    You can get false-postive if parameter `strict_match` == False, process is
    not running, but someone on server open file by path `process_file_path`
    in an editor
    :param process_file_path: path to a file which is run
    :param strict_match: we use parameter `process_file_path` as full cmd line
                         with args for comparing if `strict_match` == True.
    :return: True if it's running, False - is not,
    """
    if not os.path.exists(process_file_path):
        raise FileNotFoundError(
            f'Process file in path "{process_file_path}" does not exist'
        )
    for process in psutil.process_iter(["cmdline"]):
        try:
            if (
                not strict_match
                and process_file_path in process.cmdline()
            ) or (
                strict_match
                and process_file_path == ' '.join(process.cmdline())
            ):
                return True
        except psutil.NoSuchProcess:
            continue
    return False


def get_weekday(dt: Union[datetime.datetime, datetime.date]) -> str:
    """
    Getting string representation of weekday from datetime.datetime or datetime.date.
    Returns shortened version of weekday from WEEK_DAYS.
    """
    if not (isinstance(dt, (datetime.datetime, datetime.date))):
        raise TypeError(
            f"Require object of type datetime.datetime or datetime.date, but passed {type(dt)}"
        )
    return WEEK_DAYS[dt.weekday()]


def find_module_param_in_config(
        config_path: str,
        apache_module_name: str,
        param_name: str,
        default: int = None,
) -> Tuple[int, str]:
    """
    Helper to parse httpd config for details about mpm module used
    :param config_path: path for configuration file with modules
    :param apache_module_name: expected mpm module. Can be `event`,
        `worker`, `prefork`
    :param param_name: name of parameter to find
    :param default: default value for parameter, if there won't be record
    :return: tuple with param value and text result of operation

    Example of config file content:
    <IfModule mpm_prefork_module>
    .................
        MaxRequestWorkers   450
    </IfModule>
    --
    <IfModule mpm_worker_module>
    .................
        MaxRequestWorkers      300
    </IfModule>
    --
    <IfModule mpm_event_module>
    .................
        MaxRequestWorkers     2048
    </IfModule>
    """
    if_module_line = f"<IfModule mpm_{apache_module_name}_module>"
    section_lines = []
    mpm_lines = get_file_lines(config_path)
    is_section_found = False
    for line in mpm_lines:
        line = line.strip()
        if line == if_module_line:
            is_section_found = True
            continue
        if is_section_found and line == "</IfModule>":
            # End of section
            break
        if is_section_found:
            section_lines.append(line)
    # 2. Find directive in found section
    grep_result_list = list(
        grep(
            param_name,
            multiple_search=True,
            fixed_string=False,
            data_from_file=section_lines,
        )
    )
    mrw_list = [
        directive.strip()
        for directive in grep_result_list
        if directive.strip().startswith(param_name)
    ]
    # 3. Parse all lines with directive and find correct
    # There is no custom setting for parameter
    if not mrw_list and default is not None:
        # Plesk case, when we can use defaults
        return default, "OK"
    elif not mrw_list and default is None:
        # DA case, when we don't know about defaults
        return (
            0,
            f"MaxRequestWorkers directive not found for "
            f"mpm_{apache_module_name}_module module in {config_path}",
        )
    # There can be few records with MaxRequestWorkers, so we need
    # to take the last one
    parts = mrw_list[-1].split(" ")
    max_request_workers = int(parts[-1])
    return max_request_workers, "OK"


def get_kmodlve_module_version() -> str:
    """
    Return kmodlve module's version.
    Content of '/sys/module/kmodlve/version' looks like '2.0-30.el8'.
    """
    kmodlve_module_file_path = "/sys/module/kmodlve/version"
    if not os.path.isfile(kmodlve_module_file_path):
        return ""
    try:
        with open(kmodlve_module_file_path, "r", encoding="utf-8") as f:
            module = f.read().strip()
        module_version, _ = module.rsplit(".", 1)
    except (OSError, IOError, ValueError):
        return ""
    return module_version


def proc_can_see_other_uid_and_hidepid_synced() -> bool:
    """
    Find out if system has version of the kernel
    (according to kmodlve module's version)
    where fs.proc_can_see_other_uid and hidepid options are synchronized.
    They are only synchronized if kmodlve module's version
    is equal to version in synced_kmodlve_versions.
    """
    synced_kmodlve_versions = {"cl8": "2.0-30"}
    cl_version = get_cl_version()
    if cl_version not in synced_kmodlve_versions:
        return False
    synced_version = synced_kmodlve_versions[cl_version]
    current_version = get_kmodlve_module_version()
    if current_version == synced_version:
        return True
    return False


def get_process_pid(pid_filename: str) -> Optional[int]:
    """
    Detrmines working daemon process pid
    :param pid_filename: PID filename
    :return: PID from file or None if error (file not found, etc)
    """
    try:
        with open(pid_filename, "r", encoding="utf-8") as pf:
            pid = int(pf.read().strip())
            # check that pid is still running
            os.kill(pid, 0)
    except (OSError, IOError, ValueError):
        pid = None
    return pid


def write_pid_file(pid_filename: str):
    """
    Writes pid file
    """
    with open(pid_filename, "w+", encoding="utf-8") as pf:
        pf.write(f"{os.getpid()}")


def remove_pid_file(pid_filename: str):
    """
    Remove PID file
    """
    try:
        os.remove(pid_filename)
    except (OSError, IOError):
        pass


def demote(uid, gid):
    """
    Set user's real uid and gid to specified ones.
    Checking equality of real and effective uids is needed
    because this function may be used by root
    with effective uid dropped to user's uid.
    In that case it is needed to set effective uid back to 0 first.
    """

    def func():
        real_uid = os.getuid()
        eff_uid = os.geteuid()

        if real_uid != eff_uid:
            os.seteuid(real_uid)

        os.setgid(gid)
        os.setuid(uid)

    return func


def get_mount_point(dirpath: str) -> str:
    """
    Get mount point for dirpath directory from output of `df` utility
    """
    if not os.path.isdir(dirpath):
        raise OSError(f'Directory "{dirpath}" does not exist')

    fs_info_cmd = ['/bin/df', '-h', dirpath]
    fs_info = run_command(fs_info_cmd)
    # example stdout:
    # [root@localhost ~]  # df -h /home
    # Filesystem      Size  Used Avail Use% Mounted on
    # /dev/vda1        10G  3.1G  7.0G  31% /
    try:
        mountpoint_info = fs_info.split('\n')[1]
        mounted_on = mountpoint_info.split(' ')[-1]
    except IndexError as e:
        raise OSError(f'Utility "df" returned unexpected output:\n{fs_info}') from e

    return mounted_on


def is_may_detach_mounts_enabled() -> bool:
    """
    Detect if 'may_detach_mounts' kernel option is enabled.
    More info on the option: https://cloudlinux.atlassian.net/browse/KMODLVE-512
    """
    may_detach_mounts_file = "/proc/sys/fs/may_detach_mounts"

    if not os.path.isfile(may_detach_mounts_file):
        # Missing option means its enabled for all OS except CL6
        return get_cl_version() != "cl6"

    with open(may_detach_mounts_file, encoding="utf-8") as f:
        val = int(f.read())
        return val == 1


def get_filesystem_type(device: str) -> str:
    """
    Return the file system type for the given device.
    """
    partitions = psutil.disk_partitions()
    for partition in partitions:
        if partition.device == device:
            return partition.fstype
    return ''


def get_os_version() -> Tuple[Optional[str], Optional[str]]:
    """
    Detect system name and version
    :return: tuple (os_name, os_ver)
    """
    # # cat /etc/os-release | grep -E '^NAME|^VERSION_ID'
    # NAME="Ubuntu"
    # VERSION_ID="20.04"
    # # cat /etc/os-release | grep -E '^NAME|^VERSION_ID'
    # NAME="CloudLinux"
    # VERSION_ID="7.9"
    try:
        os_release_filename = '/etc/os-release'
        section_name = 'top'
        config = configparser.ConfigParser()
        # config.read('/etc/os-release')
        with open(os_release_filename, encoding='utf-8') as stream:
            config.read_string(f"[{section_name}]\n" + stream.read())
        os_name = config.get(section_name, 'NAME').strip('"')
        os_ver = config.get(section_name, 'VERSION_ID').strip('"')
        return os_name, os_ver
    except (OSError, IOError, configparser.Error):
        pass
    return None, None


def is_ubuntu() -> bool:
    """
    Detertmines is this system Ubuntu
    :return: bool flag is_ubuntu
    """
    os_name, _ = get_os_version()
    return os_name == 'Ubuntu'


def is_secureboot_enabled() -> bool:
    """
    Determines if secure boot is turned on
    :return: bool flag is_secureboot_enabled
    """
    enabled = False
    if os.path.exists('/sys/firmware/efi'):
        enabled = subprocess.call('mokutil --sb-state | grep enabled',
                                  shell=True, executable='/bin/bash',
                                  stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) == 0
    return enabled


def get_username():
    """
    Get username of current user.
    """
    try:
        return pwd.getpwuid(os.getuid())[0]
    except Exception:
        return None