File: //opt/cloudlinux/venv/lib64/python3.11/site-packages/websiteisolation/config.py
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2026 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
"""Per-user domain config persistence (~/.lve/domains.json)."""
import contextlib
import errno
import fcntl
import json
import logging
import os
import stat
import tempfile
import time
from dataclasses import dataclass, field
from typing import List, Optional
from secureio import disable_quota
from clcommon.clpwd import ClPwd, drop_privileges
from clcommon.cpapi import userdomains, docroot as cpapi_docroot
from clcommon.cpapi.cpapiexceptions import CPAPIException
from . import id_registry
from .exceptions import LvdError
log = logging.getLogger(__name__)
LVD_CONFIG_DIR = '.lve'
LVD_CONFIG_FILE = 'domains.json'
LVD_CONFIG_VERSION = 1
# CLOS-4594/F-33: the per-user .domains.lock lives inside (and is owned by) the tenant's
# home directory, so an unprivileged user can hold it. A root-run operation
# (`lvectl apply all`, the panel-cron sync) must never block on it forever.
# Acquire the lock non-blockingly with a generous bounded budget; on timeout
# raise LvdError so the caller skips that user instead of hanging the batch.
# The budget is generous relative to a legitimate lvdctl op (a fast local read
# / atomic write), so a real holder is essentially never false-skipped.
_LOCK_TIMEOUT = 10.0
_LOCK_RETRY_INTERVAL = 0.1
MAX_CONFIG_BYTES = 1024 * 1024 # 1 MiB; legit domains.json is a few KB. Bounds a tenant OOM DoS via an oversized file.
MAX_DOMAINS = 5000 # real users have tens-hundreds of domains; cap entry materialization.
LIMIT_FIELDS = ('cpu', 'pmem', 'io', 'nproc', 'iops', 'ep', 'vmem')
# Every limit field is stored in a signed 32-bit C struct member (T_INT:
# ls_cpu, ls_memory_phy, ... in pylve.c). A finite int outside this range
# passes int() but raises an uncaught OverflowError on struct assignment in
# lveapi.py during 'lvectl apply all', aborting the root batch. Bound to the
# T_INT range so any value the C layer can accept passes and only values that
# would OverflowError are rejected as malformed.
_INT32_MIN = -2 ** 31 # -2147483648
_INT32_MAX = 2 ** 31 - 1 # 2147483647
@dataclass
class DomainLimits:
cpu: Optional[int] = None
pmem: Optional[int] = None
io: Optional[int] = None
nproc: Optional[int] = None
iops: Optional[int] = None
ep: Optional[int] = None
vmem: Optional[int] = None
def to_dict(self):
"""Return only fields that are set (non-None)."""
return {k: v for k, v in self.__dict__.items() if v is not None}
@classmethod
def from_dict(cls, data):
if not data:
return cls()
if not isinstance(data, dict):
raise LvdError(f"limits must be an object, got {type(data).__name__}")
values = {}
for k, v in data.items():
if k not in LIMIT_FIELDS:
continue
try:
iv = int(v)
except (TypeError, ValueError, OverflowError) as e:
# OverflowError: int(float('inf')) — json.load accepts the
# bare token Infinity/-Infinity, so a tenant can smuggle a
# non-finite float into a limit.
raise LvdError(f"limit '{k}' must be an integer, got {v!r}") from e
# Bound the magnitude to the signed 32-bit T_INT range. A finite
# int beyond it (e.g. 9999999999, or 1e308 coerced to a 309-digit
# int) passes int() but raises an uncaught OverflowError when
# assigned into the C T_INT struct member downstream, aborting the
# root batch. Reject here -> LvdError -> _read returns safe default.
if not _INT32_MIN <= iv <= _INT32_MAX:
raise LvdError(
f"limit '{k}' out of range "
f"[{_INT32_MIN}, {_INT32_MAX}], got {iv}")
values[k] = iv
return cls(**values)
def update(self, **kwargs):
for k, v in kwargs.items():
if k in LIMIT_FIELDS and v is not None:
setattr(self, k, int(v))
def __bool__(self):
return any(v is not None for v in self.__dict__.values())
@dataclass
class DomainEntry:
name: str = ''
limits: DomainLimits = field(default_factory=DomainLimits)
def to_dict(self):
return {
'name': self.name,
'limits': self.limits.to_dict(),
}
@classmethod
def from_dict(cls, data):
if not isinstance(data, dict):
raise LvdError(f"domain entry must be an object, got {type(data).__name__}")
# 'name' is tenant-controlled and flows into hashable contexts
# downstream: docroot_by_domain.get(d.name) (lveapi.py) and
# find_domain/remove_domain string comparisons. A non-str (list/dict
# -> unhashable; int/float/bool -> mismatched key type) would raise an
# uncaught TypeError outside this module and abort the root
# 'lvectl apply all' batch for every other tenant. Reject it here so
# the only outcome of a malformed file is the safe default (via _read).
name = data.get('name', '')
if not isinstance(name, str):
raise LvdError(f"domain 'name' must be a string, got {type(name).__name__}")
return cls(
name=name,
limits=DomainLimits.from_dict(data.get('limits')),
)
@dataclass
class LvdConfig:
version: int = LVD_CONFIG_VERSION
domains: List[DomainEntry] = field(default_factory=list)
_lve_id: Optional[int] = field(default=None, repr=False, compare=False)
def find_domain(self, name):
"""Find domain entry by name. Returns None if not found."""
for d in self.domains:
if d.name == name:
return d
return None
def remove_domain(self, name):
"""Remove domain entry by name."""
self.domains = [d for d in self.domains if d.name != name]
def add_domain_by_docroot(self, docroot):
"""Resolve domain name from *docroot* and add it to the config.
"""
try:
domain_name = _resolve_domain_for_docroot(self._lve_id, docroot)
if domain_name and self.find_domain(domain_name) is None:
self.domains.append(DomainEntry(name=domain_name))
self.save()
except LvdError:
log.warning("add_domain_by_docroot: uid=%s docroot=%s",
self._lve_id, docroot, exc_info=True)
def remove_domain_by_docroot(self, docroot):
"""Resolve domain name from *docroot* and remove it from the config.
"""
try:
domain_name = _resolve_domain_for_docroot(self._lve_id, docroot)
if domain_name and self.find_domain(domain_name) is not None:
self.remove_domain(domain_name)
self.save()
except LvdError:
log.warning("remove_domain_by_docroot: uid=%s docroot=%s",
self._lve_id, docroot, exc_info=True)
def to_dict(self):
return {
'version': self.version,
'domains': [d.to_dict() for d in self.domains],
}
@classmethod
def from_dict(cls, data, lve_id=None):
if not isinstance(data, dict) or 'domains' not in data:
return cls(_lve_id=lve_id)
domains = data['domains']
if not isinstance(domains, list):
raise LvdError(f"'domains' must be a list, got {type(domains).__name__}")
if len(domains) > MAX_DOMAINS:
log.warning("domains.json for lve_id %s has %d domains; capping at %d",
lve_id, len(domains), MAX_DOMAINS)
domains = domains[:MAX_DOMAINS]
return cls(
version=data.get('version', LVD_CONFIG_VERSION),
domains=[DomainEntry.from_dict(d) for d in domains],
_lve_id=lve_id,
)
# --- Persistence ---
@classmethod
def load(cls, lve_id):
"""Load domains.json for a user (acquires flock for the read)."""
with cls._flock(lve_id):
return cls._read(lve_id)
def save(self):
"""Write domains.json for this user (acquires flock for the write)."""
if self._lve_id is None:
raise LvdError("cannot save: config has no associated lve_id")
with self._flock(self._lve_id):
self._write()
# --- Internal ---
@staticmethod
@contextlib.contextmanager
def _flock(lve_id):
"""Exclusive flock on a per-user sidecar file."""
with user_context(lve_id):
config_dir = ensure_config_dir(lve_id)
lock_path = os.path.join(config_dir, '.domains.lock')
fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
try:
deadline = time.monotonic() + _LOCK_TIMEOUT
while True:
try:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
break
except OSError as exc:
if exc.errno not in (errno.EAGAIN, errno.EWOULDBLOCK):
raise
if time.monotonic() >= deadline:
raise LvdError(
f"timed out after {_LOCK_TIMEOUT}s acquiring "
f"domain lock {lock_path}; another process holds it"
) from exc
time.sleep(_LOCK_RETRY_INTERVAL)
yield
finally:
fcntl.flock(fd, fcntl.LOCK_UN)
os.close(fd)
@classmethod
def _read(cls, lve_id):
path = config_path(lve_id)
if not os.path.exists(path):
return cls(_lve_id=lve_id)
try:
# domains.json is user-writable; open defensively so a symlink or
# special file (e.g. a FIFO) planted by the user cannot redirect the
# read or block the privileged apply/sync. O_NOFOLLOW rejects symlinks
# (ELOOP); O_NONBLOCK ensures open() does not block on a FIFO so the
# fstat regular-file check can reject it (O_NONBLOCK is a no-op for
# reads on a regular file).
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
try:
st = os.fstat(fd)
if not stat.S_ISREG(st.st_mode):
log.warning("ignoring non-regular config %s", path)
return cls(_lve_id=lve_id)
with os.fdopen(fd, 'rb') as f:
fd = -1 # fdopen now owns the descriptor
# st.st_size is a TOCTOU snapshot: the user can grow the file
# in place after fstat, and a read-to-EOF (json.load) would
# follow it. Bound the bytes actually consumed so the parse
# cannot be driven to OOM regardless of post-fstat growth.
raw = f.read(MAX_CONFIG_BYTES + 1)
if len(raw) > MAX_CONFIG_BYTES:
log.warning("ignoring oversized config %s (> %d bytes)",
path, MAX_CONFIG_BYTES)
return cls(_lve_id=lve_id)
data = json.loads(raw)
finally:
if fd >= 0:
os.close(fd)
except (ValueError, OSError, RecursionError) as e:
# ValueError subsumes json.JSONDecodeError and UnicodeDecodeError, so
# malformed JSON or non-UTF-8 bytes in the user-controlled file fail
# safe to an empty config instead of aborting the privileged apply.
log.warning("failed to read config %s: %s", path, e)
return cls(_lve_id=lve_id)
try:
return cls.from_dict(data, lve_id=lve_id)
except (LvdError, TypeError, ValueError, AttributeError, KeyError) as e:
log.warning("malformed config %s, ignoring: %s", path, e)
return cls(_lve_id=lve_id)
def _write(self):
config_dir = ensure_config_dir(self._lve_id)
path = config_path(self._lve_id)
content = json.dumps(self.to_dict(), indent=2) + '\n'
try:
_write_via_tmp(config_dir, path, content)
except IOError as e:
raise LvdError(f"failed to write config {path}: {e}") from e
def _clpwd():
return ClPwd(min_uid=0)
def get_homedir(uid):
"""Get home directory for a uid."""
try:
return _clpwd().get_pw_by_uid(uid)[0].pw_dir
except ClPwd.NoSuchUserException as exc:
raise LvdError(f"user with uid {uid} not found") from exc
def get_uid_by_username(username):
"""Resolve username to uid."""
try:
return _clpwd().get_uid(username)
except ClPwd.NoSuchUserException as exc:
raise LvdError(f"user '{username}' not found") from exc
def get_username(lve_id):
"""Resolve username from uid."""
try:
return _clpwd().get_names(lve_id)[0]
except (ClPwd.NoSuchUserException, IndexError) as exc:
raise LvdError(f"no user found for uid {lve_id}") from exc
def resolve_lve_id(lve_id=None, username=None):
"""
Resolve lve_id from either --lve-id or --username.
If neither, use effective UID (end-user mode).
"""
if lve_id is not None:
return int(lve_id)
if username is not None:
return get_uid_by_username(username)
uid = os.geteuid()
if uid == 0:
raise LvdError("root must specify --lve-id or --username")
return uid
def config_path(lve_id):
"""Return path to the user's domains.json config."""
homedir = get_homedir(lve_id)
return os.path.join(homedir, LVD_CONFIG_DIR, LVD_CONFIG_FILE)
def ensure_config_dir(lve_id):
"""Create ~/.lve/ directory with proper permissions if it doesn't exist.
Caller must ensure privileges are already dropped to the target user."""
homedir = get_homedir(lve_id)
config_dir = os.path.join(homedir, LVD_CONFIG_DIR)
if not os.path.isdir(config_dir):
os.makedirs(config_dir, mode=0o700, exist_ok=True)
return config_dir
def _write_via_tmp(directory, filename, content):
"""Write content to a file atomically via a temporary file."""
temp_path = None
try:
with tempfile.NamedTemporaryFile('w', dir=directory, delete=False) as tmp:
temp_path = tmp.name
tmp.write(content)
tmp.flush()
os.fsync(tmp.fileno())
os.replace(temp_path, filename)
finally:
if temp_path and os.path.exists(temp_path):
try:
os.remove(temp_path)
except OSError:
pass
@contextlib.contextmanager
def user_context(lve_id):
"""Drop privileges to the target user if running as root.
CLOS-4370: when invoked as root we pair drop_privileges with
secureio.disable_quota — creating ~/.lve/ and the lockfile / writing
domains.json must succeed even when the user is over disk quota,
otherwise OSError(EDQUOT) inside this block aborts batch operations
like `lvectl apply all` and leaves the rest of the users with stale
limits. disable_quota raises CAP_SYS_RESOURCE in the effective set
"""
if os.geteuid() == 0:
username = get_username(lve_id)
with drop_privileges(username), disable_quota():
yield
else:
yield
def find_all_lve_ids_with_config():
"""Return list of LVE IDs (UIDs) that have domain isolation configured.
Scans the id_registry directory for per-user registry files.
Users only appear here after ``lvdctl set`` or ``lvdctl apply``
has been called at least once for one of their domains.
"""
ids_dir = id_registry.LVD_IDS_DIR
if not os.path.isdir(ids_dir):
return []
result = []
for name in os.listdir(ids_dir):
try:
result.append(int(name))
except ValueError:
continue
return result
def load_config(lve_id):
"""Load the domain isolation config for a given user."""
return LvdConfig.load(lve_id)
def resolve_docroot(domain_name):
"""Resolve domain name to document root path via panel API."""
try:
return cpapi_docroot(domain_name)[0]
except Exception as exc:
raise LvdError(f"failed to resolve docroot for domain '{domain_name}'") from exc
def _resolve_domain_for_docroot(uid, docroot):
"""Return domain name whose document root matches *docroot*, or None."""
try:
username = get_username(uid)
for domain_name, dr in (userdomains(username) or []):
if dr == docroot:
return domain_name
except (LvdError, CPAPIException, OSError):
log.warning("Cannot resolve domain for docroot: uid=%s docroot=%s", uid, docroot, exc_info=True)
return None