SWAMPy

✎ CARVE FLESH

EDITING: cagefs.py

# coding=utf-8
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2026 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENCE.TXT
#
"""Helpers for CageFS interaction from lvectl / lvdctl.

Kept in python_lve to avoid a hard dependency on securelve: lve-utils
is installed on systems without CageFS, so anything we call here must
degrade to a no-op when CageFS is absent.
"""

import logging
import os
import subprocess
import tempfile

CAGEFSCTL_TOOL = "/usr/sbin/cagefsctl"
PROXY_COMMANDS_PATH = "/etc/cagefs/proxy.commands"

# proxyexec aliases that map an in-CageFS path to a host-side SUID binary.
# Without these entries, isolatectl inside CageFS fails with
# "No such file or directory: '/usr/share/lve-utils/lvd-registry-helper'".
LVD_PROXY_ENTRIES = {
    "LVD_REGISTRY_HELPER": "/usr/share/lve-utils/lvd-registry-helper",
    "LVD_LIMITS_HELPER": "/usr/share/lve-utils/lvd-limits-helper",
    "LVD_STATS_HELPER": "/usr/share/lve-utils/lvd-stats-helper",
}

def ensure_lvd_proxy_commands():
    """Register LVD helper proxyexec entries in /etc/cagefs/proxy.commands.

No-op when CageFS is not installed (cagefsctl binary absent) or when
    the entries are already present. When entries are added, runs
    ``cagefsctl --update-wrappers`` so the in-CageFS proxyexec wrappers
    appear immediately.
    """
    if not os.path.exists(CAGEFSCTL_TOOL):
        return

try:
        with open(PROXY_COMMANDS_PATH, "r", encoding="utf-8") as f:
            content = f.read()
    except FileNotFoundError:
        content = ""

new_content = content
    for key, binary in LVD_PROXY_ENTRIES.items():
        if key in new_content:
            continue
        if not os.path.exists(binary):
            continue
        if new_content and not new_content.endswith("\n"):
            new_content += "\n"
        new_content += f"{key}={binary}\n"

if new_content == content:
        return

logging.info("Registering LVD helpers in %s", PROXY_COMMANDS_PATH)

proxy_dir = os.path.dirname(PROXY_COMMANDS_PATH)
    os.makedirs(proxy_dir, exist_ok=True)
    fd, tmp_path = tempfile.mkstemp(dir=proxy_dir, prefix=".proxy.commands.")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write(new_content)
        os.replace(tmp_path, PROXY_COMMANDS_PATH)
    except BaseException:
        if os.path.exists(tmp_path):
            os.unlink(tmp_path)
        raise

subprocess.run(
        [CAGEFSCTL_TOOL, "--update-wrappers"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=False,
    )

CANCEL

Name	Type	Size	Modified	Actions
↩ ..	DIR	—	—
📁 __pycache__/	DIR	—	2026-11-02 22:52
📁 hooks/	DIR	—	2026-11-02 22:52
📄 __init__.py	PY	302 B	2026-10-16 11:34	EDIT
📄 cagefs.py	PY	2.6 KB	2026-10-16 11:34	EDIT
📄 ipc_isolation.py	PY	8.1 KB	2026-10-16 11:34	EDIT
📄 pylve_wrapper.py	PY	15.2 KB	2026-10-16 11:34	EDIT
📄 sentry.py	PY	913 B	2026-10-16 11:34	EDIT