SWAMPy

✎ CARVE FLESH

EDITING: domain.py

#!/opt/cloudlinux/venv/bin/python3 -sbb
# -*- coding: utf-8 -*-
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENCE.TXT
#
import functools
import logging
import os
import shutil
import subprocess
import tempfile
from collections import defaultdict
from pathlib import Path

from clcommon import ClPwd
from clcommon.clcagefs import setup_mount_dir_cagefs, CAGEFSCTL_TOOL
from clcommon.cpapi import cpusers, is_panel_feature_supported, userdomains
from clcommon.cpapi import docroot as get_domain_docroot
from clcommon.const import Feature
from clcommon.cpapi.cpapiexceptions import NoDomain

from lve_utils.pylve_wrapper import PyLve

from .fs import user_exists
from .exceptions import UserNotFoundError
from .webisolation import admin_config, config, jail_utils, litespeed
from .webisolation.config import DOCROOTS_ISOLATED_BASE
from .webisolation.docroot_validation import validate_docroot, validate_docroot_no_symlinks
from .webisolation.jail_config_builder import write_jail_mounts_config
from .webisolation.php import reload_processes_with_docroots
from .webisolation.service import start_monitoring_service, stop_monitoring_service
from .webisolation.triggers import trigger_xray_ini_regeneration, trigger_ssa_ini_regeneration

def is_website_isolation_allowed_server_wide():
    return os.path.isfile(admin_config.WEBSITE_ISOLATION_MARKER)

def is_website_isolation_feature_available():
    return os.path.isfile(admin_config.WEBSITE_ISOLATION_AVAILABLE_MARKER)

def get_isolation_user_mode() -> str | None:
    """Return the current user mode for website isolation.

Returns:
        ``"allow_all"`` – all users allowed, denied dir lists exceptions.
        ``"deny_all"``  – no users allowed, allowed dir lists exceptions.
        ``None``        – not initialised yet.
    """
    has_denied = os.path.isdir(admin_config.ISOLATION_DENIED_DIR)
    has_allowed = os.path.isdir(admin_config.ISOLATION_ALLOWED_DIR)
    if has_denied and has_allowed:
        # Error state – both dirs present.  Treat as allow_all and clean up.
        logging.warning(
            "Both site-isolation.users.allowed and site-isolation.users.denied "
            "directories exist. Removing allowed directory, treating as allow_all mode."
        )
        shutil.rmtree(admin_config.ISOLATION_ALLOWED_DIR, ignore_errors=True)
        return "allow_all"
    if has_denied:
        return "allow_all"
    if has_allowed:
        return "deny_all"
    return None

def is_website_isolation_allowed_for_user(user: str) -> bool:
    """Check whether *user* is allowed to use website isolation.

Combines the global marker with the two-mode user model:
    * **allow_all** – allowed unless the user is in the denied directory.
    * **deny_all**  – denied unless the user is in the allowed directory.
    """
    if not os.path.isfile(admin_config.WEBSITE_ISOLATION_MARKER):
        return False
    mode = get_isolation_user_mode()
    if mode == "allow_all":
        return not admin_config.user_in_dir(admin_config.ISOLATION_DENIED_DIR, user)
    if mode == "deny_all":
        return admin_config.user_in_dir(admin_config.ISOLATION_ALLOWED_DIR, user)
    return False  # not initialised

def _ensure_isolation_mount_and_marker():
    """Set up mount directories and the global marker if not already done."""
    if not os.path.isfile(admin_config.WEBSITE_ISOLATION_MARKER):
        setup_mount_dir_cagefs(
            str(DOCROOTS_ISOLATED_BASE), prefix="*",
            remount_cagefs=True, remount_in_background=False,
        )
        marker_path = Path(admin_config.WEBSITE_ISOLATION_MARKER)
        marker_path.parent.mkdir(parents=True, exist_ok=True)
        marker_path.touch()
        # Remount kills redis processes, restart clwpos_monitoring service if it's running
        subprocess.run(
            ["/usr/bin/systemctl", "try-restart", "clwpos_monitoring.service"],
            capture_output=True,
            text=True
        )

PROXY_COMMANDS_PATH = "/etc/cagefs/proxy.commands"
CAGEFSCTL_USER_PROXY_ENTRY = "CAGEFSCTL_USER:noproceed:nolve=root:/usr/sbin/cagefsctl-user"

CAGEFSCTL_USER_BINARIES = [
    "/usr/sbin/cagefsctl-user",
]

LVD_PROXY_ENTRIES = {
    "LVD_REGISTRY_HELPER": "/usr/share/lve-utils/lvd-registry-helper",
    "LVD_LIMITS_HELPER": "/usr/share/lve-utils/lvd-limits-helper",
}

def _add_proxy_entries(content, entries):
    """Append missing proxy entries to *content*, return (new_content, added).

Each entry in *entries* is a ``{KEY: path}`` dict.  An entry is added
    only when its KEY is not yet present AND the binary exists on disk.
    Returns the (possibly updated) content string and a bool indicating
    whether any entries were added.
    """
    added = False
    for key, binary in entries.items():
        if key in content:
            continue
        if not os.path.exists(binary):
            continue
        if content and not content.endswith("\n"):
            content += "\n"
        content += f"{key}={binary}\n"
        added = True
    return content, added

def _remove_proxy_entries(content, entries):
    """Remove proxy entries whose binaries no longer exist on disk.

Returns the (possibly updated) content string and a bool indicating
    whether any entries were removed.
    """
    removed = False
    for key, binary in entries.items():
        if key not in content:
            continue
        if os.path.exists(binary):
            continue
        lines = content.splitlines(keepends=True)
        lines = [l for l in lines if not l.startswith(f"{key}=")]
        content = "".join(lines)
        removed = True
    return content, removed

def ensure_proxyexec_command():
    """Register the ``cagefsctl-user`` proxyexec alias if not already present.

Appends the ``CAGEFSCTL_USER`` entry to ``/etc/cagefs/proxy.commands``
    and runs ``cagefsctl --update-list`` to pull the required binaries into
    the CageFS skeleton.  This is a no-op when the entry already exists.

Also registers/unregisters the LVD helper binaries
    (``lvd-registry-helper``, ``lvd-limits-helper``) depending on whether
    they are present on disk.
    """
    try:
        with open(PROXY_COMMANDS_PATH, "r", encoding="utf-8") as f:
            content = f.read()
    except FileNotFoundError:
        content = ""

changed = False

if "CAGEFSCTL_USER" not in content:
        logging.info("Registering cagefsctl-user in %s", PROXY_COMMANDS_PATH)
        if content and not content.endswith("\n"):
            content += "\n"
        content += CAGEFSCTL_USER_PROXY_ENTRY + "\n"
        changed = True

content, added = _add_proxy_entries(content, LVD_PROXY_ENTRIES)
    if added:
        logging.info("Registering LVD helpers in %s", PROXY_COMMANDS_PATH)
        changed = True

content, removed = _remove_proxy_entries(content, LVD_PROXY_ENTRIES)
    if removed:
        logging.info("Removing stale LVD helpers from %s", PROXY_COMMANDS_PATH)
        changed = True

if not changed:
        return

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(content)
        os.replace(tmp_path, PROXY_COMMANDS_PATH)
    except BaseException:
        os.unlink(tmp_path)
        raise

update_list = ("\n".join(CAGEFSCTL_USER_BINARIES) + "\n").encode()
    subprocess.run(
        [CAGEFSCTL_TOOL, "--wait-lock", "--update-list"],
        input=update_list,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=False,
    )

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

def toggle_isolation_user_mode() -> str:
    """Flip the isolation user mode without modifying any per-user state.

Unlike :func:`allow_website_isolation_server_wide` and
    :func:`deny_website_isolation_server_wide`, this function only flips
    the mode indicator directories.  It does **not** clean up existing
    user isolation or alter the per-user exception lists.

* ``allow_all`` → ``deny_all``
    * ``deny_all``  → ``allow_all``
    * not initialised → ``allow_all``

Returns:
        The new mode after toggling (``"allow_all"`` or ``"deny_all"``).
    """
    _ensure_isolation_mount_and_marker()
    current = get_isolation_user_mode()
    if current == "allow_all":
        new_mode = "deny_all"
        os.makedirs(admin_config.ISOLATION_ALLOWED_DIR, mode=admin_config.DIR_MODE, exist_ok=True)
        shutil.rmtree(admin_config.ISOLATION_DENIED_DIR, ignore_errors=True)
    else:
        # deny_all or not initialised → allow_all
        new_mode = "allow_all"
        os.makedirs(admin_config.ISOLATION_DENIED_DIR, mode=admin_config.DIR_MODE, exist_ok=True)
        shutil.rmtree(admin_config.ISOLATION_ALLOWED_DIR, ignore_errors=True)
    return new_mode

# CLOS-4614: lve-stats3 ships this marker once it can chart per-domain
# (website-isolation) LVE usage. Until then cagefs must NOT auto allow/enable
# LVE per-domain limits: the per-user isolation LVP (lvp<uid>) + domain LVEs it
# would create are misattributed by lve-stats (CLOS-4608, usage graph) and
# destabilise the user's enforced limit (CLOS-4609). CageFS website isolation
# (the filesystem separation) is independent of this and always works; only the
# trailing lvectl domain-limit calls are gated.
LVESTATS_SUPPORTS_DOMAIN_LVE_LIMITS_MARKER = (
    '/opt/cloudlinux/flags/available-flags.d/'
    'lvestats-supports-domain-lve-limits.flag'
)

@functools.lru_cache(maxsize=1)
def _domain_limits_supported():
    """Check if the running kernel supports per-domain LVE limits."""
    try:
        py = PyLve()
    except (AttributeError, OSError):
        return False

if not py.initialize():
        return False

return py.domains_supported()

def _domain_limits_auto_enabled():
    """Whether cagefs should auto allow/enable LVE per-domain limits (CLOS-4614).

True once the installed lve-stats3 can chart per-domain isolation usage (it
    ships ``LVESTATS_SUPPORTS_DOMAIN_LVE_LIMITS_MARKER``). Gates only the AUTO
    *allow/enable*; the deny/disable paths stay active regardless so any LVPs
    created by an earlier build can always be torn down.
    """
    return os.path.exists(LVESTATS_SUPPORTS_DOMAIN_LVE_LIMITS_MARKER)

def _allow_domain_limits(username):
    """Create LVP infrastructure for user via lvectl (idempotent)."""
    if not is_panel_feature_supported(Feature.LVE):
        return
    if not _domain_limits_supported():
        return
    if not _domain_limits_auto_enabled():
        # CLOS-4614: paused until lve-stats3 can chart per-domain usage.
        return
    subprocess.run(['/usr/sbin/lvectl', 'allow-domain-limits', username],
                   check=True, capture_output=True)

def _deny_domain_limits(username):
    """Tear down LVP infrastructure for user via lvectl."""
    if not is_panel_feature_supported(Feature.LVE):
        return
    if not _domain_limits_supported():
        return
    subprocess.run(['/usr/sbin/lvectl', 'deny-domain-limits', username],
                   check=True, capture_output=True)

def _register_domain_lve(user, domain):
    """Register domain LVE under user's LVP via lvectl."""
    if not is_panel_feature_supported(Feature.LVE):
        return
    if not _domain_limits_supported():
        return
    if not _domain_limits_auto_enabled():
        # CLOS-4614: paused until lve-stats3 can chart per-domain usage.
        return
    subprocess.run(
        ['/usr/sbin/lvectl', 'enable-domain-limits', domain],
        check=True, capture_output=True,
    )

def _unregister_domain_lve(user, domain):
    """Unregister domain LVE via lvectl."""
    if not is_panel_feature_supported(Feature.LVE):
        return
    if not _domain_limits_supported():
        return
    subprocess.run(
        ['/usr/sbin/lvectl', 'disable-domain-limits', domain],
        check=True, capture_output=True,
    )

def allow_website_isolation_server_wide():
    """Switch to *allow_all* mode – all users are allowed by default."""
    _ensure_isolation_mount_and_marker()
    ensure_proxyexec_command()
    # Create empty denied-users directory → allow_all mode indicator
    os.makedirs(admin_config.ISOLATION_DENIED_DIR, mode=admin_config.DIR_MODE, exist_ok=True)
    # Remove allowed-users directory (belongs to deny_all mode)
    shutil.rmtree(admin_config.ISOLATION_ALLOWED_DIR, ignore_errors=True)

for username in cpusers():
        if not user_exists(username):
            continue

try:
            _allow_domain_limits(username)
        except (subprocess.CalledProcessError, OSError):
            logging.exception(
                "Failed to enable domain limits for user %s, skipping.",
                username,
            )

def deny_website_isolation_server_wide():
    """Switch to *deny_all* mode – no users are allowed by default.

Disables domain isolation for every user and switches the mode.
    """
    _cleanup_all_users_isolation()

for username in cpusers():
        if not user_exists(username):
            continue

try:
            _deny_domain_limits(username)
        except (subprocess.CalledProcessError, OSError):
            logging.exception(
                "Failed to disable domain limits for user %s, skipping.",
                username,
            )

_ensure_isolation_mount_and_marker()
    # Create empty allowed-users directory → deny_all mode indicator
    os.makedirs(admin_config.ISOLATION_ALLOWED_DIR, mode=admin_config.DIR_MODE, exist_ok=True)
    # Remove denied-users directory (belongs to allow_all mode)
    shutil.rmtree(admin_config.ISOLATION_DENIED_DIR, ignore_errors=True)

def allow_website_isolation_for_user(username: str):
    """Allow website isolation for *username* (mode-aware).

* **allow_all** – removes *username* from the denied directory.
    * **deny_all**  – adds *username* to the allowed directory.
    * **not initialised** – sets up infrastructure in *deny_all* mode
      with *username* as the first allowed user.
    """
    _ensure_isolation_mount_and_marker()
    ensure_proxyexec_command()
    mode = get_isolation_user_mode()
    if mode == "allow_all":
        admin_config.remove_user_from_dir(admin_config.ISOLATION_DENIED_DIR, username)
    elif mode == "deny_all":
        admin_config.add_user_to_dir(admin_config.ISOLATION_ALLOWED_DIR, username)
    else:
        # Not initialised → start in deny_all mode with this user allowed
        os.makedirs(admin_config.ISOLATION_ALLOWED_DIR, mode=admin_config.DIR_MODE, exist_ok=True)
        admin_config.add_user_to_dir(admin_config.ISOLATION_ALLOWED_DIR, username)

_allow_domain_limits(username)

def deny_website_isolation_for_user(username: str):
    """Deny website isolation for *username* (mode-aware).

* **allow_all** – adds *username* to the denied directory.
    * **deny_all**  – removes *username* from the allowed directory.

Also disables all domain isolation for the user.
    """
    mode = get_isolation_user_mode()
    if mode == "allow_all":
        admin_config.add_user_to_dir(admin_config.ISOLATION_DENIED_DIR, username)
    elif mode == "deny_all":
        admin_config.remove_user_from_dir(admin_config.ISOLATION_ALLOWED_DIR, username)

# Clean up existing domain isolation for the user
    _cleanup_user_isolation(username)
    _deny_domain_limits(username)

def _cleanup_user_isolation(username: str):
    """Remove all domain isolation state for a single user. CLOS-6888."""
    if not user_exists(username):
        return
    user_cfg = config.load_user_config(username)
    domain_docroot_map = {
        d: _get_docroot_or_none(d) for d in user_cfg.enabled_websites
    }

if user_cfg.enabled_websites:
        # CLOS-4642: drop dedicated LiteSpeed pools on full teardown.
        try:
            litespeed.remove_all_dedicated_php_handlers(username)
        except Exception:
            logging.exception(
                "Failed to remove LiteSpeed dedicated PHP handlers for %s",
                username,
            )
        reload_processes_with_docroots(
            username, filter_by_docroots=list(domain_docroot_map.values())
        )

removed_leaves = set()

# First pass: current-docroot hash. Path 1 (dr=None) defers to sweep;
    # Path 2 (dr changed) silently no-ops and the sweep picks up the
    # registration-time leaf.
    for domain, dr in domain_docroot_map.items():
        if dr is None:
            continue
        try:
            leaf = (
                Path(jail_utils.get_user_var_cagefs_path(username))
                / ".cagefs/website"
                / jail_utils.get_website_id(dr)
            )
            jail_utils.remove_website_token_directory(username, dr)
            removed_leaves.add(str(leaf))
        except OSError:
            logging.exception(
                "Failed to remove website token for user %s domain %s",
                username, domain,
            )

# Second pass: sweep by the docroot each leaf recorded at
    # registration time (.cagefs.website, root:root 0444, unforgeable).
    # Covers Paths 1, 2, and 3.
    for leaf, _stored_docroot in jail_utils.iter_user_website_tokens(username):
        if str(leaf) in removed_leaves:
            continue
        try:
            shutil.rmtree(leaf, ignore_errors=False)
            removed_leaves.add(str(leaf))
        except OSError:
            logging.exception(
                "Failed to sweep website token %s for user %s",
                leaf, username,
            )

# Paired overlay under <homedir>/.cagefs/websites/<hash>/ is
        # NOT swept here: everything under the user's home is
        # tenant-owned, so rename permission on ``websites`` lets a
        # tenant swap it for a symlink into another account's home. A
        # root rmtree would follow that symlink. Overlay leftover is
        # cosmetic; token leaf is what drives the lsphp re-exec loop.
        # A future overlay-cleanup MR needs an fd-bound O_NOFOLLOW walk.

# Persist the registry LAST. Because we never wiped it or
    # isolates.mounts during the passes, a surviving leaf means their
    # on-disk state is already consistent with what remains — do
    # nothing. Rewriting from the stale user_cfg would retarget
    # mounts to the wrong hash (Path 2) or re-assert isolation for
    # already-cleaned domains. AC2 satisfied by inaction. Only unlink
    # when every leaf is gone.
    if not any(True for _ in jail_utils.iter_user_website_tokens(username)):
        config.save_user_config(username, config=None)
        write_jail_mounts_config(username, user_config=None)

def cleanup_deleted_domain(user: str, domain: str):
    """Drop the state a domain the panel just deleted leaves behind. CLOS-6993.

Enabling isolation on a domain and giving it its own PHP version creates
    five ``<website_id>/`` directories -- the token/ns-cache leaf and the
    ``cl.selector`` / ``cl.php.d`` farms under /var/cagefs, plus
    ``.cl.selector`` and ``.cagefs/websites`` in the user's home. Every one of
    them is keyed on the FNV-1a hash of the *docroot*, so as soon as the panel
    forgets the domain its docroot stops resolving and no ``--domain`` call can
    address them any more: they pile up as unmappable hashes in the account,
    and a new domain that reuses the docroot silently inherits the dead
    domain's PHP version.

The deleted domain's own id is therefore not computable here. Instead the
    keep-set is built from the docroots the panel *still* reports for the
    account and every other on-disk id is swept. That also picks up ids
    orphaned by earlier deletions, before this handler existed.
    """
    if not user_exists(user):
        return

# The registry keeps the domain name, so this part is directly addressable.
    # Leave it in place and `--isolates-list` keeps reporting a domain that no
    # longer exists, and a later domain of the same name comes back isolated
    # without an admin ever enabling it.
    user_config = config.load_user_config(user)
    if domain in user_config.enabled_websites:
        user_config.enabled_websites.remove(domain)
        config.save_user_config(user, user_config)
        write_jail_mounts_config(user, user_config)
        try:
            _unregister_domain_lve(user, domain)
        except (subprocess.CalledProcessError, OSError):
            logging.exception(
                "Failed to unregister domain LVE for %s, skipping.", domain)
        # Mirrors the tail of disable_website_isolation. The regeneration the
        # delete handler runs next only stops the monitor when its own
        # revalidation rejected a domain, so without this
        # cagefs-docroot-monitor keeps running with nothing to watch once the
        # last isolated website on the server is deleted. Nothing between here
        # and then starts it again -- start_monitoring_service is called only
        # from enable_website_isolation.
        if not users_with_enabled_domain_isolation():
            stop_monitoring_service()

try:
        live_docroots = [dr for _domain, dr in userdomains(user)]
    except Exception:
        logging.exception(
            "Cannot list domains of user %s, skipping per-website cleanup", user)
        return
    if not live_docroots:
        # An account always keeps at least its primary domain, so an empty
        # listing means the panel query came back broken rather than that
        # every website id on disk is stale. Sweeping on that basis would
        # wipe the live per-domain PHP configuration of every other domain.
        logging.warning(
            "Panel reports no domains for user %s, skipping per-website "
            "cleanup after deletion of %s", user, domain)
        return

removed = jail_utils.purge_stale_website_dirs(
        user, {jail_utils.get_website_id(dr) for dr in live_docroots})
    if removed:
        logging.info(
            "Removed per-website directories left by deleted domain(s) of "
            "user %s: %s", user, removed)

def _cleanup_all_users_isolation():
    """Remove domain isolation state for every user that has it.

CLOS-6888 Path 3: iteration widens to "users named by the registry"
    UNION "users with a per-site token on disk" so accounts whose
    registry entry is already gone still get visited.
    """
    users_with_registry = set(users_with_enabled_domain_isolation())
    users_with_tokens = {u for u in cpusers() if jail_utils.has_on_disk_website_tokens(u)}
    for username in sorted(users_with_registry | users_with_tokens):
        try:
            _cleanup_user_isolation(username)
        except Exception:
            logging.exception(
                "Unable to disable website isolation for user %s, skipping.",
                username,
            )
    users_left = users_with_enabled_domain_isolation()
    if not users_left:
        stop_monitoring_service()

def _get_docroot_or_none(domain: str):
    try:
        return get_domain_docroot(domain)[0]
    except (NoDomain, IndexError):
        return None

def _panel_domains_or_none(user: str):
    """Names of the domains the panel reports for ``user``, or None when
    the panel cannot tell (query failed or came back empty -- an account
    always keeps at least its primary domain)."""
    try:
        return {name for name, _docroot in userdomains(user)} or None
    except Exception:
        logging.exception("Cannot list domains of user %s", user)
        return None

def is_isolation_enabled(user):
    if not is_website_isolation_allowed_server_wide():
        return False
    try:
        domains_config_path = jail_utils.get_jail_config_path(user)
    except UserNotFoundError:
        return False
    return os.path.exists(domains_config_path)

def users_with_enabled_domain_isolation() -> dict:
    users = [u for u in cpusers() if user_exists(u) and is_isolation_enabled(u)]
    user_domain_pairs = {}
    for user in users:
        domains_with_isolation = get_websites_with_enabled_isolation(user)
        if domains_with_isolation:
            user_domain_pairs[user] = domains_with_isolation
    return user_domain_pairs

def get_websites_with_enabled_isolation(user: str):
    if not user_exists(user):
        logging.warning(
            "User %s not found, cannot get websites with enabled isolation", user)
        return []
    return config.load_user_config(user).enabled_websites

def isolated_websites_or_none(user: str):
    """The user's isolated websites, or None when the registry cannot tell:
    it cannot be read, or it disagrees with the jail config, which
    write_jail_mounts_config writes with the first website and removes with
    the last. load_user_config reads a corrupt registry as empty, and a
    change made on that reading writes it back and tears down all of the
    user's isolation (found in review, CLOS-4727)."""
    try:
        websites = get_websites_with_enabled_isolation(user)
        if bool(websites) != is_isolation_enabled(user):
            return None
    except Exception:  # pylint: disable=broad-except
        logging.warning("Isolated websites of user %s could not be read",
                        user, exc_info=True)
        return None
    return websites

def isolation_moves(user: str, domains=None, enabling=False):
    """Whether an isolation change moves a website of `user`'s to another
    PHP Selector version: an enable of one of `domains` not isolated yet,
    or a disable of one that is (every website of the user's when
    `domains` is None: a deny). A registry that cannot tell counts as a
    move (isolated_websites_or_none). For the MultiPHP sync, which steps
    aside before a website moves (CLOS-4727).

Every toggle rewrites the user's whole jail config from the registry
    (write_jail_mounts_config), so it also moves a website it does not
    name: one whose document root no longer validates (F-08) or moved, or
    one the registry lists that is not isolated in effect (_rewrite_moves;
    found in review: enabling an isolated website again dropped another
    one F-08 rejects, with no step aside)."""
    websites = isolated_websites_or_none(user)
    if websites is None:
        return True
    if domains is None:
        after = []
    elif enabling:
        after = list(websites) + [d for d in domains if d not in websites]
    else:
        after = [d for d in websites if d not in domains]
    return _rewrite_moves(user, websites, after)

def _rewrite_moves(user, websites, after):
    """Whether rewriting `user`'s jail config for the registry `after`
    (`websites` now) changes a website's isolation in effect: a website
    isolated under a website id it would not keep, or one that would be
    isolated under one it is not now (_classify_isolated_websites), or one
    not isolated now that would be. The isolation in effect is the record
    of it (isolated_website_ids), or, for a jail config with none, the
    registry: a token tells only where a website of the registry's is (found
    in review: a first isolation at a root whose token outlived a website
    that moved away read as no move). A domain the panel no longer reports
    moves nowhere, whether the change keeps or clears it: nothing serves
    it. True when that cannot be read."""
    try:
        home = ClPwd().get_pw_by_name(user).pw_dir
        valid, _rejected, removed, moved = _classify_isolated_websites(
            user, after, home)
        in_effect = jail_utils.isolated_website_ids(user)
    except Exception:  # pylint: disable=broad-except
        logging.warning("Whether an isolation change of user %s moves a "
                        "website could not be read", user, exc_info=True)
        return True
    if moved:
        return True
    before = set(websites) if in_effect is None else set(in_effect)
    gone = before - set(valid) - set(removed)
    if gone:
        # one the change clears moves nowhere either, when the panel no
        # longer reports it (found in review: a leftover's disable, or a
        # deny, waited on the sync, which could refuse it)
        panel = _panel_domains_or_none(user)
        if panel:
            gone = {name for name in gone if name in panel}
    return bool(set(valid) - before or gone)

def forget_panel_caches():
    """Drop what clcommon keeps of the panel's domains for the life of the
    process: cPanel's userdomains() lists every account's domains with
    their document roots on its first call, and the jail config is written
    from it. A change that waited for the MultiPHP sync to step aside, which
    can take minutes, reads them again (found in review: a domain made and
    isolated meanwhile was written out of the jail config)."""
    try:
        from clcommon.cpapi.plugins import cpanel  # pylint: disable=import-outside-toplevel
    except Exception:  # pylint: disable=broad-except
        return
    cache = getattr(cpanel, "_user_to_domains_map_cpanel", None)
    if isinstance(cache, dict):
        cache.clear()

def users_with_jail_config():
    """The accounts whose website isolation is in effect: their jail config
    stands, whatever their registry reads (isolated_websites_or_none), on a
    box that allows isolation server-wide (is_isolation_enabled reads the
    server-wide flag first). These are the accounts --isolates-deny-all
    moves websites of."""
    return [u for u in cpusers() if user_exists(u) and is_isolation_enabled(u)]

def get_docroots_of_isolated_websites() -> dict:
    """
    Returns pairs user: set(docroots) for all users with website isolation enabled
    Used by monitoring service to watch docroots changes to load actual list of docroot paths
    instead of stale storage
    """
    users_with_isolation = users_with_enabled_domain_isolation()
    pairs = defaultdict(set)
    for user, domains in users_with_isolation.items():
        for domain in domains:
            try:
                dr = get_domain_docroot(domain)[0]
            except (NoDomain, IndexError):
                continue
            pairs[user].add(dr)
    return pairs

def enable_website_isolation(user, domain):
    """Isolate ``domain`` for ``user``.

Returns ``None`` when the domain was isolated, or a human-readable
    reason string when it was SKIPPED.

CLOS-7001: the two skip paths below used to ``return`` after a bare
    ``logging.warning``, so callers could not distinguish "isolated"
    from "skipped". ``cagefsctl --isolates-enable`` consequently printed
    every requested domain as enabled and exited 0 while nothing had
    been isolated — a panel or script reading the exit status and stdout
    saw success, and the only trace was a warning on stderr. The skip
    itself is deliberate (one bad domain must not abort the loop over
    the user's other domains), so the reason is RETURNED rather than
    raised; every caller is responsible for surfacing it.

Failures after this point are not skips and still propagate as
    exceptions — ``cagefsctl.main()`` turns those into exit 1.
    """
    if not user_exists(user):
        logging.warning(
            "User %s not found, cannot enable website isolation", user)
        return f"user {user} not found"
    user_config = config.load_user_config(user)

# if it crashes just let the command fail with NoDomain
    # exception, it should be a very rare case because
    # we validate input domain name in cagefsctl.py
    # Validate at the trust boundary: docroot is panel-supplied and is
    # consumed verbatim by the jail mount config writer (raw bind
    # source/target and bracketed section header). See docroot_validation.
    # Skip-and-log on invalid docroot (matches the user-not-found path
    # above) so a single bad domain does not abort the multi-domain loop
    # in cagefsctl_user.py's site-isolation-enable handler. The
    # symlink-rejecting check on top of the lexical one blocks the tenant
    # from pre-aiming the bind source at an arbitrary host path via a
    # symlink in their own writable ancestor tree.
    try:
        document_root = validate_docroot(get_domain_docroot(domain)[0])
        validate_docroot_no_symlinks(document_root, ClPwd().get_pw_by_name(user).pw_dir)
    except ValueError as exc:
        logging.warning(
            "Skipping website isolation for user %s domain %s: %s",
            user, domain, exc)
        return str(exc)

if domain not in user_config.enabled_websites:
        user_config.enabled_websites.append(domain)

# Create website token directory and overlay storage
    jail_utils.create_website_token_directory(user, document_root)
    jail_utils.create_overlay_storage_directory(user, document_root)

config.save_user_config(user, user_config)

# regenerate alt-php ini configuration for selector for the specific domain
    # scanner-triage: bare-name cagefsctl is safe. The only tenant-reachable
    # path is the CAGEFSCTL_USER proxyexec alias, which is marked `root:` in
    # proxy.commands and thus goes through scrub_env_for_root_exec — PATH is
    # reset before execv. The other caller (root admin CLI) is root-to-root.
    subprocess.run(["cagefsctl", "--rebuild-alt-php-ini", "--domain", domain], check=True)

write_jail_mounts_config(user, user_config)

# CLOS-4642: configure the dedicated per-domain lsphp pool BEFORE reloading
    # workers so that on standalone LiteSpeed the reloaded workers come up in the
    # dedicated pool (per-website namespace) instead of the shared per-user pool.
    # No-op on Apache+mod_lsapi / Plesk / DirectAdmin.
    try:
        litespeed.enable_dedicated_php_handler(user, document_root)
    except Exception:
        logging.exception(
            "Failed to apply LiteSpeed dedicated PHP handler for %s/%s", user, domain)

reload_processes_with_docroots(user, filter_by_docroots=[_get_docroot_or_none(domain)])
    start_monitoring_service()

# Trigger xray/ssa ini regeneration for per-domain PHP selector
    trigger_xray_ini_regeneration(user, domain)
    trigger_ssa_ini_regeneration(user)

_register_domain_lve(user, domain)

# Isolated. Explicit so the contract stays visible at the exit point
    # anything appended below has to keep.
    return None

def _classify_isolated_websites(user, enabled_websites, home):
    """Sort `user`'s registry for a regenerate: (valid, rejected, removed,
    moved). `rejected` holds (domain, docroot, why) for a website whose
    document root no longer validates (F-08), `removed` a domain the panel
    no longer reports, and `moved` the valid websites whose document root
    is not the one they are isolated under: read against the isolation in
    effect (isolated_website_ids), or, for a jail config with no record of
    it (an older cagefs wrote it), by the website's token. Logs nothing of
    its own (the panel's domain listing it may ask for logs a failure): a
    regenerate that steps the MultiPHP sync aside sorts twice, and
    isolation_moves sorts for every toggle."""
    # F-08 (CLOS-5946) DiD: filter enabled_websites down to domains whose
    # CURRENTLY resolved docroot still passes the lexical + no-symlink
    # validation before rewriting the jail mount config. A docroot that
    # passed at enable time can change on disk (panel edit, tenant plants
    # a symlink under an owned ancestor); without this re-check the root-
    # run bind mounts in write_jail_mounts_config would follow the
    # symlink. Mirrors the shape used at enable_website_isolation above.
    # A missing on-disk component here short-circuits the walk as safe
    # — see validate_docroot_no_symlinks (jail.c refuses to follow a
    # tenant-planted symlink at bind time and skips ENOENT sources).
    valid_websites = []
    rejected_domains = []
    removed_domains = []
    moved_domains = []
    panel_domains = None
    in_effect = jail_utils.isolated_website_ids(user)
    for domain in enabled_websites:
        current_docroot = _get_docroot_or_none(domain)
        if current_docroot is None:
            # CLOS-7514: a domain the panel no longer reports was removed
            # without reaching post_delete_domain (most cPanel removal paths
            # before the Whostmgr hooks), so its entry would stay listed
            # forever and --isolates-disable refuses it. Drop it, but only
            # when the panel answered and positively lacks the name;
            # otherwise keep the entry so downstream code can log/handle the
            # disappearing-domain case below.
            if panel_domains is None:
                panel_domains = _panel_domains_or_none(user) or set()
            if panel_domains and domain not in panel_domains:
                removed_domains.append(domain)
                continue
            valid_websites.append(domain)
            continue
        try:
            validate_docroot(current_docroot)
            validate_docroot_no_symlinks(current_docroot, home)
        except ValueError as exc:
            # Capture (domain, docroot) so the teardown below can
            # remove the stale token, unregister LVE, and reload
            # workers for the rejected docroot — mirrors the
            # disable_website_isolation sequence. Without this,
            # workers keep the stale isolation mount for a docroot
            # that no longer appears in the mounts config.
            rejected_domains.append((domain, current_docroot, exc))
            continue
        valid_websites.append(domain)
        if in_effect is None:
            if not jail_utils.website_token_exists(user, current_docroot):
                moved_domains.append(domain)
        elif in_effect.get(domain) != jail_utils.get_website_id(current_docroot):
            moved_domains.append(domain)
    return valid_websites, rejected_domains, removed_domains, moved_domains

class NotRegenerated(Exception):
    """regenerate_isolation_configuration left the user's isolation as it
    was: a registry that cannot tell, which would tear every website down,
    and a MultiPHP sync that could not step aside for it (CLOS-4727). Not
    an answer: one that read as False said nothing of it, and cagefsctl
    reported a teardown found only on the second reading as regenerated
    (found in the final review)."""

def regenerate_isolation_configuration(user, before_move=None, stop_monitor=True):
    """Rewrite `user`'s isolation from the registry, dropping a website
    whose document root no longer validates (F-08) and one the panel no
    longer reports (CLOS-7514).

before_move(user, domains): called before anything is written, with
    the websites whose PHP Selector version this moves (CLOS-4727: the
    MultiPHP sync steps aside first): one dropped for its document root,
    which folds back onto the account's version as a disable folds it, and
    one whose document root moved, whose settings are kept per website id
    (found in review: the new one runs the account's version). Moved means
    isolated under another website id than its document root's now, back
    to a root it had or onto another website's included (found in review:
    the token of a root a website left outlives the move). [] when the
    registry cannot tell: read as empty while the jail config stands, it
    tears every website of the user's down. A removed domain is not named:
    nothing serves it any more. before_move answers whether the sync
    stepped aside (False: it refused). The registry, the panel's document
    roots and the account's home are read again after it, which can wait
    minutes for the account's
    pass lock: an isolation change landing meanwhile is not written over
    (found in review). A refusal does not
    keep a website F-08 rejects, nor one whose document root moved (its
    isolation follows it), but a registry that cannot tell is torn down
    only once the sync stepped aside: refused, nothing is written and
    NotRegenerated is raised (found in review: every website went back to
    the account's version while the pins of the sync's stood, for no F-08
    reason), also where only the reading after a refusal for another
    website finds it so (found in review).

stop_monitor: stop the docroot monitor when no website is left to
    watch. The monitor runs this through cagefsctl, and stopping its unit
    kills that cagefsctl before it is done (found in review); a caller
    that passes False stops it itself, last, when this returns True."""
    if not user_exists(user):
        logging.warning(
            "User %s not found, cannot regenerate website isolation configuration", user)
        return False
    home = ClPwd().get_pw_by_name(user).pw_dir
    user_config = config.load_user_config(user)
    valid_websites, rejected_domains, removed_domains, moved_domains = \
        _classify_isolated_websites(user, user_config.enabled_websites, home)
    if before_move is not None:
        teardown = not user_config.enabled_websites and is_isolation_enabled(user)
        if teardown or rejected_domains or moved_domains:
            aside = before_move(user, [] if teardown else [
                domain for domain, _docroot, _why in rejected_domains] + moved_domains)
            if teardown and aside is False:
                logging.warning(
                    "Website isolation of user %s not regenerated: its registry "
                    "reads empty while its jail config stands, and the MultiPHP "
                    "sync could not step aside", user)
                raise NotRegenerated(user)
            forget_panel_caches()
            # the home too: an account moved to another partition meanwhile
            # had its websites checked against the one it left, and dropped
            # as F-08 rejects (found in review)
            home = ClPwd().get_pw_by_name(user).pw_dir
            user_config = config.load_user_config(user)
            valid_websites, rejected_domains, removed_domains, moved_domains = \
                _classify_isolated_websites(user, user_config.enabled_websites, home)
            # read again after a refusal, a registry that reads empty now
            # tears every website down with the sync's pins standing (found
            # in review)
            if aside is False and not user_config.enabled_websites \
                    and is_isolation_enabled(user):
                logging.warning(
                    "Website isolation of user %s not regenerated: its registry "
                    "reads empty while its jail config stands, and the MultiPHP "
                    "sync could not step aside", user)
                raise NotRegenerated(user)
    for domain in removed_domains:
        logging.info(
            "Dropping isolation entry of user %s for domain %s "
            "the panel no longer reports", user, domain)
    for domain, _docroot, exc in rejected_domains:
        logging.warning(
            "Skipping website isolation regeneration for user %s domain %s: %s",
            user, domain, exc)
    if len(valid_websites) != len(user_config.enabled_websites):
        # Persist the filtered list to disk. Without this, dropped
        # domains linger in the saved config and reappear on the next
        # load — the F-08 revalidation would then have to reject them
        # again on every regeneration cycle. Mirrors the persist-then-
        # write-mounts pattern used in enable/disable/cleanup above.
        user_config.enabled_websites = valid_websites
        config.save_user_config(user, user_config)
    else:
        user_config.enabled_websites = valid_websites
    write_jail_mounts_config(user, user_config)

# F-08 (CLOS-5946) DiD: unregister LVE for each rejected domain so
    # workers stop associating requests with the stale isolation entry.
    # LVE unregister is domain-keyed (no filesystem touch), safe even
    # for a rejected/symlinked docroot.
    #
    # Post-review Bugbot MED on !211: do NOT call
    # jail_utils.remove_website_token_directory(user, rejected_docroot)
    # or reload_processes_with_docroots(filter=rejected_docroots) here.
    # The rejected docroot value is *why* revalidation refused the
    # domain (attacker-planted symlink under the tenant home); feeding
    # it back into filesystem-touching helpers gives that untrusted
    # string a second bite at the sink. The stale mount and token dir
    # will be cleaned up on the next legitimate disable_website_
    # isolation or user-teardown pass, where the operator (or the
    # regenerate path with a now-clean docroot) has a trusted value
    # in hand. Log an advisory so operators can intervene manually.
    if rejected_domains:
        for domain, _docroot, _why in rejected_domains:
            try:
                _unregister_domain_lve(user, domain)
            except (subprocess.CalledProcessError, OSError):
                logging.exception(
                    "Failed to unregister domain LVE for %s, skipping.", domain)
            logging.warning(
                "Website isolation rejected for user %s domain %s; "
                "on-disk token and worker reload skipped for the "
                "untrusted docroot. Run disable_website_isolation with "
                "a trusted docroot value if manual cleanup is required.",
                user, domain,
            )

# Mirrors cleanup_deleted_domain, which a working removal would have run.
    for domain in removed_domains:
        try:
            _unregister_domain_lve(user, domain)
        except (subprocess.CalledProcessError, OSError):
            logging.exception(
                "Failed to unregister domain LVE for %s, skipping.", domain)

document_roots = []
    for domain in user_config.enabled_websites:
        document_root = _get_docroot_or_none(domain)
        if document_root is None:
            logging.warning(
                "Unable to find document root for domain %s, "
                "please contact CloudLinux support if the issue persists.",
                domain,
            )
            continue

document_roots.append(document_root)
        try:
            # recreate tokens and storage e.g. when username changes
            jail_utils.create_website_token_directory(user, document_root)
            jail_utils.create_overlay_storage_directory(user, document_root)
            # CLOS-6923: refresh the per-website cl.php.d/<website_id>/
            # side too. Without this, a docroot change after enable
            # leaves the jail bind rule pointing at a nonexistent
            # source and alt-php scans zero ini files inside the cage.
            # Mirror the shell-out enable_website_isolation runs at its
            # own domain step above.
            #
            # scanner-triage: bare-name cagefsctl is safe. See the same
            # comment on the matching call in enable_website_isolation.
            subprocess.run(
                ["cagefsctl", "--rebuild-alt-php-ini", "--domain", domain],
                check=True,
            )
        except Exception as e:
            logging.error("Unable to recreate token/storage for domain=%s, Error=%s", domain, e)
            continue

# CLOS-4642: reconcile dedicated LiteSpeed pools to the current isolated
    # docroots -- adds missing includes and sweeps stale ones for removed,
    # unresolvable, or deleted vhosts. Runs after tokens exist so a reload never
    # precedes a token. No-op off standalone LiteSpeed.
    try:
        litespeed.reconcile_dedicated_php_handlers(user, document_roots)
    except Exception:
        logging.exception(
            "Failed to reconcile LiteSpeed dedicated PHP handlers for %s", user)

# F-48 / CLOS-5435: defense-in-depth. If every enabled_websites entry
    # failed docroot resolution (e.g. the panel-side domain was just deleted
    # but is still listed in the CageFS user config), document_roots is
    # empty. Do NOT tail-call reload_processes_with_docroots with an empty
    # filter here -- there is nothing to reload, and the empty-filter path
    # is treated as "reload all isolated processes for this user", not
    # "reload every process this user owns".
    if not document_roots:
        # Post-review Bugbot LOW on !211: if the F-08 filter dropped
        # every isolated domain and no user has isolation left, stop
        # the docroot-monitor service — mirrors the same tail in
        # disable_website_isolation. Without this cagefs-docroot-monitor
        # can keep running with nothing to watch.
        if (rejected_domains or removed_domains) and not users_with_enabled_domain_isolation():
            if not stop_monitor:
                return True
            stop_monitoring_service()
        return False

reload_processes_with_docroots(user, filter_by_docroots=document_roots)
    return False

def disable_website_isolation(user: str, domain: str | None = None):
    if not user_exists(user):
        logging.warning(
            "User %s not found, cannot disable website isolation", user)
        return
    user_config = config.load_user_config(user)
    reload_docroots = None
    domains_to_unregister = []

if domain is None:
        domains_to_unregister = list(user_config.enabled_websites)
        reload_docroots = [
            _get_docroot_or_none(website) for website in user_config.enabled_websites
        ]
        user_config.enabled_websites = []
    elif domain in user_config.enabled_websites:
        domains_to_unregister = [domain]
        reload_docroots = [_get_docroot_or_none(domain)]
        user_config.enabled_websites.remove(domain)

config.save_user_config(user, user_config)
    write_jail_mounts_config(user, user_config)
    if reload_docroots:
        # CLOS-4642: reconcile dedicated LiteSpeed pools to the docroots that are
        # STILL isolated, BEFORE reloading workers (so respawned workers do not
        # re-enter a stale pool). Reconcile is robust to full teardown, docroots
        # shared by several domains, and docroots that no longer resolve.
        # No-op off standalone LiteSpeed.
        try:
            litespeed.reconcile_dedicated_php_handlers(
                user,
                [_get_docroot_or_none(d) for d in user_config.enabled_websites],
            )
        except Exception:
            logging.exception(
                "Failed to reconcile LiteSpeed dedicated PHP handlers for %s", user)
        reload_processes_with_docroots(user, filter_by_docroots=reload_docroots)
        for document_root in reload_docroots:
            if document_root is None:
                continue
            jail_utils.remove_website_token_directory(user, document_root)

for d in domains_to_unregister:
        try:
            _unregister_domain_lve(user, d)
        except (subprocess.CalledProcessError, OSError):
            logging.exception(
                "Failed to unregister domain LVE for %s, skipping.", d)

# get actual docroots for all users with website isolation enabled
    users_with_isolation = users_with_enabled_domain_isolation()
    if not users_with_isolation:
        stop_monitoring_service()

CANCEL

Name	Type	Size	Modified	Actions
↩ ..	DIR	—	—
📁 __pycache__/	DIR	—	2026-11-02 22:52
📁 selector/	DIR	—	2026-11-02 22:52
📁 webisolation/	DIR	—	2026-11-02 22:52
📄 __init__.py	PY	180 B	2026-10-16 11:42	EDIT
📄 cli.py	PY	3.6 KB	2026-10-16 11:42	EDIT
📄 const.py	PY	1 KB	2026-10-16 11:42	EDIT
📄 domain.py	PY	50.4 KB	2026-10-16 11:42	EDIT
📄 exceptions.py	PY	402 B	2026-10-16 11:42	EDIT
📄 fs.py	PY	1020 B	2026-10-16 11:42	EDIT
📄 io.py	PY	5.3 KB	2026-10-16 11:42	EDIT
📄 multiphp_sync.py	PY	9.7 KB	2026-10-16 11:42	EDIT