Gecko [ tanishkshah.com ]

Name	Size	Permission	Date
lvestats-cm-helper	1.31 KB	-rwxr-xr-x	2026-08-20 13:34
lvestats-cpapi-helper	15.83 KB	-rwxr-xr-x	2026-08-20 13:34
lvestats-gov-helper	7.26 KB	-rwxr-xr-x	2026-08-20 13:34
lvestats-jwt-helper	1.19 KB	-rwxr-xr-x	2026-08-20 13:34
lvestats-plugin-runner	47.76 KB	-rwxr-xr-x	2026-08-20 13:34

Rename

#!/opt/cloudlinux/venv/bin/python3
# Copyright Cloud Linux Software, Inc 2010-2026 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

"""
lvestats-plugin-runner -- long-lived subprocess host for Python LveStatsPlugin classes.

The Rust daemon (lvestats-server) spawns ONE runner for the whole daemon lifetime
and drives it over stdin/stdout with line-delimited JSON. Every Python plugin
discovered in the plugins directory is hosted as an instance inside this single
process, so plugin state (counters, caches, DB engines) survives across ticks,
matching the semantics of the Python lve-stats daemon's in-process executor.

Protocol (one request per line, one response per line, stderr = logs):

{"op":"list","plugins_dir":"<dir>"}
        -> {"status":"ok","plugins":[{"file":..,"class":..,"order":..,"period":..,"timeout":..}, ...]}

{"op":"init","class":"<Name>","plugin_config":{...},"db_url":"..."|null,"is_user_plugin":true}
        -> {"status":"ok"} | {"status":"error","message":..,"traceback":..}

{"op":"execute","class":"<Name>","now":<float>,"lve_data":{...}}
        -> {"status":"ok","lve_data":{...}}
         | {"status":"terminated"}
         | {"status":"error","message":..,"traceback":..}

{"op":"shutdown"}
        -> {"status":"ok"}   (process exits after writing response)

SIGUSR1 dumps stack traces of all threads to stderr.
SIGUSR2 is ignored outside of execute() and raises LveStatsPluginTerminated inside,
so the Rust daemon can interrupt a sleeping plugin during graceful shutdown.
"""

import ctypes
import importlib.util
import inspect
import io
import json
import logging
import os
import signal
import stat
import sys
import traceback

# Import the real base class and termination exception from the installed
# lve-stats package so a plugin's `except LveStatsPluginTerminated:` catches
# the same class our SIGUSR2 handler raises. Fall back to local stubs so the
# runner still works on systems where the package is not installed
# (e.g. development / CI environments that rely only on this runner).
try:
    from lvestats.core.plugin import LveStatsPlugin, LveStatsPluginTerminated  # noqa: F401
except ImportError:
    class LveStatsPluginTerminated(Exception):
        pass

class LveStatsPlugin:
        # No `order` here, mirroring the real lvestats.core.plugin.LveStatsPlugin:
        # `order` is documented as an attribute a plugin *may* define, and its
        # absence is what lets the daemon apply its last-in-pipeline default
        # (CLOS-6853). An `order = 0` on the base class makes every orderless
        # plugin claim the front of the tick instead.
        period = None
        timeout = None

def execute(self, lve_data):
            pass

def set_config(self, config):
            pass

def set_db_engine(self, engine):
            pass

# ---------------------------------------------------------------------------
# lve_data hydrate/dehydrate
#
# Convert `lve_data` between the Rust JSON shape and the rich Python object
# shape that user plugins ported from legacy lve-stats expect.
#
# Legacy lve-stats (Python-only) passed `lve_data` containing rich objects:
#
#     lve_data['stats'][uid]         -> LVEStat               (lvestat module)
#     lve_data['old_stats'][uid]     -> LVEStat
#     lve_data['lve_usage'][uid]     -> AggregatedLveUsage    (lvestats.plugins.generic.aggregators)
#     lve_data['lve_usages_5s'][i]   -> {uid: LVEUsage}       (lvestats.plugins.generic.analyzers)
#
# In lve-stats3 the daemon serializes `LveData` to JSON and we receive a
# plain dict of dicts with string keys. Several field names also differ:
# `ProcLveEntry` and `AggregatedLveUsage` use `*_limit`-suffixed names
# instead of legacy short names like `cpu`, `lep`, `lmem`, `io`. We wrap
# each entry into the original legacy class so plugin code that does
# `isinstance(s, LVEStat)`, `usage.has_interesting_values()`, or simply
# attribute reads/writes keeps working unchanged. The runner is shebanged
# to `/opt/cloudlinux/venv/bin/python3`, where `lvestat` and `lvestats`
# are always installed; we import them at module load and fail fast if
# missing.
#
# Two of the three classes need a small extension because the Rust struct
# carries fields legacy didn't model:
#
#     * LVEStat              + cpu_fault, io_fault, iops_fault, cpu_max_limit
#     * AggregatedLveUsage   + id, count, created
#     * LVEUsage             — no extension (slots match Rust keys 1:1)
#
# The subclasses keep `isinstance(_, LVEStat)` etc. true for plugin code
# that does type checks.
#
# Errors are non-fatal: malformed values are passed through unconverted
# with a warning to stderr — a single bad key from one plugin must not
# corrupt the pipeline for the rest. Matches the policy in
# `crates/lvestats-core/src/types.rs::lve_data_from_python_json`.
# ---------------------------------------------------------------------------

from lvestat import LVEStat  # noqa: E402
from lvestats.plugins.generic.analyzers import LVEUsage, LVEUSAGESLOTS  # noqa: E402
from lvestats.plugins.generic.aggregators import AggregatedLveUsage  # noqa: E402

# Field-name translation tables (Rust JSON key <-> legacy slot name)

# `LveStat` (Rust ProcLveEntry) <-> legacy `LVEStat`
_LVESTAT_RUST_TO_LEGACY = {
    "cpu_limit":    "cpu",       # legacy: cpu = CPU limit
    "mep_limit":    "lep",
    "mem_limit":    "lmem",
    "io_limit":     "io",
    "memphy_limit": "lmemphy",
    "nproc_limit":  "lnproc",
    "iops_limit":   "liops",
}
_LVESTAT_LEGACY_TO_RUST = {v: k for k, v in _LVESTAT_RUST_TO_LEGACY.items()}

# `LveUsage` (Rust) <-> `LVEUsage` (legacy) — names match 1:1, no table needed.

# `AggregatedLveUsage` (Rust DB shape) <-> `AggregatedLveUsage` (legacy LVEUSAGESLOTS)
_AGG_RUST_TO_LEGACY = {
    "cpu":       "cpu_usage",    # rust: avg cpu / legacy: cpu_usage
    "cpu_limit": "lcpu",
    "mep_limit": "lep",
    "io":        "io_usage",     # rust: avg io / legacy: io_usage
    "io_limit":  "io",           # rust: io limit / legacy: io
    "mem":       "mem_usage",
    "mem_limit": "lmem",
}
_AGG_LEGACY_TO_RUST = {v: k for k, v in _AGG_RUST_TO_LEGACY.items()}

# Real-class extensions: add Rust-only fields that legacy slots don't model.

# Slots on LVEStat we DON'T serialize back to the Rust ProcLveEntry shape:
# helper-method refs, plus legacy-only fields with no Rust counterpart.
_LVESTAT_NON_WIRE_SLOTS = frozenset((
    "_get_attributes", "_set_attributes",
    "ncpu", "lcpuw",
))

class _LVEStatExtended(LVEStat):
    """Real `LVEStat` extended with the Rust ProcLveEntry fields legacy
    didn't model: per-resource `*_fault` counters and `cpu_max_limit`.
    Plugins still see `isinstance(s, LVEStat) == True`."""

__slots__ = ("cpu_fault", "io_fault", "iops_fault", "cpu_max_limit")

def __init__(self, version=8):
        # `LVEStat(None, version)` zero-inits all slots; `None` means
        # "do not parse a /proc/lve line" (the documented escape hatch).
        LVEStat.__init__(self, None, version)
        self.cpu_fault = 0
        self.io_fault = 0
        self.iops_fault = 0
        self.cpu_max_limit = 0

# Slots emitted on dehydrate — base LVEStat slots minus internals/legacy-only,
# plus the four extended slots.
_LVESTAT_WIRE_SLOTS = tuple(
    s for s in LVEStat.__slots__ if s not in _LVESTAT_NON_WIRE_SLOTS
) + _LVEStatExtended.__slots__

# Extra fields lve-stats3's AggregatedLveUsage carries that legacy
# `LVEUSAGESLOTS` doesn't model.
_AGG_EXTRA_SLOTS = ("id", "count", "created")

class _AggregatedExtended(AggregatedLveUsage):
    """Real `AggregatedLveUsage` extended with the lve-stats3 DB-shape
    fields (`id`, `count`, `created`)."""

__slots__ = _AGG_EXTRA_SLOTS

def __init__(self, lve_version=8):
        AggregatedLveUsage.__init__(self, lve_version=lve_version)
        self.id = 0
        self.count = 0
        self.created = 0

_AGG_WIRE_SLOTS = tuple(LVEUSAGESLOTS) + _AGG_EXTRA_SLOTS

# Legacy scratch keys user plugins read/write but Rust doesn't model.
SCRATCH_KEYS = ("old_now", "lve_usages", "lve_active_ids")

def _hydrate_warn(msg):
    sys.stderr.write("Warning: lvestats-plugin-runner: " + msg + "\n")

def _make_lve_stat(d):
    """Build a (extended) `LVEStat` instance from a Rust ProcLveEntry dict."""
    obj = _LVEStatExtended(version=8)
    for rust_key, val in d.items():
        legacy_attr = _LVESTAT_RUST_TO_LEGACY.get(rust_key, rust_key)
        try:
            setattr(obj, legacy_attr, val)
        except AttributeError:
            _hydrate_warn("LVEStat: unknown field %r dropped during hydrate" % (rust_key,))
    return obj

def _lve_stat_to_dict(obj):
    """Serialize an `LVEStat` (or extended subclass) to a Rust ProcLveEntry dict."""
    out = {}
    for slot in _LVESTAT_WIRE_SLOTS:
        try:
            val = getattr(obj, slot)
        except AttributeError:
            continue
        if val is None:
            continue
        rust_key = _LVESTAT_LEGACY_TO_RUST.get(slot, slot)
        out[rust_key] = val
    return out

def _make_lve_usage(d):
    """Build an `LVEUsage` from a Rust LveUsage dict (slots match 1:1)."""
    obj = LVEUsage(lve_version=d.get("lve_version", 8))
    for k, val in d.items():
        try:
            setattr(obj, k, val)
        except AttributeError:
            _hydrate_warn("LVEUsage: unknown field %r dropped during hydrate" % (k,))
    return obj

def _lve_usage_to_dict(obj):
    """Serialize an `LVEUsage` to a Rust LveUsage dict (1:1, no rename)."""
    out = {}
    for slot in LVEUSAGESLOTS:
        try:
            val = getattr(obj, slot)
        except AttributeError:
            continue
        if val is None:
            continue
        out[slot] = val
    return out

def _make_aggregated(d):
    """Build a (extended) `AggregatedLveUsage` from a Rust dict."""
    obj = _AggregatedExtended(lve_version=d.get("lve_version", 8))
    for rust_key, val in d.items():
        legacy_attr = _AGG_RUST_TO_LEGACY.get(rust_key, rust_key)
        try:
            setattr(obj, legacy_attr, val)
        except AttributeError:
            _hydrate_warn("AggregatedLveUsage: unknown field %r dropped during hydrate"
                          % (rust_key,))
    return obj

def _aggregated_to_dict(obj):
    """Serialize an `AggregatedLveUsage` to a Rust AggregatedLveUsage dict."""
    out = {}
    for slot in _AGG_WIRE_SLOTS:
        try:
            val = getattr(obj, slot)
        except AttributeError:
            continue
        if val is None:
            continue
        rust_key = _AGG_LEGACY_TO_RUST.get(slot, slot)
        out[rust_key] = val
    return out

def _to_int_uid(k):
    if isinstance(k, int):
        return k
    try:
        return int(k)
    except (TypeError, ValueError):
        return None

def _hydrate_uid_map(d, builder, label):
    """Hydrate {uid_str: dict} into {int(uid): rich_object}."""
    if not isinstance(d, dict):
        _hydrate_warn("expected dict for %s, got %s" % (label, type(d).__name__))
        return d
    out = {}
    for k, v in d.items():
        uid = _to_int_uid(k)
        if uid is None:
            _hydrate_warn("non-int %s key %r dropped during hydrate" % (label, k))
            continue
        if isinstance(v, dict):
            out[uid] = builder(v)
        else:
            _hydrate_warn("non-dict %s entry for uid %r passed through" % (label, k))
            out[uid] = v
    return out

def hydrate(lve_data, scratch=None):
    """Convert the JSON-decoded `lve_data` into a rich-object form for plugins.

- `stats` / `old_stats`: {uid_str: dict} -> {int(uid): LVEStat}
    - `lve_usage`:           {uid_str: dict} -> {int(uid): AggregatedLveUsage}
    - `lve_usages_5s`:       {uid_str: dict} -> [{int(uid): LVEUsage}]
                              (1-element list to match legacy aggregator API)
    - `faults`, `users`, `dbgov_data`, scalars: passed through unchanged
    - `scratch` (legacy keys persisted across ticks): merged in if not present

Returns a *new* dict; the caller's `lve_data` is not mutated.
    """
    if not isinstance(lve_data, dict):
        return lve_data

out = dict(lve_data)

if "stats" in out:
        out["stats"] = _hydrate_uid_map(out["stats"], _make_lve_stat, "stats")
    if "old_stats" in out:
        out["old_stats"] = _hydrate_uid_map(
            out["old_stats"], _make_lve_stat, "old_stats"
        )
    if "lve_usage" in out:
        out["lve_usage"] = _hydrate_uid_map(
            out["lve_usage"], _make_aggregated, "lve_usage"
        )
    if "dbgov_data_for_cm" in out:
        # Python keys this dict by INT uid (dbgov_saver.py: `data_for_cm[uid]`),
        # and its values are plain dicts, so only the keys need converting
        # (CLOS-6853).
        out["dbgov_data_for_cm"] = _hydrate_uid_map(
            out["dbgov_data_for_cm"], lambda d: d, "dbgov_data_for_cm"
        )
    if "lve_usages_5s" in out:
        flat = _hydrate_uid_map(out["lve_usages_5s"], _make_lve_usage, "lve_usages_5s")
        # Promote to legacy list-of-dicts shape so legacy aggregator code
        # (`for it in lve_data['lve_usages_5s']: for uid, usage in it.items()`)
        # iterates correctly.
        out["lve_usages_5s"] = [flat] if isinstance(flat, dict) else flat

if scratch:
        for k, v in scratch.items():
            out.setdefault(k, v)

return out

def _unwrap_stats_value(v):
    if isinstance(v, LVEStat):
        return _lve_stat_to_dict(v)
    return v

def _unwrap_lve_usage_value(v):
    # AggregatedLveUsage is a subclass of LVEUsage — check it first so we
    # serialize the Rust DB-shape keys (cpu, mem, io with *_limit twins).
    if isinstance(v, AggregatedLveUsage):
        return _aggregated_to_dict(v)
    if isinstance(v, LVEUsage):
        # Plugin assigned a bare LVEUsage where AggregatedLveUsage was
        # expected — best-effort: serialize the LVEUsage portion via the
        # AggregatedLveUsage key namespace.
        return _aggregated_to_dict(v)
    return v

def _unwrap_lve_usages_5s_value(v):
    if isinstance(v, LVEUsage):
        return _lve_usage_to_dict(v)
    return v

def _dehydrate_uid_map(d, unwrap, label):
    if not isinstance(d, dict):
        _hydrate_warn("expected dict for %s during dehydrate, got %s"
                      % (label, type(d).__name__))
        return d
    out = {}
    for k, v in d.items():
        out[str(k)] = unwrap(v)
    return out

def dehydrate(lve_data, scratch=None):
    """Convert plugin-mutated `lve_data` back to the JSON shape Rust expects.

Inverse of `hydrate`. Strips `SCRATCH_KEYS` into `scratch` (if provided)
    so they survive into the next tick's hydrate but don't reach Rust.
    """
    if not isinstance(lve_data, dict):
        return lve_data

out = dict(lve_data)

if "stats" in out:
        out["stats"] = _dehydrate_uid_map(out["stats"], _unwrap_stats_value, "stats")
    if "old_stats" in out:
        out["old_stats"] = _dehydrate_uid_map(
            out["old_stats"], _unwrap_stats_value, "old_stats"
        )
    if "lve_usage" in out:
        out["lve_usage"] = _dehydrate_uid_map(
            out["lve_usage"], _unwrap_lve_usage_value, "lve_usage"
        )

if "dbgov_data_for_cm" in out:
        out["dbgov_data_for_cm"] = _dehydrate_uid_map(
            out["dbgov_data_for_cm"], lambda v: v, "dbgov_data_for_cm"
        )

if "lve_usages_5s" in out:
        v = out["lve_usages_5s"]
        if isinstance(v, list):
            tail = v[-1] if v else {}
            if isinstance(tail, dict):
                out["lve_usages_5s"] = _dehydrate_uid_map(
                    tail, _unwrap_lve_usages_5s_value, "lve_usages_5s"
                )
            else:
                out["lve_usages_5s"] = {}
        elif isinstance(v, dict):
            out["lve_usages_5s"] = _dehydrate_uid_map(
                v, _unwrap_lve_usages_5s_value, "lve_usages_5s"
            )
        # Anything else: pass through; Rust deserializer skips on type mismatch.

for k in SCRATCH_KEYS:
        if k in out:
            if scratch is not None:
                scratch[k] = out[k]
            del out[k]

return out

# ---------------------------------------------------------------------------
# Plugin host
# ---------------------------------------------------------------------------

# Schema-level SQLAlchemy exceptions we treat specially so the Rust daemon
# can recreate the schema and retry (mirrors Python daemon's recover_db()
# path in `lvestats/eventloop/plugin_executors.py:106-114`). If SQLAlchemy
# isn't installed, the tuple is empty and no classification happens.
try:
    from sqlalchemy.exc import (
        NoSuchColumnError, NoSuchTableError, NoReferenceError,
    )
    _DB_SCHEMA_ERRORS = (NoSuchColumnError, NoSuchTableError, NoReferenceError)
except ImportError:
    _DB_SCHEMA_ERRORS = ()

# Single-threaded request loop — no locking needed.
_classes = {}    # class_name -> class object, populated by op_list
_instances = {}  # class_name -> plugin instance, populated by op_init
# Per-instance scratch dict for legacy keys (`old_now`, `lve_usages`,
# `lve_active_ids`) that user plugins read/write but Rust doesn't model.
# Preserved across ticks so legacy semantics hold (e.g. LVEUsageAnalyzer
# stores `self.now` in `lve_data['old_now']` for the next tick to read).
_scratch = {}    # class_name -> dict of legacy scratch keys

def _sigusr1_handler(signum, frame):
    lines = ["--- lvestats-plugin-runner thread traces ---"]
    for tid, stack in sys._current_frames().items():
        lines.append("# Thread %d" % tid)
        lines.extend(traceback.format_stack(stack))
    sys.stderr.write("\n".join(lines) + "\n")
    sys.stderr.flush()

def _sigusr2_handler(signum, frame):
    raise LveStatsPluginTerminated("SIGUSR2")

# Filesystem types whose inode attributes are asserted by a user-space server
# instead of by the kernel. Matched as a prefix, so this covers `fuse`,
# `fuseblk`, `fusectl` and every `fuse.<subtype>` (sshfs, glusterfs, ...).
_UNTRUSTED_FSTYPE_PREFIXES = ("fuse",)

def _mount_fstype(st_dev, path):
    """Return the filesystem type backing `path`, or None if undeterminable.

Resolved primarily from the kernel-assigned device number `st_dev` against
    the `major:minor` field of /proc/self/mountinfo -- both sides are kernel
    data that a user-space filesystem server cannot influence, unlike the uid
    and mode it answers stat with. Filesystems that hand out an anonymous device
    per subvolume (btrfs) expose st_dev values absent from mountinfo, so fall
    back to the longest mount point that is a prefix of `path`, which is the
    mount governing it. Relative paths match nothing and yield None.

That fallback makes this a PATH search, sound only for a path the caller
    resolved itself. A superblock can leave mountinfo while it keeps serving
    descriptors already opened against it (MNT_DETACH, which the mount's own
    owner can issue), and for such a device the fallback credits the inode with
    the enclosing mount's type -- "/" prefixes every absolute path. So never
    ask this about an fd whose bytes are about to be read; ask the superblock
    itself, via _is_trusted_fd_filesystem.
    """
    want = "%d:%d" % (os.major(st_dev), os.minor(st_dev))
    enclosing_fstype = None
    enclosing_len = -1
    try:
        # surrogateescape, not strict: the kernel escapes only space, tab,
        # newline and backslash in the mount-point field and passes every other
        # byte through raw, so one mount whose name is not valid UTF-8 would
        # otherwise raise UnicodeDecodeError -- a ValueError, which the guard
        # below does not catch -- and fail every plugin closed. It also matches
        # how os.listdir/os.path.realpath decode such names, so `path` and
        # `mount_point` stay comparable below.
        with open("/proc/self/mountinfo", "r", errors="surrogateescape") as fh:
            entries = fh.readlines()
    except OSError:
        return None
    for entry in entries:
        # "<id> <parent> <major:minor> <root> <mount point> <opts> [tags] - <fstype> ..."
        head, separator, tail = entry.partition(" - ")
        if not separator:
            continue
        head_fields = head.split()
        tail_fields = tail.split()
        if len(head_fields) < 5 or not tail_fields:
            continue
        fstype = tail_fields[0]
        if head_fields[2] == want:
            return fstype
        mount_point = head_fields[4]
        if len(mount_point) >= enclosing_len and (
            path == mount_point or path.startswith(mount_point.rstrip("/") + "/")
        ):
            enclosing_fstype = fstype
            enclosing_len = len(mount_point)
    return enclosing_fstype

def _is_trusted_filesystem(st, path):
    """Return (trusted, reason) for the mount backing an already-stat'ed inode.

The uid/mode gates below only prove something when the KERNEL is the one
    asserting uid and mode. On a filesystem served from user space a single
    process answers both the getattr behind our stat and the read that yields
    the plugin bytes, so `uid 0` and `mode 0644` are values the mount's owner
    chose rather than observations -- and an unprivileged user may mount FUSE on
    a directory it owns. Refuse those mounts; kernel-backed filesystems are
    accepted, including the network exports and bind mounts an operator may
    deliberately point plugins_dir at (that is root trusting its own storage).

For PATH components only -- the ancestry walk, whose paths this process
    resolved itself, so the mount governing each one is listed. The lookup below
    therefore answers None only when /proc/self/mountinfo cannot be read at all,
    which fails closed; it does NOT fail closed for a device missing from
    mountinfo, because a listed enclosing mount answers instead (_mount_fstype).
    The leaf inode whose bytes reach exec() is gated on its own superblock
    instead, by _is_trusted_fd_filesystem, which needs no such fallback.
    """
    fstype = _mount_fstype(st.st_dev, path)
    if fstype is None:
        return False, "cannot identify the filesystem backing %s (device %d:%d)" % (
            path,
            os.major(st.st_dev),
            os.minor(st.st_dev),
        )
    if fstype.startswith(_UNTRUSTED_FSTYPE_PREFIXES):
        return False, "%s lives on a %s mount, whose owner asserts its uid and mode" % (
            path,
            fstype,
        )
    return True, ""

# Superblock magic numbers (statfs f_type) of the filesystems whose inode
# attributes a user-space server asserts. FUSE_SUPER_MAGIC is shared by `fuse`,
# `fuseblk` and every `fuse.<subtype>` alike -- one superblock type serves them
# all -- and FUSE_CTL_SUPER_MAGIC is the `fusectl` control filesystem, so the
# pair covers exactly the set _UNTRUSTED_FSTYPE_PREFIXES names.
_FUSE_SUPER_MAGIC = 0x65735546
_FUSE_CTL_SUPER_MAGIC = 0x65735543
_UNTRUSTED_SUPER_MAGICS = (_FUSE_SUPER_MAGIC, _FUSE_CTL_SUPER_MAGIC)

class _StructStatfs(ctypes.Structure):
    """glibc `struct statfs` as laid out on LP64 Linux (x86_64, aarch64).

`__fsword_t`, `fsblkcnt_t` and `fsfilcnt_t` are all 64-bit words there, which
    also makes `statfs` and `statfs64` the same layout, so one declaration serves
    both. Spelled out in full rather than as "f_type plus opaque padding" so the
    buffer the kernel fills is provably the right size (120 bytes); only f_type
    is ever read. _load_fstatfs refuses to bind on any non-LP64 build.
    """

_fields_ = [
        ("f_type", ctypes.c_long),
        ("f_bsize", ctypes.c_long),
        ("f_blocks", ctypes.c_ulong),
        ("f_bfree", ctypes.c_ulong),
        ("f_bavail", ctypes.c_ulong),
        ("f_files", ctypes.c_ulong),
        ("f_ffree", ctypes.c_ulong),
        ("f_fsid", ctypes.c_int * 2),
        ("f_namelen", ctypes.c_long),
        ("f_frsize", ctypes.c_long),
        ("f_flags", ctypes.c_long),
        ("f_spare", ctypes.c_long * 4),
    ]

def _load_fstatfs():
    """Bind libc's fstatfs(2), or return None when it cannot be bound safely."""
    if ctypes.sizeof(ctypes.c_long) != 8:
        # _StructStatfs is LP64-only; guessing another ABI's layout would read
        # the wrong word and mis-identify the filesystem.
        return None
    try:
        # dlopen(NULL): the interpreter is already linked against libc, so this
        # resolves fstatfs without depending on a soname or on ldconfig.
        fstatfs = ctypes.CDLL(None, use_errno=True).fstatfs
    except (OSError, AttributeError):
        return None
    fstatfs.argtypes = [ctypes.c_int, ctypes.POINTER(_StructStatfs)]
    fstatfs.restype = ctypes.c_int
    return fstatfs

_FSTATFS = _load_fstatfs()

def _fd_super_magic(fd):
    """Return the 32-bit superblock magic (statfs f_type) behind `fd`.

Read with fstatfs(2), i.e. from the SUPERBLOCK the kernel already has open
    for this very inode, never from a mount table -- a mount table entry is not
    a property of the inode. Its owner can drop the entry with MNT_DETACH (setuid
    /usr/bin/fusermount exposes exactly that as `-u -z`) while the superblock
    keeps serving descriptors already opened against it, so a device missing from
    /proc/self/mountinfo proves nothing about the filesystem, and resolving it by
    enclosing mount point -- as _mount_fstype must, so a btrfs subvolume's
    anonymous device still resolves -- would credit such an inode with the type
    of "/". The magic stays right in both cases: detached FUSE still reads
    FUSE_SUPER_MAGIC and a btrfs subvolume still reads BTRFS_SUPER_MAGIC, so this
    needs no fallback at all.

ctypes because os.statvfs exposes no f_type and the stdlib wraps no other
    route to it. Raises OSError when fstatfs cannot answer -- a failing call, an
    unbindable libc, or a non-LP64 build -- which callers must treat as untrusted.
    `f_type` is a signed 64-bit word here while the magics are 32-bit constants,
    hence the mask.
    """
    if _FSTATFS is None:
        raise OSError("fstatfs is unavailable on this build")
    buf = _StructStatfs()
    if _FSTATFS(fd, ctypes.byref(buf)) != 0:
        err = ctypes.get_errno()
        raise OSError(err, os.strerror(err))
    return buf.f_type & 0xFFFFFFFF

def _is_trusted_fd_filesystem(fd, path):
    """Return (trusted, reason) for the filesystem serving an open descriptor.

The fd counterpart of _is_trusted_filesystem, and the authority for the leaf
    inode whose bytes reach exec(): it identifies the filesystem from that
    inode's own superblock (_fd_super_magic) rather than from a mount table, so
    a mount detached out of /proc/self/mountinfo cannot launder its type into
    something trusted. FAIL-CLOSED when the superblock cannot be read, since an
    unidentified filesystem is precisely the state this gate exists to refuse.
    """
    try:
        magic = _fd_super_magic(fd)
    except OSError as exc:
        return False, "cannot identify the filesystem backing %s: %s" % (path, exc)
    if magic in _UNTRUSTED_SUPER_MAGICS:
        return False, "%s lives on a user-space-served mount (fs magic 0x%x), whose owner asserts its uid and mode" % (
            path,
            magic,
        )
    return True, ""

def _is_trusted_fd(fd, path):
    """Return (trusted, reason) for an already-open file descriptor.

The decision is made against the inode behind `fd` (via fstat), so it binds
    to the exact object that will be read/executed -- not to whatever the path
    string re-resolves to afterwards. `path` is used only for messages. A plugin
    is trusted only when owned by root (uid 0), not group/world-writable, and
    backed by a filesystem whose attributes the kernel asserts rather than a
    user-space server (_is_trusted_fd_filesystem, which reads that from the fd's
    own superblock so a detached mount cannot launder it). Fail-closed on any
    error.
    """
    try:
        st = os.fstat(fd)
    except OSError as exc:
        return False, "cannot stat %s: %s" % (path, exc)
    if st.st_uid != 0:
        return False, "%s is owned by uid %d, not root" % (path, st.st_uid)
    if st.st_mode & 0o022 != 0:
        return False, "%s is group- or world-writable (mode %o)" % (
            path,
            st.st_mode & 0o7777,
        )
    # The uid and mode above are only evidence if the kernel produced them.
    return _is_trusted_fd_filesystem(fd, path)

def _is_trusted_plugin_file(path):
    """Return (trusted, reason) for a plugin .py before it is imported.

Mirrors the Rust check_plugin_path_trust helper. Opens `path` (following a
    symlink once, so the product's supported opt-in enablement -- a
    plugins.other/*.py symlinked into plugins_dir -- keeps working) and
    validates the opened fd's inode, so the check applies to the exact file
    behind the descriptor. Binding to that fd already defeats a swap after open.
    Fail-closed on any error.
    """
    try:
        # O_NONBLOCK for the same reason as the discovery open: a FIFO must not
        # be able to block the gate itself.
        fd = os.open(path, os.O_RDONLY | os.O_NONBLOCK)
    except OSError as exc:
        return False, "cannot open %s: %s" % (path, exc)
    try:
        return _is_trusted_fd(fd, path)
    finally:
        os.close(fd)

def _is_trusted_ancestry(plugins_dir):
    """Return (trusted, reason) for plugins_dir's whole ancestor chain.

Defense-in-depth pre-check (F-23) layered on top of the per-inode fd gate
    (_is_trusted_fd): canonicalize plugins_dir with realpath, then walk every
    directory from the resolved plugins_dir up to "/" and require each to be
    owned by root (uid 0) and not group/world-writable. A writable ancestor
    lets an unprivileged user rename the trusted plugins_dir aside and drop in
    their own tree, so we refuse to scan at all when any ancestor is
    attacker-writable, rather than relying solely on the per-file gate.

Exception: a group/world-writable ancestor that carries the sticky bit
    (S_ISVTX) is accepted. The sticky bit stops non-owners from renaming or
    deleting entries they do not own, which is exactly the rename-swap the walk
    guards against; a shared sticky root (e.g. the system temp dirs) therefore
    cannot be used to swap the root-owned plugins_dir aside. A writable ancestor
    WITHOUT the sticky bit stays rejected. Fail-closed on any error.

Filesystem-origin trust IS enforced, on every component (F-07): a mount
    served from user space answers both the getattr behind the stat below and
    the read that later yields the plugin bytes, so on such a mount the
    root-ownership gate proves nothing -- and creating a directory inside a
    shared sticky root is exactly how an unprivileged user obtains a mount point
    it owns, which is what makes the carve-out above safe only in combination
    with this check. os.statvfs carries no filesystem type, but the inode's
    kernel-assigned st_dev resolved through /proc/self/mountinfo does; see
    _is_trusted_filesystem.
    """
    path = os.path.realpath(plugins_dir)
    while True:
        try:
            st = os.stat(path)
        except OSError as exc:
            return False, "cannot stat ancestor %s: %s" % (path, exc)
        if st.st_uid != 0:
            return False, "ancestor %s is owned by uid %d, not root" % (
                path,
                st.st_uid,
            )
        if st.st_mode & 0o022 != 0 and (st.st_mode & 0o1000) == 0:
            # Writable and NOT sticky: a non-owner could rename/swap entries.
            return False, "ancestor %s is group- or world-writable without the sticky bit (mode %o)" % (
                path,
                st.st_mode & 0o7777,
            )
        # `reason` already names the offending component.
        trusted, reason = _is_trusted_filesystem(st, path)
        if not trusted:
            return False, reason
        parent = os.path.dirname(path)
        if parent == path:
            return True, ""
        path = parent

def _discover_classes(plugins_dir):
    """Import every .py file in plugins_dir and collect all LveStatsPlugin subclasses.

Mirrors lvestats.core.plugin_loader.PluginLoader: loads every subclass by
    isinstance check (not by filename), registers modules in sys.modules so
    sibling imports work, and skips broken symlinks with a warning.
    """
    descriptors = []
    classes = {}

# Defense-in-depth (F-23): before opening the directory, refuse to scan when
    # any ancestor of the canonicalized plugins_dir is not root-owned or is
    # group/world-writable -- an attacker-writable ancestor would let an
    # unprivileged user swap the whole directory out. The per-inode fd gate
    # below stays the authoritative check; this only fails closed earlier.
    trusted, reason = _is_trusted_ancestry(plugins_dir)
    if not trusted:
        sys.stderr.write(
            "Warning: refusing untrusted plugins directory %s: %s\n"
            % (plugins_dir, reason)
        )
        return descriptors, classes

# Open the directory itself with O_DIRECTORY and validate the opened inode;
    # every candidate is then opened relative to this fd (openat) and executed
    # from the bytes read off that fd -- NOT by re-opening the path string. This
    # binds the trust check and the code that actually runs to the same inodes,
    # closing the stat-then-open TOCTOU window an attacker could exploit by
    # swapping the directory or a plugin file between check and use. Symlinks are
    # followed once and judged on their resolved inode (so the product's opt-in
    # plugins.other/*.py symlinks still load); the fd binding makes a
    # swap-after-open ineffective, so O_NOFOLLOW is not needed here.
    try:
        dir_fd = os.open(plugins_dir, os.O_RDONLY | os.O_DIRECTORY)
    except OSError as exc:
        sys.stderr.write(
            "Warning: cannot open plugins directory %s: %s\n" % (plugins_dir, exc)
        )
        return descriptors, classes

try:
        trusted, reason = _is_trusted_fd(dir_fd, plugins_dir)
        if not trusted:
            sys.stderr.write(
                "Warning: refusing untrusted plugins directory %s: %s\n"
                % (plugins_dir, reason)
            )
            return descriptors, classes

for filename in sorted(os.listdir(dir_fd)):
            if not filename.endswith(".py"):
                continue
            full_path = os.path.join(plugins_dir, filename)

try:
                # O_NONBLOCK so a planted FIFO cannot park the runner inside
                # open() before the file-type check below can reject it; it has
                # no effect on the regular files we actually load.
                fd = os.open(
                    filename, os.O_RDONLY | os.O_NONBLOCK, dir_fd=dir_fd
                )
            except OSError as exc:
                sys.stderr.write(
                    "Warning: refusing untrusted plugin %s: %s\n" % (full_path, exc)
                )
                continue
            try:
                trusted, reason = _is_trusted_fd(fd, full_path)
                if not trusted:
                    sys.stderr.write(
                        "Warning: refusing untrusted plugin %s: %s\n"
                        % (full_path, reason)
                    )
                    continue
                # The fd gate above trusts only the target inode; a symlink can
                # point at a root-owned .py whose ancestor directory is
                # attacker-writable. Walk the resolved file's ancestry (matching
                # Rust check_ancestor_chain_trust) and refuse before exec so we
                # never run a plugin the Rust ancestor filter would drop.
                trusted, reason = _is_trusted_ancestry(full_path)
                if not trusted:
                    sys.stderr.write(
                        "Warning: refusing untrusted plugin %s: %s\n"
                        % (full_path, reason)
                    )
                    continue
                # Mirror Rust plugin_entry_trusted (main.rs:640): only a regular
                # file may be exec'd, so a directory, device or FIFO named *.py
                # is refused rather than read.
                if not stat.S_ISREG(os.fstat(fd).st_mode):
                    sys.stderr.write(
                        "Warning: refusing untrusted plugin %s: not a regular file\n"
                        % full_path
                    )
                    continue
                try:
                    with os.fdopen(os.dup(fd), "rb") as fh:
                        source = fh.read()
                except OSError as exc:
                    sys.stderr.write(
                        "Warning: failed to read %s: %s\n" % (full_path, exc)
                    )
                    continue
            finally:
                os.close(fd)

module_name = filename[:-3]
            try:
                spec = importlib.util.spec_from_file_location(module_name, full_path)
                if spec is None:
                    continue
                module = importlib.util.module_from_spec(spec)
                # Register before exec so sibling modules can find each other.
                sys.modules[module_name] = module
                # Execute the bytes read from the validated fd, not the path, so
                # the code that runs is the inode we trust-checked above.
                exec(compile(source, full_path, "exec"), module.__dict__)
            except Exception as exc:
                sys.stderr.write("Warning: failed to import %s: %s\n" % (full_path, exc))
                sys.modules.pop(module_name, None)
                continue

for name, obj in inspect.getmembers(module, inspect.isclass):
                if not issubclass(obj, LveStatsPlugin) or obj is LveStatsPlugin:
                    continue
                # Skip classes re-exported from other modules (match plugin_loader.py).
                if getattr(obj, "__module__", None) != module_name:
                    continue
                classes[name] = obj
                descriptors.append({
                    "file": filename,
                    "class": name,
                    # CLOS-6853: `order` MUST default to None, not 0. An absent
                    # `order` means "the daemon decides", and the daemon's
                    # DEFAULT_PYTHON_PLUGIN_ORDER (15000) puts the plugin after
                    # every built-in — matching lvestats.main.init_plugins,
                    # which sorts orderless user plugins with
                    # LOW_PRIORITY = sys.maxsize. Reporting a concrete 0 made
                    # that default unreachable and ran user plugins ahead of the
                    # collectors, analyzer and aggregator whose output they
                    # consume, so every lve_data key they read was a tick stale
                    # and `lve_usage` was always the aggregator's cleared value.
                    "order": getattr(obj, "order", None),
                    "period": getattr(obj, "period", None),
                    "timeout": getattr(obj, "timeout", None),
                })
    finally:
        os.close(dir_fd)

return descriptors, classes

def op_list(req):
    plugins_dir = req.get("plugins_dir")
    if not plugins_dir:
        return {"status": "error", "message": "list: plugins_dir missing"}
    descriptors, classes = _discover_classes(plugins_dir)
    _classes.clear()
    _classes.update(classes)
    # Instances tied to previously-loaded classes are no longer valid.
    _instances.clear()
    _scratch.clear()
    return {"status": "ok", "plugins": descriptors}

def op_init(req):
    class_name = req.get("class")
    if not class_name:
        return {"status": "error", "message": "init: class missing"}
    cls = _classes.get(class_name)
    if cls is None:
        return {
            "status": "error",
            "message": "class %r not loaded; send list first" % class_name,
        }

try:
        cls.__is_user_plugin__ = bool(req.get("is_user_plugin", True))
        instance = cls()

plugin_config = req.get("plugin_config") or {}
        if hasattr(instance, "set_config"):
            instance.set_config(plugin_config)

db_url = req.get("db_url")
        if db_url and hasattr(instance, "set_db_engine"):
            try:
                from sqlalchemy import create_engine  # noqa: WPS433
                engine = create_engine(db_url)
                instance.set_db_engine(engine)
            except Exception as exc:
                # Plugins that actually use the engine will fail later; log and continue.
                sys.stderr.write(
                    "Warning: set_db_engine failed for %s: %s\n" % (class_name, exc)
                )

_instances[class_name] = instance
        # Re-initializing a plugin discards any prior scratch state.
        _scratch.pop(class_name, None)
        return {"status": "ok"}
    except Exception as exc:
        return {
            "status": "error",
            "message": str(exc),
            "traceback": traceback.format_exc(),
        }

def op_execute(req):
    class_name = req.get("class")
    if not class_name:
        return {"status": "error", "message": "execute: class missing"}
    instance = _instances.get(class_name)
    if instance is None:
        return {
            "status": "error",
            "message": "plugin %r not initialized; send init first" % class_name,
        }

now = req.get("now", 0.0)
    raw_lve_data = req.get("lve_data") or {}

try:
        instance.now = now
    except Exception:
        pass

# Hydrate dict -> rich-object form so user plugins ported from legacy
    # lve-stats see the LVEStat / LVEUsage / AggregatedLveUsage shape they
    # expect (attribute access, int uid keys, list-shaped lve_usages_5s).
    # Per-plugin scratch dict carries legacy-only keys (old_now, lve_usages,
    # lve_active_ids) across ticks since Rust doesn't model them.
    scratch = _scratch.setdefault(class_name, {})
    lve_data = hydrate(raw_lve_data, scratch=scratch)

# Arm SIGUSR2 only around execute() so an out-of-band signal between
    # requests doesn't kill the runner.
    signal.signal(signal.SIGUSR2, _sigusr2_handler)
    try:
        instance.execute(lve_data)
    except LveStatsPluginTerminated:
        return {"status": "terminated"}
    except _DB_SCHEMA_ERRORS as exc:
        # Tell the Rust side to run ensure_schema() and respawn us; matches
        # the `recover_db()` flow in Python lve-stats
        # (plugin_executors.py:106-114 + plugin_context.py:34-35).
        return {
            "status": "db_schema_error",
            "message": str(exc),
            "traceback": traceback.format_exc(),
        }
    except Exception as exc:
        return {
            "status": "error",
            "message": str(exc),
            "traceback": traceback.format_exc(),
        }
    finally:
        signal.signal(signal.SIGUSR2, signal.SIG_IGN)

# Convert rich objects back to JSON-serializable form. Legacy scratch
    # keys are pulled into `scratch` (already the same dict registered in
    # `_scratch[class_name]`) and stripped from the outbound payload.
    out_lve_data = dehydrate(lve_data, scratch=scratch)
    return {"status": "ok", "lve_data": out_lve_data}

# ---------------------------------------------------------------------------
# Plugin logging bridge
#
# Plugins use the stdlib `logging` module (the Python daemon ran them in a
# process whose root logger `setup_logging()` had already configured). This
# process has no such setup, so without a handler here every record escapes
# through `logging.lastResort`: bare text, WARNING and above only. The daemon
# then sees an untagged stderr line and records it at DEBUG — so `log.error()`
# became invisible at the default log level and `log.info()`/`log.debug()` never
# left the process at all (CLOS-6853).
#
# Fix: tag every line with its severity and logger name so the daemon's stderr
# drainer can re-emit it at the matching `tracing` level. Filtering stays with
# the daemon (its EnvFilter follows the config `log_level`), which is why the
# root logger is opened all the way to DEBUG — mirroring the Python daemon's
# `logging.root.setLevel(logging.DEBUG)`.
# ---------------------------------------------------------------------------

_PLUGIN_LOG_PREFIX = "LVESTATS_PLUGIN_LOG"

class _RustBridgeLogHandler(logging.Handler):
    """Write each record to stderr as `PREFIX|LEVELNAME|logger|text` lines.

A record whose formatted text spans several lines (anything logged via
    `log.exception()`) gets the tag repeated on EVERY line: the daemon
    classifies line by line, so an untagged continuation line would be demoted
    back to DEBUG and split a traceback across two severities.

The message body is written last and unescaped, so pipes inside it are
    harmless — the daemon splits into exactly four fields.
    """

def emit(self, record):
        try:
            text = self.format(record)
            # `or [""]` keeps an empty message as one tagged line instead of
            # vanishing.
            for line in text.splitlines() or [""]:
                sys.stderr.write(
                    "%s|%s|%s|%s\n"
                    % (_PLUGIN_LOG_PREFIX, record.levelname, record.name, line)
                )
            sys.stderr.flush()
        except Exception:
            self.handleError(record)

def _install_plugin_log_bridge():
    """Attach the bridge handler to the root logger. Idempotent."""
    root = logging.getLogger()
    for handler in root.handlers:
        if isinstance(handler, _RustBridgeLogHandler):
            return handler
    handler = _RustBridgeLogHandler()
    # Bare message: the daemon's own log layer supplies timestamp and level.
    handler.setFormatter(logging.Formatter("%(message)s"))
    handler.setLevel(logging.NOTSET)
    root.addHandler(handler)
    root.setLevel(logging.DEBUG)
    return handler

def op_shutdown(_req):
    return {"status": "ok", "_exit": True}

_DISPATCH = {
    "list": op_list,
    "init": op_init,
    "execute": op_execute,
    "shutdown": op_shutdown,
}

def main():
    # Isolate the JSON-RPC channel from anything a plugin (or a library it
    # imports) might write to stdout. The Rust daemon reads this process's
    # stdout line-by-line and parses each line as JSON. A single stray
    # `print(...)`, `os.write(1, ...)`, or C-extension `fprintf(stdout, ...)`
    # corrupts the stream. Clone fd 1 to a private fd used only for protocol
    # writes, then point fd 1 at fd 2 (stderr) so all other writes — including
    # those from C extensions and subprocesses that inherit fd 1 — land in
    # stderr, which the daemon already drains into its logs. Done here (not
    # at import time) so the module can be loaded by tests without clobbering
    # the test process's stdout.
    global _protocol
    _protocol_fd = os.dup(1)
    os.dup2(2, 1)
    _protocol = io.TextIOWrapper(
        os.fdopen(_protocol_fd, "wb", buffering=0),
        encoding="utf-8",
        write_through=True,
    )
    # Rebind Python-level stdout too so `print()` without an explicit `file=`
    # goes to stderr instead of leaving the stream partly redirected.
    sys.stdout = sys.stderr

# Route plugin log records to the daemon with their severity intact. Done
    # after the stdio rebinding above so the handler captures the final stderr.
    _install_plugin_log_bridge()

signal.signal(signal.SIGUSR1, _sigusr1_handler)
    signal.signal(signal.SIGUSR2, signal.SIG_IGN)

for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            req = json.loads(line)
        except json.JSONDecodeError as exc:
            resp = {"status": "error", "message": "invalid json: %s" % exc}
            _protocol.write(json.dumps(resp) + "\n")
            _protocol.flush()
            continue

op = req.get("op")
        handler = _DISPATCH.get(op)
        if handler is None:
            resp = {"status": "error", "message": "unknown op %r" % op}
        else:
            try:
                resp = handler(req)
            except Exception as exc:
                resp = {
                    "status": "error",
                    "message": str(exc),
                    "traceback": traceback.format_exc(),
                }

exit_after = resp.pop("_exit", False)
        _protocol.write(json.dumps(resp) + "\n")
        _protocol.flush()
        if exit_after:
            sys.exit(0)

if __name__ == "__main__":
    main()