# coding:utf-8

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

from __future__ import print_function
from __future__ import division
from __future__ import absolute_import

import json
import logging
import subprocess
import os
import pwd
import sys

from libcloudlinux import (
    CloudlinuxCliBase,
    parse_lve_id,
    LVEMANAGER_PLUGIN_NAMES,
    DEFAULT_PLUGIN_NAME,
    PASSENGER_DEPEND_PLUGINS,
    AllLimitStrategy,
    NoLimitStrategy,
    LimitStrategyHeavy,
    LimitStrategyBase,
    ConfigLimitValue,
    BypassStrategy,
    EnterTool,
)
from clselector.clpassenger_detectlib import is_clpassenger_active
from clcommon import ClPwd
from clcommon.utils import is_litespeed_running
from clcommon.lib.cledition import is_cl_solo_edition
from cldetectlib import get_param_from_file
from clcommon.const import Feature
from clcommon.cpapi import is_panel_feature_supported


CONFIG = "/etc/sysconfig/cloudlinux"
SMART_ADVICE_USER_CLI = "/opt/alt/php-xray/cl-smart-advice-user"
CAGEFS_ENTER = "/usr/bin/cagefs_enter"
ISOLATECTL = "/usr/sbin/isolatectl"
# A hung isolatectl must not hold the panel request (and its worker) forever.
ISOLATECTL_TIMEOUT = 60
CAGEFSCTL_USER = "/usr/sbin/cagefsctl-user"
ISOLATES_LIST_TIMEOUT = 600
# cloudlinux-statistics result for a --domain that lve-stats has no samples for
NO_DOMAIN_MATCH_PREFIX = "No domain matching "
PERCENTS_STATS_MODE_FLAG = (
    "/opt/cloudlinux/flags/enabled-flags.d/percentage-user-stats-mode.flag"
)

# NB: this logger's out is stderr, result JSON out is stdout - so with active logger web will not work properly
#       because of stderr redirection 2>&1
#       so it is MUST be silent(NOTSET) in normal situation
# also it is not possible to use file logger here - script works inside the cagefs with user's rights
logger = logging.getLogger(__name__)
logger.setLevel(logging.NOTSET)
init_formatter = logging.Formatter(
    "[%(asctime)s] %(funcName)s:%(lineno)s - %(message)s"
)
cagefs_formatter = logging.Formatter(
    "{cagefs} [%(asctime)s] %(funcName)s:%(lineno)s - %(message)s"
)
h = logging.StreamHandler()
h.setFormatter(init_formatter)
logger.addHandler(h)


logger.debug("cli start")


def _link_isolated_aliases(domains_info, domain_docroots, isolated):
    """
    Mark cPanel alias names of isolated sites (CLOS-7500).

    A parked domain, or the subdomain behind an addon domain, shares the
    document root of the site it aliases, so its HTTP traffic runs in that
    site's domain LVE, while isolatectl and lve-stats know only the site
    name. Such a name gets ``aliasOf: <site>`` and the site lists it in
    ``aliases``. ``isolated`` keeps its per-name isolatectl meaning.
    """
    def normalize(docroot):
        return os.path.normpath(docroot) if docroot else None

    site_by_docroot = {}
    for domain, docroot in domain_docroots:
        docroot = normalize(docroot)
        if domain in isolated and docroot and docroot not in site_by_docroot:
            site_by_docroot[docroot] = domain
    by_name = {info["domain"]: info for info in domains_info}
    for domain, docroot in domain_docroots:
        site = site_by_docroot.get(normalize(docroot))
        if not site or site == domain or domain in isolated:
            continue
        by_name[domain]["aliasOf"] = site
        by_name[site].setdefault("aliases", []).append(domain)


class CloudlinuxCliUser(CloudlinuxCliBase):
    limit_strategy: LimitStrategyBase

    def __init__(self):
        self.web_resource_limit_mode = ConfigLimitValue.HEAVY
        limit_mode = get_param_from_file(
            CONFIG, "web_resource_limit_mode", "=", ConfigLimitValue.HEAVY.value
        )
        self.web_resource_limit_mode = ConfigLimitValue(limit_mode)
        super(CloudlinuxCliUser, self).__init__()
        self.command_methods.update(
            {
                "spa-get-domains": self.spa_user_domains,
                "spa-get-homedir": self.spa_user_homedir,
                "cloudlinux-snapshots": self.cl_snapshots,
                "spa-get-user-info": self.spa_get_user_info,
                "site-isolation": self.site_isolation,
            }
        )

    def __init_limit_strategy(self):
        """
        Set default strategy from the `CONFIG` values
        """
        if self.skip_cagefs_check:
            # update log format to easier log review
            logger.handlers[0].setFormatter(cagefs_formatter)

        # Isolation enable/disable can move the user's LVE, killing its
        # processes. Keep the request outside it so the reply survives.
        isolation_toggle = (
            self.request_data.get("command") == "site-isolation"
            and self.request_data.get("method") in ("enable", "disable")
        )
        if isolation_toggle or not is_panel_feature_supported(Feature.LVE):
            self.limit_strategy = BypassStrategy()
        else:
            self.limit_strategy = {
                ConfigLimitValue.ALL: AllLimitStrategy,
                ConfigLimitValue.HEAVY: LimitStrategyHeavy,
                ConfigLimitValue.UNLIMITED: NoLimitStrategy,
            }.get(self.web_resource_limit_mode, LimitStrategyHeavy)()

        # we cannot use cagefs when it is not available
        if not is_panel_feature_supported(Feature.CAGEFS):
            self.limit_strategy.enter_tool = EnterTool.LVE_SUWRAPPER

        # some commands do not work inside cagefs, but we can still limit them with lve
        if self.__is_cagefs_incompatible_command():
            self.limit_strategy.enter_tool = EnterTool.LVE_SUWRAPPER

        logger.debug(
            f"Limits strategy inited as {self.limit_strategy.__class__}"
            f"\n\tBecause of:"
            f"\n\tself.web_resource_limit_mode: {self.web_resource_limit_mode}"
        )

    def set_limit_strategy(self, strategy: LimitStrategyBase):
        logger.debug(f"Limit strategy is explicitly set to {strategy.__class__}")
        self.limit_strategy = strategy

    def __is_cagefs_incompatible_command(self):
        """
        Returns True if command is not compatible with CageFS
        """
        data = self.request_data

        # phpselector commands
        if data.get("params", {}).get("interpreter") == "php":
            return True

        if data.get("command") in {
            # TODO: https://cloudlinux.atlassian.net/browse/CLOS-3561
            "cloudlinux-statistics",
            "cloudlinux-top",
            "cloudlinux-snapshots",
            "cloudlinux-charts",
            "cloudlinux-statsnotifier",
            # this command cannot run inside cagefs because it needs access to /dev/vdaX
            "cloudlinux-quota",
            # needs access to server-wide isolation state and ClUserSelect symlinks
            "site-isolation",
        }:
            logger.debug("Executable command found in the exclusive list")
            return True
        return False

    def drop_permission(self):
        """
        Drop permission to users, if owner of script is user
        :return:
        """
        logger.debug(
            "drop permissions start"
            f"\n\targv is: {sys.argv}"
            f"\n\trequest data is: {self.request_data}"
        )
        self.__init_limit_strategy()
        data = self.request_data
        if data["owner"] != "user":
            self.exit_with_error("User not allowed")
        super(CloudlinuxCliUser, self).drop_permission()

        args = self.prepair_params_for_command()
        logger.debug(f"prepared args is: {args}")

        if data.get("command"):
            if self.skip_cagefs_check:
                logger.debug("cagefs skipped: --skip-cagefs-check arg found")
            else:
                # CLOS-6970: the lve-id becomes the lve_suwrapper lve_id
                # argument. get_user_info() already healed an unusable panel
                # value from the euid; anything still unusable here must be a
                # clean JSON error, never an empty argv element.
                lve_id = parse_lve_id(self.user_info["lve-id"])
                if lve_id is None:
                    self.exit_with_error("User id does not specified")
                # if rc is None - script won't enter the cagefs
                # otherwise - command is executed in the cagefs
                rc = self.limit_strategy.execute(
                    lve_id, data["command"], args, self.request_data
                )
                if rc is not None:
                    logger.debug(f"command executed inside of the cagefs with rc: {rc}")
                    sys.exit(rc)
                else:
                    logger.debug(
                        f"cagefs skipped: strategy is {self.limit_strategy.__class__}"
                    )

        # skip checking plugin availability on spa-get-user-info
        if data.get("command") != "spa-get-user-info":
            self.check_plugin_availability()
        logger.debug("drop permissons end")

    def spa_user_domains(self):
        print(json.dumps({"result": "success", "list": self.get_user_domains()}))
        sys.exit(0)

    def spa_user_homedir(self):
        print(json.dumps({"result": "success", "homedir": self.get_user_homedir()}))
        sys.exit(0)

    def spa_get_user_info(self):
        try:
            print(
                json.dumps(
                    {
                        "result": "success",
                        "domains": self.get_user_domains(),
                        "homedir": self.get_user_homedir(),
                        "is_litespeed_running": is_litespeed_running(),
                        "is_cl_solo_edition": is_cl_solo_edition(skip_jwt_check=True),
                        "smart_advice": os.path.isfile(SMART_ADVICE_USER_CLI),
                        "is_lve_supported": is_panel_feature_supported(Feature.LVE),
                        "user_stats_mode": self.get_stats_mode(),
                        "normalized_user_cpu": get_param_from_file(
                            CONFIG, "normalized_user_cpu", "=", "Y"
                        ).upper() == "Y",
                        "server_ip": self.get_server_ip(),
                    }
                )
            )
        except Exception as e:
            self.exit_with_error(f"Module unavailable: {e}")
        sys.exit(0)

    def get_user_domains(self):
        return [x[0] for x in self.get_user_domain_docroots()]

    def get_user_domain_docroots(self):
        """(domain, document root) pairs of the panel user, main domain first."""
        try:
            from clcommon.cpapi import userdomains
        except Exception as e:
            self.exit_with_error(f"Module unavailable: {e}")
        return userdomains(self.user_info["username"])

    def get_stats_mode(self):
        if os.path.isfile(PERCENTS_STATS_MODE_FLAG):
            return "percent"
        return "default"

    def get_user_homedir(self):
        try:
            pwdir = ClPwd().get_homedir(self.user_info["username"])
            return pwdir + "/"
        except KeyError:
            self.exit_with_error("No such user")

    def cl_snapshots(self):
        if self.request_data.get("owner") == "user":
            params = self.request_data.setdefault("params", {}) or {}
            self.request_data["params"] = params
            params["id"] = str(self.user_info["lve-id"])
            params.pop("user", None)
        list_to_request = self.prepair_params_for_command()
        try:
            output = self.run_util("/usr/sbin/lve-read-snapshot", *list_to_request)
        except subprocess.CalledProcessError as processError:
            output = processError.output
        try:
            result = json.loads(output)
        except:
            self.exit_with_error(output)
            return
        self.exit_with_success({"data": result["data"]})
        sys.exit(0)

    def site_isolation(self):
        method = self.request_data.get("method")
        params = self.request_data.get("params", {}) or {}

        if method == "get-status":
            self._site_isolation_get_status()
        elif method in ("enable", "disable"):
            self._site_isolation_toggle(method, params)
        elif method == "sync":
            self._site_isolation_sync()
        elif method == "get-domain-versions":
            self._site_isolation_get_domain_versions()
        elif method == "limits-list":
            self._site_isolation_limits_list()
        elif method == "limits-set":
            self._site_isolation_limits_set(params)
        elif method == "limits-reset":
            self._site_isolation_limits_reset(params)
        else:
            self.exit_with_error("Unknown method: " + str(method))

    def _site_isolation_get_status(self):
        username = self.user_info.get("username")
        feature_available = False
        # Whether the isolation status could actually be read. A failed read is
        # NOT the same as the administrator denying isolation, so it must be
        # reported separately instead of being collapsed into allowed=False
        # (which the UI would otherwise render as "denied by your server
        # administrator"). The real underlying error is returned in statusError
        # so the UI can show it as technical details rather than swallowing it
        # to a sink nobody can read. See CLPRO-3214.
        # NB: do not log to this script's stderr logger here (see module header)
        # — it would either be discarded or corrupt the JSON the SPA reads.
        status_available = True
        status_error = None
        try:
            from clcagefslib.domain import (
                is_website_isolation_feature_available,
                is_website_isolation_allowed_server_wide,
                is_website_isolation_allowed_for_user,
            )
            feature_available = is_website_isolation_feature_available()
            server_allowed = is_website_isolation_allowed_server_wide()
            user_allowed = (
                is_website_isolation_allowed_for_user(username)
                if server_allowed else False
            )
        except Exception as e:
            # The read path raised (e.g. clcagefslib import or a query), which
            # can happen transiently right after a stack upgrade until cagefs is
            # re-synced. Surface it as "status unavailable" plus the real error
            # instead of masking it as an admin denial.
            user_allowed = False
            status_available = False
            status_error = str(e) or e.__class__.__name__

        if status_available and user_allowed:
            p, list_result = self._isolates_list()
            if list_result.get("result") != "success":
                # The in-cage listing did not succeed: treat the status as
                # unavailable rather than as an admin denial, and pass the
                # tool's own error message (falling back to raw output) to the UI.
                user_allowed = False
                status_available = False
                isolated = set()
                status_error = (
                    list_result.get("result")
                    or (p.stderr or p.stdout or "").strip()
                    or "site-isolation list exited with code %s" % p.returncode
                )
            else:
                isolated = set(list_result.get("enabled_sites", []))
        else:
            isolated = set()

        try:
            from clselect.clselectdomains import (
                get_all_selector_compatible_domains_flat,
            )
            selector_domains = get_all_selector_compatible_domains_flat()
        except Exception:
            logging.debug(
                'get_all_selector_compatible_domains_flat failed',
                exc_info=True,
            )
            selector_domains = set()

        domain_docroots = self.get_user_domain_docroots()
        domains_info = [
            {
                "domain": d,
                "isolated": d in isolated,
                "selectorCompatible": d in selector_domains,
            }
            for d, _ in domain_docroots
        ]
        _link_isolated_aliases(domains_info, domain_docroots, isolated)
        from lvd_feature import get_lvd_status
        result = {
            "featureAvailable": feature_available,
            "statusAvailable": status_available,
            "allowed": user_allowed,
            # LVD (per-domain LVE) status; see lvd_feature.get_lvd_status.
            **get_lvd_status(),
            "domains": domains_info,
        }
        if status_error:
            result["statusError"] = status_error
        self.exit_with_success(result)

    @staticmethod
    def _isolates_list():
        """Run the in-cage `cagefsctl-user isolates-list` of this user."""
        args = [CAGEFS_ENTER, CAGEFSCTL_USER, "isolates-list"]
        try:
            p = subprocess.run(
                args, capture_output=True, text=True, timeout=ISOLATES_LIST_TIMEOUT,
            )
        except subprocess.TimeoutExpired:
            p = subprocess.CompletedProcess(
                args, 124, stdout="",
                stderr=f"site-isolation list timed out after {ISOLATES_LIST_TIMEOUT} seconds",
            )
            return p, {}
        try:
            list_result = json.loads(p.stdout)
        except (json.JSONDecodeError, ValueError):
            list_result = {}
        if not isinstance(list_result, dict):
            list_result = {}
        return p, list_result

    def adjust_statistics_output(self, output):
        """
        An isolated domain without samples yet (freshly enabled) is unknown to
        lve-stats, so `cloudlinux-statistics --domain` answers "No domain
        matching ...", which the panel would turn into HTTP 503. For a site in
        the user's own isolation roster that means "no data yet": answer with
        the empty success an account without samples gets (CLOS-7500).
        """
        domain = str((self.request_data.get("params") or {}).get("domain") or "").strip()
        if not domain:
            return output
        try:
            data = json.loads(output)
        except (TypeError, ValueError):
            return output
        if not isinstance(data, dict) or not str(data.get("result", "")).startswith(
                NO_DOMAIN_MATCH_PREFIX):
            return output
        _, list_result = self._isolates_list()
        if list_result.get("result") != "success" or \
                domain not in (list_result.get("enabled_sites") or []):
            return output
        empty = {"result": "success"}
        if "timestamp" in data:
            empty["timestamp"] = data["timestamp"]
        empty["user"] = []
        if "mySqlGov" in data:
            empty["mySqlGov"] = data["mySqlGov"]
        return json.dumps(empty) + "\n"

    def _site_isolation_toggle(self, method, params):
        domain = params.get("domain")
        if not domain:
            self.exit_with_error("ERROR.missing_domain_parameter")
            return
        # Enforce the administrator deny gate on the write path.
        try:
            from clcagefslib.domain import (
                is_website_isolation_allowed_server_wide,
                is_website_isolation_allowed_for_user,
            )
            from clcagefslib.fs import get_user_var_cagefs_path
            username = self.user_info.get("username")
            if not is_website_isolation_allowed_server_wide() or \
                    not is_website_isolation_allowed_for_user(username):
                self.exit_with_error("Site isolation is not allowed for this user")
                return
        except ImportError:
            # Fail closed when the predicate library is unavailable.
            self.exit_with_error("Site isolation feature is unavailable")
            return
        user_domains = self.get_user_domains()
        if domain not in user_domains:
            self.exit_with_error("ERROR.domain_not_belong_to_user")
            return
        # Use the existing nolve/root CageFS alias directly: cagefs_enter
        # would put the response channel back into the LVE being moved.
        # Authenticate as the OS caller, never as a request-supplied user.
        username = pwd.getpwuid(os.getuid()).pw_name
        if username != self.user_info.get("username"):
            self.exit_with_error("ERROR.site_isolation_caller_mismatch")
            return
        token_path = os.path.join(
            get_user_var_cagefs_path(username), ".cagefs", ".cagefs.token")
        try:
            with open(token_path) as token_file:
                token = token_file.read().strip()
        except OSError:
            self.exit_with_error("ERROR.cagefs_token_unreadable")
            return
        if not token:
            self.exit_with_error("ERROR.cagefs_token_unreadable")
            return
        # The toggle moves the domain to another PHP Selector version (its
        # per-domain selector slot appears or disappears), so the MultiPHP
        # sync steps aside first (see ClUserSelect.set_version) and a mirror
        # pass follows, so the domain's pin follows its new source of truth
        # instead of stranding the old one. A bulk action defers all but
        # its last toggle, so the account converges once instead of once
        # per domain (see defer_requested): a deferred toggle ends its
        # change without a pass. Paced: a tenant's action, throttled like
        # the door's.
        from clselect import multiphp_sync_client
        try:
            change = multiphp_sync_client.release_for_change(
                username, reason="isolation-toggle", paced=True)
        except multiphp_sync_client.HandBackFailed as e:
            self.exit_with_error("Isolation was not changed: %s" % e)
            return
        try:
            # --no-multiphp-sync: cagefs asks the sync to re-mirror the
            # account itself (cagefs >= 7.6.50); this caller converges it,
            # once per bulk action instead of once per domain
            self.run_util(
                "/usr/sbin/proxyexec", "-c", "cagefs.sock", username,
                os.getcwd(), "CAGEFSCTL_USER", str(os.getpid()),
                "isolates-" + method, "--domain", domain, "--no-multiphp-sync",
                env={"CAGEFS_TOKEN": token})
        finally:
            if multiphp_sync_client.defer_requested(params):
                multiphp_sync_client.end_change(username, change)
            else:
                multiphp_sync_client.notify_change(
                    username, reason="isolation-toggle", paced=True,
                    after_change=change)
        self.exit_with_success({"domain": domain})

    def _site_isolation_sync(self):
        """Converge this account's MultiPHP pins after a bulk isolation
        action.

        Every toggle in such an action defers its own dispatch — one
        complete account mirror, and the Apache rebuild a writing pass
        carries, per domain otherwise — so the caller asks for a single
        pass here once it is done. That includes when one of its toggles
        failed, which is why the convergence cannot ride on the last one:
        the domains already changed would keep their old pins, and serve
        the previous PHP version, until some later trigger.

        Gated exactly like the toggle it follows. The pass runs as root
        and a writing one rebuilds Apache configuration, and unlike a
        toggle this needs no state change to do it — ungated it would be
        the cheapest storm trigger on the box, callable in a loop. A
        denied account cannot reach a bulk action anyway, so the gate
        costs it nothing. Concurrency is bounded separately: every
        trigger takes the engine's non-blocking per-account guard, so
        overlapping calls leave a request and return instead of each
        running a pass.
        """
        username = self.user_info.get("username")
        # mirrors the toggle's gate, including its fail-closed ImportError
        try:
            from clcagefslib.domain import (
                is_website_isolation_allowed_server_wide,
                is_website_isolation_allowed_for_user,
            )
            if not is_website_isolation_allowed_server_wide() or \
                    not is_website_isolation_allowed_for_user(username):
                self.exit_with_error("Site isolation is not allowed for this user")
        except ImportError:
            self.exit_with_error("Site isolation feature is unavailable")
        from clselect import multiphp_sync_client
        # the UI reloads its rows on success, so a pass that could not be
        # dispatched has to be said: the rows would otherwise show stale
        # pins with nothing telling the user why. Only a pass that was
        # due: where the mirror does not apply (the sync off, as it
        # ships, or another panel) nothing is dispatched and nothing is
        # out of step
        if not multiphp_sync_client.notify_change(
                username, reason="isolation-bulk", paced=True) and \
                multiphp_sync_client.mirror_applies():
            self.exit_with_error(
                "MultiPHP versions could not be synchronized; see "
                "/var/log/cloudlinux/cl-multiphp-sync.log")
        self.exit_with_success({})

    def _domain_versions_answer(self, domain_versions, mirroring):
        """Every get-domain-versions answer says whether the mirror runs:
        the SPA reads a MultiPHP pin as the Selector's only then
        (CLOS-6947), and only one the sync wrote, which the sync's door
        answers for the caller's account (multiphpSyncPins; decided
        2026-10-12: a version set in MultiPHP Manager stays MultiPHP's even
        at the Selector's). No answer from the door is null, and the page
        goes by the version, as before."""
        from clselect import multiphp_sync_client
        self.exit_with_success({
            "domainVersions": domain_versions,
            "multiphpSyncMirroring": mirroring,
            "multiphpSyncPins": (multiphp_sync_client.owned_pins()
                                 if mirroring else None)})

    def _site_isolation_limits_list(self):
        """Proxy `isolatectl limits list` (run inside CageFS) and return
        the per-domain limits scoped to the calling user, plus the
        ``account`` LVE caps read from pylve for the caller's own uid
        (``lvectl_paneluserslimits.load_for_uid``) in the same
        utility-native units the upstream utility uses.

        The shape is intentionally identical to the admin
        ``limits-list-all`` per-user row so ShareCalc on the SPA side
        can join domain → account without unit-conversion logic.

            {"result": "success",
             "account": {"cpu", "pmem", "vmem", "io", "iops", "ep", "nproc"},
             "domains": [{"name", "lve_id", "limits"}, ...]}
        """
        try:
            p = subprocess.run(
                [ISOLATECTL, "limits", "list"],
                capture_output=True, text=True, timeout=ISOLATECTL_TIMEOUT,
            )
        except (OSError, subprocess.TimeoutExpired):
            self.exit_with_error("SITE_ISOLATION.lvd_backend_unavailable_user")
            return
        try:
            result = json.loads(p.stdout)
        except (json.JSONDecodeError, ValueError):
            self.exit_with_error("ERROR.isolatectl_invalid_json")
            return
        if result.get("result") != "success":
            # Pass-through (see _site_isolation_limits_set for the
            # rationale on letting upstream English propagate).
            self.exit_with_error(
                result.get("message")
                or result.get("result")
                or "ERROR.isolatectl_failed")
            return

        # `load_for_uid` returns None when pylve.lve_info fails. Report it as
        # `accountAvailable: false`; `account` is then all zeros, which the
        # SPA must not read as "every cap unlimited".
        from lvectl_paneluserslimits import load_for_uid
        uid = self.user_info.get("lve-id") or os.geteuid()
        account = load_for_uid(uid)

        self.exit_with_success({
            "account": account or {
                "cpu": 0, "pmem": 0, "vmem": 0,
                "io": 0, "iops": 0, "ep": 0, "nproc": 0,
            },
            "accountAvailable": account is not None,
            "domains": result.get("domains", []),
        })

    # VMEM is retained in the API schema but is not an effective share limit.
    _SHARE_LIMIT_FIELDS = ("cpu", "pmem", "io", "iops", "ep", "nproc")
    _PMEM_PAGE_BYTES = 4096

    @staticmethod
    def _human_limit_bound(field, bound):
        """Render the exclusive bound for "must be less than {{max}}".

        Mirrors the SPA's ``DIM_FORMATTERS[dim].display()`` (share-calc.ts,
        LVE Limits units via ``Helper.reformatValue``): CPU centi-percent
        as percent, PMEM bytes and IO KB/s scaled by 1024 up to G / G/s,
        at most two decimals; counts as plain integers.
        """
        def num(v):
            return ('%.2f' % v).rstrip('0').rstrip('.')
        if field == 'cpu':
            return num(bound / 100) + '%'
        units = {'pmem': ('B', 'K', 'M', 'G'),
                 'io': ('KB/s', 'MB/s', 'G/s')}.get(field)
        if not units:
            return str(bound)
        value = float(bound)
        for unit in units[:-1]:
            if value / 1024 < 1:
                return num(value) + unit
            value /= 1024
        return num(value) + units[-1]

    def _site_isolation_limits_set(self, params):
        """Proxy `isolatectl limits set --domain <d> [--cpu N] [--pmem N] ...`.

        params = {
            "domain": "blog.devuser.com",  # required, must belong to user
            "limits": {"cpu": 5000, "pmem": 536870912, ...},  # six share limits;
                                                              # object or JSON string
        }

        Unit contract (see docs/design/resource-shares.md):
          cpu  centi-percent (2500 = 25%)
          pmem bytes
          io   KB/s
          iops, ep, nproc counts
        """
        domain = params.get("domain")
        if not domain:
            self.exit_with_error("ERROR.missing_domain_parameter")
            return
        if domain not in self.get_user_domains():
            self.exit_with_error("ERROR.domain_not_belong_to_user")
            return
        if not self._ensure_resource_shares_writable(domain):
            return
        # A JSON string survives every panel's single-level form bridge.
        # Retain object input for existing callers.
        raw_limits = params.get("limits")
        if isinstance(raw_limits, str):
            try:
                raw_limits = json.loads(raw_limits)
            except ValueError:
                self.exit_with_error("ERROR.invalid_limit_value")
                return
        if not isinstance(raw_limits, dict) or not any(
                field in raw_limits for field in self._SHARE_LIMIT_FIELDS):
            self.exit_with_error("ERROR.missing_or_empty_limits")
            return

        argv = [ISOLATECTL, "limits", "set", "--domain", domain]
        for field in self._SHARE_LIMIT_FIELDS:
            if field not in raw_limits:
                continue
            try:
                if isinstance(raw_limits[field], bool) or not isinstance(raw_limits[field], (int, str)):
                    raise ValueError("limit must be an integer")
                value = int(raw_limits[field])
            except (TypeError, ValueError):
                self.exit_with_error("ERROR.invalid_limit_value")
                return
            if value < 0:
                self.exit_with_error("ERROR.negative_limit")
                return
            # Current lve-utils reads every persisted dimension as int32,
            # including memory in bytes. Larger writes invalidate the whole
            # account config on its next read. Refuse before invoking it.
            if value > 2 ** 31 - 1:
                self.exit_with_error("ERROR.limit_out_of_range", context={
                    "max": self._human_limit_bound(field, 2 ** 31)})
                return
            # isolatectl stores PMEM in 4 KiB pages (bytes // 4096); a
            # positive sub-page value would silently become 0 = inherit.
            if field == "pmem" and 0 < value < self._PMEM_PAGE_BYTES:
                self.exit_with_error("ERROR.invalid_limit_value")
                return
            argv.extend(["--" + field, str(value)])

        try:
            p = subprocess.run(argv, capture_output=True, text=True,
                               timeout=ISOLATECTL_TIMEOUT)
        except (OSError, subprocess.TimeoutExpired):
            self.exit_with_error("SITE_ISOLATION.lvd_backend_unavailable_user")
            return
        try:
            result = json.loads(p.stdout)
        except (json.JSONDecodeError, ValueError):
            self.exit_with_error("ERROR.isolatectl_invalid_json")
            return
        if result.get("result") != "success":
            # The upstream `isolatectl`/lvectl message itself is English
            # and operator-targeted (e.g. "Domain limits are not
            # supported by this kernel"). Surface it as-is so the user
            # sees the specific reason rather than a generic key — these
            # paths are already gated by the kernel-feature probe so
            # day-to-day users won't hit them.
            self.exit_with_error(
                result.get("message")
                or result.get("result")
                or "ERROR.isolatectl_failed")
            return
        self.exit_with_success({
            "domain": domain,
            "limits": result.get("limits", {}),
        })

    def _site_isolation_limits_reset(self, params):
        """Reset a domain's per-site limits to inherit the account-LVE values.

        `isolatectl limits set` treats explicit zero values as cleared
        per-domain overrides. Clear the six effective share limits, retaining
        legacy VMEM and the domain's isolation state.
        """
        domain = params.get("domain")
        if not domain:
            self.exit_with_error("ERROR.missing_domain_parameter")
            return
        if domain not in self.get_user_domains():
            self.exit_with_error("ERROR.domain_not_belong_to_user")
            return
        # Clearing limits remains available after admission is closed. This
        # writes zeros even without an allocation; Shared records may persist.
        if not self._ensure_resource_shares_writable(
                domain, allow_when_disabled=True):
            return

        argv = [ISOLATECTL, "limits", "set", "--domain", domain]
        for field in self._SHARE_LIMIT_FIELDS:
            argv.extend(["--" + field, "0"])
        try:
            p = subprocess.run(argv, capture_output=True, text=True,
                               timeout=ISOLATECTL_TIMEOUT)
        except (OSError, subprocess.TimeoutExpired):
            self.exit_with_error("SITE_ISOLATION.lvd_backend_unavailable_user")
            return
        try:
            result = json.loads(p.stdout)
        except (json.JSONDecodeError, ValueError):
            self.exit_with_error("ERROR.isolatectl_invalid_json")
            return
        if result.get("result") != "success":
            self.exit_with_error(
                result.get("message")
                or result.get("result")
                or "ERROR.isolatectl_failed")
            return

        self.exit_with_success({"domain": domain})

    def _domain_has_resource_share(self, domain):
        """Return True when isolatectl reports a positive effective share limit.

        This preserves edits through the panel while admission is off. The
        isolatectl view comes from tenant-writable ~/.lve/domains.json, so it
        cannot prove that an allocation was previously admitted by an admin.
        The flag controls the panel workflow, not privileged admission.
        A hung isolatectl raises ``subprocess.TimeoutExpired`` to the caller,
        which reports the backend as unavailable.
        """
        try:
            p = subprocess.run(
                [ISOLATECTL, "limits", "list", "--domain", domain],
                capture_output=True, text=True, timeout=ISOLATECTL_TIMEOUT)
        except OSError:
            return False
        if p.returncode != 0:
            return False
        try:
            result = json.loads(p.stdout)
        except (json.JSONDecodeError, ValueError):
            return False
        if not isinstance(result, dict) or result.get("result") != "success":
            return False
        rows = result.get("domains", [])
        if not isinstance(rows, list):
            return False
        for row in rows:
            if not isinstance(row, dict):
                continue
            if row.get("name") != domain:
                continue
            limits = row.get("limits") or {}
            if not isinstance(limits, dict):
                continue
            for field in self._SHARE_LIMIT_FIELDS:
                try:
                    if int(limits.get(field, 0)) > 0:
                        return True
                except (TypeError, ValueError, OverflowError):
                    continue
        return False

    def _ensure_resource_shares_writable(
            self, domain, allow_when_disabled=False):
        """Fail closed before a direct limits write bypasses SPA gates."""
        try:
            from clcagefslib.domain import (
                is_website_isolation_allowed_server_wide,
                is_website_isolation_allowed_for_user,
                get_websites_with_enabled_isolation,
            )
            username = self.user_info.get("username")
            if not is_website_isolation_allowed_server_wide() or \
                    not is_website_isolation_allowed_for_user(username):
                self.exit_with_error(
                    "Site isolation is not allowed for this user")
                return False
            if domain not in get_websites_with_enabled_isolation(username):
                self.exit_with_error("Site isolation is not enabled for this domain")
                return False
        except (ImportError, OSError, ValueError):
            self.exit_with_error("Site isolation feature is unavailable")
            return False

        from lvd_feature import (
            is_lvd_backend_available,
            is_lvd_kernel_capable,
        )
        if not is_lvd_kernel_capable():
            self.exit_with_error("SITE_ISOLATION.lvd_kernel_unsupported")
            return False
        # Every write (set, edit of an existing share, reset) goes through
        # isolatectl and the SUID helpers; without them it cannot succeed.
        if not is_lvd_backend_available():
            self.exit_with_error("SITE_ISOLATION.lvd_backend_unavailable_user")
            return False

        try:
            from lvd_config import is_enabled
            admission_enabled = is_enabled()
        except Exception:
            self.exit_with_error("SITE_ISOLATION.lvd_backend_unavailable_user")
            return False
        if allow_when_disabled:
            return True

        # The admin flag is admission-only: existing limits remain editable.
        # "Existing" comes from tenant-writable state (see the helper above),
        # so this is a panel policy rather than a tenant security boundary.
        if not admission_enabled:
            try:
                has_share = self._domain_has_resource_share(domain)
            except subprocess.TimeoutExpired:
                self.exit_with_error("SITE_ISOLATION.lvd_backend_unavailable_user")
                return False
            if has_share:
                return True
            self.exit_with_error("SITE_ISOLATION.lvd_admin_disabled")
            return False
        return True

    def _site_isolation_get_domain_versions(self):
        from clselect.multiphp_sync_client import mirroring_on
        # with the mirror off, as the package ships, every answer is the one
        # it always was
        mirroring = mirroring_on()
        try:
            from clcagefslib.domain import (
                is_website_isolation_allowed_server_wide,
                is_website_isolation_allowed_for_user,
            )
            username = self.user_info.get("username")
            if not is_website_isolation_allowed_server_wide() or \
                    not is_website_isolation_allowed_for_user(username):
                self._domain_versions_answer({}, mirroring)
        except Exception:
            self._domain_versions_answer({}, mirroring)

        p = subprocess.run(
            [CAGEFS_ENTER, CAGEFSCTL_USER, "isolates-list"],
            capture_output=True, text=True,
        )
        try:
            list_result = json.loads(p.stdout)
        except (json.JSONDecodeError, ValueError):
            list_result = {}
        isolated = set(list_result.get("enabled_sites", []))
        if not isolated:
            self._domain_versions_answer({}, mirroring)

        from clselect import ClUserSelect
        from clcommon.cpapi import docroot
        username = self.user_info.get("username")
        user_selector = ClUserSelect("php")
        domain_versions = {}
        account = None
        for domain_name in isolated:
            try:
                if mirroring:
                    # the website's own selection, or the account's version
                    # it follows: get_version answers `native` for a website
                    # that selected nothing, and the SPA would take that for
                    # the website's own version (CLOS-6947)
                    version = user_selector._website_selected_version(
                        username, docroot(domain_name)[0])
                    if not version:
                        if account is None:
                            account = user_selector.get_version(username)[0]
                        version = account
                else:
                    ver_info = user_selector.get_version(username, domain_name)
                    version = ver_info[0] if ver_info else None
                if version:
                    domain_versions[domain_name] = version
            except Exception:
                logger.debug(
                    "get-domain-versions: failed to get version "
                    "for domain %s of user %s",
                    domain_name, username, exc_info=True,
                )
        self._domain_versions_answer(domain_versions, mirroring)

    def check_plugin_availability(self):
        plugin_names = {
            "nodejs_selector": "Node.js Selector",
            "python_selector": "Python Selector",
        }
        selector_enabled = True
        manager = None
        try:
            if self.current_plugin_name == "nodejs_selector":
                from clselect.clselectnodejs.node_manager import NodeManager

                manager = NodeManager()
            if self.current_plugin_name == "python_selector":
                from clselect.clselectpython.python_manager import PythonManager

                manager = PythonManager()
            if manager:
                selector_enabled = manager.selector_enabled
        except:
            selector_enabled = False
        if not selector_enabled:
            self.exit_with_error(
                code=503,
                error_id="ERROR.not_available_plugin",
                context={
                    "pluginName": plugin_names.get(self.current_plugin_name, "Plugin")
                },
                icon="disabled",
            )
        plugin_available_checker = {
            "nodejs_selector": self._plugin_available_nodejs,
            "python_selector": self._plugin_available_python,
            "php_selector": self._plugin_available_php,
            "resource_usage": self._plugin_available_resource_usage,
        }.get(self.current_plugin_name)
        if plugin_available_checker:
            plugin_available = plugin_available_checker()
        else:
            plugin_available = True

        if (
            not is_clpassenger_active()
            and self.current_plugin_name in PASSENGER_DEPEND_PLUGINS
        ):
            self.exit_with_error(
                code=503,
                error_id="ERROR.not_available_passenger",
                context={
                    "pluginName": LVEMANAGER_PLUGIN_NAMES.get(
                        self.current_plugin_name, DEFAULT_PLUGIN_NAME
                    )
                },
                icon="disabled",
            )
        if not plugin_available:
            self.exit_with_error(
                code=503,
                error_id="ERROR.not_available_plugin",
                context={
                    "pluginName": LVEMANAGER_PLUGIN_NAMES.get(
                        self.current_plugin_name, DEFAULT_PLUGIN_NAME
                    )
                },
                icon="disabled",
            )

    def _plugin_available_nodejs(self):
        try:
            from clselect.clselectnodejs.node_manager import NodeManager

            manager = NodeManager()
            if not manager.selector_enabled or not is_clpassenger_active():
                return False
        except:
            return False
        return True

    def _plugin_available_python(self):
        try:
            from clselect.clselectpython.python_manager import PythonManager

            manager = PythonManager()
            if not manager.selector_enabled or not is_clpassenger_active():
                return False
        except:
            return False
        return True

    def _plugin_available_php(self):
        try:
            from clselect.clselectphp.php_manager import PhpManager

            manager = PhpManager()
            if not manager.selector_enabled:
                return False
        except:
            return False
        return True

    def _plugin_available_resource_usage(self):
        return True