Edit

[EDIT] FILE: website_isolation.py

# -*- coding: utf-8 -*-

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

"""
Website isolation support for SSA (clos_ssa.ini) files.

This module provides functions to manage clos_ssa.ini files in per-website
directories when CageFS website isolation is enabled.
"""

import logging
import os
import stat
import subprocess
from glob import iglob

from secureio import disable_quota

from .clos_ssa_ini import (
    INI_FILE_NAME,
    INI_USER_LOCATIONS_BASE,
    INI_USER_LOCATIONS_WEBSITE_ISOLATION,
    is_excluded_path,
    extract_php_version,
)

# Try to import website isolation check from securelve (cagefs)
try:
    from clcagefslib.domain import is_website_isolation_allowed_server_wide, is_isolation_enabled
except ImportError:

def is_website_isolation_allowed_server_wide():
        return False

def is_isolation_enabled(user):
        return False

logger = logging.getLogger(__name__)

def _write_isolation_ini(ini_file, content, uid, gid, user_context_func) -> None:
    """
    Write content into a per-website clos_ssa.ini under the tenant context.

O_NOFOLLOW refuses a tenant-planted symlink at the ini path (raises OSError
    ELOOP on a final-component symlink). O_NONBLOCK stops a tenant-planted FIFO
    from blocking the shared regen thread forever on open() (a reader-less
    O_WRONLY FIFO open fails ENXIO instead of hanging). The fstat check then
    refuses any non-regular target (FIFO/device/socket) by raising OSError, so
    it is skipped rather than written (and, like the O_NOFOLLOW/O_NONBLOCK
    refusals, is never counted by callers as a created ini). O_NONBLOCK has no
    effect on regular-file writes. Callers handle the raised OSError per-target.
    """
    with user_context_func(uid, gid), disable_quota():
        fd = os.open(ini_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW | os.O_NONBLOCK, 0o644)
        with os.fdopen(fd, 'w') as f:
            if not stat.S_ISREG(os.fstat(f.fileno()).st_mode):
                # Refuse a non-regular target (a FIFO with a reader, a device or a
                # socket) that opened OK. RAISE rather than return so callers treat
                # it exactly like the O_NOFOLLOW/O_NONBLOCK refusals: the file is not
                # counted as a created ini and no spurious cagefsctl regenerate runs.
                raise OSError('refusing to write non-regular ini target (not a regular file)')
            f.write(content)

def _read_isolation_ini(ini_file, uid, gid, user_context_func):
    """
    Read a base clos_ssa.ini under the tenant context, refusing non-regular sources.

The base ini path is tenant-controlled, so the same hardening as the write
    helper applies: O_NOFOLLOW refuses a final-component symlink (ELOOP), and
    O_NONBLOCK makes an O_RDONLY open of a tenant-planted FIFO return immediately
    instead of blocking the shared regen thread forever waiting for a writer. The
    fstat check on the opened fd then refuses any non-regular source (FIFO/device/
    socket): it is skipped rather than read. Returns the file content, or None if
    the source is non-regular. Callers handle the raised OSError per-source.
    """
    with user_context_func(uid, gid):
        fd = os.open(ini_file, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
        with os.fdopen(fd) as f:
            if not stat.S_ISREG(os.fstat(f.fileno()).st_mode):
                logger.warning('Refusing to read non-regular ini %s', ini_file)
                return None
            return f.read()

def copy_inis_to_website_isolation_paths(user_context_func) -> None:
    """
    Copy clos_ssa.ini files from base user paths to per-website directories.

:param user_context_func: Context manager function for user permissions
    """
    if not is_website_isolation_allowed_server_wide():
        return

# Collect all base ini files: {(user, php_ver): (content, uid, gid)}
    base_ini_files = {}
    for location in INI_USER_LOCATIONS_BASE:
        for dir_path in iglob(location['path']):
            if is_excluded_path(dir_path):
                continue
            try:
                pw_record = location['user'](dir_path)
            except Exception:
                logger.debug("Cannot get pw_record for path: %s", dir_path)
                continue

ini_file = os.path.join(dir_path, INI_FILE_NAME)
            try:
                content = _read_isolation_ini(ini_file, pw_record.pw_uid, pw_record.pw_gid, user_context_func)
            except FileNotFoundError:
                continue
            except Exception:
                logger.warning('Failed to read %s', ini_file)
                continue
            if content is None:
                continue
            php_ver = extract_php_version(dir_path)
            if php_ver:
                base_ini_files[(pw_record.pw_name, php_ver)] = (
                    content,
                    pw_record.pw_uid,
                    pw_record.pw_gid,
                )

if not base_ini_files:
        return

created_ini = set()

# Copy to per-website directories
    for location in INI_USER_LOCATIONS_WEBSITE_ISOLATION:
        for dir_path in iglob(location['path']):
            if is_excluded_path(dir_path):
                continue
            try:
                pw_record = location['user'](dir_path)
            except Exception:
                logger.debug("Cannot get pw_record for path: %s", dir_path)
                continue
            if not is_isolation_enabled(pw_record.pw_name):
                continue

php_ver = extract_php_version(dir_path)
            if not php_ver:
                continue

key = (pw_record.pw_name, php_ver)
            if key not in base_ini_files:
                continue

content, uid, gid = base_ini_files[key]
            ini_file = os.path.join(dir_path, INI_FILE_NAME)
            if not os.path.exists(os.path.dirname(ini_file)):
                continue
            try:
                _write_isolation_ini(ini_file, content, uid, gid, user_context_func)
                created_ini.add(pw_record.pw_name)
            except Exception as e:
                logger.warning('Failed to create %s: %s', ini_file, str(e))
                continue
    for username in created_ini:
        _regenerate_user_website_isolation(username)

def remove_inis_from_website_isolation_paths(user_context_func) -> None:
    """
    Remove clos_ssa.ini files from all per-website directories.

:param user_context_func: Context manager function for user permissions
    """
    if not is_website_isolation_allowed_server_wide():
        return

removed_ini = set()

for location in INI_USER_LOCATIONS_WEBSITE_ISOLATION:
        for dir_path in iglob(location['path']):
            if is_excluded_path(dir_path):
                continue
            try:
                pw_record = location['user'](dir_path)
            except Exception:
                continue

ini_file = os.path.join(dir_path, INI_FILE_NAME)
            if os.path.exists(ini_file):
                try:
                    with user_context_func(pw_record.pw_uid, pw_record.pw_gid):
                        os.unlink(ini_file)
                    removed_ini.add(pw_record.pw_name)
                except Exception as e:
                    logger.warning('Failed to remove %s: %s', ini_file, str(e))
                    continue
        for username in removed_ini:
            _regenerate_user_website_isolation(username)

def _regenerate_user_website_isolation(user: str) -> None:
    """
    Needed to terminate php processes to immediately apply clos_ssa.ini creation/deletion.
    """
    try:
        subprocess.run(
            ["/usr/sbin/cagefsctl", "--site-isolation-regenerate", user], capture_output=True, check=True, text=True
        )
    except subprocess.CalledProcessError as e:
        logger.warning("Failed to trigger cagefsctl site isolation regeneration for %s: %s", user, e.stdout)

def regenerate_inis_for_user(user: str, user_context_func) -> None:
    """
    Regenerate clos_ssa.ini files for a specific user's website isolation directories.

This is called by cagefsctl when enabling website isolation for a user.
    Only creates per-website ini files if base per-user ini exists.

:param user: Username to regenerate ini files for
    :param user_context_func: Context manager function for user permissions
    """
    if not is_website_isolation_allowed_server_wide():
        return

logger.info('Regenerating clos_ssa.ini for user %s website isolation...', user)

# First, collect existing base ini files for this user: {php_ver: (content, uid, gid)}
    base_ini_files = {}
    for location in INI_USER_LOCATIONS_BASE:
        for dir_path in iglob(location['path']):
            if is_excluded_path(dir_path):
                continue
            try:
                pw_record = location['user'](dir_path)
                if pw_record.pw_name != user:
                    continue
            except Exception:
                logger.debug("Cannot get pw_record for path: %s", dir_path)
                continue

ini_file = os.path.join(dir_path, INI_FILE_NAME)
            try:
                content = _read_isolation_ini(ini_file, pw_record.pw_uid, pw_record.pw_gid, user_context_func)
            except FileNotFoundError:
                continue
            except Exception:
                logger.warning('Failed to read %s', ini_file)
                continue
            if content is None:
                continue
            php_ver = extract_php_version(dir_path)
            if php_ver:
                base_ini_files[php_ver] = (content, pw_record.pw_uid, pw_record.pw_gid)

if not base_ini_files:
        logger.info('No base clos_ssa.ini files found for user %s', user)
        return

# Copy to per-website directories
    for location in INI_USER_LOCATIONS_WEBSITE_ISOLATION:
        for dir_path in iglob(location['path']):
            if is_excluded_path(dir_path):
                continue
            try:
                pw_record = location['user'](dir_path)
                if pw_record.pw_name != user:
                    continue
            except Exception:
                logger.debug("Cannot get pw_record for path: %s", dir_path)
                continue

php_ver = extract_php_version(dir_path)
            if not php_ver:
                continue

if php_ver not in base_ini_files:
                continue

content, uid, gid = base_ini_files[php_ver]
            ini_file = os.path.join(dir_path, INI_FILE_NAME)
            if not os.path.exists(os.path.dirname(ini_file)):
                continue
            try:
                _write_isolation_ini(ini_file, content, uid, gid, user_context_func)
                logger.info('Created %s', ini_file)
            except Exception as e:
                logger.warning('Failed to create %s: %s', ini_file, str(e))
                continue

logger.info('Finished regenerating for user %s!', user)

CANCEL