disk.py

#! /usr/bin/env python3
# -*- coding: utf-8; py-indent-offset: 4 -*-
#
# Author:  Linuxfabrik GmbH, Zurich, Switzerland
# Contact: info (at) linuxfabrik (dot) ch
#          https://www.linuxfabrik.ch/
# License: The Unlicense, see LICENSE file.

# https://github.com/Linuxfabrik/lib/blob/main/CONTRIBUTING.md

"""Offers file and disk related functions, like getting a list of
partitions, grepping a file, etc.
"""

__author__ = 'Linuxfabrik GmbH, Zurich/Switzerland'
__version__ = '2026101203'

import csv
import errno
import glob as _glob
import hashlib
import os
import re
import shutil
import stat as _stat
import struct
import tempfile

from . import shell, txt

try:
    import pwd
except ImportError:
    # Not on Windows; resolve_trusted_path() refuses to decide there.
    pwd = None


def bd2dmd(device):
    """
    Retrieve the mapped device name for a device-mapper block device.

    This function reads the sysfs entry directly instead of using `dmsetup ls`, thus avoiding
    elevated privileges. ("bd2dmd" = block device to device-mapper device).

    Parameters
    ----------
    device : str
        The block device name or path (e.g., 'dm-0', '/dev/dm-0').

    Returns
    -------
    str
        The full path to the mapped device (e.g., '/dev/mapper/rl_rocky8-root'),
        or an empty string if not a device-mapper device.

    Examples
    --------
    >>> bd2dmd('dm-0')
    '/dev/mapper/rl_rocky8-root'
    >>> bd2dmd('sda')
    ''
    """
    device = os.path.basename(device)
    success, name = read_file(f'/sys/class/block/{device}/dm/name')

    if not success or not name:
        return ''

    mapped_device = f'/dev/mapper/{name.strip()}'
    return mapped_device if os.path.islink(mapped_device) else ''


def copy_dir(src, dst):
    """
    Recursively copy a directory tree.

    Wraps `shutil.copytree()` and reports the outcome in the same
    `(success, error)` style as the other disk helpers, so callers do not have to
    handle exceptions themselves.

    Parameters
    ----------
    src : str
        Source directory.
    dst : str
        Destination directory (must not exist yet).

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True if the copy succeeded, otherwise False.
          - tuple[1] (**None or str**): None on success, otherwise an error message.

    Examples
    --------
    >>> copy_dir('/usr/share/lynis', '/tmp/lynis')
    (True, None)
    """
    try:
        shutil.copytree(src, dst)
        return True, None
    except shutil.Error as e:
        # copytree() carries on after a failed file and raises one Error at the end,
        # holding a (src, dst, reason) tuple for every file, which can run to pages
        errors = e.args[0] if e.args and isinstance(e.args[0], list) else []
        if not errors:
            return False, f'Error copying directory {src} to {dst}: {e}'
        more = f' (and {len(errors) - 1} more)' if len(errors) > 1 else ''
        return False, f'Error copying directory {src} to {dst}: {errors[0][2]}{more}'
    except OSError as e:
        return False, f'Error copying directory {src} to {dst}: {e}'
    except Exception as e:
        return False, f'Unknown error copying directory {src} to {dst}: {e}'


def copy_file(src, dst):
    """
    Copy a single file, preserving its metadata.

    Wraps `shutil.copy2()` and reports the outcome in the same `(success, error)`
    style as the other disk helpers.

    Parameters
    ----------
    src : str
        Source file.
    dst : str
        Destination file or directory.

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True if the copy succeeded, otherwise False.
          - tuple[1] (**None or str**): None on success, otherwise an error message.

    Examples
    --------
    >>> copy_file('/etc/lynis/default.prf', '/tmp/lynis/default.prf')
    (True, None)
    """
    try:
        shutil.copy2(src, dst)
        return True, None
    except OSError as e:
        return False, f'OS error "{e.strerror}" while copying {src} to {dst}'
    except Exception as e:
        return False, f'Unknown error copying {src} to {dst}: {e}'


def dir_exists(path):
    """
    Check if a directory exists at the given path.

    Use this when you specifically want a directory check. `file_exists()`
    only returns `True` for regular files (it is `os.path.isfile()` under
    the hood), so passing a directory to it always returns `False`.

    Parameters
    ----------
    path : str
        The path to the directory to check.

    Returns
    -------
    bool
        `True` if the path exists and is a directory (or a symlink to one),
        `False` otherwise.

    Examples
    --------
    >>> dir_exists('/etc')
    True
    >>> dir_exists('/etc/passwd')
    False
    >>> dir_exists('/path/that/does/not/exist')
    False
    """
    return os.path.isdir(path)


def file_exists(path, allow_empty=False):
    """
    Check if a regular file exists at the given path, optionally allowing
    empty files.

    This wraps `os.path.isfile()`, so it only matches regular files and
    symlinks pointing at regular files. Directories return `False`; use
    `dir_exists()` for directory checks.

    Parameters
    ----------
    path : str
        The path to the file to check.
    allow_empty : bool, optional
        If True, consider empty files as existing.
        If False, empty files are treated as non-existent. Defaults to False.

    Returns
    -------
    bool
        True if the file exists (and is non-empty unless `allow_empty` is True), otherwise False.

    Examples
    --------
    >>> file_exists('/path/to/file')
    True
    >>> file_exists('/path/to/empty_file', allow_empty=False)
    False
    >>> file_exists('/path/to/empty_file', allow_empty=True)
    True
    """
    if not os.path.isfile(path):
        return False

    if allow_empty:
        return True

    return os.path.getsize(path) > 0


# /proc/mounts escapes a space, tab, newline, backslash and `#` in the device and the
# mount point as a backslash and three octal digits (fs/proc_namespace.c, mangle(), and
# show_vfsmnt(); Linux 7.3).
_MOUNT_ESCAPE = re.compile(r'\\([0-7]{3})')


def _get_block_mounts():
    """
    Return the mounts of `/proc/mounts` whose source is below `/dev/`, as a list of
    `(device, mount point, mount point as listed)`, the first two with the escapes
    decoded, or an empty list where there is no `/proc/mounts`.
    """
    success, content = read_file('/proc/mounts')
    if not success:
        return []
    mounts = []
    for line in content.splitlines():
        parts = line.split()
        if len(parts) < 2 or not parts[0].startswith('/dev/'):
            continue
        device, mount_point = (
            _MOUNT_ESCAPE.sub(lambda m: chr(int(m.group(1), 8)), part)
            for part in parts[:2]
        )
        mounts.append((device, mount_point, parts[1]))
    return mounts


def _get_kernel_device(device):
    """
    Return the kernel's name for a block device node, such as `/dev/dm-0` for
    `/dev/mapper/rl-root` or a `/dev/disk/by-uuid/` link, from its device number in
    `/sys/dev/block/`. Needs no udev. Returns `device` unchanged where the node or sysfs
    is missing, for example in a container.
    """
    try:
        st = os.stat(device)
    except (OSError, ValueError):
        return device
    if not _stat.S_ISBLK(st.st_mode):
        return device
    link = f'/sys/dev/block/{os.major(st.st_rdev)}:{os.minor(st.st_rdev)}'
    if not os.path.exists(link):
        return device
    return f'/dev/{os.path.basename(os.path.realpath(link))}'


def get_block_devices():
    """
    Return all local block devices that expose I/O counters, mounted or not.

    Unlike `get_real_disks()`, which is limited to block devices that currently have a mounted
    filesystem, this also includes devices without a mounted filesystem, for example raw devices
    backing a database or storage layer, or unmounted multipath/SAN volumes. Devices are
    enumerated from `/proc/diskstats`, so their names line up with the per-device I/O counters
    exposed there.

    Each device is represented as a dictionary with:
    - 'bd' : Block device path (e.g. '/dev/sda' or '/dev/dm-7').
    - 'dmd': Device-mapper path if the device is a device-mapper target (e.g.
      '/dev/mapper/data'), otherwise an empty string.
    - 'mount_points': List of the mount points, with escaped characters such as spaces
      decoded, empty if the device is not mounted.
    - 'mp' : Mount point(s) as `/proc/mounts` lists them, escapes included, space-separated
      if mounted in several places, or an empty string if the device is not mounted. Kept
      for compatibility, use 'mount_points' instead.

    Pseudo devices that never carry meaningful I/O are skipped by name prefix: loopback (`loop`),
    RAM disks (`ram`), compressed RAM (`zram`), floppy (`fd`) and optical (`sr`) devices.

    Returns
    -------
    list of dict
        One entry per block device, including unmounted ones. Empty list on
        systems without `/proc/diskstats` (e.g. non-Linux).

    Examples
    --------
    >>> get_block_devices()
    [{'bd': '/dev/dm-7', 'dmd': '/dev/mapper/data', 'mount_points': [], 'mp': ''},
     {'bd': '/dev/sda1', 'dmd': '', 'mount_points': ['/boot'], 'mp': '/boot'}]
    """
    success, diskstats = read_file('/proc/diskstats')
    if not success:
        return []

    # map every mounted block device to its mount points
    mountpoints = {}
    for device, mount_point, listed in _get_block_mounts():
        mountpoints.setdefault(_get_kernel_device(device), []).append(
            (mount_point, listed)
        )

    disks = []
    for line in diskstats.splitlines():
        parts = line.split()
        if len(parts) < 3:
            continue
        name = parts[2]
        if name.startswith(_PSEUDO_DEVICE_PREFIXES):
            continue
        bd = f'/dev/{name}'
        dmd = bd2dmd(name)
        # where sysfs could not resolve a mount, it stays under its device-mapper path
        mounts = mountpoints.get(bd) or (mountpoints.get(dmd, []) if dmd else [])
        disks.append(
            {
                'bd': bd,
                'dmd': dmd,
                'mount_points': [mount_point for mount_point, _ in mounts],
                'mp': ' '.join(listed for _, listed in mounts),
            }
        )

    return disks


def _fingerprint_handle(f, fingerprint, length):
    """Hash the slice of an open binary file that `get_fingerprint()` asks for."""
    if length == 0:
        hashed = 0
        # Read in chunks: the file is hashed as a whole, but never held as a whole
        # in memory.
        for chunk in iter(lambda: f.read(65536), b''):
            fingerprint.update(chunk)
            hashed += len(chunk)
        return True, (fingerprint.hexdigest(), hashed)
    if length < 0:
        # Seek to the start of the tail. Clamped to 0, so a file shorter than the
        # tail is read from its beginning instead of raising.
        f.seek(max(0, os.fstat(f.fileno()).st_size + length))
    data = f.read(abs(length))
    fingerprint.update(data)
    return True, (fingerprint.hexdigest(), len(data))


def get_fingerprint(filename, length=256, algorithm='sha256'):
    """
    Hash a slice of a file, to recognize the file by its content instead of by its metadata.

    Metadata answers few questions about content. Two files of the same size are not
    necessarily the same, and a file that was rewritten in place keeps its inode and can end
    up with the very same size, so nothing the filesystem reports has changed although
    everything in the file has. A hash over the content answers those questions, and hashing
    a fixed-size slice of it keeps the cost independent of the file size.

    Which slice to take depends on the question:

    - The **head** is the part that stays as it is while a file is appended to, which makes it
      an identity marker: it changes when the file has become a different one (rewritten,
      replaced, truncated), not when something was added to it. Use it to tell "still the same
      file I was reading" from "a new file under the same name".
    - The **tail** changes with every write. Use it to tell whether anything was added since
      last time, not to identify the file.
    - The **whole file** is the exact answer to "is this byte-for-byte the same content", and
      the only one that also catches a change in the middle. It costs a full read.

    Compare a fingerprint only against one taken over a slice of the same size and side,
    otherwise a file that has merely grown looks like a different one. Pass the returned byte
    count back as `length` for that, and treat a count below the one requested as "the file no
    longer holds that many bytes", i.e. it was truncated.

    Parameters
    ----------
    filename : str or binary file object
        Path to the file to fingerprint, or a file already opened in binary mode.
        Pass the handle when the file has to be the very one that is read afterwards
        (see `open_file()`); it is not closed, and its position is not restored.
    length : int, optional
        How many bytes to hash, and from which side:

        - `> 0`: the first `length` bytes (the head). Defaults to 256, which is enough to tell
          two lines of text apart.
        - `< 0`: the last `abs(length)` bytes (the tail).
        - `0`: the whole file, read in chunks.
    algorithm : str, optional
        Name of the hash algorithm, as accepted by `hashlib.new()`, for example `'md5'`,
        `'sha1'`, `'sha256'` or `'sha512'`. Defaults to `'sha256'`.

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True if reading succeeded, otherwise False.
          - tuple[1] (**tuple | str**):

            - If successful, a `(fingerprint, hashed)` tuple: the hexdigest of the slice, and
              the number of bytes it was taken over. `hashed` is less than the requested number
              of bytes if the file is shorter than that.
            - If unsuccessful, an error message string.

    Notes
    -----
    - A file shorter than the requested slice is hashed as a whole, so head, tail and whole
      file yield the same fingerprint for it.
    - Pick `algorithm` to match whatever the fingerprint is compared against. Where the
      comparison is against a digest somebody else published, the publisher decides; where it
      is against a digest taken by an earlier call of this function, the default is the right
      answer. `'md5'` and `'sha1'` are not collision resistant and say nothing about a file an
      attacker was able to choose the content of, so use them only to reproduce a foreign
      digest, never to decide on your own that two files are the same.
    - The algorithm is requested from OpenSSL through `hashlib.new()`. A host in FIPS mode
      refuses MD5 there, which surfaces as a failure of this function rather than as a wrong
      answer.

    Examples
    --------
    >>> success, (fingerprint, hashed) = get_fingerprint('/var/log/messages')
    >>> success, (fingerprint, hashed) = get_fingerprint('/tmp/export', length=-4096)
    >>> success, (fingerprint, hashed) = get_fingerprint('/tmp/export', length=0)
    >>> success, (fingerprint, hashed) = get_fingerprint(
    ...     '/tmp/export', length=0, algorithm='md5'
    ... )
    """
    try:
        # usedforsecurity=False keeps a weak algorithm available on a host that
        # enforces FIPS, which is correct here: the digest is only ever compared
        # against one somebody else published, it is not a security decision of
        # our own. Unknown to Python 3.8 and older, hence the fallback.
        try:
            fingerprint = hashlib.new(algorithm, usedforsecurity=False)
        except TypeError:
            fingerprint = hashlib.new(algorithm)
    except ValueError as e:
        return False, f'Unsupported hash algorithm "{algorithm}": {e}'
    try:
        if hasattr(filename, 'read'):
            filename.seek(0)
            return _fingerprint_handle(filename, fingerprint, length)
        success, handle = open_file(filename, binary=True)
        if not success:
            return False, handle
        with handle as f:
            return _fingerprint_handle(f, fingerprint, length)
    except OSError as e:
        return False, f'I/O error "{e.strerror}" while opening or reading {filename}'
    except Exception as e:
        return False, f'Unknown error opening or reading {filename}: {e}'


def get_inode_usage(mount):
    """
    Return the inode usage of the filesystem mounted at `mount`, via `os.statvfs()`.

    Inodes are a Unix filesystem concept, so this relies on `os.statvfs()`, which does not
    exist on Windows; there the call reports failure instead of raising. Some filesystems
    (btrfs, FAT, many network mounts) allocate inodes dynamically and report a total of
    zero. Such a filesystem has no meaningful inode-usage percentage, which is signalled
    with a `None` result, distinct from a read failure.

    Parameters
    ----------
    mount : str
        Mount point to inspect, e.g. `/` or `/boot`.

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True if the mount point could be read, otherwise False (for
            example permission denied, an I/O error, or a platform without `os.statvfs()`).
          - tuple[1] (**dict | None | str**):

            - On success, a dict with the inode counts `total`, `free` and `used`, plus
              `percent` (used inodes in percent, rounded to one decimal); or `None` if the
              filesystem does not report inodes (total is zero).
            - On failure, an error message string.

    Examples
    --------
    >>> get_inode_usage('/boot')
    (True, {'total': 65536, 'free': 65427, 'used': 109, 'percent': 0.2})
    >>> get_inode_usage('/')  # btrfs, no fixed inode count
    (True, None)
    """
    if not hasattr(os, 'statvfs'):
        return False, 'os.statvfs() is not available on this platform'
    try:
        st = os.statvfs(mount)
    except OSError as e:
        return False, f'I/O error "{e.strerror}" while reading inodes of {mount}'
    total = st.f_files
    if not total:
        return True, None
    free = st.f_ffree
    used = total - free
    return True, {
        'total': total,
        'free': free,
        'used': used,
        'percent': round(used / total * 100, 1),
    }


def get_owner(file):
    """
    Get the numeric user ID (UID) of the owner of a filesystem entry.

    Parameters
    ----------
    file : str | os.PathLike
        Path to the file or directory whose owner UID should be retrieved.

    Returns
    -------
    int
        The owner's UID if available; `-1` if the call fails or ownership cannot be determined.

    Notes
    -----
    - This function is POSIX-oriented. On Windows, `st_uid` may be `0` for all files
      and not reflect the real owner. If you need the account name on Windows,
      consider platform-specific APIs (e.g., `win32security`).
    - All exceptions are caught and result in `-1`.

    Examples
    --------
    >>> get_owner('/etc/passwd')  # doctest: +SKIP (system-dependent)
    0
    >>> get_owner('/path/does/not/exist')
    -1
    """
    try:
        return os.stat(file).st_uid
    except OSError:
        return -1


def get_package(path):
    """
    Return the package a path belongs to, or the empty string when none does.

    Tells software the distribution installed apart from software an administrator
    put there by hand, which is what decides whether a version is the package
    manager's business or the operator's.

    Asks `rpm` or `dpkg`, whichever the host has. Both answer an unowned path with
    a non-zero return code, so "no package" and "the lookup failed" are the same
    answer here on purpose: neither means the path belongs to a package. A host
    with neither package manager also yields the empty string.

    Parameters
    ----------
    path : str
        The path to ask about, for example
        `/usr/share/icingaweb2/modules/director`.

    Returns
    -------
    str
        The package name, or the empty string.

    Notes
    -----
    - The path is refused when a package manager could read it as one of its own
      options. `rpm --query --file --version` prints the rpm version and succeeds,
      so an unguarded lookup on a path named that way would report the tool's
      version string as a package name. Verified against rpm 6.0.2 on Fedora 43.

    Examples
    --------
    >>> get_package('/usr/share/icingaweb2/modules/director')
    'icinga-director-php-1.11.9-1.fc43.noarch'

    >>> get_package('/usr/share/icingaweb2/modules/reporting')
    ''
    """
    if not path:
        return ''
    success, _ = shell.safe_cli_value(path, 'path')
    if not success:
        return ''

    if shell.which('rpm'):
        cmd = ['rpm', '--query', '--file', path]
    elif shell.which('dpkg'):
        cmd = ['dpkg', '--search', path]
    else:
        return ''

    success, result = shell.shell_exec(cmd)
    if not success:
        return ''
    stdout, _stderr, retc = result
    if retc != 0 or not stdout.strip():
        return ''

    # rpm answers with the package name alone, dpkg with "package: /path".
    return stdout.splitlines()[0].split(':')[0].strip()


# Who may own and access a private directory besides the account itself: SYSTEM,
# Administrators, and the placeholders CREATOR OWNER and OWNER RIGHTS, which stand for the
# owner and grant nobody else anything. Python 3.13 creates `os.mkdir(mode=0o700)` and
# `tempfile.mkdtemp()` directories with exactly SYSTEM, Administrators and OWNER RIGHTS.
_WINDOWS_TRUSTED_SIDS = frozenset(('S-1-3-0', 'S-1-3-4', 'S-1-5-18', 'S-1-5-32-544'))
# TrustedInstaller owns system directories such as `C:\\`, which a temporary directory can
# fall back to
_WINDOWS_TRUSTED_OWNERS = frozenset(
    ('S-1-5-80-956008885-3418522649-1831038044-1853292631-2271478464',)
)

_windows_api_cache = {}


def _windows_api():
    """Load and type the Win32 security functions `get_private_dir()` needs, once."""
    if _windows_api_cache:
        return _windows_api_cache
    # Windows-only, and not worth its import time anywhere else
    import ctypes
    from ctypes import wintypes

    class ACL_SIZE_INFORMATION(ctypes.Structure):
        """ACL_SIZE_INFORMATION from winnt.h."""

        _fields_ = (
            ('AceCount', wintypes.DWORD),
            ('AclBytesInUse', wintypes.DWORD),
            ('AclBytesFree', wintypes.DWORD),
        )

    class ACE_HEADER(ctypes.Structure):
        """ACE_HEADER from winnt.h."""

        _fields_ = (
            ('AceType', wintypes.BYTE),
            ('AceFlags', wintypes.BYTE),
            ('AceSize', wintypes.WORD),
        )

    class SECURITY_ATTRIBUTES(ctypes.Structure):
        """SECURITY_ATTRIBUTES from minwinbase.h."""

        _fields_ = (
            ('nLength', wintypes.DWORD),
            ('lpSecurityDescriptor', wintypes.LPVOID),
            ('bInheritHandle', wintypes.BOOL),
        )

    advapi32 = ctypes.WinDLL('advapi32', use_last_error=True)
    kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
    pointer = ctypes.POINTER
    signatures = (
        (kernel32.CloseHandle, (wintypes.HANDLE,), wintypes.BOOL),
        (
            kernel32.CreateDirectoryW,
            (wintypes.LPCWSTR, pointer(SECURITY_ATTRIBUTES)),
            wintypes.BOOL,
        ),
        (kernel32.GetCurrentProcess, (), wintypes.HANDLE),
        (kernel32.LocalFree, (wintypes.LPVOID,), wintypes.LPVOID),
        (
            advapi32.ConvertSidToStringSidW,
            (wintypes.LPVOID, pointer(wintypes.LPWSTR)),
            wintypes.BOOL,
        ),
        (
            advapi32.ConvertStringSecurityDescriptorToSecurityDescriptorW,
            (
                wintypes.LPCWSTR,
                wintypes.DWORD,
                pointer(wintypes.LPVOID),
                pointer(wintypes.ULONG),
            ),
            wintypes.BOOL,
        ),
        (
            advapi32.GetAce,
            (wintypes.LPVOID, wintypes.DWORD, pointer(wintypes.LPVOID)),
            wintypes.BOOL,
        ),
        (
            advapi32.GetAclInformation,
            (wintypes.LPVOID, wintypes.LPVOID, wintypes.DWORD, ctypes.c_int),
            wintypes.BOOL,
        ),
        (
            advapi32.GetNamedSecurityInfoW,
            (
                wintypes.LPCWSTR,
                ctypes.c_int,
                wintypes.DWORD,
                pointer(wintypes.LPVOID),
                pointer(wintypes.LPVOID),
                pointer(wintypes.LPVOID),
                pointer(wintypes.LPVOID),
                pointer(wintypes.LPVOID),
            ),
            wintypes.DWORD,
        ),
        (
            advapi32.GetTokenInformation,
            (
                wintypes.HANDLE,
                ctypes.c_int,
                wintypes.LPVOID,
                wintypes.DWORD,
                pointer(wintypes.DWORD),
            ),
            wintypes.BOOL,
        ),
        (
            advapi32.OpenProcessToken,
            (wintypes.HANDLE, wintypes.DWORD, pointer(wintypes.HANDLE)),
            wintypes.BOOL,
        ),
    )
    for function, argtypes, restype in signatures:
        function.argtypes = argtypes
        function.restype = restype
    _windows_api_cache.update(
        ACE_HEADER=ACE_HEADER,
        ACL_SIZE_INFORMATION=ACL_SIZE_INFORMATION,
        SECURITY_ATTRIBUTES=SECURITY_ATTRIBUTES,
        advapi32=advapi32,
        ctypes=ctypes,
        kernel32=kernel32,
        wintypes=wintypes,
    )
    return _windows_api_cache


def _windows_sid_string(psid):
    """Return the SID at `psid` in its string form, `S-1-5-18` for example."""
    api = _windows_api()
    ctypes = api['ctypes']
    string_sid = api['wintypes'].LPWSTR()
    if not api['advapi32'].ConvertSidToStringSidW(psid, ctypes.byref(string_sid)):
        raise ctypes.WinError(ctypes.get_last_error())
    try:
        return string_sid.value
    finally:
        api['kernel32'].LocalFree(ctypes.cast(string_sid, ctypes.c_void_p))


def _windows_current_sid():
    """Return the SID of the account the process runs as."""
    api = _windows_api()
    ctypes = api['ctypes']
    wintypes = api['wintypes']
    token = wintypes.HANDLE()
    # 0x0008 is TOKEN_QUERY
    if not api['advapi32'].OpenProcessToken(
        api['kernel32'].GetCurrentProcess(), 0x0008, ctypes.byref(token)
    ):
        raise ctypes.WinError(ctypes.get_last_error())
    try:
        size = wintypes.DWORD(0)
        # 1 is TokenUser; the first call only asks for the size
        api['advapi32'].GetTokenInformation(token, 1, None, 0, ctypes.byref(size))
        buffer = ctypes.create_string_buffer(size.value)
        if not api['advapi32'].GetTokenInformation(
            token, 1, buffer, size, ctypes.byref(size)
        ):
            raise ctypes.WinError(ctypes.get_last_error())
        # TOKEN_USER starts with the pointer to the SID
        return _windows_sid_string(
            ctypes.cast(buffer, ctypes.POINTER(ctypes.c_void_p))[0]
        )
    finally:
        api['kernel32'].CloseHandle(token)


def _windows_security(path):
    """Return the owner SID and the DACL entries of `path`.

    The entries are `(type, flags, mask, sid)` tuples, or None for a missing DACL, which
    grants everybody everything. Raises OSError if the security descriptor cannot be read.
    """
    api = _windows_api()
    ctypes = api['ctypes']
    wintypes = api['wintypes']
    owner = wintypes.LPVOID()
    dacl = wintypes.LPVOID()
    descriptor = wintypes.LPVOID()
    # 1 is SE_FILE_OBJECT, 0x1 | 0x4 are OWNER_ and DACL_SECURITY_INFORMATION
    error = api['advapi32'].GetNamedSecurityInfoW(
        path,
        1,
        0x1 | 0x4,
        ctypes.byref(owner),
        None,
        ctypes.byref(dacl),
        None,
        ctypes.byref(descriptor),
    )
    if error:
        raise ctypes.WinError(error)
    try:
        owner_sid = _windows_sid_string(owner)
        if not dacl.value:
            return owner_sid, None
        info = api['ACL_SIZE_INFORMATION']()
        # 2 is AclSizeInformation
        if not api['advapi32'].GetAclInformation(
            dacl, ctypes.byref(info), ctypes.sizeof(info), 2
        ):
            raise ctypes.WinError(ctypes.get_last_error())
        entries = []
        for index in range(info.AceCount):
            ace = wintypes.LPVOID()
            if not api['advapi32'].GetAce(dacl, index, ctypes.byref(ace)):
                raise ctypes.WinError(ctypes.get_last_error())
            header = api['ACE_HEADER'].from_address(ace.value)
            # ACCESS_ALLOWED_ACE and ACCESS_DENIED_ACE: the 4-byte header, the 4-byte
            # access mask, then the SID. Other types are reported without mask and SID.
            if header.AceType in (0, 1):
                mask = wintypes.DWORD.from_address(ace.value + 4).value
                sid = _windows_sid_string(ace.value + 8)
                entries.append((header.AceType, header.AceFlags, mask, sid))
            else:
                entries.append((header.AceType, header.AceFlags, 0, None))
        return owner_sid, entries
    finally:
        api['kernel32'].LocalFree(descriptor)


def _windows_reparse_problem(path):
    """Return why `path` cannot be a directory to rely on, or None."""
    try:
        st = os.lstat(path)
    except OSError as e:
        return f'cannot be inspected ({e.strerror})'
    # 0x400 is FILE_ATTRIBUTE_REPARSE_POINT: a symlink or a junction
    if getattr(st, 'st_file_attributes', 0) & 0x400:
        return 'is a symlink or junction'
    if not _stat.S_ISDIR(st.st_mode):
        return 'is not a directory'
    return None


def _windows_private_problem(path, sid):
    """Return what keeps `path` from being private to `sid`, or None if it is private.

    Private means: a directory and not a symlink or junction, owned by `sid` or one of
    `_WINDOWS_TRUSTED_SIDS`, and a DACL that grants access to nobody else.
    """
    problem = _windows_reparse_problem(path)
    if problem:
        return problem
    try:
        owner_sid, entries = _windows_security(path)
    except OSError as e:
        return f'has a security descriptor that cannot be read ({e})'
    trusted = _WINDOWS_TRUSTED_SIDS | {sid}
    if owner_sid not in trusted:
        return 'has the wrong owner'
    if entries is None:
        return 'is too permissive'
    for ace_type, _flags, _mask, ace_sid in entries:
        # 1 is ACCESS_DENIED_ACE_TYPE: taking rights away is never a problem
        if ace_type == 1:
            continue
        # 0 is ACCESS_ALLOWED_ACE_TYPE, any other kind of grant is not expected here
        if ace_type != 0 or ace_sid not in trusted:
            return 'is too permissive'
    return None


def _windows_parent_problem(path, sid):
    """Return why the private directory must not be created below `path`, or None.

    Whoever may delete or rename the entries of `path`, or take it over, can swap the
    private directory for one of their own after it was checked. `C:\\Windows\\Temp` lets
    every user create entries, but not delete those of others, and passes.
    """
    problem = _windows_reparse_problem(path)
    if problem:
        return problem
    try:
        owner_sid, entries = _windows_security(path)
    except OSError as e:
        return f'has a security descriptor that cannot be read ({e})'
    trusted = _WINDOWS_TRUSTED_SIDS | _WINDOWS_TRUSTED_OWNERS | {sid}
    if owner_sid not in trusted:
        return 'belongs to another account'
    if entries is None:
        return 'lets every account replace its entries'
    for ace_type, flags, mask, ace_sid in entries:
        # 0x08 is INHERIT_ONLY_ACE: it applies to what is created inside, not to `path`
        if ace_type == 1 or flags & 0x08 or ace_sid in trusted:
            continue
        # FILE_DELETE_CHILD, WRITE_DAC, WRITE_OWNER and GENERIC_ALL
        if ace_type != 0 or mask & (0x40 | 0x40000 | 0x80000 | 0x10000000):
            return 'lets other accounts replace its entries'
    return None


def _windows_make_private_dir(path, sid):
    """Create `path` with a DACL that admits `sid`, SYSTEM and Administrators only.

    An existing `path` is left alone, the caller validates it either way.
    """
    api = _windows_api()
    ctypes = api['ctypes']
    wintypes = api['wintypes']
    descriptor = wintypes.LPVOID()
    # D:P protects the DACL from inheriting the entries of the parent
    sddl = f'D:P(A;OICI;FA;;;SY)(A;OICI;FA;;;BA)(A;OICI;FA;;;{sid})'
    # 1 is SDDL_REVISION_1
    if not api['advapi32'].ConvertStringSecurityDescriptorToSecurityDescriptorW(
        sddl, 1, ctypes.byref(descriptor), None
    ):
        raise ctypes.WinError(ctypes.get_last_error())
    try:
        attributes = api['SECURITY_ATTRIBUTES'](
            ctypes.sizeof(api['SECURITY_ATTRIBUTES']), descriptor, False
        )
        if not api['kernel32'].CreateDirectoryW(path, ctypes.byref(attributes)):
            error = ctypes.get_last_error()
            # 183 is ERROR_ALREADY_EXISTS
            if error != 183:
                raise ctypes.WinError(error)
    finally:
        api['kernel32'].LocalFree(descriptor)


def get_private_dir(path, prefix='linuxfabrik-monitoring-plugins'):
    """
    Return a directory below `path` that only the current account can use, creating it if
    necessary.

    Files at predictable names in a shared temporary directory can be planted by another
    local user: a symlink on POSIX that redirects a write to a file of the attacker's choice,
    a file on Windows that a process running as SYSTEM then trusts (CWE-377,
    GHSA-r35r-fpx2-jgr4). Such files belong in a directory nobody else can write to.

    - POSIX: always the subdirectory `<prefix>-uid<euid>`, created with `0700`, and refused
      if it is a symlink, belongs to someone else or grants group or other permissions.
      `path` itself has to belong to root or the account, and if group or others may write
      to it, it needs the sticky bit, as `/tmp` has.
    - Windows: `path` itself if it is private already, which the temporary directory of an
      ordinary account or of NetworkService is. Otherwise, as in `C:\\Windows\\Temp`, which
      SYSTEM uses and where every user may create files, the subdirectory `<prefix>-<SID>`,
      created with a DACL that admits the account, SYSTEM and Administrators only. Private
      means: not a symlink or junction, owned by the account, SYSTEM or Administrators, and a
      DACL that grants access to nobody else. An existing subdirectory that is not private is
      refused, and so is a `path` that is a symlink or junction, belongs to another account,
      or lets another account delete or rename its entries, since the subdirectory could be
      swapped there. The SID is part of the name because several accounts can share one
      temporary directory.

    Parameters
    ----------
    path : str
        The base directory, typically the system temporary directory.
    prefix : str, optional
        Start of the subdirectory's name. Defaults to `'linuxfabrik-monitoring-plugins'`.

    Returns
    -------
    tuple (bool, str)
        - First element (`bool`): `True` on success, `False` on failure.
        - Second element (`str`): The directory on success, an error message otherwise.

    Examples
    --------
    >>> get_private_dir('/tmp')
    (True, '/tmp/linuxfabrik-monitoring-plugins-uid1000')

    >>> get_private_dir('C:\\\\Windows\\\\Temp')
    (True, 'C:\\\\Windows\\\\Temp\\\\linuxfabrik-monitoring-plugins-S-1-5-18')
    """
    if os.name == 'nt':
        try:
            sid = _windows_current_sid()
            if _windows_private_problem(path, sid) is None:
                return True, path
            problem = _windows_parent_problem(path, sid)
            if problem:
                return False, f'Directory {path} {problem}, refusing to use it'
            private_dir = os.path.join(path, f'{prefix}-{sid}')
            _windows_make_private_dir(private_dir, sid)
            problem = _windows_private_problem(private_dir, sid)
        except OSError as e:
            return (
                False,
                f'Creating the private directory below {path} failed, Error: {e}',
            )
        if problem:
            return False, f'Directory {private_dir} {problem}, refusing to use it'
        return True, private_dir

    euid = os.geteuid()
    # Whoever may rename the entries of `path` can swap the private directory for one of
    # their own after it was checked. /tmp lets everybody create entries, but its sticky bit
    # keeps them from renaming or deleting those of others. stat() follows a symlink, so the
    # directory it leads to is the one judged.
    try:
        base = os.stat(path)
    except OSError as e:
        return False, f'Inspecting directory {path} failed, Error: {e}'
    if base.st_uid not in (0, euid):
        return False, f'Directory {path} belongs to another account, refusing to use it'
    if base.st_mode & 0o022 and not base.st_mode & _stat.S_ISVTX:
        return False, (
            f'Directory {path} lets other accounts replace its entries, refusing to use it'
        )
    private_dir = os.path.join(path, f'{prefix}-uid{euid}')
    # Reject a pre-existing symlink outright: makedirs(exist_ok=True) would either follow it
    # (when it resolves to a directory) or fail with a confusing "File exists" (when it dangles).
    # Either way it must not be used. os.path.islink() does not follow the link.
    if os.path.islink(private_dir):
        return False, f'Directory {private_dir} is a symlink, refusing to use it'
    try:
        # 0o700: only the owner may access the directory. An existing directory is fine and gets
        # validated below; any other error (e.g. an unwritable temp dir) aborts.
        os.makedirs(private_dir, mode=0o700, exist_ok=True)
    except OSError as e:
        return False, f'Creating directory {private_dir} failed, Error: {e}'
    # lstat() does not follow symlinks, so a symlink planted at private_dir is caught here
    # instead of silently redirecting every file to the attacker's target.
    try:
        st = os.lstat(private_dir)
    except OSError as e:
        return False, f'Inspecting directory {private_dir} failed, Error: {e}'
    if not _stat.S_ISDIR(st.st_mode):
        return False, f'Directory {private_dir} is not a directory, refusing to use it'
    if st.st_uid != euid:
        return False, f'Directory {private_dir} has the wrong owner, refusing to use it'
    if st.st_mode & 0o077:
        return False, f'Directory {private_dir} is too permissive, refusing to use it'
    return True, private_dir


def get_real_disks():
    """
    Return a list of real local block devices that are mounted and have a filesystem.

    Each device is represented as a dictionary with:
    - 'bd': Block device name (e.g., '/dev/sda1' or '/dev/dm-0').
    - 'dmd': Device-mapper name if available (e.g., '/dev/mapper/rl-root'), otherwise an
      empty string.
    - 'mount_points': List of the mount points, with escaped characters such as spaces
      decoded.
    - 'mp' : Mount point(s) as `/proc/mounts` lists them, escapes included, space-separated
      if mounted in multiple places. Kept for compatibility, use 'mount_points' instead.

    Devices are discovered by parsing /proc/mounts, and mapped to the kernel's device name
    and the device-mapper name through sysfs, so neither udev nor udevadm is needed. A
    device whose node or sysfs entry is missing (as in a container) keeps the path it is
    mounted from. Devices under /dev/loop* (loopback devices) are ignored.

    Returns
    -------
    list of dict
        List of mounted devices and their details.

    Examples
    --------
    >>> get_real_disks()
    [{'bd': '/dev/dm-0', 'dmd': '/dev/mapper/rl-root', 'mount_points': ['/', '/home'],
      'mp': '/ /home'}]
    """
    disks = {}
    for device, mount_point, listed in _get_block_mounts():
        bd = _get_kernel_device(device)
        if device.startswith('/dev/loop') or bd.startswith('/dev/loop'):
            continue
        if bd not in disks:
            dmd = bd2dmd(bd)
            if not dmd and device.startswith('/dev/mapper/'):
                dmd = device
            disks[bd] = {'bd': bd, 'dmd': dmd, 'mount_points': [], 'mp': listed}
        else:
            disks[bd]['mp'] += f' {listed}'
        disks[bd]['mount_points'].append(mount_point)
    return list(disks.values())


def get_tmpdir():
    """
    Return the absolute path of the directory used for temporary files, without a trailing '/'.

    Thin wrapper around `tempfile.gettempdir()`, which picks a usable temporary directory by
    trying to create, write and delete a test file in each candidate and returning the first one
    that works. The candidates are tried in this order:

    - The directories named by the `TMPDIR`, `TEMP` and `TMP` environment variables.
    - Platform-specific defaults: on Windows `~\\AppData\\Local\\Temp`, `%SYSTEMROOT%\\Temp`,
      `c:\\temp`, `c:\\tmp`, `\\temp`, `\\tmp`; on other systems `/tmp`, `/var/tmp`, `/usr/tmp`.
    - As a last resort, the current working directory.

    The literal `/tmp` is only returned as a fallback if `tempfile.gettempdir()` itself raises.

    Returns
    -------
    str
        The absolute path to the temporary directory.

    Notes
    -----
    - `tempfile.gettempdir()` computes the result once and caches it; changing `TMPDIR` and
      friends afterwards has no effect for the rest of the process.
    - The path is made absolute but not symlink-resolved (`os.path.abspath`, not
      `os.path.realpath`), so a caller that needs a trusted location must validate the final path
      itself.

    Examples
    --------
    >>> get_tmpdir()
    '/tmp'

    >>> get_tmpdir()
    'C:\\Users\\vagrant\\AppData\\Local\\Temp\\2'
    """
    tmpdir = None
    try:
        tmpdir = tempfile.gettempdir()
    except Exception:
        pass

    return tmpdir or '/tmp'  # nosec B108 - fallback when tempfile.gettempdir() fails


def glob(pattern, recursive=True):
    """
    Return a sorted list of paths matching a shell glob pattern.

    Wraps the standard-library glob with `recursive=True` by default, so a `**`
    segment spans directories. Absolute and relative patterns both work. A
    pattern that matches nothing yields an empty list; the call never raises for
    a non-matching pattern.

    Parameters
    ----------
    pattern : str
        The glob pattern, e.g. `/var/log/**/*.log` or `*.txt`.
    recursive : bool, optional
        Whether `**` should match across directory boundaries. Defaults to True.

    Returns
    -------
    list
        The matching paths, sorted for a deterministic order.

    Examples
    --------
    >>> glob('/path/does/not/exist/*')
    []
    """
    return sorted(_glob.glob(pattern, recursive=recursive))


def grep_file(filename, pattern):
    """
    Search for a regex pattern in a file, similar to the `grep` command.

    Returns the first group of the first match, or the whole match where the pattern has
    no group.

    Parameters
    ----------
    filename : str
        Path to the file to search.
    pattern : str
        A Python regular expression pattern to search for.

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True if the file could be read and `pattern` is a valid
            regular expression, otherwise False.
          - tuple[1] (**str**): The first group of the match, the whole match where
            `pattern` has no group, the empty string where nothing matches, or an error
            message if unsuccessful.

    Examples
    --------
    >>> success, nc_version = grep_file('version.php', r'\\$OC_version=array\\((.*)\\)')
    """
    try:
        success, handle = open_file(filename)
        if not success:
            return False, handle
        with handle as f:
            data = f.read()
    except OSError as e:
        return False, f'I/O error "{e.strerror}" while opening or reading {filename}'
    except Exception as e:
        return False, f'Unknown error opening or reading {filename}: {e}'

    try:
        match = re.search(pattern, data)
    except (re.error, TypeError) as e:
        return False, f'Invalid regular expression {pattern!r}: {e}'
    if match:
        return True, match.group(1) if match.re.groups else match.group(0)
    return True, ''


def is_symlink(path):
    """
    Return whether `path` is a symbolic link.

    An exception-safe wrapper around `os.path.islink()`, so a plugin that must refuse a
    symlink (a root process about to open a caller-supplied path, for example) does not
    have to import `os` itself. It sits with `dir_exists()` and `file_exists()` as the
    filesystem-predicate a plugin reaches for. A path that cannot be examined counts as
    not a symlink; the caller's own open or read then reports the real trouble.

    Parameters
    ----------
    path : str | os.PathLike
        The path to test.

    Returns
    -------
    bool
        True if `path` is a symbolic link, otherwise False.

    Examples
    --------
    >>> is_symlink('/path/does/not/exist')
    False
    """
    try:
        return os.path.islink(path)
    except OSError:
        return False


def is_within(path, roots):
    """
    Return whether `path` resolves inside one of the directories in `roots`.

    Both `path` and each root are canonicalized with `os.path.realpath()` before
    comparison, so symlinks and `..` segments are resolved and cannot be used to
    step outside a root. A caller that must restrict which files it touches (for
    example something running with elevated privileges that should stay inside an
    operator-controlled directory) can use this to reject any path that escapes
    the intended roots. Because symlinks are resolved, a symlink pointing out of
    a root is rejected; to legitimately reach a location stored elsewhere,
    bind-mount it into a root instead of symlinking it.

    Parameters
    ----------
    path : str
        The path to check. It need not exist; only its resolved location matters.
    roots : iterable of str
        The directories `path` is allowed to resolve into. A single path given as a
        string is not a list of roots and yields False, as does an empty iterable.

    Returns
    -------
    bool
        True if `path` resolves to one of the roots or a location below it,
        otherwise False.

    Examples
    --------
    >>> is_within('/var/log/app/today.log', ['/var/log'])
    True
    >>> is_within('/var/log/../etc/shadow', ['/var/log'])
    False
    """
    if isinstance(roots, (str, bytes, os.PathLike)):
        # iterated, a string yields characters, and `/` would contain every path
        return False
    # normcase() is a no-op on POSIX but lowercases and normalizes separators on
    # Windows, where paths are case-insensitive; without it the containment check
    # would wrongly reject `C:\Var\Log\...` against a `C:\var\log` root.
    real = os.path.normcase(os.path.realpath(path))
    for root in roots:
        real_root = os.path.normcase(os.path.realpath(root))
        # The filesystem root already ends in a separator (`/`, `C:\\`), so do not
        # add a second one, or nothing would lie inside it.
        prefix = real_root if real_root.endswith(os.sep) else real_root + os.sep
        if real == real_root or real.startswith(prefix):
            return True
    return False


# Block-device name prefixes that never carry meaningful I/O for monitoring
# (loopback, RAM disks, compressed RAM, floppy and optical devices). They are
# skipped by get_block_devices().
_PSEUDO_DEVICE_PREFIXES = ('fd', 'loop', 'ram', 'sr', 'zram')


def make_temp_dir(prefix=''):
    """
    Create a unique temporary directory and return its path.

    Wraps `tempfile.mkdtemp()` and reports the outcome in the same
    `(success, result)` style as the other disk helpers.

    Parameters
    ----------
    prefix : str, optional
        Prefix for the directory name. Defaults to ''.

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True on success, otherwise False.
          - tuple[1] (**str**): The created directory path on success, otherwise an
            error message.

    Examples
    --------
    >>> make_temp_dir(prefix='myapp-')
    (True, '/tmp/myapp-abcd1234')
    """
    try:
        return True, tempfile.mkdtemp(prefix=prefix)
    except OSError as e:
        return False, f'OS error "{e.strerror}" while creating a temporary directory'
    except Exception as e:
        return False, f'Unknown error creating a temporary directory: {e}'


def mkdir(path, mode=0o755, exist_ok=True):
    """
    Create a directory, including any missing parent directories.

    Wraps `os.makedirs()` and reports the outcome in the same `(success, error)`
    style as the other disk helpers, so callers do not have to handle exceptions
    themselves.

    Parameters
    ----------
    path : str
        Directory path to create.
    mode : int, optional
        Permission bits for newly created directories.
        Defaults to `0o755`.
    exist_ok : bool, optional
        If `True`, an already existing directory is
        not an error. Defaults to `True`.

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True if the directory exists afterwards, else False.
          - tuple[1] (**None or str**): None on success, otherwise an error message.

    Examples
    --------
    >>> mkdir('/tmp/example/sub')
    (True, None)
    """
    try:
        os.makedirs(path, mode=mode, exist_ok=exist_ok)
        return True, None
    except OSError as e:
        return False, f'OS error "{e.strerror}" while creating {path}'
    except Exception as e:
        return False, f'Unknown error creating {path}: {e}'


def open_file(filename, binary=False, allowed_roots=None, nofollow=False, newline=None):
    """
    Open a file for reading, optionally confined to a set of directories.

    A process that runs with more privileges than whoever chose the path (for example
    root via sudo) must not be steerable into reading a file of the caller's choice.
    With `allowed_roots` the file that is actually opened is checked, not just the
    name: after the open, the handle is compared with what the path resolves to, so a
    directory anywhere in the path that is swapped for a symlink between the check and
    the open is caught as well. `O_NOFOLLOW` alone protects only the last component.

    Parameters
    ----------
    filename : str
        Path to open.
    binary : bool, optional
        Open in binary mode. Defaults to False (UTF-8 text).
    allowed_roots : iterable of str, optional
        When given, only a regular file whose real path lies inside one of these
        roots is opened. A symlink pointing out of a root, a path that changed while
        it was being opened, and anything that is not a regular file (a FIFO that
        would block the reader, a device) are refused. An empty iterable allows
        nothing, and a single path given as a string is refused rather than read as
        its characters. Defaults to None (no containment).
    nofollow : bool, optional
        Refuse a symlink as the last path component (`O_NOFOLLOW`). Like
        `allowed_roots`, it also refuses anything that is not a regular file. Defaults
        to False.
    newline : str, optional
        Passed to the text mode of the file as `newline`, like `open()`. Defaults to
        None.

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True on success, otherwise False.
          - tuple[1] (**file object or str**): The open file object, or an error
            message.

    Notes
    -----
    - With `allowed_roots` or `nofollow`, only a regular file is opened: a FIFO or any
      other non-regular file is refused at once instead of hanging the caller. Without
      either, a FIFO is opened too, non-blocking: one with a writer delivers what it is
      fed, one without a writer reads as empty instead of hanging the caller for ever.
      A writer that shows up after the open is not waited for.
    - A hard link inside a root that points to a file elsewhere is indistinguishable
      from a regular file. On Linux, `fs.protected_hardlinks=1` (the default on
      current distributions) keeps an unprivileged user from creating one to a file
      they do not own.

    Examples
    --------
    >>> success, handle = open_file('/var/log/messages', allowed_roots=['/var/log'])
    """
    if allowed_roots is not None:
        # A single path is a string, which iterates as characters: `'/var/log'` made
        # `/` a root and opened every file. An empty list allowed everything as well,
        # because it is false. Neither is what a caller who asked for confinement
        # wants, so the first is refused and the second allows nothing.
        if isinstance(allowed_roots, (str, bytes, os.PathLike)):
            return False, (
                f'Refusing to read "{filename}": the allowed roots have to be a list '
                'of directories, not a single path.'
            )
        allowed_roots = list(allowed_roots)
        if not allowed_roots:
            return False, (
                f'Refusing to read "{filename}": no directory is allowed to read from.'
            )
    if allowed_roots and not is_within(filename, allowed_roots):
        return False, (
            f'Refusing to read "{filename}": resolved path is outside the allowed '
            f'roots ({", ".join(allowed_roots)}); bind-mount it in if intended.'
        )
    flags = os.O_RDONLY | getattr(os, 'O_BINARY', 0)
    if nofollow:
        # POSIX-only. Where it is missing getattr() yields 0 and the flag is absent.
        flags |= getattr(os, 'O_NOFOLLOW', 0)
    # POSIX-only as well. Windows has neither the flag nor FIFOs in the filesystem.
    # Set in every case: a FIFO named by a caller would hang a plain `open()` for ever
    # when nobody writes to it. Non-blocking, the open returns at once, and the reads
    # that follow are blocking again.
    nonblock = getattr(os, 'O_NONBLOCK', 0)
    if nonblock:
        # Opening a FIFO for reading blocks until a writer shows up. Non-blocking,
        # the open returns at once and the type check below refuses it.
        flags |= nonblock
    try:
        fd = os.open(filename, flags)
    except OSError as e:
        return False, f'I/O error "{e.strerror}" while opening or reading {filename}'
    except Exception as e:
        return False, f'Unknown error opening or reading {filename}: {e}'
    try:
        opened = os.fstat(fd)
        # Without confinement the file is read as it always was, a pipe from a process
        # substitution or `/dev/stdin` included. A FIFO nobody writes to then reads as
        # empty instead of hanging the open, which is why it is opened non-blocking.
        if (allowed_roots or nofollow) and not _stat.S_ISREG(opened.st_mode):
            os.close(fd)
            return False, f'Refusing to read "{filename}": not a regular file.'
        if allowed_roots:
            # The file that was opened has to be the one the path names now, and
            # the path still has to lie inside the roots. A path that was
            # redirected between the check and the open fails one of the two.
            try:
                current = os.stat(os.path.realpath(filename))
            except OSError:
                current = None
            if (
                current is None
                or (current.st_dev, current.st_ino) != (opened.st_dev, opened.st_ino)
                or not is_within(filename, allowed_roots)
            ):
                os.close(fd)
                return False, (
                    f'Refusing to read "{filename}": the path changed while it was '
                    'being opened.'
                )
        if nonblock:
            # Only where the flag was set: on Windows `os.set_blocking()` works on
            # pipes alone and fails on a file with WinError 87.
            os.set_blocking(fd, True)
        if binary:
            return True, os.fdopen(fd, 'rb')
        return True, os.fdopen(fd, mode='r', encoding='utf-8', newline=newline)
    except OSError as e:
        os.close(fd)
        return False, f'I/O error "{e.strerror}" while opening or reading {filename}'
    except Exception as e:
        os.close(fd)
        return False, f'Unknown error opening or reading {filename}: {e}'


def read_csv(
    filename,
    delimiter=',',
    quotechar='"',
    newline='',
    as_dict=False,
    skip_empty_rows=False,
):
    """
    Read a CSV file and return its content as a list or dictionary.

    Parameters
    ----------
    filename : str
        Path to the CSV file.
    delimiter : str, optional
        The field delimiter character. Defaults to ','.
    quotechar : str, optional
        The character used to quote fields. Defaults to '"'.
    newline : str, optional
        Controls how universal newlines mode works while opening the
        file. Defaults to ''.
    as_dict : bool, optional
        If True, return each row as a dictionary using the CSV header.
        Defaults to False.
    skip_empty_rows : bool, optional
        If True, skip rows that contain only empty or
        whitespace fields. Defaults to False.

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True if reading succeeded, otherwise False.
          - tuple[1] (**list or str**):

            - If successful, a list of rows (as lists or dicts depending on `as_dict`).
            - If unsuccessful, an error message string.

    Examples
    --------
    >>> success, data = read_csv('data.csv')
    >>> success, data = read_csv('data.csv', as_dict=True, skip_empty_rows=True)
    """
    reader = None
    try:
        success, csvfile = open_file(filename, newline=newline)
        if not success:
            return False, csvfile
        with csvfile:
            reader = (
                csv.DictReader(csvfile, delimiter=delimiter, quotechar=quotechar)
                if as_dict
                else csv.reader(csvfile, delimiter=delimiter, quotechar=quotechar)
            )
            data = [
                row
                for row in reader
                if not (
                    skip_empty_rows
                    and all(
                        not str(v).strip()
                        for v in (row.values() if isinstance(row, dict) else row)
                    )
                )
            ]
        return True, data
    except csv.Error as e:
        line_num = getattr(reader, 'line_num', 'unknown')
        return False, f'CSV error in file {filename}, line {line_num}: {e}'
    except OSError as e:
        return False, f'I/O error "{e.strerror}" while opening or reading {filename}'
    except Exception as e:
        return False, f'Unknown error opening or reading {filename}: {e}'


def read_env(filename, delimiter='=', allowed_roots=None, nofollow=False):
    """
    Read a shell script that sets environment variables and return a dictionary with the
    extracted variables.

    Lines starting with '#' are treated as comments and ignored. Only lines that set variables
    (optionally prefixed with 'export') are processed. More complex shell logic (e.g., conditional
    reads) is ignored.

    Parameters
    ----------
    filename : str
        Path to the environment file to read.
    delimiter : str, optional
        The character that separates keys and values.
        Defaults to '='.
    allowed_roots : iterable of str, optional
        Confine the read to these
        roots; a file resolving outside them is refused. See `open_file()`.
        Defaults to None.
    nofollow : bool, optional
        Refuse to follow a symlink at the final path
        component. See `open_file()`. Defaults to False.

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True if reading succeeded, otherwise False.
          - tuple[1] (**dict or str**):

            - If successful, a dictionary of environment variable names and values.
            - If unsuccessful, an error message string.

    Examples
    --------
    Example shell script 'env.sh':

        export OS_AUTH_URL="https://api/v3"
        export OS_PROJECT_NAME=mypro
        # comment
        OS_PASSWORD='linuxfabrik'
        [ -z "$OS_PASSWORD" ] && read -e -p "Pass: " OS_PASSWORD
        export OS_PASSWORD

    >>> read_env('env.sh')
    {'OS_AUTH_URL': 'https://api/v3', 'OS_PROJECT_NAME': 'mypro', 'OS_PASSWORD': 'linuxfabrik'}
    """
    success, handle = open_file(
        filename, allowed_roots=allowed_roots, nofollow=nofollow
    )
    if not success:
        return False, handle
    try:
        with handle as envfile:
            data = {}
            for raw_line in envfile:
                line = raw_line.strip()
                if not line or line.startswith('#'):
                    continue
                if line.startswith('export '):
                    line = line[7:]
                if delimiter not in line:
                    continue
                key, value = line.split(delimiter, 1)
                data[key.strip()] = value.strip().strip('\'"')
        return True, data
    except OSError as e:
        return False, f'I/O error "{e.strerror}" while opening or reading {filename}'
    except Exception as e:
        return False, f'Unknown error opening or reading {filename}: {e}'


def read_file(
    filename, binary=False, max_bytes=None, allowed_roots=None, nofollow=False
):
    """
    Read the contents of a file and return them.

    Parameters
    ----------
    filename : str
        Path to the file to read.
    binary : bool, optional
        If True, read in binary mode and return the
        contents as `bytes`; otherwise decode as UTF-8 text. Defaults to False.
    max_bytes : int, optional
        Stop after this many bytes (characters in text
        mode) instead of reading the whole file. Use it whenever only a bounded part of
        the file is of interest, for example a metadata header at the top, so a file that
        unexpectedly grew to gigabytes cannot exhaust memory. Defaults to None, which
        reads the file completely.
    allowed_roots : iterable of str, optional
        Confine the read to these
        roots; a file whose real path resolves outside them is refused. A process
        that runs with more privileges than whoever chose the path must pass this, so
        a planted symlink cannot redirect the read to `/etc/shadow` or a private key.
        See `open_file()`. Defaults to None.
    nofollow : bool, optional
        Refuse to follow a symlink at the final path
        component (`O_NOFOLLOW`). See `open_file()`. Defaults to False.

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True if reading succeeded, otherwise False.
          - tuple[1] (**str | bytes**):

            - If successful, the contents of the file (`str`, or `bytes` when
              `binary` is True).
            - If unsuccessful, an error message string.

    Notes
    -----
    - With `max_bytes` in text mode the read may end in the middle of a multi-byte
      character sequence. Prefer `binary=True` plus an explicit decode when the file's
      encoding is not reliably known.

    Examples
    --------
    >>> success, content = read_file('example.txt')
    >>> success, raw = read_file('cert.der', binary=True)
    >>> success, header = read_file('plugin.php', binary=True, max_bytes=8192)
    >>> success, content = read_file(dmesg, allowed_roots=[dump_dir], nofollow=True)
    """
    success, handle = open_file(
        filename, binary=binary, allowed_roots=allowed_roots, nofollow=nofollow
    )
    if not success:
        return False, handle
    try:
        with handle as f:
            return True, f.read() if max_bytes is None else f.read(max_bytes)
    except OSError as e:
        return False, f'I/O error "{e.strerror}" while opening or reading {filename}'
    except Exception as e:
        return False, f'Unknown error opening or reading {filename}: {e}'


# ACL formats that can grant write access beyond the mode bits but are not evaluated
# here, so an object carrying one is refused: NFSv4 (fs/nfs/nfs4proc.c) and SMB
# (fs/smb/client/xattr.c). The POSIX ACL entry tags and the write bit are from
# include/uapi/linux/posix_acl.h, the xattr layout (a version header, then 8-byte
# entries) from include/uapi/linux/posix_acl_xattr.h. Read in the source of Linux 7.3.
_FOREIGN_ACL_XATTRS = (
    'system.cifs_acl',
    'system.nfs4_acl',
    'system.nfs4_dacl',
    'system.smb3_acl',
)
_POSIX_ACL_GROUP = 0x08
_POSIX_ACL_GROUP_OBJ = 0x04
_POSIX_ACL_MASK = 0x10
_POSIX_ACL_USER = 0x02
_POSIX_ACL_WRITE = 0x02
_POSIX_ACL_XATTR_VERSION = 0x0002


def _get_xattr(path, name):
    """Return the value of an extended attribute, None if there is none, or raise."""
    try:
        return os.getxattr(path, name, follow_symlinks=False)
    except OSError as e:
        if e.errno in (errno.ENODATA, errno.EOPNOTSUPP):
            return None
        raise


def _check_acl(path, st, uids, gids):
    """
    Look for an ACL entry that lets an account other than root, the trusted owners and
    their primary groups write to `path`. The kernel grants a named user or group, and
    the owning group, only what the mask allows, so the mask is applied first.

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True if `path` has a POSIX ACL with a mask, in which
            case the group bits of its mode are that mask and the owning group was
            checked here.
          - tuple[1] (**None or str**): None if no other account can write, otherwise
            the reason.
    """
    if not hasattr(os, 'getxattr'):
        return False, 'cannot be checked for ACLs on this platform'
    try:
        for name in _FOREIGN_ACL_XATTRS:
            if _get_xattr(path, name) is not None:
                return False, f'has an ACL ({name}) that cannot be checked'
        raw = _get_xattr(path, 'system.posix_acl_access')
    except OSError as e:
        return False, f'has an ACL that cannot be read ({e.strerror})'
    if raw is None:
        return False, None
    if (
        len(raw) < 4
        or (len(raw) - 4) % 8
        or struct.unpack_from('<I', raw)[0] != _POSIX_ACL_XATTR_VERSION
    ):
        return False, 'has an ACL that cannot be read'
    entries = list(struct.iter_unpack('<HHI', raw[4:]))
    masks = [perm for tag, perm, _ in entries if tag == _POSIX_ACL_MASK]
    mask = masks[0] if masks else 0x07
    for tag, perm, acl_id in entries:
        if not perm & mask & _POSIX_ACL_WRITE:
            continue
        if tag == _POSIX_ACL_USER and acl_id not in uids:
            return bool(masks), (
                f'is writable by uid {acl_id} through its ACL, which is not trusted '
                'for this'
            )
        if tag == _POSIX_ACL_GROUP and acl_id not in gids:
            return bool(masks), (
                f'is writable by group {acl_id} through its ACL, which is not '
                'trusted for this'
            )
        if tag == _POSIX_ACL_GROUP_OBJ and st.st_gid not in gids:
            return bool(masks), (
                f'is writable by group {st.st_gid}, which is not trusted for this'
            )
    return bool(masks), None


def _trusted_ids(owners):
    """Return the UIDs and GIDs `resolve_trusted_path()` and `resolve_trusted_tree()`
    accept: root, the given owners, and the primary groups of all of them."""
    uids = {0}
    for owner in owners or ():
        if isinstance(owner, int) and not isinstance(owner, bool) and owner >= 0:
            uids.add(owner)
    gids = {0}
    for uid in uids:
        try:
            gids.add(pwd.getpwuid(uid).pw_gid)
        except KeyError:
            pass
    return uids, gids


def _untrusted_reason(current, st, uids, gids):
    """Return why an account other than root and the trusted owners can change the
    object `current` whose `lstat()` result is `st`, or None if nobody else can. The
    single check behind `resolve_trusted_path()` and `resolve_trusted_tree()`."""
    if st.st_uid not in uids:
        return f'belongs to uid {st.st_uid}, which is not trusted for this'
    if _stat.S_ISSOCK(st.st_mode) or _stat.S_ISLNK(st.st_mode):
        # Who may connect to a socket or follow a symlink says nothing about who may
        # replace it; that takes write access to the directory, which is checked.
        return None
    if st.st_mode & _stat.S_IWOTH:
        return 'is writable by everyone'
    has_mask, reason = _check_acl(current, st, uids, gids)
    if reason:
        return reason
    if not has_mask and st.st_mode & _stat.S_IWGRP and st.st_gid not in gids:
        return f'is writable by group {st.st_gid}, which is not trusted for this'
    return None


def resolve_trusted_path(path, owners=None):
    """
    Resolve a path and confirm that nobody but root and the given owners can change
    what it points to.

    Use this before a process running with more privileges than whoever chose the path
    executes a file, connects to a socket, or derives an identity from a file there.
    Unlike a read, none of these can be checked after the fact on an open handle, so
    the path itself has to be out of reach: every directory from `/` down and the
    object itself must belong to root or one of `owners`, and none of them may be
    writable by anybody else. Once that holds, nobody else can swap a component, so
    the resolved path that is returned stays valid. Use the returned path, not the one
    that was passed in; a symlink on the way is fine because it is already resolved.

    Parameters
    ----------
    path : str
        The path to check. It has to exist.
    owners : iterable of int, optional
        UIDs trusted besides root, for example the account a service runs as.
        Group write access is accepted only for the primary groups of root and these
        accounts. Defaults to None (root only).

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True if the path is trusted, otherwise False.
          - tuple[1] (**str**): The resolved path, or an error message.

    Notes
    -----
    - The mode of a socket decides who may connect to it, not who may replace it, so
      it is not checked on a socket itself; replacing one takes write access to its
      directory, which is checked.
    - A directory writable by everyone (such as `/tmp`, sticky bit or not) is never
      trusted.
    - A POSIX ACL counts like the mode: an entry that lets another user or group write
      is refused. An object with an NFSv4 or SMB ACL is refused, since those cannot be
      evaluated here.
    - Linux only. Where ownership or ACLs cannot be checked (Windows, macOS), the path is
      refused.

    Examples
    --------
    >>> resolve_trusted_path('/usr/sbin/httpd')
    (True, '/usr/sbin/httpd')
    >>> resolve_trusted_path('/tmp/httpd')
    (False, 'Refusing "/tmp/httpd": "/tmp" is writable by everyone.')
    """
    if not isinstance(path, str) or not path:
        return False, f'Refusing "{path}": not a path.'
    if pwd is None:
        return (
            False,
            f'Refusing "{path}": ownership cannot be checked on this platform.',
        )
    uids, gids = _trusted_ids(owners)
    real = os.path.realpath(path)
    # Walk from the top down: once a directory is trusted, nobody else can change
    # its entries, so the check of the next level cannot be undermined from above.
    current = os.sep
    components = [part for part in real.split(os.sep) if part]
    for index in range(len(components) + 1):
        if index:
            current = os.path.join(current, components[index - 1])
        try:
            st = os.lstat(current)
        except OSError as e:
            return (
                False,
                f'Refusing "{path}": cannot examine "{current}" ({e.strerror}).',
            )
        if _stat.S_ISLNK(st.st_mode):
            return False, f'Refusing "{path}": "{current}" changed while being checked.'
        reason = _untrusted_reason(current, st, uids, gids)
        if reason:
            return False, f'Refusing "{path}": "{current}" {reason}.'
    return True, real


def resolve_trusted_tree(path, owners=None, prune=None, name_objects=True):
    """
    Resolve a directory and confirm that nobody but root and the given owners can change
    anything below it.

    `resolve_trusted_path()` covers a single file. A program that loads code from a
    whole tree, an interpreter running an application, can be given someone else's code
    through any file in it. This checks the path itself like `resolve_trusted_path()`,
    and then every directory, file and symlink below it the same way, from the top
    down. A symlink is followed, and its target has to pass as well, a directory
    target with everything below it.

    Parameters
    ----------
    path : str
        The directory to check. It has to exist.
    owners : iterable of int, optional
        UIDs trusted besides root, see `resolve_trusted_path()`.
    prune : callable, optional
        Called with the path of each directory and each symlink found below `path`,
        as the walk reaches it: its trusted parent directory and its name, never a
        resolved target. When it returns True, that entry is left out: a directory
        with everything below it, a symlink with its target and the path to it. For
        example a data directory that holds no code but may hold millions of files,
        possibly linked in from a volume of its own. The entry itself cannot be
        replaced by anybody else, since the directory holding it is checked. Never
        decide on a resolved path: whoever controls a component of it could point the
        decision at any directory in the tree.
    name_objects : bool, optional
        Name the object below `path` that failed the check. Pass False where the
        caller chose `path` but may not be able to list what is below it, such as a
        consumer running as root on a path an unprivileged account passed: the
        message would otherwise reveal names the caller cannot see. Defaults to True.

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True if the tree is trusted, otherwise False.
          - tuple[1] (**str**): The resolved path, or an error message naming the
            first object somebody else can change.

    Notes
    -----
    - Top-down order is what makes the result hold: once a directory is trusted,
      nobody else can add, remove or replace its entries, so nothing checked below it
      can be swapped afterwards either.
    - A directory that cannot be listed is refused, since what it holds is unknown.
    - Linux only, like `resolve_trusted_path()`.

    Examples
    --------
    >>> resolve_trusted_tree('/usr/share/nextcloud', owners=[48])
    (True, '/usr/share/nextcloud')
    """
    success, real = resolve_trusted_path(path, owners=owners)
    if not success:
        return False, real
    uids, gids = _trusted_ids(owners)

    def refuse(detail):
        if name_objects:
            return False, f'Refusing "{path}": {detail}.'
        return False, (
            f'Refusing "{path}": something below it can be changed by an account other '
            'than root and the trusted owners, or cannot be examined.'
        )

    pending = [real]
    seen = {real}
    while pending:
        directory = pending.pop()
        try:
            with os.scandir(directory) as entries:
                names = sorted(entry.name for entry in entries)
        except OSError as e:
            return refuse(f'cannot list "{directory}" ({e.strerror})')
        for name in names:
            current = os.path.join(directory, name)
            try:
                st = os.lstat(current)
            except OSError as e:
                return refuse(f'cannot examine "{current}" ({e.strerror})')
            if (
                (_stat.S_ISDIR(st.st_mode) or _stat.S_ISLNK(st.st_mode))
                and prune
                and prune(current)
            ):
                continue
            reason = _untrusted_reason(current, st, uids, gids)
            if reason:
                return refuse(f'"{current}" {reason}')
            if _stat.S_ISLNK(st.st_mode):
                # the link itself is safe in a trusted directory, its target has to be
                # trusted on its own, all the way from `/`
                success, target = resolve_trusted_path(current, owners=owners)
                if not success:
                    return refuse(target.removeprefix('Refusing ').rstrip('.'))
                if os.path.isdir(target) and target not in seen:
                    seen.add(target)
                    pending.append(target)
                continue
            if _stat.S_ISDIR(st.st_mode):
                pending.append(current)
    return True, real


def rm_dir(path):
    """
    Recursively delete a directory tree.

    Wraps `shutil.rmtree()` and reports the outcome in the same `(success, error)`
    style as `rm_file()`.

    Parameters
    ----------
    path : str
        Directory tree to delete.

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True if deletion succeeded, otherwise False.
          - tuple[1] (**None or str**): None on success, otherwise an error message.

    Examples
    --------
    >>> rm_dir('/tmp/lynis')
    (True, None)
    """
    try:
        shutil.rmtree(path)
        return True, None
    except OSError as e:
        return False, f'OS error "{e.strerror}" while deleting {path}'
    except Exception as e:
        return False, f'Unknown error deleting {path}: {e}'


def rm_file(filename):
    """
    Delete or remove a file.

    Parameters
    ----------
    filename : str
        Path to the file to delete.

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True if deletion succeeded, otherwise False.
          - tuple[1] (**None or str**):

            - None if the file was successfully deleted.
            - An error message string if unsuccessful.

    Examples
    --------
    >>> rm_file('test.txt')
    (True, None)
    """
    try:
        os.remove(filename)
        return True, None
    except OSError as e:
        return False, f'OS error "{e.strerror}" while deleting {filename}'
    except Exception as e:
        return False, f'Unknown error deleting {filename}: {e}'


def shorten_path(path, max_len=None, truncate=True):
    """
    Shorten a path for display.

    Abbreviate a slash-separated path by reducing every component except the last one to its
    first character, the way the zsh prompt shortens a long working directory. The last
    component is kept in full because it usually carries the identifying information. This is
    a pure string transformation; the path is not resolved and the filesystem is not touched.

    With `max_len`, a path that already fits within `max_len` characters is returned
    unchanged, so short and readable paths are not abbreviated needlessly; only a longer path
    is shortened. If reducing the parents to their initial still leaves the result longer than
    `max_len` (a very long final component, for example a Kubernetes CSI volume handle), the
    result is middle-truncated with `...`, keeping its head and tail so distinct paths stay
    distinguishable. The `...` marker is plain ASCII so it renders in monospace tables on any
    terminal and transport.

    Parameters
    ----------
    path : str
        A slash-separated path, for example `/etc/pki/tls/certs/002c0b4f.0`.
    max_len : int, optional
        Maximum length of the returned string. When `None` (the
        default), the path is always abbreviated and never truncated.
    truncate : bool, optional
        Whether an abbreviated path that is still longer than
        `max_len` may be middle-truncated. `False` keeps the result whole, which is what a
        caller wants where the final component is the identifying part and a name cut in the
        middle would no longer name anything - `max_len` is then the length at which
        abbreviating starts rather than a limit on the result. Defaults to True.

    Returns
    -------
    str
        The abbreviated path, for example `/e/p/t/c/002c0b4f.0`. A value without a
        slash (a bare basename, an empty string, a sentinel like `-`) is returned unchanged.

    Examples
    --------
    >>> shorten_path('/etc/pki/tls/certs/002c0b4f.0')
    '/e/p/t/c/002c0b4f.0'

    >>> shorten_path('index.php')
    'index.php'

    >>> shorten_path('/var/log/audit', max_len=40)  # already short enough
    '/var/log/audit'

    >>> shorten_path('/a/very/long/path/' + 'x' * 60, max_len=20)
    '/a/v/l/p...xxxxxxxxx'

    >>> shorten_path(
    ...     '/var/log/httpd/www.example.com-error.log', max_len=32, truncate=False
    ... )
    '/v/l/h/www.example.com-error.log'
    """
    if not path or '/' not in path:
        return path
    if max_len is not None and len(path) <= max_len:
        return path
    head, _, tail = path.rpartition('/')
    abbrev = '/'.join(part[:1] for part in head.split('/'))
    result = f'{abbrev}/{tail}'
    if max_len is not None and truncate:
        result = txt.shorten(result, max_len)
    return result


def stat(file):
    """
    Return the `os.stat()` result for a filesystem entry, or `None` on failure.

    A single stat() syscall exposes every field (size, modification time, mode,
    owner, ...), so a caller that needs more than one of them should use this
    instead of calling `get_size()`, `get_owner()` etc. separately. The common
    fields (`st_size`, `st_mtime`, `st_mode`, `st_uid`) are available on every
    supported platform, including Windows.

    Parameters
    ----------
    file : str | os.PathLike
        Path to stat.

    Returns
    -------
    os.stat_result | None
        The stat result on success; `None` if the call fails (for example the path
        does not exist or is not accessible).

    Examples
    --------
    >>> stat('/path/does/not/exist') is None
    True
    """
    try:
        return os.stat(file)
    except OSError:
        return None


def udevadm(device, _property):
    """
    Run `udevadm info` and extract a specific property manually.

    To support older systems, the function does not use the `--property=` option
    and instead parses all properties manually to find the desired one.

    Parameters
    ----------
    device : str
        Path to the device (e.g., '/dev/dm-0' or '/dev/mapper/rl-root').
    _property : str
        The property name to retrieve (e.g., 'DEVNAME', 'DM_NAME').

    Returns
    -------
    str
        The value of the requested property if found, otherwise an empty string.

    Examples
    --------
    >>> udevadm('/dev/mapper/rl_rocky8-root', 'DEVNAME')
    '/dev/dm-0'

    >>> udevadm('/dev/dm-0', 'DM_NAME')
    'rl_rocky8-root'

    >>> udevadm('/dev/linuxfabrik', 'DEVNAME')
    ''
    """
    # Only query real device nodes under /dev. Resolving the path first defends
    # against a crafted `device` that points elsewhere or smuggles arguments.
    device = os.path.realpath(device)
    if not device.startswith('/dev/'):
        return ''
    success, result = shell.shell_exec(
        ['udevadm', 'info', '--query=property', f'--name={device}']
    )
    if not success:
        return ''
    stdout, _, _ = result
    for line in stdout.strip().splitlines():
        if '=' not in line:
            continue
        key, value = line.split('=', maxsplit=1)
        if key == _property:
            return value
    return ''


def under_root(root, path):
    """
    Return `path` relocated below `root`, whether it is absolute or relative.

    Meant for a consumer that reads a set of well-known paths and has to be able to
    read them from somewhere else instead: a test fixture standing in for the host's
    `/etc`, a mounted image, an extracted backup. The caller keeps writing the paths
    it means, and one root decides where they are read from.

    No result ever leaves `root`: an empty segment, `.` and `..` are dropped rather
    than followed, so a path assembled from data cannot walk out of the root through
    its own text. That is not a sandbox, though, because a symlink inside the root
    still points wherever it points. Where the containment has to hold against the
    filesystem too, check the result with `is_within()`.

    Parameters
    ----------
    root : str
        The directory every path is read below.
    path : str
        The path to relocate. An absolute one loses its anchor, a
        relative one is taken as it is.

    Returns
    -------
    str
        The relocated path, or `root` itself where `path` names no segment.

    Examples
    --------
    >>> under_root('/fixtures/rhel', '/etc/os-release')
    '/fixtures/rhel/etc/os-release'

    >>> under_root('/fixtures/rhel', '/../etc/shadow')
    '/fixtures/rhel/etc/shadow'
    """
    # splitdrive() takes a Windows drive letter off; on POSIX it is a no-op. The
    # backslash is translated as well, because a path handed over on Windows uses it
    # as its separator and would otherwise survive as part of a single segment.
    _, tail = os.path.splitdrive(str(path))
    parts = [
        part
        for part in tail.replace('\\', '/').split('/')
        if part not in ('', '.', '..')
    ]
    if not parts:
        return root
    return os.path.join(root, *parts)


def walk_directory(path, exclude_pattern=r'', include_pattern=r'', relative=True):
    """
    Walk recursively through a directory and create a list of files.

    If an `exclude_pattern` (regex) is specified, files matching this pattern
    are ignored. If an `include_pattern` (regex) is specified, only files matching
    this pattern are included. Exclude filtering is applied before include filtering.

    Parameters
    ----------
    path : str
        The root directory to walk.
    exclude_pattern : str, optional
        Regex pattern to exclude files. Defaults to ''.
    include_pattern : str, optional
        Regex pattern to include files. Defaults to ''.
    relative : bool, optional
        Return relative paths if True, else absolute.
        Defaults to True.

    Returns
    -------
    list of str
        List of matching file paths.

    Examples
    --------
    >>> walk_directory('/tmp')
    ['cpu-usage.db', 'segv_output.MCiVt9']

    >>> walk_directory('/tmp', exclude_pattern='.*Temp-.*', relative=False)
    ['/tmp/cpu-usage.db', '/tmp/segv_output.MCiVt9']
    """
    if exclude_pattern:
        exclude_pattern = re.compile(exclude_pattern, re.IGNORECASE)
    if include_pattern:
        include_pattern = re.compile(include_pattern, re.IGNORECASE)

    result = []
    for current, _, files in os.walk(path):
        for file in files:
            full_path = os.path.join(current, file)
            if exclude_pattern and exclude_pattern.match(full_path):
                continue
            if include_pattern and not include_pattern.match(full_path):
                continue
            # relpath() rather than stripping the root off the front as a string: the
            # latter needs the separator spelled the way the platform spells it, and
            # removes the root again wherever it occurs a second time further along the
            # path (`/srv/wp/wp-content/srv/wp/...`).
            result.append(os.path.relpath(full_path, path) if relative else full_path)

    return result


def write_file(filename, content, append=False):
    """
    Write a string to a file.

    If `append` is True, the content is appended to the file instead of overwriting it.

    Parameters
    ----------
    filename : str
        Path to the file to write to.
    content : str
        The string content to write into the file.
    append : bool, optional
        If True, append to the file; if False, overwrite the file.
        Defaults to False.

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True if writing succeeded, otherwise False.
          - tuple[1] (**None or str**):

            - None if the file was written successfully.
            - An error message string if unsuccessful.

    Examples
    --------
    >>> write_file('test.txt', 'First line\\nSecond line')
    (True, None)
    """
    try:
        mode = 'a' if append else 'w'
        with open(filename, mode, encoding='utf-8') as f:
            f.write(content)
        return True, None
    except OSError as e:
        return False, f'I/O error "{e.strerror}" while writing {filename}'
    except Exception as e:
        return False, f'Unknown error writing {filename}: {e}'