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