# 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: `` 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 == "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 self.run_util( "/usr/sbin/proxyexec", "-c", "cagefs.sock", username, os.getcwd(), "CAGEFSCTL_USER", str(os.getpid()), "isolates-" + method, "--domain", domain, env={"CAGEFS_TOKEN": token}) self.exit_with_success({"domain": domain}) 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 [--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): 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.exit_with_success({"domainVersions": {}}) except Exception: self.exit_with_success({"domainVersions": {}}) 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.exit_with_success({"domainVersions": {}}) from clselect import ClUserSelect username = self.user_info.get("username") user_selector = ClUserSelect("php") domain_versions = {} for domain_name in isolated: try: ver_info = user_selector.get_version(username, domain_name) if ver_info and ver_info[0]: domain_versions[domain_name] = ver_info[0] except Exception: logger.debug( "get-domain-versions: failed to get version " "for domain %s of user %s", domain_name, username, exc_info=True, ) self.exit_with_success({"domainVersions": domain_versions}) 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