Glass Manager

✏️ EDITING: cagefsctl-user

#!/opt/cloudlinux/venv/bin/python3 -sbb
# -*- coding: utf-8 -*-
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2025 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENCE.TXT
#
"""
User-level CLI utility for managing website isolation.

This script runs via proxyexec with root privileges but operates on behalf
of the calling user. It validates that the user only manages domains they own.

Commands:
    site-isolation-enable --domain DOMAIN[,DOMAIN2,...]   Enable site isolation for domain(s)
    site-isolation-disable --domain DOMAIN[,DOMAIN2,...]  Disable site isolation for domain(s)
    site-isolation-list                                   List domains with site isolation enabled

All commands output JSON:
    Success: {"result": "success", "enabled_sites": ["domain1", "domain2"]}
    Error:   {"result": "ERROR_CODE"}
"""

import argparse
import json
import logging
import os
import sys
import pwd

from clcommon.cpapi import domain_owner, userdomains
from clcommon.cpapi.cpapiexceptions import NoDomain

from clcagefslib.cli import (
    call_via_proxyexec,
    in_cagefs,
    is_running_via_proxyexec,
)
from clcagefslib.domain import (
    enable_website_isolation,
    disable_website_isolation,
    get_websites_with_enabled_isolation,
    isolation_moves,
    forget_panel_caches,
    is_website_isolation_allowed_server_wide,
    is_website_isolation_allowed_for_user,
)
from clcagefslib.fs import user_exists
from clcagefslib.multiphp_sync import (
    converge,
    refusal,
    notify_account,
    step_aside,
    sync_steps_aside,
)

# Logging configuration
LOG_FILE = "/var/log/cloudlinux/cagefsctl-user.log"

# Proxyexec alias for all cagefsctl-user commands
PROXYEXEC_ALIAS = "CAGEFSCTL_USER"

def _safe_for_log(value):
    """Escape control characters before interpolating untrusted values into a log line.

The cagefsctl-user log uses a single-prefix formatter ('[LEVEL | ts]: %(message)s'),
    so an unescaped '\\n' inside a user-supplied value (e.g. `domain`) yields a forged
    second line that an admin reads as a separately-logged event. Apply to every
    user-controlled value reaching a logger.* call in this module.
    """
    if value is None:
        return value
    if isinstance(value, (list, tuple)):
        return type(value)(_safe_for_log(v) for v in value)
    if not isinstance(value, str):
        value = str(value)
    return value.encode("unicode_escape").decode("ascii")

def setup_logger():
    """
    Set up logging to file only (no console output).

Returns:
        logging.Logger: Configured logger instance
    """
    logger = logging.getLogger("cagefsctl-user")
    logger.setLevel(logging.INFO)
    # Disable propagation to root logger to prevent console output
    logger.propagate = False
    try:
        fh = logging.FileHandler(LOG_FILE)
        fh.setFormatter(logging.Formatter(
            "[%(levelname)s | %(asctime)s]: %(message)s"
        ))
        logger.addHandler(fh)
    except (IOError, OSError):
        # Cannot write to log file, continue without file logging
        pass
    return logger

logger = setup_logger()

class ErrorCodes:
    """Error codes for JSON responses."""
    SITE_ISOLATION_NOT_ALLOWED = "Site isolation feature is not allowed"
    # CLOS-7001: distinct from NOT_ALLOWED and INTERNAL_ERROR — the
    # request was well-formed and permitted, but one or more domains
    # were skipped, so isolation is not in effect for them. `message`
    # carries "<domain>: <reason>" per skipped domain.
    SITE_ISOLATION_SKIPPED = "Site isolation was not applied for one or more domains"
    DOMAIN_NOT_FOUND = "Specified domain is not found"
    USER_NOT_FOUND = "User not found"
    INTERNAL_ERROR = "Internal error"
    MISSING_DOMAIN = "Domain is not specified"
    ROOT_NOT_ALLOWED = "Utility cannot be run as root"

# CLOS-7520: a machine-readable `code` for each ErrorCodes value. The values
# above are sentences that callers also show to people, so they are kept as
# `result` unchanged; `code` is what a caller branches on. Add entries, never
# rename one -- the same rule as isolatectl's codes, whose responses carry
# these through `isolatectl site-isolation`.
STABLE_CODES = {
    ErrorCodes.SITE_ISOLATION_NOT_ALLOWED: "not_allowed",
    ErrorCodes.SITE_ISOLATION_SKIPPED: "not_applied",
    ErrorCodes.DOMAIN_NOT_FOUND: "domain_not_found",
    ErrorCodes.USER_NOT_FOUND: "user_not_found",
    ErrorCodes.INTERNAL_ERROR: "internal_error",
    ErrorCodes.MISSING_DOMAIN: "missing_domain",
    ErrorCodes.ROOT_NOT_ALLOWED: "root_not_allowed",
}

def get_calling_user():
    """
    Get the username of the calling user from proxyexec environment.

When running via proxyexec, PROXYEXEC_UID contains the original user's UID.
    Falls back to current process UID if not set.

Returns:
        str: Username of the calling user
        None: If user cannot be determined
    """
    proxyexec_uid = os.environ.get("PROXYEXEC_UID")
    if not proxyexec_uid:
        return None
    try:
        uid = int(proxyexec_uid)
        pw = pwd.getpwuid(uid)
        return pw.pw_name
    except (ValueError, KeyError):
        return None

def json_response(result, enabled_sites=None, message=None):
    """
    Create a JSON response dictionary.

Args:
        result: "success" or error code
        enabled_sites: Optional list of the account's currently isolated
            sites. Carried on SITE_ISOLATION_SKIPPED responses too, not
            only on success — a partial run has to report both halves
            (CLOS-7001).
        message: Optional error message with additional details

Returns:
        dict: Response dictionary; an error also carries its STABLE_CODES
        ``code``
    """
    response = {"result": result}
    if result != "success":
        response["code"] = STABLE_CODES.get(result, "internal_error")
    if enabled_sites is not None:
        response["enabled_sites"] = enabled_sites
    if message is not None:
        response["message"] = message
    return response

def output_json(response):
    """Print JSON response to stdout."""
    print(json.dumps(response))

def validate_domain_ownership(username, domain):
    """
    Validate that a domain belongs to the specified user.

Args:
        username: The username to check ownership for
        domain: The domain to validate

Returns:
        tuple: (is_valid, error_code)
            is_valid: True if domain belongs to user
            error_code: Error code if validation fails, None otherwise
    """
    try:
        owner = domain_owner(domain)
        if owner is None:
            # Fallback: domain_owner() may not resolve subdomains
            # on some panels (e.g. DirectAdmin). Check via userdomains().
            user_domains = [d for d, _ in userdomains(username)]
            if domain in user_domains:
                return True, None
            return False, ErrorCodes.DOMAIN_NOT_FOUND
        if owner != username:
            return False, ErrorCodes.DOMAIN_NOT_FOUND
        return True, None
    except NoDomain:
        return False, ErrorCodes.DOMAIN_NOT_FOUND
    except Exception:
        return False, ErrorCodes.INTERNAL_ERROR

def get_validated_user():
    """
    Get the calling user and validate they exist.

Returns:
        tuple: (username, error_code)
            username: The validated username, or None if validation failed
            error_code: Error code if validation failed, or None if successful
    """
    username = get_calling_user()
    if not username:
        logger.error("User not found")
        return None, ErrorCodes.USER_NOT_FOUND

if not user_exists(username):
        logger.error("User %s does not exist", username)
        return None, ErrorCodes.USER_NOT_FOUND

return username, None

def validate_domain_for_user(username, domain):
    """
    Validate domain argument and ownership for a user.

Args:
        username: The username to check ownership for
        domain: The domain to validate

Returns:
        tuple: (is_valid, error_code)
            is_valid: True if domain is valid and belongs to user
            error_code: Error code if validation failed, None otherwise
    """
    if not domain:
        logger.error("Missing domain argument")
        return False, ErrorCodes.MISSING_DOMAIN

is_valid, error_code = validate_domain_ownership(username, domain)
    if not is_valid:
        logger.error("Domain validation failed: user=%s, domain=%s, error=%s",
                     username, _safe_for_log(domain), error_code)
        return False, error_code

return True, None

def parse_domains(domain_arg):
    """
    Parse comma-separated domain argument into a list of domains.

Args:
        domain_arg: Comma-separated domain string (e.g., "domain1.com,domain2.com")

Returns:
        list: List of domain names, with whitespace stripped
    """
    if not domain_arg:
        return []
    return [d.strip() for d in domain_arg.split(",") if d.strip()]

def _step_aside(args, username, enabling):
    """CLOS-4727: have the MultiPHP sync hand its pins on the account back
    before an isolation change (a no-op without it), unless the caller
    converges the account itself. Returns the change to converge() after
    the toggle, and whether the toggle must not happen: a pin of the sync's
    left on the version a website is leaving serves it with the server-wide
    extensions until the sync re-mirrors it. Only for a toggle that moves a
    website (isolation_moves), the one it names or one the rewrite of the
    account's jail config moves: found in review, a no-op one was handed
    back and re-pinned, and could be refused where it used to succeed. A
    refusal names the domains whose toggle moves a website only: one that
    moves nothing is never refused, beside one that moves either (found in
    review). With no sync to step aside from, nothing is weighed, as in
    cagefsctl (found in the final review)."""
    if getattr(args, "no_multiphp_sync", False) or not sync_steps_aside():
        return None, False
    moving = [domain for domain in parse_domains(args.domain)
              if isolation_moves(username, [domain], enabling=enabling)]
    if not moving:
        return None, False
    change, refused = step_aside([username], paced=True)
    # the paced wait can take minutes: what the panel said of the document
    # roots before it is stale (found in review)
    forget_panel_caches()
    if refused:
        # CLOS-7001's skip format: "<domain>: <reason>" per domain
        # the sync's reason, not root's log, which the tenant cannot read
        # (found in review)
        why = refusal(refused[username], root=False)
        message = "; ".join(f"{domain}: {why}" for domain in moving)
        # CLOS-7001: a SKIPPED answer says what stays isolated; guarded as
        # the error paths read it
        try:
            enabled_sites = get_websites_with_enabled_isolation(username)
        except Exception:  # pylint: disable=broad-except
            enabled_sites = None
        logger.error("Site isolation not changed: user=%s, %s", username, why)
        output_json(json_response(ErrorCodes.SITE_ISOLATION_SKIPPED,
                                  enabled_sites, message=message))
    return change, bool(refused)

def cmd_site_isolation_enable(args):
    """Handle site-isolation-enable command."""
    domains = parse_domains(args.domain)
    logger.info("site-isolation-enable called: domains=%s", _safe_for_log(domains))

if not domains:
        logger.error("No domains specified")
        output_json(json_response(ErrorCodes.MISSING_DOMAIN))
        return 1

username, error = get_validated_user()
    if error:
        output_json(json_response(error))
        return 1

if not is_website_isolation_allowed_server_wide():
        logger.error("Site isolation not allowed server-wide")
        output_json(json_response(ErrorCodes.SITE_ISOLATION_NOT_ALLOWED))
        return 1

if not is_website_isolation_allowed_for_user(username):
        logger.error("Site isolation not allowed for user %s", username)
        output_json(json_response(ErrorCodes.SITE_ISOLATION_NOT_ALLOWED))
        return 1

# Validate all domains first
    for domain in domains:
        is_valid, error = validate_domain_for_user(username, domain)
        if not is_valid:
            output_json(json_response(error))
            return 1

change, refused = _step_aside(args, username, enabling=True)
    if refused:
        return 1
    # CLOS-7001: enable_website_isolation returns a reason string for
    # a domain it SKIPPED (invalid docroot, missing user). This path
    # already echoed the real enabled set rather than the requested
    # one, but still answered "success" with status 0 — so a panel
    # keying off either saw the operation as done.
    # Collected outside the `try` so the error path below can still
    # report what was skipped before the failure.
    skipped = []
    try:
        if change is not None and \
                not is_website_isolation_allowed_for_user(username):
            # denied while the sync stepped aside, a paced wait: the deny
            # stands (found in review), and the change still ends below
            logger.error("Site isolation not allowed for user %s", username)
            output_json(json_response(ErrorCodes.SITE_ISOLATION_NOT_ALLOWED))
            return 1
        for domain in domains:
            skip_reason = enable_website_isolation(username, domain)
            if skip_reason:
                skipped.append(f"{domain}: {skip_reason}")
        if change is None and len(skipped) < len(domains) and \
                not getattr(args, "no_multiphp_sync", False):
            # the domains now run PHP Selector versions of their own: have
            # lvemanager's MultiPHP sync mirror them (a no-op without it);
            # once per invocation, whatever the number of domains. A domain
            # CLOS-7001 skipped changed nothing to mirror, so a run that
            # skipped every one of them asks for no pass at all.
            notify_account(username, paced=True)
        enabled_sites = get_websites_with_enabled_isolation(username)
        if skipped:
            logger.error("Site isolation skipped: user=%s, skipped=%s, enabled_sites=%s",
                         username, _safe_for_log(skipped), _safe_for_log(enabled_sites))
            output_json(json_response(ErrorCodes.SITE_ISOLATION_SKIPPED,
                                      enabled_sites, message='; '.join(skipped)))
            return 1
        logger.info("Site isolation enabled: user=%s, domains=%s, enabled_sites=%s",
                    username, _safe_for_log(domains), _safe_for_log(enabled_sites))
        output_json(json_response("success", enabled_sites))
        return 0
    except Exception as e:
        # CLOS-7001: a hard failure on one domain must not discard the
        # report for the domains already processed — the guarantee the
        # `finally` gives cagefsctl's --isolates-enable. enabled_sites is
        # read back from config rather than echoed, so it is accurate
        # mid-loop; guarded, because the config the loop choked on is the
        # one this read touches and a second failure here must not
        # replace the failure being reported.
        try:
            enabled_sites = get_websites_with_enabled_isolation(username)
        except Exception:
            enabled_sites = None
        # Use logger.error (not logger.exception) so the auto-appended traceback
        # line — ExceptionType: str(exc) — does not bypass _safe_for_log when
        # the exception carries an attacker-controlled value in its message.
        logger.error("Failed to enable site isolation: user=%s, domains=%s, "
                     "skipped=%s, enabled_sites=%s, error=%s",
                     username, _safe_for_log(domains), _safe_for_log(skipped),
                     _safe_for_log(enabled_sites), _safe_for_log(e))
        output_json(json_response(ErrorCodes.INTERNAL_ERROR, enabled_sites,
                                  message='; '.join(skipped + [str(e)])))
        return 1
    finally:
        # the change the sync stepped aside for ends with its pass,
        # whatever happened to the toggle
        if change is not None:
            converge(change, paced=True)

def cmd_site_isolation_disable(args):
    """Handle site-isolation-disable command."""
    domains = parse_domains(args.domain)
    logger.info("site-isolation-disable called: domains=%s", _safe_for_log(domains))

if not domains:
        logger.error("No domains specified")
        output_json(json_response(ErrorCodes.MISSING_DOMAIN))
        return 1

username, error = get_validated_user()
    if error:
        output_json(json_response(error))
        return 1

change, refused = _step_aside(args, username, enabling=False)
    if refused:
        return 1
    try:
        for domain in domains:
            disable_website_isolation(username, domain)
        if change is None and not getattr(args, "no_multiphp_sync", False):
            # the domains fold back onto the account's PHP Selector
            # version: have the MultiPHP sync re-mirror them
            notify_account(username, paced=True)
        enabled_sites = get_websites_with_enabled_isolation(username)
        logger.info("Site isolation disabled: user=%s, domains=%s, enabled_sites=%s",
                    username, _safe_for_log(domains), _safe_for_log(enabled_sites))
        output_json(json_response("success", enabled_sites))
        return 0
    except Exception as e:
        logger.error("Failed to disable site isolation: user=%s, domains=%s, error=%s",
                     username, _safe_for_log(domains), _safe_for_log(e))
        output_json(json_response(ErrorCodes.INTERNAL_ERROR, message=str(e)))
        return 1
    finally:
        if change is not None:
            converge(change, paced=True)

def cmd_site_isolation_list(args):
    """Handle site-isolation-list command."""
    logger.info("site-isolation-list called")

username, error = get_validated_user()
    if error:
        output_json(json_response(error))
        return 1

try:
        enabled_sites = get_websites_with_enabled_isolation(username)
        logger.info("Site isolation list: user=%s, enabled_sites=%s",
                    username, _safe_for_log(enabled_sites))
        output_json(json_response("success", enabled_sites))
        return 0
    except Exception as e:
        logger.error("Failed to list site isolation: user=%s, error=%s",
                     username, _safe_for_log(e))
        output_json(json_response(ErrorCodes.INTERNAL_ERROR, message=str(e)))
        return 1

def _wrap_deprecated(func, old_name, new_name):
    """Return a wrapper that prints a deprecation warning then delegates."""
    def wrapper(args):
        print(f"WARNING: {old_name} is deprecated, use {new_name}",
              file=sys.stderr)
        return func(args)
    return wrapper

def _add_no_multiphp_sync(parser):
    parser.add_argument(
        "--no-multiphp-sync",
        action="store_true",
        help="Do not have lvemanager's MultiPHP <-> PHP Selector sync hand "
             "its versions back before the change, nor re-mirror the "
             "account after it; the caller does both itself (the LVE "
             "Manager UI does, once per bulk action)",
    )

def create_parser():
    """Create argument parser for cagefsctl-user."""
    parser = argparse.ArgumentParser(
        prog="cagefsctl-user",
        description="User-level CLI utility for managing CloudLinux Isolates.",
    )

subparsers = parser.add_subparsers(
        title="commands",
        dest="command",
        help="Available commands",
    )

# isolates-enable command (primary)
    enable_parser = subparsers.add_parser(
        "isolates-enable",
        help="Enable isolation for domain(s)",
    )
    enable_parser.add_argument(
        "--domain",
        required=True,
        help="Domain name(s) to enable isolation for (comma-separated)",
    )
    _add_no_multiphp_sync(enable_parser)
    enable_parser.set_defaults(func=cmd_site_isolation_enable)

# isolates-disable command (primary)
    disable_parser = subparsers.add_parser(
        "isolates-disable",
        help="Disable isolation for domain(s)",
    )
    disable_parser.add_argument(
        "--domain",
        required=True,
        help="Domain name(s) to disable isolation for (comma-separated)",
    )
    _add_no_multiphp_sync(disable_parser)
    disable_parser.set_defaults(func=cmd_site_isolation_disable)

# isolates-list command (primary)
    list_parser = subparsers.add_parser(
        "isolates-list",
        help="List domains with isolation enabled",
    )
    list_parser.set_defaults(func=cmd_site_isolation_list)

# Deprecated aliases (backward compatible, print warning)
    dep_enable = subparsers.add_parser(
        "site-isolation-enable",
        help="(deprecated, use isolates-enable)",
    )
    dep_enable.add_argument("--domain", required=True, help=argparse.SUPPRESS)
    _add_no_multiphp_sync(dep_enable)
    dep_enable.set_defaults(func=_wrap_deprecated(
        cmd_site_isolation_enable, "site-isolation-enable", "isolates-enable"))

dep_disable = subparsers.add_parser(
        "site-isolation-disable",
        help="(deprecated, use isolates-disable)",
    )
    dep_disable.add_argument("--domain", required=True, help=argparse.SUPPRESS)
    _add_no_multiphp_sync(dep_disable)
    dep_disable.set_defaults(func=_wrap_deprecated(
        cmd_site_isolation_disable, "site-isolation-disable", "isolates-disable"))

dep_list = subparsers.add_parser(
        "site-isolation-list",
        help="(deprecated, use isolates-list)",
    )
    dep_list.set_defaults(func=_wrap_deprecated(
        cmd_site_isolation_list, "site-isolation-list", "isolates-list"))

return parser

def main(argv=None):
    """Main entry point."""
    parser = create_parser()
    args = parser.parse_args(argv)

# Guard: do not allow running as root unless via proxyexec
    # When running via proxyexec, PROXYEXEC_UID is set
    if os.getuid() == 0 and not is_running_via_proxyexec():
        logger.error("Direct root invocation not allowed")
        output_json(json_response(ErrorCodes.ROOT_NOT_ALLOWED))
        return 1

# If running as user (not root via proxyexec)
    if os.getuid() != 0:
        if not in_cagefs():
            print("This utility is only available inside CageFS.\n"
                  "Please run it via: cagefs_enter cagefsctl-user <command>",
                  file=sys.stderr)
            return 1

# Inside CageFS - call via proxyexec to get root privileges
        if not args.command:
            parser.print_help()
            return 1

# Build args list for proxyexec
        args_list = sys.argv[1:]  # Pass all original args
        result = call_via_proxyexec(PROXYEXEC_ALIAS, args_list)
        if result is None:
            output_json(json_response(
                ErrorCodes.INTERNAL_ERROR,
                message="Failed to execute via proxyexec"
            ))
            return 1
        return result

# Running as root via proxyexec - execute the command
    if not hasattr(args, "func"):
        parser.print_help()
        return 1

return args.func(args)

if __name__ == "__main__":
    sys.exit(main())

CANCEL