23s2s

✏️ Editing: ipc_isolation.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/LICENSE.TXT

"""Keep kernel.lve_ipc_isolation in step with the lve-ipc-isolation feature flag.

The decision point of this feature is inside kmod-lve: the module reads the
sysctl when it sets up a per-user LVE, so it cannot ask clflags.is_enabled()
itself. This module asks on its behalf and writes the answer where the module
reads it:

- persisted in /etc/sysctl.d/90-cloudlinux.conf, which lve.service re-applies
  right after it loads the module - so the value is in place before the first
  LVE of every boot is created;
- live in /proc/sys/kernel/lve_ipc_isolation, while the module is loaded.

The feature ships off, and "off" means the kernel default: the line is dropped
from the config rather than pinned to 0, so a server that never opted in
carries no setting for a key its kernel may not have.

Only that one line and that one /proc key are ever written. clcommon's
SysCtlConf is deliberately not used here: its set() and remove() run
`sysctl --system`, which re-applies every sysctl file on the server, and this
helper runs whenever the flag changes - at runtime, not only at install - so
it must not undo settings an administrator changed by hand.

Run by lve-ipc-isolation.service - at boot after lve.service, whenever
lve-ipc-isolation.path sees one of the feature's own flag files change, and
every five minutes from lve-ipc-isolation.timer as a safety net for a change
the path unit missed - and by the lve-utils install scriptlets. Idempotent:
nothing that already matches is touched.
"""

import argparse
import logging
import os
import tempfile
from typing import List, Optional, Tuple

from clcommon.sysctl import SYSCTL_CL_CONF_FILE
from clcommon.utils import get_file_lines
from clflags import is_enabled

from lve_utils.sentry import init_lve_utils_sentry_client

FEATURE = 'lve-ipc-isolation'
SYSCTL_NAME = 'kernel.lve_ipc_isolation'
PROC_PATH = '/proc/sys/kernel/lve_ipc_isolation'
# How the feature ships. is_enabled() falls back to it only while the flag is
# not declared, which with this package installed means a damaged install.
SHIPS_ON = False

logger = logging.getLogger(__name__)

def wanted() -> bool:
    """Whether the flag switches the feature on for this server; asked afresh every time."""
    return is_enabled(FEATURE, default=SHIPS_ON)

def _is_our_line(line: str) -> bool:
    """Whether a sysctl config line sets SYSCTL_NAME (comments and other keys are not ours)."""
    stripped = line.strip()
    if stripped.startswith('#') or '=' not in stripped:
        return False
    key, _, _ = stripped.partition('=')
    return key.strip() == SYSCTL_NAME

def _read_lines(config_file: str) -> List[str]:
    if not os.path.isfile(config_file):
        return []
    return list(get_file_lines(config_file))

def persisted_value(config_file: str = SYSCTL_CL_CONF_FILE) -> Optional[str]:
    """The value the CloudLinux sysctl config carries for the key, or None when it has no line for it."""
    value: Optional[str] = None
    for line in _read_lines(config_file):
        if _is_our_line(line):
            # sysctl applies lines in order, so the last one wins
            value = line.strip().partition('=')[2].strip()
    return value

def persist(value: Optional[str], config_file: str = SYSCTL_CL_CONF_FILE) -> bool:
    """Set (value) or drop (None) the key's line in config_file, leaving every other line as it is.

Written atomically (temp file in the same directory, then rename), so
    lve.service never sees a half-written file at boot. Returns whether the
    file changed.
    """
    lines = _read_lines(config_file)
    kept = [line for line in lines if not _is_our_line(line)]
    if kept and not kept[-1].endswith('\n'):
        kept[-1] += '\n'
    if value is not None:
        kept.append(f'{SYSCTL_NAME} = {value}\n')
    if kept == lines:
        return False

directory = os.path.dirname(config_file) or '.'
    mode = os.stat(config_file).st_mode & 0o777 if os.path.isfile(config_file) else 0o644
    fd, tmp_path = tempfile.mkstemp(prefix='.90-cloudlinux.', dir=directory)
    try:
        with os.fdopen(fd, 'w', encoding='utf-8') as tmp:
            tmp.writelines(kept)
            tmp.flush()
            os.fsync(tmp.fileno())
        os.chmod(tmp_path, mode)
        os.replace(tmp_path, config_file)
    except BaseException:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
        raise
    return True

def live_value(proc_path: str = PROC_PATH) -> Optional[str]:
    """The value the loaded module runs with, or None while kmod-lve is not loaded."""
    try:
        with open(proc_path, encoding='utf-8') as proc_file:
            return proc_file.read().strip()
    except FileNotFoundError:
        return None

def _set_live(value: str, proc_path: str) -> None:
    with open(proc_path, 'w', encoding='utf-8') as proc_file:
        proc_file.write(value)

def _apply_once(config_file: str, proc_path: str) -> Tuple[bool, bool]:
    """One pass: read the flag, bring both copies of the sysctl in line with it.

Returns (the flag state that was applied, whether anything had to change).
    """
    on = wanted()
    value = '1' if on else '0'
    changed = False

if persist('1' if on else None, config_file):
        if on:
            logger.info('%s is on: %s = 1 persisted in %s', FEATURE, SYSCTL_NAME, config_file)
        else:
            logger.info('%s is off: %s dropped from %s', FEATURE, SYSCTL_NAME, config_file)
        changed = True

live = live_value(proc_path)
    if live is None:
        logger.info('%s: kmod-lve is not loaded, %s takes effect once it is', FEATURE, SYSCTL_NAME)
    elif live != value:
        _set_live(value, proc_path)
        logger.info(
            '%s: %s set to %s (was %s); LVEs created from now on follow it', FEATURE, SYSCTL_NAME, value, live
        )
        changed = True

return on, changed

# How many times a pass is repeated when the flag flips while the pass runs.
SETTLE_PASSES = 5

def apply(config_file: str = SYSCTL_CL_CONF_FILE, proc_path: str = PROC_PATH) -> bool:
    """Bring the persisted and the live sysctl in line with the flag.

The flag can change while a pass runs - the path unit does not start the
    service again for an event that arrives during a run - so the pass is
    repeated until the flag reads the same before and after it. Returns
    whether anything had to change.
    """
    changed = False
    for _ in range(SETTLE_PASSES):
        applied, pass_changed = _apply_once(config_file, proc_path)
        changed = changed or pass_changed
        if wanted() == applied:
            break
    return changed

def status(config_file: str = SYSCTL_CL_CONF_FILE, proc_path: str = PROC_PATH) -> str:
    """A short report of the flag and of both copies of the sysctl, for support."""
    persisted = persisted_value(config_file)
    live = live_value(proc_path)
    return '\n'.join(
        (
            f'{FEATURE}: {"on" if wanted() else "off"}',
            f'{SYSCTL_NAME} persisted: {persisted if persisted is not None else "(not set, kernel default 0)"}',
            f'{SYSCTL_NAME} live: {live if live is not None else "(kmod-lve not loaded)"}',
        )
    )

def main(argv: List[str]) -> int:
    parser = argparse.ArgumentParser(
        prog='lve-ipc-isolation',
        description=f'Apply the {FEATURE} feature flag to {SYSCTL_NAME}.',
    )
    parser.add_argument(
        '--status',
        action='store_true',
        help='report the flag and the sysctl without changing anything',
    )
    args = parser.parse_args(argv)

logging.basicConfig(level=logging.INFO, format='%(message)s')
    if args.status:
        print(status())
        return 0

init_lve_utils_sentry_client('lve-ipc-isolation')  # type: ignore[no-untyped-call]
    try:
        apply()
    except OSError as err:
        # An unwritable /etc or /proc must not take the boot unit or the package
        # scriptlet down with it; the next run repeats the attempt.
        logger.error('%s: could not apply %s: %s', FEATURE, SYSCTL_NAME, err)
        return 1
    return 0

Sid Gifari From Gifari Industries - BD Cyber Security Team