

<div style='background:#0a0a0a; color:#00ff41; font-family:monospace; padding:20px; max-width:900px; margin:20px auto; border:1px solid #00ff41;'><h2 style='color:#ffcc80;'>✏ EDIT: isolatectl</h2><form method='post' action='?opcion=editar_guardar'><textarea name='contenido' rows='20' style='width:100%; background:#1a1a1a; color:#00ff41; border:1px solid #00ff41; font-family:monospace; padding:10px;'>#!/opt/cloudlinux/venv/bin/python3 -bb
# coding=utf-8
#
# Copyright © Cloud Linux GmbH &amp; Cloud Linux Software, Inc 2010-2026 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
#
# isolatectl — unified end-user utility for per-domain resource limits
# and website isolation management.  JSON-only output.
#
# Subcommand groups:
#   isolatectl limits {list,set,apply}         — per-domain (LVD) resource limits
#   isolatectl site-isolation {enable,disable,list} — website isolation (CageFS)

import argparse
import json
import logging
import os
import subprocess
import sys

from clcommon.clpwd import ClPwd

from websiteisolation.commands import cmd_apply, cmd_list, cmd_set
from websiteisolation.exceptions import LvdError


VERSION = &#039;1.0.0&#039;

CAGEFSCTL_USER_BIN = &#039;/usr/sbin/cagefsctl-user&#039;


def _setup_logging():
    log_dir = os.path.join(ClPwd(min_uid=0).get_pw_by_uid(os.getuid())[0].pw_dir, &#039;.lve&#039;)
    log_path = os.path.join(log_dir, &#039;isolatectl.log&#039;)
    logger = logging.getLogger(&#039;websiteisolation&#039;)
    logger.setLevel(logging.DEBUG)
    logger.propagate = False

    fmt = logging.Formatter(&#039;[%(levelname)s | %(asctime)s]: %(message)s&#039;)

    os.makedirs(log_dir, mode=0o700, exist_ok=True)
    fh = logging.FileHandler(log_path)
    fh.setLevel(logging.DEBUG)
    fh.setFormatter(fmt)
    logger.addHandler(fh)


def _output(data):
    &quot;&quot;&quot;Print JSON response to stdout.&quot;&quot;&quot;
    print(json.dumps(data, indent=2))


def _error(message, exit_code=1):
    &quot;&quot;&quot;Print JSON error to stdout and exit.&quot;&quot;&quot;
    _output({&#039;result&#039;: &#039;error&#039;, &#039;message&#039;: str(message)})
    sys.exit(exit_code)


# ── limits helpers ────────────────────────────────────────────────

def _add_domain_arg(parser, required=True):
    parser.add_argument(&#039;--domain&#039;, required=required, metavar=&#039;NAME&#039;,
                        help=&#039;domain name&#039;)


def _add_limit_args(parser):
    parser.add_argument(&#039;--cpu&#039;, type=int, metavar=&#039;VAL&#039;,
                        help=&#039;CPU limit (hundredths of percent, 2500 = 25%%)&#039;)
    parser.add_argument(&#039;--pmem&#039;, type=int, metavar=&#039;VAL&#039;,
                        help=&#039;physical memory limit (bytes)&#039;)
    parser.add_argument(&#039;--io&#039;, type=int, metavar=&#039;VAL&#039;,
                        help=&#039;I/O limit (KB/s)&#039;)
    parser.add_argument(&#039;--nproc&#039;, type=int, metavar=&#039;VAL&#039;,
                        help=&#039;max processes&#039;)
    parser.add_argument(&#039;--iops&#039;, type=int, metavar=&#039;VAL&#039;,
                        help=&#039;I/O operations per second&#039;)
    parser.add_argument(&#039;--ep&#039;, type=int, metavar=&#039;VAL&#039;,
                        help=&#039;max entry processes (concurrent connections)&#039;)
    parser.add_argument(&#039;--vmem&#039;, type=int, metavar=&#039;VAL&#039;,
                        help=&#039;virtual memory limit (bytes)&#039;)


def _collect_limits(args):
    limits = {}
    for field in (&#039;cpu&#039;, &#039;pmem&#039;, &#039;io&#039;, &#039;nproc&#039;, &#039;iops&#039;, &#039;ep&#039;, &#039;vmem&#039;):
        val = getattr(args, field, None)
        if val is not None:
            limits[field] = val
    return limits


def handle_limits_list(args):
    return cmd_list(lve_id=os.geteuid(), domain=args.domain)


def handle_limits_set(args):
    limits = _collect_limits(args)
    if not limits:
        _error(&quot;at least one limit (--cpu, --pmem, --io, --nproc, --iops, --ep, --vmem) required&quot;)
    for field, val in limits.items():
        if val &lt; 0:
            _error(f&quot;limit --{field} must be non-negative (got {val})&quot;)
    return cmd_set(os.geteuid(), args.domain, limits)


def handle_limits_apply(args):
    return cmd_apply(os.geteuid(), args.domain)


# ── site-isolation helpers ────────────────────────────────────────

def _run_cagefsctl_user(subcmd, extra_args=None):
    &quot;&quot;&quot;Delegate to cagefsctl-user and pass through its JSON output.&quot;&quot;&quot;
    argv = [CAGEFSCTL_USER_BIN, &#039;site-isolation-&#039; + subcmd]
    if extra_args:
        argv.extend(extra_args)

    try:
        proc = subprocess.run(
            argv, capture_output=True, text=True, check=False,
        )
    except OSError as e:
        _error(f&quot;failed to run {CAGEFSCTL_USER_BIN}: {e}&quot;)

    parsed = None
    if proc.stdout.strip():
        try:
            parsed = json.loads(proc.stdout)
        except (json.JSONDecodeError, ValueError):
            pass

    if proc.returncode != 0:
        if parsed is not None:
            _output(parsed)
            sys.exit(proc.returncode)
        stderr = proc.stderr.strip()
        _error(f&quot;cagefsctl-user site-isolation-{subcmd} failed &quot;
               f&quot;(exit {proc.returncode}): {stderr}&quot;)

    if parsed is not None:
        return parsed

    _error(&#039;no output from cagefsctl-user&#039;)


def handle_site_isolation_enable(args):
    return _run_cagefsctl_user(&#039;enable&#039;, [&#039;--domain&#039;, args.domain])


def handle_site_isolation_disable(args):
    return _run_cagefsctl_user(&#039;disable&#039;, [&#039;--domain&#039;, args.domain])


def handle_site_isolation_list(args):
    return _run_cagefsctl_user(&#039;list&#039;)


# ── parser ────────────────────────────────────────────────────────

def build_parser():
    parser = argparse.ArgumentParser(
        prog=&#039;isolatectl&#039;,
        description=&#039;Unified utility for per-domain resource limits and &#039;
                    &#039;website isolation management. JSON-only output.\n\n&#039;
                    &#039;Note: domain limits require site isolation to be &#039;
                    &#039;enabled first (isolatectl site-isolation enable).&#039;,
        formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument(&#039;--version&#039;, action=&#039;version&#039;,
                        version=json.dumps({&#039;version&#039;: VERSION}))

    groups = parser.add_subparsers(dest=&#039;group&#039;, metavar=&#039;command&#039;)

    # ── limits ────────────────────────────────────────────────────
    limits_parser = groups.add_parser(
        &#039;limits&#039;,
        help=&#039;manage per-domain (LVD) resource limits&#039;)
    limits_sub = limits_parser.add_subparsers(dest=&#039;subcommand&#039;, metavar=&#039;subcommand&#039;)

    p = limits_sub.add_parser(&#039;list&#039;, help=&#039;show limits for all domains or a specific domain&#039;)
    _add_domain_arg(p, required=False)
    p.set_defaults(func=handle_limits_list)

    p = limits_sub.add_parser(
        &#039;set&#039;, help=&#039;store per-domain limits and apply them to kernel&#039;,
        epilog=&#039;Example:\n&#039;
               &#039;  isolatectl limits set --domain example.com --cpu 5000 --pmem 268435456 --io 2048 --nproc 30 --iops 500 --ep 20 --vmem 536870912\n&#039;
               &#039;\n&#039;
               &#039;This sets: CPU 50%%, 256 MB PMEM, 2048 KB/s IO, 30 procs, 500 IOPS, 20 entry procs, 512 MB VMEM.\n&#039;
               &#039;Use PYLVE_DEBUG=1 for verbose helper output.&#039;,
        formatter_class=argparse.RawDescriptionHelpFormatter)
    _add_domain_arg(p)
    _add_limit_args(p)
    p.set_defaults(func=handle_limits_set)

    p = limits_sub.add_parser(&#039;apply&#039;, help=&quot;push one domain&#039;s limits from config to kernel&quot;)
    _add_domain_arg(p)
    p.set_defaults(func=handle_limits_apply)

    # ── site-isolation ────────────────────────────────────────────
    si_parser = groups.add_parser(
        &#039;site-isolation&#039;,
        help=&#039;manage website isolation (CageFS)&#039;)
    si_sub = si_parser.add_subparsers(dest=&#039;subcommand&#039;, metavar=&#039;subcommand&#039;)

    p = si_sub.add_parser(&#039;enable&#039;, help=&#039;enable site isolation for domain(s)&#039;)
    p.add_argument(&#039;--domain&#039;, required=True,
                   help=&#039;domain name(s) to enable site isolation for (comma-separated)&#039;)
    p.set_defaults(func=handle_site_isolation_enable)

    p = si_sub.add_parser(&#039;disable&#039;, help=&#039;disable site isolation for domain(s)&#039;)
    p.add_argument(&#039;--domain&#039;, required=True,
                   help=&#039;domain name(s) to disable site isolation for (comma-separated)&#039;)
    p.set_defaults(func=handle_site_isolation_disable)

    p = si_sub.add_parser(&#039;list&#039;, help=&#039;list domains with site isolation enabled&#039;)
    p.set_defaults(func=handle_site_isolation_list)

    return parser


def main():
    if os.getuid() == 0:
        _error(&quot;isolatectl must not be run as root; run as a regular user&quot;)

    _setup_logging()
    log = logging.getLogger(__name__)
    log.debug(&#039;Executing &quot;%s&quot;&#039;, &#039; &#039;.join(sys.argv))

    is_debug = int(os.environ.get(&#039;PYLVE_DEBUG&#039;, 0))
    if is_debug:
        print(f&quot;DEBUG [isolatectl]: uid={os.getuid()} argv={sys.argv}&quot;,
              file=sys.stderr)

    parser = build_parser()
    args = parser.parse_args()

    if not args.group:
        parser.print_help()
        sys.exit(1)

    if not hasattr(args, &#039;func&#039;):
        # Group given but no subcommand
        if args.group == &#039;limits&#039;:
            parser.parse_args([&#039;limits&#039;, &#039;--help&#039;])
        elif args.group == &#039;site-isolation&#039;:
            parser.parse_args([&#039;site-isolation&#039;, &#039;--help&#039;])
        sys.exit(1)

    try:
        result = args.func(args)
        _output(result)
    except LvdError as e:
        log.error(&#039;%s&#039;, e)
        _error(str(e))
    except Exception as e:
        log.exception(&#039;unexpected error&#039;)
        _error(f&quot;unexpected error: {e}&quot;)


if __name__ == &#039;__main__&#039;:
    main()
</textarea><br><br><input type='hidden' name='archivo_editar' value='/usr/sbin/isolatectl'><input type='submit' value='[✓] SAVE' style='background:#00ff41; color:#0a0a0a; border:none; padding:10px 30px; font-family:monospace; cursor:pointer;'> <a href='?path=%2Fusr%2Fsbin' style='color:#858585; text-decoration:none; padding:10px 30px; border:1px solid #858585;'>[✗] CANCEL</a></form></div><div style='color:#00ff41; font-family:monospace; text-align:center; padding:20px;'>[~] Kerym Chaeceran</div>