ꦫꦣꦺꦤ꧀ꦄꦤ꧀ꦠꦱꦺꦤ

PATH: opt / cloudlinux / venv / lib / python3.11 / site-packages / xray / reconfiguration

# -*- 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 X-Ray INI files.

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

import logging
import os
import pwd
from glob import iglob
from typing import Optional

from secureio import disable_quota
from clcommon.cpapi import docroot as get_docroot

from xray.internal.utils import user_context, cagefsctl_get_prefix
from .xray_ini import (
    is_excluded_path,
    get_domain_php_version_from_selector,
    INI_USER_LOCATIONS,
)

logger = logging.getLogger(__name__)

# Try to import website isolation check from securelve (cagefs)
# This is optional - if securelve is not installed, we assume website isolation is not available
try:
    from clcagefslib.domain import (
        is_website_isolation_allowed_server_wide,
        is_isolation_enabled,
        get_websites_with_enabled_isolation,
    )
    from clcagefslib.webisolation.jail_utils import get_website_id
except ImportError:
    def is_website_isolation_allowed_server_wide():
        return False

def is_isolation_enabled(user):
        return False

def get_websites_with_enabled_isolation(user):
        return []

get_website_id = None

def is_per_website_php_selector(user: str, domain: str):
    if not is_website_isolation_allowed_server_wide():
        return False
    return domain in get_websites_with_enabled_isolation(user)

def _get_per_website_ini_path(user: str, website_id: str, php_ver_dir: str) -> Optional[str]:
    """
    Build path to xray.ini in per-website directory.
    :param user: Username
    :param website_id: Website ID hash
    :param php_ver_dir: PHP version directory (e.g., 'alt-php80')
    :return: Full path to xray.ini or None if cagefs prefix not available
    """
    prefix = cagefsctl_get_prefix(user)
    if prefix is None:
        return None
    return f'/var/cagefs/{prefix}/{user}/etc/cl.php.d/{website_id}/{php_ver_dir}/xray.ini'

def regenerate_ini_for_website_isolation(user: str, domain: str) -> None:
    """
    Copy xray.ini files from base user locations to per-website directories.

This function is called by cagefsctl when enabling website isolation for a user.
    If xray.ini exists in base location (meaning user has active tasks), copy it
    to per-website directories. Overwrites existing files for consistency.

Each domain may have a different PHP version set via cloudlinux-selector,
    so we determine the domain's actual PHP version from cl.selector symlinks.

:param user: Username to regenerate ini files for
    """
    if not is_per_website_php_selector(user, domain):
        return

# Collect existing base ini files for this user: {php_ver_dir: content}
    base_ini_files = {}
    uid = None
    gid = None
    for location in INI_USER_LOCATIONS:
        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
                uid = pw_record.pw_uid
                gid = pw_record.pw_gid
            except:
                logger.debug("Cannot get pw_record for path: %s", dir_path)
                continue

ini_file = os.path.join(dir_path, 'xray.ini')
            if not os.path.exists(ini_file):
                continue

try:
                with open(ini_file) as f:
                    php_ver_dir = os.path.basename(dir_path)
                    base_ini_files[php_ver_dir] = f.read()
            except OSError as e:
                logger.error("Cannot read xray.ini for path: %s, error=%s", dir_path, str(e))
                continue

if not base_ini_files or uid is None:
        return

docroot_result = get_docroot(domain)
    document_root = docroot_result[0]
    website_id = get_website_id(document_root)

# Get domain's actual PHP version from cl.selector symlinks
    domain_php_ver = get_domain_php_version_from_selector(user, website_id)

content = base_ini_files.get(domain_php_ver)
    if not content:
        # Fallback: use any available base ini content for domain-specific version
        content = next(iter(base_ini_files.values()), None)

ini_path = _get_per_website_ini_path(user, website_id, domain_php_ver)

# Ensure directory exists: it is configured normally once enabling per domain php version
    ini_dir = os.path.dirname(ini_path)
    if not os.path.exists(ini_dir):
        logger.info(f"Per-website ini directory does not exist: {ini_dir}")
        return
    try:
        with user_context(uid, gid), disable_quota(), open(ini_path, 'w') as f:
            f.write(content)
        logger.debug('Created %s for domain %s', ini_path, domain)
    except Exception as e:
        logger.error('Failed to create %s: %s', ini_path, e)

def _generate_ini_with_counter(
        existing_contents: Optional[list],
        counter: int,
        php_version: str = None
) -> str:
    """
    Generate xray.ini content with a specific task counter value.

:param existing_contents: Existing ini file lines or None for new file
    :param counter: Task counter value to set
    :param php_version: PHP version for extension path (used only for new files)
    :return: Generated ini file content
    """
    # Determine extension path based on PHP version
    # Short 2-digit versions get full path, others use generic xray.so
    if php_version is None or len(php_version) > 2:
        so_path = "xray.so"
    else:
        so_path = f"/opt/alt/php{php_version}/usr/lib64/php/modules/xray.so"

if existing_contents is None:
        return f"""extension={so_path}
;xray.tasks={counter}\n"""

def update_line():
        for line in existing_contents:
            if "xray.tasks" in line:
                yield f";xray.tasks={counter}\n"
            else:
                yield line + "\n"

return "".join(list(update_line()))

def update_website_isolation_ini(
        user: str,
        uid: int,
        gid: int,
        domain: str,
        domain_task_count: int,
        existing_contents: Optional[list] = None,
        php_version: str = None
) -> None:
    """
    Update xray.ini file in per-website directory for a SPECIFIC domain.

This function is called when a tracing task is added/updated for a domain.
    The ini file is placed in the directory matching the domain's PHP version
    as configured via cloudlinux-selector.

Generates the ini content with domain-specific task counter, which may differ
    from the per-user counter when a user has tasks for multiple domains.

:param user: Username
    :param uid: User ID for file ownership
    :param gid: Group ID for file ownership
    :param domain: Domain name (e.g., 'example.com') - REQUIRED
    :param domain_task_count: Number of tasks for this specific domain
    :param existing_contents: Existing per-user ini file lines (to preserve settings)
    :param php_version: PHP version for extension path (used only for new files)
    """
    if not is_per_website_php_selector(user, domain):
        return

# Get website_id from domain's docroot
    try:
        docroot_result = get_docroot(domain)
        if not docroot_result:
            logger.debug('Failed to get docroot for domain %s', domain)
            return
        document_root = docroot_result[0]
        website_id = get_website_id(document_root)
    except Exception as e:
        logger.error('Failed to get website_id for domain %s: %s', domain, e)
        return

if not website_id:
        return

# Get domain's actual PHP version from cl.selector symlinks
    domain_php_ver = get_domain_php_version_from_selector(user, website_id)
    if not domain_php_ver:
        logger.debug('No specific PHP version set for domain %s, skipping', domain)
        return

ini_path = _get_per_website_ini_path(user, website_id, domain_php_ver)
    if not ini_path:
        return

# Ensure directory exists
    ini_dir = os.path.dirname(ini_path)
    if not os.path.isdir(ini_dir):
        logger.debug('Per-website ini directory does not exist: %s', ini_dir)
        return

# Generate ini content with domain-specific task counter
    content = _generate_ini_with_counter(existing_contents, domain_task_count, php_version)

try:
        with user_context(uid, gid), disable_quota(), open(ini_path, 'w') as f:
            f.write(content)
        logger.debug(
            'Updated %s for domain %s (PHP %s, tasks=%d)',
            ini_path, domain, domain_php_ver, domain_task_count
            )
    except OSError as e:
        logger.error('Failed to update %s: %s', ini_path, e)

def remove_website_isolation_ini(user: str, domain: str) -> None:
    """
    Remove xray.ini file from per-website directory for a SPECIFIC domain.

This function is called when the last tracing task for a domain is removed.
    The ini file is removed from the directory matching the domain's PHP version
    as configured via cloudlinux-selector.

:param user: Username
    :param domain: Domain name (e.g., 'example.com') - REQUIRED
    """
    if not is_per_website_php_selector(user, domain):
        return

if not website_id:
        return

ini_path = _get_per_website_ini_path(user, website_id, domain_php_ver)
    if not ini_path:
        return

if not os.path.exists(ini_path):
        return

try:
        pw_record = pwd.getpwnam(user)
        with user_context(pw_record.pw_uid, pw_record.pw_gid):
            os.unlink(ini_path)
        logger.debug('Removed %s for domain %s (PHP %s)', ini_path, domain, domain_php_ver)
    except Exception as e:
        logger.error('Failed to remove %s: %s', ini_path, e)

[+] ..
[+] __pycache__
[-] system_id_shift.py [open]
[-] website_isolation.py [open]
[-] global_ini.py [open]
[-] xray_ini.py [open]
[-] __init__.py [open]
[-] migrate.py [open]