23s2s

✏️ Editing: custom.py

# 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

"""
This module contains classes implementing X-Ray Manager behaviour
for custom integration
"""

import os
from collections import ChainMap
from typing import ClassVar

from clcommon.cpapi.plugins.vendors import PublicApi

from xray import gettext as _
from xray.internal import phpinfo_utils

from ..internal.constants import is_allowed_ini_path
from ..internal.exceptions import XRayManagerError, XRayMissingDomain
from ..internal.types import DomainInfo
from ..internal.user_plugin_utils import user_mode_verification, with_fpm_reload_restricted
from .base import BaseManager

class CustomManager(BaseManager):
    """
    Manager supporting integration scripts
    """

VERSIONS_CUSTOM: ClassVar[dict] = {
        '54': None,
        '55': None,
        '56': None,
        '70': None,
        '71': None,
        '72': None,
        '73': None,
        '74': None,
        '80': None,
        '81': None,
        '82': None,
        '83': None,
        '84': None,
        '85': None,
    }

def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

self.api = PublicApi()
        if self.is_xray_integrated:
            self.all_domains = self.get_all_domains()
        else:
            raise XRayManagerError(_('X-Ray is not supported by control panel vendor'))

def supported_versions(self) -> ChainMap:
        """
        Get supported PHP versions
        :return: dict with custom supported versions
        """
        return ChainMap(self.VERSIONS, self.VERSIONS_CUSTOM)

@property
    def is_xray_integrated(self):
        """
        Check the X-Ray feature status through the panel_info script
        """
        _info = self.api.panel_info()  # PanelInfo instance
        features = _info.supported_cl_features
        if features is None:
            return True

return features.get('xray', False)

def get_all_domains(self) -> dict:
        """
        Collect domains from integration script
        """
        return self.api.domains(with_php=True)

@with_fpm_reload_restricted
    @user_mode_verification
    def get_domain_info(self, domain_name: str) -> DomainInfo:
        """
        Retrieve PHP setting for given domain_name
        """
        try:
            domain_conf = self.all_domains[domain_name]  # DomainData instance
        except KeyError:
            self.logger.warning('Domain does not exist on the server', extra={'domain_name': domain_name})
            raise XRayMissingDomain(domain_name)

if self.phpinfo_mode:
            config = phpinfo_utils.get_php_configuration(
                domain_conf.owner,
                domain=domain_name,
                # Trusted (server-side) PHP version for this domain. The custom
                # vendor API reports a BARE number ('80') that carries no distro
                # family, so the global per-distro+version INI_LOCATION cannot be
                # uniquely derived: the global accept branch (3) default-DENIES
                # for custom (own-CageFS / own per-domain ini dirs still accepted).
                # The probe value is attacker-controlled and must not be an anchor.
                trusted_php_version=domain_conf.php.version,
            )

return DomainInfo(
                name=domain_name,
                panel_php_version=config.get_full_php_version(''),
                php_ini_scan_dir=config.absolute_ini_scan_dir,
                # indicates that there is no need to apply selector
                # and try to resolve php version, the one given in
                # php_version is final one
                is_selector_applied=True,
                user=domain_conf.owner,
                panel_fpm=config.is_php_fpm,
            )
        else:
            ini_path = self._validate_ini_path(domain_conf.php.ini_path, domain_conf.owner)
            domain_info = DomainInfo(
                name=domain_name,
                user=domain_conf.owner,
                panel_php_version=domain_conf.php.version,
                panel_fpm=domain_conf.php.fpm,
                is_native=domain_conf.php.is_native,
                ini_path=ini_path,
            )

self.logger.info(
            'Retrieved domain info: domain %s owned by %s uses php version %s',
            domain_name,
            domain_info.user,
            domain_info.panel_php_version,
        )
        return domain_info

def panel_specific_selector_enabled(self, domain_info: DomainInfo) -> bool:
        """
        Check if selector is enabled specifically for custom panels
        Required to be implemented by child classes
        :param domain_info: a DomainInfo object
        :return: True if yes, False otherwise
        """
        return domain_info.is_native and not domain_info.panel_fpm

def fpm_service_name(self, dom_info: DomainInfo) -> str | None:
        """
        Retrieve FPM service name
        """
        return dom_info.panel_fpm

def _validate_ini_path(self, ini_path: str, owner: str) -> str:
        """Owner-bind the vendor-supplied ini_path before it steers a root write.

In non-phpinfo mode the custom integration reports ini_path verbatim and
        the only downstream guard is the broad is_allowed_ini_path prefix
        allowlist, which admits serverwide (/etc/php.d) and per-tenant
        (/var/cagefs/..., the per-domain ini roots) trees. Without owner-binding an
        allowlisted-but-foreign path would let the root xray.ini write land in
        another tenant's jail. Accept the path only when it is the owner's own
        subtree or an operator-controlled root-owned INI directory; reject a path
        that resolves outside the allowlist, into a foreign tenant's per-user
        tree, or onto a dir owned by a different unprivileged user.

Returns the CANONICAL (realpath'd) directory, not the raw vendor string.
        The write-time owner binding (Task._ini_dir_validator / unified_write)
        classifies ini_location as per-tenant vs global from this value and then
        re-checks the PINNED inode against it, so it must be canonical here — a
        raw symlink path would misclassify and either over-reject a legitimate
        own-subtree symlink or fail to bind a per-tenant location.
        """
        if not ini_path:
            return ini_path
        resolved = os.path.realpath(ini_path)
        if not is_allowed_ini_path(resolved):
            self.logger.warning('Rejected custom ini_path outside allowlist: %s', ini_path)
            raise XRayManagerError(_('ini_location outside allowed paths: %s') % ini_path)
        # The owner's OWN CageFS subtree or per-domain ini dir
        # (<root>/<owner>/... for any root in phpinfo_utils._PER_DOMAIN_INI_ROOTS).
        if phpinfo_utils._is_own_cagefs_subtree(owner, resolved) or phpinfo_utils._is_own_per_domain_ini_dir(
            owner, resolved
        ):
            return resolved
        # Per-tenant roots are legitimate only as the owner's own subtree
        # (accepted above); any other /var/cagefs or per-domain-panel-root target
        # is a foreign tenant's jail.
        if phpinfo_utils._is_within(
            phpinfo_utils._CAGEFS_ROOT, resolved
        ) or phpinfo_utils._is_under_per_domain_ini_root(resolved):
            self.logger.warning('Rejected custom ini_path bound to a foreign tenant: %s', ini_path)
            raise XRayManagerError(_('ini_location outside allowed paths: %s') % ini_path)
        # A global, operator-controlled allowlist dir (/opt/alt/phpNN,
        # /usr/local/phpNN, /etc/php.d, ...). If present, require root ownership;
        # a dir owned by an unprivileged user here is unexpected. If absent (the
        # handler is not installed on this server) it is not a foreign tenant's
        # existing subtree, so accept it under the allowlist bound.
        try:
            st_uid = os.stat(resolved).st_uid
        except OSError:
            return resolved
        if st_uid == 0:
            return resolved
        self.logger.warning('Rejected custom ini_path owned by uid %s (not root): %s', st_uid, ini_path)
        raise XRayManagerError(_('ini_location outside allowed paths: %s') % ini_path)

def _ini_path(self, domain_info: DomainInfo) -> str:
        """
        Path to additional .ini files specific custom panel getter
        """
        return domain_info.ini_path

def get_ini_path(self, domain_info: DomainInfo) -> str:
        """
        Resolve a path to directory for additional ini file.
        It depends on version set for domain and on selector
        NOTE:
        This method is overrided to manage php.d.location=selector resolving.
        In custom integration we do not know if PHP version is alt or not,
        it is set as just two digits.
        Thus, we only could rely on resolved path -- if it is '/opt/alt'.
        :param domain_info: a DomainInfo object
        :return: path to directory for ini files
        """
        if domain_info.php_ini_scan_dir:
            return domain_info.php_ini_scan_dir

# here follows the hack to resolve php.d.location=selector
        # for custom integration
        ini_path = super().get_ini_path(domain_info)
        if ini_path.startswith('/opt/alt') and not domain_info.panel_php_version.startswith('alt-php'):
            saved_panel_php = domain_info.panel_php_version
            domain_info.panel_php_version = f'alt-php{domain_info.panel_php_version}'
            try:
                ini_path = domain_info.phpd_location_ini_path or ini_path
            except ValueError:
                # failed to resolve CageFS prefix for user
                pass
            domain_info.panel_php_version = saved_panel_php
            self.logger.info('Ini path re-resolved as %s', ini_path)
        return ini_path

Sid Gifari From Gifari Industries - BD Cyber Security Team