OwlCyberSecurity - MANAGER

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 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 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:
                logger.debug("Cannot get pw_record for path: %s", dir_path)
                continue

ini_file = os.path.join(dir_path, INI_FILE_NAME)
            if not os.path.exists(ini_file):
                continue

try:
                with open(ini_file) as f:
                    php_ver = extract_php_version(dir_path)
                    if php_ver:
                        base_ini_files[(pw_record.pw_name, php_ver)] = (
                            f.read(), pw_record.pw_uid, pw_record.pw_gid
                        )
            except Exception:
                logger.warning('Failed to read %s', ini_file)
                continue

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:
                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:
                with user_context_func(uid, gid), disable_quota():
                    with open(ini_file, 'w') as f:
                        f.write(content)
                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:
                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:
                logger.debug("Cannot get pw_record for path: %s", dir_path)
                continue

ini_file = os.path.join(dir_path, INI_FILE_NAME)
            if not os.path.exists(ini_file):
                continue

try:
                with open(ini_file) as f:
                    php_ver = extract_php_version(dir_path)
                    if php_ver:
                        base_ini_files[php_ver] = (f.read(), pw_record.pw_uid, pw_record.pw_gid)
            except Exception:
                logger.warning('Failed to create %s', ini_file)
                continue

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:
                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:
                with user_context_func(uid, gid), disable_quota():
                    with open(ini_file, 'w') as f:
                        f.write(content)
                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)