File Manager

✏️ ctadmin.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

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

import os
import re
import subprocess
import urllib.parse
from collections import ChainMap
from glob import glob

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

from .base import BaseManager
from ..internal.exceptions import XRayManagerError, XRayMissingDomain, XRayManagerExit
from ..internal.types import DomainInfo
from ..internal.user_plugin_utils import (
    user_mode_verification,
    with_fpm_reload_restricted
)

class DirectAdminManager(BaseManager):
    """
    Class implementing an X-Ray manager behaviour for DirectAdmin
    """

da_options_conf = '/usr/local/directadmin/custombuild/options.conf'
    da_domain_pattern = '/usr/local/directadmin/data/users/*/domains/*.conf'
    da_subdomain_pattern = '/usr/local/directadmin/data/users/*/domains/*.subdomains'
    da_alias_pattern = '/usr/local/directadmin/data/users/*/domains/*.pointers'
    da_docroot_override_pattern = '/usr/local/directadmin/data/users/*/domains/*.subdomains.docroot.override'

VERSIONS_DA = {
        'php54': '/usr/local/php54/lib/php.conf.d',
        'php55': '/usr/local/php55/lib/php.conf.d',
        'php56': '/usr/local/php56/lib/php.conf.d',
        'php70': '/usr/local/php70/lib/php.conf.d',
        'php71': '/usr/local/php71/lib/php.conf.d',
        'php72': '/usr/local/php72/lib/php.conf.d',
        'php73': '/usr/local/php73/lib/php.conf.d',
        'php74': '/usr/local/php74/lib/php.conf.d',
        'php80': '/usr/local/php80/lib/php.conf.d',
        'php81': '/usr/local/php81/lib/php.conf.d',
        'php82': '/usr/local/php82/lib/php.conf.d',
        'php83': '/usr/local/php83/lib/php.conf.d',
        'php84': '/usr/local/php84/lib/php.conf.d',
        'php85': '/usr/local/php85/lib/php.conf.d',
    }

def supported_versions(self) -> ChainMap:
        """
        Get supported PHP versions
        :return: a chained map with basic supported versions
                and DirectAdmin supported versions
        """
        return ChainMap(self.VERSIONS,
                        self.VERSIONS_DA)

def file_readlines(self, filename: str) -> list:
        """
        Read lines from file
        :param filename: a name of file to read
        :return: list of stripped lines
        """

def get_file_encoding():
            """
            Retrieve file encoding
            """
            with open(filename, 'rb') as f:
                result = chardet.detect(f.read())
            return result['encoding']

try:
            with open(filename, encoding=get_file_encoding()) as f:
                return [line.strip() for line in f.readlines()]
        except OSError as e:
            self.logger.error('Failed to read [DA conf] file',
                              extra={'fname': filename,
                                     'err': str(e)})
            raise XRayManagerExit(_('Failed to read file %s') % filename) from e

@property
    def php_options(self) -> dict:
        """
        Retrieve DirectAdmin PHP settings
        :return: dict of format {'1': {ver, fpm}, '2': {ver, fpm}...}
                where '1', '2' etc is an ordinal number of a handler as
                it is defined in options.conf
        """
        parsed_options = dict()
        opts = self.file_readlines(self.da_options_conf)

def inner_filter(seq, marker):
            """
            Filter PHP release|mode items in seq by marker
            :param seq: initial sequence
            :param marker: should be contained in seq item
            :return: all items from seq containing marker
            """
            return [l for l in seq if
                    marker in l and 'php' in l and not l.startswith('#')]

for index, o in enumerate(zip(inner_filter(opts, 'release'),
                                      inner_filter(opts, 'mode')),
                                  start=1):
            release, mode = o
            if 'no' not in release:
                parsed_options[str(index)] = {
                    'ver': f"php{''.join(release.split('=')[-1].split('.'))}",
                    'fpm': 'fpm' in mode,
                    'handler': mode.split('=')[-1]
                }
        return parsed_options

@property
    def main_domains(self) -> dict:
        """
        Retrieve main domains configuration files
        """
        domains = dict()
        for dom_conf in glob(self.da_domain_pattern):
            name = os.path.basename(dom_conf).split('.conf')[0]
            domains[name] = dom_conf
        return domains

@property
    def subdomains(self) -> dict:
        """
        Retrieve subdomains configuration files
        """
        subdomains = dict()
        for sub_conf in glob(self.da_subdomain_pattern):
            for subdom in self.file_readlines(sub_conf):
                sub_parent = f"{os.path.basename(sub_conf).split('.subdomains')[0]}"
                sub_name = f"{subdom}.{sub_parent}"
                subdomains[
                    sub_name] = f"{sub_conf.split('.subdomains')[0]}.conf"
        return subdomains

@property
    def aliases(self) -> dict:
        """
        Retrieve aliases configuration files
        """
        aliases = dict()
        for alias_conf in glob(self.da_alias_pattern):
            parent_domain_name = alias_conf.split('.pointers')[0]
            for alias in self.file_readlines(alias_conf):
                alias_info = alias.split('=')
                alias_name = alias_info[0]
                _type = alias_info[-1]
                if _type == 'pointer':
                    # pointers are not considered as domains,
                    # because they just perform a redirect to parent domain
                    continue
                aliases[alias_name] = f"{parent_domain_name}.conf"
                try:
                    for sub in self.file_readlines(
                            f"{parent_domain_name}.subdomains"):
                        aliases[
                            f"{sub}.{alias_name}"] = f"{parent_domain_name}.conf"
                except XRayManagerError:
                    # there could be no subdomains
                    pass
        return aliases

@property
    def subdomains_php_settings(self) -> dict:
        """
        Retrieve subdomains_docroot_override configuration files
        """
        sub_php_set = dict()
        for sub_doc_override in glob(self.da_docroot_override_pattern):
            for subdomline in self.file_readlines(sub_doc_override):
                subdompart, data = urllib.parse.unquote(
                    subdomline).split('=', maxsplit=1)
                php_select_value = re.search(r'(?<=php1_select=)\d(?=&)',
                                             data)
                if php_select_value is not None:
                    domname = f"{os.path.basename(sub_doc_override).split('.subdomains.docroot.override')[0]}"
                    subdomname = f"{subdompart}.{domname}"
                    sub_php_set[subdomname] = php_select_value.group()
        return sub_php_set

@property
    def all_sites(self) -> dict:
        """
        Retrieve all domains and subdomains, existing on DA server,
        including aliases
        in the form of dict {domain_name: domain_config}
        :return: {domain_name: domain_config} including subdomains
        """
        da_sites = dict()
        for bunch in self.main_domains, self.subdomains, self.aliases:
            da_sites.update(bunch)
        return da_sites

@user_mode_verification
    @with_fpm_reload_restricted
    def get_domain_info(self, domain_name: str) -> DomainInfo:
        """
        Retrieve information about given domain from control panel environment:
        PHP version, user of domain, fpm status
        :param domain_name: name of domain
        :return: a DomainInfo object
        """
        try:
            domain_conf = self.all_sites[domain_name]
        except KeyError:
            self.logger.warning(
                'Domain does not exist on the server or is a pointer (no task allowed for pointers)',
                extra={'domain_name': domain_name})
            raise XRayMissingDomain(domain_name,
                                    message=_("Domain '%(domain_name)s' does not exist on this server "
                                            "or is a pointer (no task allowed for pointers)"))

data = self.file_readlines(domain_conf)

def find_item(item: str) -> str:
            """
            Get config value of item (e.g. item=value)
            :param item: key to get value of
            :return: value of item
            """
            found = [line.strip() for line in data if item in line]
            try:
                return found[0].split('=')[-1]
            except IndexError:
                return '1'

opts = self.php_options

# Trying to get the subdomain handler first,
        # get main domain handler if nothing is set for subdomain
        php_selected = self.subdomains_php_settings.get(
            domain_name) or find_item('php1_select')

# if user config has disabled php version selected ->
        # DirectAdmin uses the 1st one
        # grep php1 /usr/local/directadmin/data/users/auser/domains/auser.com.conf
        # php1_select=2
        # cat /usr/local/directadmin/custombuild/options.conf | grep -i php
        # # PHP Settings
        # php1_release=7.4
        # php1_mode=lsphp
        # php2_release=no
        if not opts.get(php_selected):
            php_selected = '1'

if self.phpinfo_mode:
            config = phpinfo_utils.get_php_configuration(
                find_item('username'), domain=domain_name)
            domain_info = DomainInfo(
                name=domain_name,
                panel_php_version=config.get_full_php_version('php'),
                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=find_item('username'),
                panel_fpm=config.is_php_fpm,
                handler=php_selected
            )
        else:
            domain_info = DomainInfo(name=domain_name,
                                     panel_php_version=opts[php_selected]['ver'],
                                     user=find_item('username'),
                                     panel_fpm=opts[php_selected]['fpm'],
                                     handler=php_selected)
        self.logger.info(
            'Retrieved domain info: domain %s owned by %s uses php version %s',
            domain_name, domain_info.user, domain_info.handler)
        return domain_info

def panel_specific_selector_enabled(self, domain_info: DomainInfo) -> bool:
        """
        Check if selector is enabled specifically for DirectAdmin
        Required to be implemented by child classes
        :param domain_info: a DomainInfo object
        :return: True if yes, False otherwise
        """
        compatible_handlers = ('suphp', 'lsphp', 'fastcgi')
        current_handler = self.php_options[domain_info.handler]['handler']
        return domain_info.handler == '1' and current_handler in compatible_handlers

def fpm_service_name(self, dom_info: DomainInfo) -> str:
        """
        Get DirectAdmin FPM service name
        :param dom_info: a DomainInfo object
        :return: FPM service name
        """
        return f'php-fpm{dom_info.panel_php_version[-2:]}'

def php_procs_reload(self, domain_info: DomainInfo) -> None:
        """
        Copy xray.so for current version, create ini_location directory
        Reload FPM service or kill all *php* processes of user
        :param domain_info: a DomainInfo object
        """
        try:
            subprocess.run(['/usr/share/alt-php-xray/da_cp_xray',
                            domain_info.panel_php_version[-2:]],
                           capture_output=True, text=True)
        except self.subprocess_errors as e:
            self.logger.error('Failed to copy xray.so',
                              extra={'err': str(e),
                                     'info': domain_info})
        if self.php_options[domain_info.handler]['handler'] == 'mod_php':
            try:
                subprocess.run(['/usr/sbin/service',
                                'httpd',
                                'restart'],
                               capture_output=True, text=True)
                self.logger.info('httpd restarted')
            except self.subprocess_errors as e:
                self.logger.error('Failed to restart httpd',
                                  extra={'err': str(e),
                                         'info': domain_info})
        super().php_procs_reload(domain_info)

Batal

⬅ xray 7 item

Nama	Tipe	Ukuran	Diubah	Aksi
📁__pycache__	dir	—	2026-03-11 19:11	📂 Buka
🐍__init__.py	py	1.2 KB	2026-02-23 14:52	✏️ 👁️
🐍base.py	py	30.6 KB	2026-02-23 14:52	✏️ 👁️
🐍cpanel.py	py	8.3 KB	2026-02-23 14:52	✏️ 👁️
🐍custom.py	py	6 KB	2026-02-23 14:52	✏️ 👁️
🐍directadmin.py	py	12.9 KB	2026-02-23 14:52	✏️ 👁️
🐍plesk.py	py	6.9 KB	2026-02-23 14:52	✏️ 👁️

✏️ Rename