?HAMA

Edit File Size: 2.52 KB

/opt/cloudlinux/venv/lib64/python3.11/site-packages/lve_utils/cagefs.py

Text Base64

# 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",
}

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,
    )

Exit & Reset Text mode: syntax highlighting auto-detects file type.

Name	Size	Perms	Modified	Actions
hooks DIR	-	drwxr-xr-x	2026-06-03 07:00:45	Edit Download recursive recursive
__pycache__ DIR	-	drwxr-xr-x	2026-06-03 07:00:46	Edit Download recursive recursive
cagefs.py	2.52 KB	lrw-r--r--	2026-05-20 18:04:05	Edit Download recursive recursive
pylve_wrapper.py	10.75 KB	lrw-r--r--	2026-05-20 18:04:05	Edit Download recursive recursive
sentry.py	913 B	lrw-r--r--	2026-05-20 18:04:05	Edit Download recursive recursive
__init__.py	302 B	lrw-r--r--	2026-05-20 18:04:05	Edit Download recursive recursive

Directory Contents