编辑文件 - linux

✏️ 正在编辑: linux_cpanel.py

路径: /var/opt/nydus/ops/customer_local_ops/control_panel/linux_cpanel.py
提示: 您可以编辑任何文件（包括二进制文件），但请注意不当修改可能导致文件损坏。

from datetime import datetime
from typing import Dict, Any, List, Tuple
import functools
import logging
import os
import pwd
import re
import socket
import subprocess
import time
import json

from shortuuid import ShortUUID

from customer_local_ops import OpType, ResourceType
from customer_local_ops.operating_system.linux import (
    AlmaLinux8, AlmaLinux9, AlmaLinux10, Linux, CentOS, CentOS6, CentOS7, Debian, Debian8, Ubuntu1604)
from customer_local_ops.control_panel.cpanel import CPanelException, OSCPanel
from customer_local_ops.util import random_password
from customer_local_ops.util.execute import runCommand, run_uapi_command
from customer_local_ops.util.helpers import replace_line, edit_file_lines, create_file, replace_file_lines_multiple
from customer_local_ops.util.retry import Retry, RETRY

LOG = logging.getLogger(__name__)

SHORT_UUID_DEFAULT_LENGTH = 30

class LinuxCPanel(Linux, OSCPanel):
    """
    CPanel Customer Local Ops for the Linux OS.
    All function names should contain 'cpanel' so as not to override the OS ops
    """

AUTO_RESTART_EXCLUDE_SERVICES = ['nydus-ex', 'nydus-ex-api']
    AUTO_RESTART_EXCLUSION_FILE = '/etc/cpanel/local/ignore_outdated_services'
    # ^ as of cPanel 11.76 / Jan 13 2019
    HOSTNAME_CHANGE_LOCK_FILE = '/var/cpanel/.application-locks/UpdateHostname'
    HOSTNAME_LOCK_WAIT_TIMEOUT = 120  # seconds to wait for hostname lock to clear
    HOSTNAME_LOCK_POLL_INTERVAL = 5  # seconds between lock file checks
    TRANSIENT_LICENSE_MESSAGES = (
        'a license check appears to already be running',
        'cannot read license file',
    )
    CPKEYCLT_WAIT_TIMEOUT = 120  # seconds to wait for existing cpkeyclt to finish
    CPKEYCLT_POLL_INTERVAL = 5  # seconds between checks
    CPKEYCLT_ACTIVATE_MAX_ATTEMPTS = 6
    CPKEYCLT_ACTIVATE_RETRY_INTERVAL = 30  # seconds between internal cpkeyclt retries

op_type = OpType.CONTROL_PANEL_OPERATING_SYSTEM

# this is a bit different. If the user picks cpanel, we don't want to do the regular os_op configureMTA
    # (exim conflicts with sendmail)
    def configure_mta_cpanel(self, payload: Dict[str, Any], *args: Any) -> Any:
        """Configures the mail transfer agent for cPanel

:param payload: A dict containing input data
        """
        LOG.info("LinuxCPanel.configure_mta_cpanel start")
        op_name = 'configure_mta_cpanel'
        try:
            set_tgt = functools.partial(replace_line,
                                        match='defaultmailaction',
                                        replace='defaultmailaction=fail\n',
                                        firstword=False)
            edit_file_lines('/var/cpanel/cpanel.config', set_tgt)
            relay = payload.get('relay_address')
            if relay is not None:
                exim_conf_path = '/etc/exim.conf.local'
                LOG.info("LinuxCPanel.configure_mta_cpanel writing %s", exim_conf_path)
                create_file(exim_conf_path, """
@AUTH@
@BEGINACL@
@CONFIG@
@DIRECTOREND@
@DIRECTORMIDDLE@
@DIRECTORSTART@
@ENDACL@
@RETRYEND@
@RETRYSTART@
@REWRITE@
@ROUTEREND@
@ROUTERSTART@
send_to_smart_host:
driver = manualroute
route_list = !+local_domains %s
transport = remote_smtp
@TRANSPORTEND@
@TRANSPORTMIDDLE@
@TRANSPORTSTART@
""" % relay)
        except Exception as ex:  # pylint: disable=broad-except
            LOG.error('cp_os_op LinuxCPanel configure_mta_cpanel result(fail): %s', str(ex))
            return False, self.build_result_dict('', str(ex), op_name)

LOG.info("LinuxCPanel.configure_mta_cpanel running buildeximconf")
        return self._run_uapi_command(['/scripts/buildeximconf'], 'build exim config', op_name)

def _run_uapi_command(self, cmd_list: List[str], description: str, op_name: str) -> Tuple[bool, Dict[str, Any]]:
        """Runs a local cPanel command

:param cmd_list: A list of commands to run
        :param description: A description of the command(s) to be run
        :param op_name: The name of the op calling this function
        """
        exit_code, outs, errs = run_uapi_command(cmd_list, description, op_name)
        if exit_code == 0:
            # cpanel's uapi returns 0 even when it errors. For example, if you try to add the same site name twice.
            LOG.debug("cp_op_result: success! %s", errs)
            if 'uapi --user=' in ' '.join(cmd_list):
                # The errors key is always returned. A '~' value indicates no errors occurred.
                if 'errors: ~' not in outs:
                    return False, self.build_result_dict(outs, errs, op_name)
            if 'whmapi1 ' in ' '.join(cmd_list):
                # If result is 1, that denotes success. 0 denotes failure.
                if 'result: 1' not in outs:
                    return False, self.build_result_dict(outs, errs, op_name)
            return True, self.build_result_dict(outs, errs, op_name)
        return False, self.build_result_dict(outs, errs, op_name)

def _wait_for_hostname_lock(self):
        """Wait for the cPanel hostname change lock file to clear.

If the lock does not clear within HOSTNAME_LOCK_WAIT_TIMEOUT, it is
        forcefully removed — a cPanel hostname change takes seconds, so a lock
        older than the timeout is either stale (leftover from a previous boot)
        or held by a hung process that will never finish.
        """
        deadline = time.time() + self.HOSTNAME_LOCK_WAIT_TIMEOUT
        while os.path.exists(self.HOSTNAME_CHANGE_LOCK_FILE):
            if time.time() >= deadline:
                LOG.warning("Hostname lock file %s did not clear in %ds, "
                            "force-removing",
                            self.HOSTNAME_CHANGE_LOCK_FILE,
                            self.HOSTNAME_LOCK_WAIT_TIMEOUT)
                try:
                    os.remove(self.HOSTNAME_CHANGE_LOCK_FILE)
                except OSError:
                    pass
                return True
            LOG.info("Hostname lock file exists (%s), waiting %ds (timeout in %ds)",
                     self.HOSTNAME_CHANGE_LOCK_FILE,
                     self.HOSTNAME_LOCK_POLL_INTERVAL, int(deadline - time.time()))
            time.sleep(self.HOSTNAME_LOCK_POLL_INTERVAL)
        return True

def change_hostname_cpanel(self, hostname: str, *_: Any,
                               intermediate_result: Dict[str, Any] = None) -> Any:
        """Changes the server hostname via a command for the local operating system

:param hostname: the new server hostname
        :param intermediate_result: Nydus intermediate result for storing the set-hostname
            result while waiting for the cPanel lock file to clear
        :returns: the result of the set-hostname cPanel call, in Nydus/CLO result format
        """
        LOG.info("LinuxCPanel.change_hostname_cpanel start")

if intermediate_result is None:
            intermediate_result = {}

set_hostname_result = intermediate_result.get('set_hostname_result')

if set_hostname_result is None:
            if not self._wait_for_hostname_lock():
                LOG.warning("Hostname lock did not clear in %ds, requesting retry",
                            self.HOSTNAME_LOCK_WAIT_TIMEOUT)
                return Retry(intermediate_result=intermediate_result)

result = self._run_uapi_command(
                ['/usr/local/cpanel/bin/set_hostname', hostname],
                'set cpanel host',
                'change_hostname_cpanel')

if not result[0]:
                outs = result[1].get('outs', '') if isinstance(result[1], dict) else ''
                if 'hostname change in progress' in str(outs).lower():
                    LOG.info('cPanel set_hostname reports hostname change already in progress, will retry.')
                    return Retry(intermediate_result=intermediate_result)
                return result

intermediate_result['set_hostname_result'] = result
            set_hostname_result = result

if not self._wait_for_hostname_lock():
            LOG.warning("Hostname lock did not clear after set_hostname in %ds, requesting retry",
                        self.HOSTNAME_LOCK_WAIT_TIMEOUT)
            return Retry(intermediate_result=intermediate_result)

return set_hostname_result

def get_public_ip_cpanel(self, *args: Any) -> Any:
        """Gets the cPanel public IP for this server"""
        op_name = 'get_public_ip_cpanel'
        command = ['wget', '-q', '-O', '-', 'http://cpanel.net/showip.cgi']
        exit_code, outs, errs = runCommand(command, 'Get cPanel outbound IP')
        if exit_code == 0:
            public_ip = outs
            return public_ip

return False, self.build_result_dict(outs, errs, op_name)

def mark_internal_addresses_cpanel(self, private_addrs: List[str], *args: Any) -> Tuple[bool, Dict]:
        """Marks the server IPs as reserved

:param private_addrs: A list of IPs to mark
        """
        op_name = 'mark_internal_addresses_cpanel'
        try:
            create_file('/etc/reservedips', '\n'.join(private_addrs))
            create_file('/etc/reservedipreasons', 'Internal datacenter-local addresses not publicly accessible.')
        except OSError as ex:
            return False, self.build_result_dict('', str(ex), op_name)
        return True, self.build_result_dict('Marking internal addresses succeeded', '', op_name)

def cpanel_enable(self, cpanel_public_ip: str, *args: Any) -> Any:
        """Enables cPanel functionality for this server

:param cpanel_public_ip: The cPanel public IP for the server
        """
        op_name = 'cpanel_enable'
        LOG.debug("cpanel_public_ip- %s", cpanel_public_ip)
        # update /etc/wwwacct.conf ADDR
        try:
            edit_file_lines('/etc/wwwacct.conf',
                            functools.partial(replace_line,
                                              match='ADDR',
                                              replace='ADDR %s\n' % cpanel_public_ip,
                                              firstword=True))
            # update /var/cpanel/mainip
            try:
                os.remove('/var/cpanel/mainip')
            except OSError:
                pass
            create_file('/var/cpanel/mainip', '%s' % cpanel_public_ip)
            create_file('/var/cpanel/activate/2012-07.v01.EULACPWHM', '')
            create_file('/etc/.whostmgrft', '')
            exit_code, outs, errs = runCommand(['sed', '-i', 's/rpmup_allow_kernel=0/rpmup_allow_kernel=1/g',
                                                '/var/cpanel/cpanel.config'], 'enable automatic kernel updates')
            LOG.debug("Automatic kernel updates result: %s: %s -- %s", exit_code, errs, outs)
            exit_code, outs, errs = runCommand(['sed', '-i',
                                                's/allow_deprecated_accesshash=0/allow_deprecated_accesshash=1/g',
                                                '/var/cpanel/cpanel.config'], 'enable accesshash')
            LOG.debug("Enable accesshash result: %s: %s -- %s", exit_code, errs, outs)
            exit_code, outs, errs = runCommand(['service', 'cpanel', 'restart'], 'restart cpanel')
            LOG.debug("restart cpanel result: %s: %s -- %s", exit_code, errs, outs)

for feature in ('appconfig',
                            'email_archiving',
                            'email_autodiscovery',
                            'log_archiving',
                            'query_apache_for_nobody_senders',
                            'server_usage_analytics',
                            'servers_usage_analytics',
                            'smtp_restrictions',
                            'trust_x_php_script'):
                create_file(os.path.join('/var/cpanel/activate/features/', feature),
                            """USER=root
MODIFIED=%s
TIMESTAMP=%s
INTERFACE=GUI
""" % (datetime.now().ctime(), datetime.now().ctime()))

# The following section is an attempt to reduce queueprocd errors.
            exit_code, outs, errs = runCommand(['/scripts/restartsrv_queueprocd'], 'restart queueprocd')
            LOG.debug("Restart queueprocd result: %s: %s -- %s", exit_code, errs, outs)
        except OSError as ex:
            return False, self.build_result_dict('', str(ex), op_name)
        return True, self.build_result_dict('CPanel enabled successfully', '', op_name)

def _wait_for_cpkeyclt(self):
        """Wait for any running cpkeyclt process to finish before starting a new one.

Returns True if safe to proceed, False if timed out waiting.
        """
        deadline = time.time() + self.CPKEYCLT_WAIT_TIMEOUT
        while time.time() < deadline:
            try:
                result = subprocess.run(
                    ['pgrep', '-f', 'cpkeyclt'],
                    stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                    check=False, timeout=10
                )
                if result.returncode != 0:
                    return True
            except (subprocess.TimeoutExpired, OSError):
                pass
            LOG.info("cpkeyclt process still running, waiting %ds (timeout in %ds)",
                     self.CPKEYCLT_POLL_INTERVAL, int(deadline - time.time()))
            time.sleep(self.CPKEYCLT_POLL_INTERVAL)

LOG.warning("Timed out waiting %ds for cpkeyclt to finish", self.CPKEYCLT_WAIT_TIMEOUT)
        return False

def cpanel_activate(self, vm_resource: str, *args: Any,
                        intermediate_result: Dict[str, Any] = None) -> Any:
        """Activates cPanel license for this server.

Waits for any existing cpkeyclt process to finish before attempting
        license activation, then retries internally on transient failures
        (e.g. "A License check appears to already be running").

:param vm_resource: The resource name for the third-party hosting provider
        :param intermediate_result: an intermediate result
        """
        op_name = self.OP_CPANEL_ACTIVATE

if not self._wait_for_cpkeyclt():
            LOG.warning("Proceeding with cpanel_activate despite cpkeyclt wait timeout")

for attempt in range(self.CPKEYCLT_ACTIVATE_MAX_ATTEMPTS):
            exit_code, outs, errs = runCommand(['/usr/local/cpanel/cpkeyclt'], 'activate license')
            output = '{outs}\n{errs}'.format(outs=str(outs), errs=str(errs)).lower()

if not any(message in output for message in self.TRANSIENT_LICENSE_MESSAGES):
                break

if attempt < self.CPKEYCLT_ACTIVATE_MAX_ATTEMPTS - 1:
                LOG.info("cpkeyclt reported transient license state (attempt %d/%d), "
                         "waiting %ds before retry: %s",
                         attempt + 1, self.CPKEYCLT_ACTIVATE_MAX_ATTEMPTS,
                         self.CPKEYCLT_ACTIVATE_RETRY_INTERVAL, output)
                time.sleep(self.CPKEYCLT_ACTIVATE_RETRY_INTERVAL)
            else:
                LOG.info("cpkeyclt transient state persisted after %d attempts, "
                         "requesting nydus retry: %s",
                         self.CPKEYCLT_ACTIVATE_MAX_ATTEMPTS, output)
                return RETRY

if exit_code != 0:
            return False, self.build_result_dict(outs, errs, op_name)

ops_map = self.CPANEL_OPS_RESOURCE_ATTRIBUTE_MAP[ResourceType(vm_resource)][op_name]

if ops_map[self.RUN_INSTALLATRON_REPAIR]:
            # We need to re-initialize installatron after getting ne wkey (Only on VMs)
            exit_code, outs, errs = runCommand(
                ['rm', '-fr', '/usr/local/installatron/lib', '/usr/local/installatron/etc/php.ini'],
                'rm installatron php.ini')
            LOG.debug("rm installatron php.ini results: %s: %s -- %s", exit_code, errs, outs)
            exit_code, outs, errs = runCommand(['curl', '-O', 'https://data.installatron.com/installatron-plugin.sh'],
                                               'curl installatron-plugin')
            LOG.debug("curl installatron-plugin result: %s: %s -- %s", exit_code, errs, outs)
            exit_code, outs, errs = runCommand(['chmod', '+x', 'installatron-plugin.sh'], 'chmod plugin')
            LOG.debug("chmod plugin result: %s: %s -- %s", exit_code, errs, outs)
            exit_code, outs, errs = runCommand(['./installatron-plugin.sh', '-f', '--quick'], 'rebuild plugin')
            LOG.debug("rebuild plugin result: %s: %s -- %s", exit_code, errs, outs)

def _cleanup_pyinstaller_temp_dirs(self) -> None:
        """Clean up stale PyInstaller extraction directories

PyInstaller extracts files to /opt/nydus/tmp/_MEI* directories at runtime.
        These should be cleaned up automatically, but sometimes orphaned directories
        remain and cause "already exists but should not" errors on AlmaLinux 8.

This method removes stale _MEI directories older than 1 hour.
        """
        nydus_tmp_dir = '/opt/nydus/tmp'
        if not os.path.exists(nydus_tmp_dir):
            return

try:
            # Find and remove _MEI directories older than 1 hour (3600 seconds)
            current_time = datetime.now().timestamp()
            for entry in os.listdir(nydus_tmp_dir):
                if entry.startswith('_MEI'):
                    dir_path = os.path.join(nydus_tmp_dir, entry)
                    if os.path.isdir(dir_path):
                        # Check if directory is older than 1 hour
                        dir_mtime = os.path.getmtime(dir_path)
                        if current_time - dir_mtime > 3600:
                            LOG.debug("Cleaning up stale PyInstaller directory: %s", dir_path)
                            exit_code, _, errs = runCommand(
                                ['rm', '-rf', dir_path],
                                f'cleanup pyinstaller temp {entry}')
                            if exit_code != 0:
                                LOG.warning("Failed to cleanup %s: %s", dir_path, errs)
        except (OSError, IOError) as ex:
            LOG.warning("Error during PyInstaller temp cleanup: %s", str(ex))

def _ensure_mysql_running(self):
        """Start MariaDB/MySQL service if it is not already active.

On cPanel AlmaLinux 9 images, MariaDB is masked in systemd at boot
        and the datadir may not exist. We unmask the service, then either
        start it directly (if system tables exist) or run mariadb-install-db
        to initialize from scratch.
        """
        datadir = '/var/lib/mysql'
        if not os.path.isdir(datadir):
            LOG.info("Creating MySQL datadir %s", datadir)
            os.makedirs(datadir, mode=0o755, exist_ok=True)
            try:
                mysql_uid = pwd.getpwnam('mysql').pw_uid
                mysql_gid = pwd.getpwnam('mysql').pw_gid
                os.chown(datadir, mysql_uid, mysql_gid)
            except KeyError:
                pass

# Check if already running
        for svc in ('mariadb', 'mysql', 'mysqld'):
            rc, _, _ = runCommand(
                ['systemctl', 'is-active', '--quiet', svc],
                'check %s active' % svc)
            if rc == 0:
                return

# Unmask (required on cPanel AlmaLinux 9 images)
        for svc in ('mariadb', 'mysql', 'mysqld'):
            runCommand(['systemctl', 'unmask', svc], 'unmask %s' % svc)

# If system tables don't exist, initialize the database.
        # Stale InnoDB files without system tables block install-db, so
        # remove them first. Then disable unix_socket=OFF in my.cnf
        # (install-db creates root with unix_socket auth; the plugin must
        # be loaded for mysqlpasswd to connect).
        if not os.path.isdir(os.path.join(datadir, 'mysql')):
            LOG.info("No mysql system tables, initializing database")
            for f in ('ibdata1', 'ib_logfile0', 'ib_logfile1', 'ibtmp1',
                      'aria_log.00000001', 'aria_log_control',
                      'ib_buffer_pool'):
                path = os.path.join(datadir, f)
                if os.path.exists(path):
                    os.remove(path)
            # Remove unix_socket=OFF so mysqlpasswd can connect after init
            mycnf = '/etc/my.cnf'
            if os.path.isfile(mycnf):
                try:
                    with open(mycnf, 'r') as fh:
                        lines = fh.readlines()
                    filtered = [l for l in lines
                                if 'unix_socket' not in l.lower()
                                or l.strip().startswith('#')]
                    if len(filtered) != len(lines):
                        with open(mycnf, 'w') as fh:
                            fh.writelines(filtered)
                        LOG.info("Removed unix_socket=OFF from %s", mycnf)
                except (IOError, OSError) as exc:
                    LOG.warning("Could not update %s: %s", mycnf, exc)
            runCommand(['mariadb-install-db', '--user=mysql',
                        '--datadir=' + datadir], 'init mysql db')
            runCommand(['systemctl', 'start', 'mariadb'], 'start mariadb')
            return

# System tables exist — safe to start directly
        for svc in ('mariadb', 'mysql', 'mysqld'):
            rc, _, _ = runCommand(
                ['systemctl', 'start', svc], 'start %s' % svc)
            if rc == 0:
                LOG.info("Started %s service", svc)
                return

def set_mysql_password_cpanel(self, *args: Any,
                                  intermediate_result: Dict[str, Any] = None) -> Any:
        """Generates and sets a random mysql password for cPanel

:param intermediate_result: Intermediate result for retry state tracking
        """
        op_name = 'set_mysql_password_cpanel'

# Clean up stale PyInstaller directories that can cause extraction errors
        # on AlmaLinux 8 (fixes "already exists but should not" errors)
        self._cleanup_pyinstaller_temp_dirs()

# Start MariaDB if not already running (cPanel may not start it on boot)
        self._ensure_mysql_running()

password = random_password()
        exit_code, outs, errs = runCommand(
            ['/scripts/mysqlpasswd', 'root', password],
            'mysql password', omitString=password)

# Check for MySQL connection errors that indicate service not ready
        if exit_code != 0:
            mysql_not_ready_errors = [
                'CR_CONNECTION_ERROR',
                "Can't connect to local server",
                "Can't connect to MySQL server",
                'ConnectError',
                'mysql.sock',
                'Connection refused'
            ]

if any(error in errs for error in mysql_not_ready_errors):
                LOG.warning("MySQL not ready yet, will retry: %s", errs)
                return RETRY  # Trigger retry via decorator in parent class

return False, self.build_result_dict(outs, errs, op_name)

exit_code, outs, errs = runCommand(['/scripts/mysqlconnectioncheck'], 'restart mysql')
        if exit_code != 0:
            LOG.warning("mysqlconnectioncheck failed (exit %d), will retry: %s", exit_code, errs)
            return RETRY

def enable_secure_tmp_cpanel(self, *args: Any) -> Any:
        """Re-secures the /tmp directory"""
        op_name = 'enable_secure_tmp_cpanel'
        exit_code, outs, errs = runCommand(['rm', '-f', '/var/cpanel/version/securetmp_disabled'], 're-secure tmp dir')
        if exit_code != 0:
            return False, self.build_result_dict(outs, errs, op_name)

def cpanel_prep(self, *args: Any) -> Any:
        """Pre-installs and prepares cPanel on the server"""
        op_name = 'cpanel_prep'

try:
            # Clean up stale PyInstaller directories before starting prep
            self._cleanup_pyinstaller_temp_dirs()

self._check_hostname_cpanel()
            self._yum_update_cpanel()
            self._download_cpanel()
            self._exclude_nydus_from_auto_restart_cpanel()
            self._install_cpanel()
            self._install_installatron_cpanel()
            self._config_cpanel()
            self._disable_secure_tmp_cpanel()
        except CPanelException as ex:
            return False, self.build_result_dict(ex.outs, ex.errs, op_name)

return self.build_result_dict('cpanel_prep complete', '', op_name)

def _check_hostname_cpanel(self) -> bool:
        """Checks that the server's fully qualified hostname is valid for CPanel

`hostname` -> "PT-Test.secureserver.net" but `hostname -f` -> "localhost".
        /etc/hostname: pt-test.secureserver.net
        /etc/hosts:
            127.0.0.1 localhost PT-Test.secureserver.net # cloud-controlled; do not change
            10.192.28.50 pt-test.secureserver.net pt-test
          CHANGE: removed the PT-Test on the 127.0.0.1 line.
          `hostname` -> "PT-Test.secureserver.net"
          `hostname -f` -> "pt-test.secureserver.net"
        """
        hostname = socket.gethostbyaddr(socket.gethostname())[0]
        if len(hostname.split('.')) < 3:
            raise CPanelException(
                '',
                'cPanel installation requires a fully-qualified hostname ({} is not enough)'.format(hostname))
        return True

def _yum_update_cpanel(self) -> bool:
        """Performs a yum update on the server

:raises CPanelException: If the command fails
        """
        exit_code, outs, errs = self._yum_update()
        if exit_code != 0:
            raise CPanelException(outs, errs)
        return True

def _download_cpanel(self) -> bool:
        """Downloads cPanel to the server

:raises CPanelException: If the command fails
        """
        exit_code, outs, errs = runCommand(
            ['wget', 'https://securedownloads.cpanel.net/latest', '-O',
             '/root/cpinstall'],
            'download cpinstall')
        if exit_code != 0:
            raise CPanelException(outs, errs)
        return True

def _exclude_nydus_from_auto_restart_cpanel(self) -> bool:  # pylint:disable=invalid-name
        """Exclude Nydus services from cPanel's auto restarts.

cPanel restarts services with outdated dependencies during install. It
        does this very generally, and Nydus services are included and stopped
        in the middle of install, causing terminal failure.

This method adds Nydus services to the exclusion list so they are not
        restarted.

See also, on a cPanel system:
            - /usr/local/cpanel/scripts/find_outdated_services
            - /usr/local/cpanel/Cpanel/ProcessCheck/Outdated.pm
        """
        os.makedirs(os.path.dirname(self.AUTO_RESTART_EXCLUSION_FILE), exist_ok=True)
        content = '%s\n' % '\n'.join(self.AUTO_RESTART_EXCLUDE_SERVICES)
        create_file(self.AUTO_RESTART_EXCLUSION_FILE, content)
        return True

def _install_cpanel(self) -> bool:
        """Installs cPanel on the server

:raises CPanelException: If the command fails
        """
        exit_code, outs, errs = runCommand(['bash', '/root/cpinstall'], 'install cpanel')
        if exit_code != 0:
            raise CPanelException(outs, errs)
        return True

def _install_installatron_cpanel(self) -> bool:
        """Installs Installatron plugin on the server

:raises CPanelException: If the command fails
        """
        exit_code, outs, errs = runCommand(
            ['rpm', '-U', '-h', '-v',
             'http://data.installatron.com/installatron-plugin-cpanel-latest.noarch.rpm'],
            'install installatron')
        if exit_code != 0:
            raise CPanelException(outs, errs)
        return True

def _config_cpanel(self) -> bool:
        """Adds initial cPanel configuration

:raises CPanelException: If there is a problem running any commands
        """
        try:
            replace_dict = {'NS2': 'NS2 ns2.secureserver.net\n',
                            'NS': 'NS ns1.secureserver.net\n',
                            'ETHDEV': 'ETHDEV eth0\n',
                            'CONTACTEMAIL': 'CONTACTEMAIL root@cpaneltmp.secureserver.net\n',
                            '^ADDR .*': 'ADDR\n'}
            replace_file_lines_multiple('/etc/wwwacct.conf', replace_dict)
            mainipfile = '/var/cpanel/mainip'
            if os.path.exists(mainipfile):
                os.unlink(mainipfile)
        except OSError as ex:
            raise CPanelException('', str(ex)) from ex
        return True

def _disable_secure_tmp_cpanel(self) -> bool:
        """Disables noexec on the /tmp directory

:raises CPanelException: If the command fails
        """
        exit_code, outs, errs = runCommand(['touch', '/var/cpanel/version/securetmp_disabled'], 'disable noexec /tmp')
        if exit_code != 0:
            raise CPanelException(outs, errs)
        return True

def hulk_whitelist_cpanel(self, from_ip_addr: str, *args: Any) -> Tuple[bool, Dict[str, Any]]:
        """Allow cPanel access from customer IP

:param from_ip_addr: The IP address from which the customer will access the cPanel instance
        """
        op_name = 'hulk_whitelist_cpanel'
        exit_code, outs, errs = runCommand(['/scripts/cphulkdwhitelist', from_ip_addr], 'cphulk whitelist')
        return exit_code == 0, self.build_result_dict(outs, errs, op_name)

def get_hash_cpanel(self, *args: Any) -> Any:
        """Set cPanel hash"""

op_name = 'get_hash_cpanel'
        path = '/root/.accesshash'
        if os.path.exists(path) and os.path.getsize(path) == 0:
            os.unlink(path)
        if not os.path.exists(path):
            exit_code, outs, errs = runCommand(['/usr/local/cpanel/whostmgr/bin/whostmgr', 'setrhash'], 'generate hash')
            if exit_code != 0:
                return False, self.build_result_dict(outs, errs, op_name)

with open(path, 'r', encoding='utf-8') as hashf:
            cphash = hashf.read()
        return ''.join(cphash.split('\n'))

LETSENCRYPT_TOS_URL_PATTERN = re.compile(
        r'^https://letsencrypt\.org/documents/LE-SA-v\d+\.\d+-[A-Za-z]+-\d{1,2}-\d{4}\.pdf$'
    )
    LETSENCRYPT_DEFAULT_TOS_URL = 'https://letsencrypt.org/documents/LE-SA-v1.6-August-18-2025.pdf'

def accept_autossl_terms_cpanel(self, *args: Any, tos_url: str = None) -> Tuple[bool, Dict[str, Any]]:
        """Accepts the Let's Encrypt Terms of Service and re-registers the AutoSSL provider.

Runs whmapi1 reset_autossl_provider to re-register the Let's Encrypt provider
        with the accepted Terms of Service. This resolves cases where Let's Encrypt
        updated their Terms of Service and certificates cannot be renewed until the
        new ToS is accepted and the provider is re-registered.

:param tos_url: The Let's Encrypt Terms of Service URL. If not provided, uses the default URL.
        """
        op_name = 'accept_autossl_terms_cpanel'

if tos_url is None:
            tos_url = self.LETSENCRYPT_DEFAULT_TOS_URL

# Validate the ToS URL matches the expected Let's Encrypt format
        if not self.LETSENCRYPT_TOS_URL_PATTERN.match(tos_url):
            errs = 'Invalid Let\'s Encrypt ToS URL: {}'.format(tos_url)
            LOG.error("LinuxCPanel.accept_autossl_terms_cpanel %s", errs)
            return False, self.build_result_dict('', errs, op_name)

LOG.info("LinuxCPanel.accept_autossl_terms_cpanel re-registering Let's Encrypt provider")
        exit_code, outs, errs = runCommand(
            ['/usr/local/cpanel/bin/whmapi1', '--output=jsonpretty',
             'reset_autossl_provider',
             'provider=LetsEncrypt',
             'x_terms_of_service_accepted={}'.format(tos_url)],
            'reset autossl provider with accepted tos')
        if exit_code != 0:
            return False, self.build_result_dict(outs, errs, op_name)

# Verify the whmapi1 call succeeded by parsing the JSON output
        try:
            parsed_outs = json.loads(outs)
        except (json.JSONDecodeError, TypeError):
            LOG.error("LinuxCPanel.accept_autossl_terms_cpanel failed to parse whmapi1 JSON output: %s", outs)
            return False, self.build_result_dict(outs, errs, op_name)

metadata = parsed_outs.get('metadata', {})
        if not isinstance(metadata, dict) or metadata.get('result') != 1:
            return False, self.build_result_dict(outs, errs, op_name)

return True, self.build_result_dict(outs, errs, op_name)

def renew_ssl_certs_cpanel(self, *args: Any) -> Tuple[bool, Dict[str, Any]]:
        """Triggers SSL certificate validation and renewal using cPanel's built-in AutoSSL mechanism.

Executes /usr/local/cpanel/bin/checkallsslcerts which causes cPanel to check,
        renew, and reinstall SSL certificates as applicable.
        """
        op_name = 'renew_ssl_certs_cpanel'
        LOG.info("LinuxCPanel.renew_ssl_certs_cpanel running checkallsslcerts")
        exit_code, outs, errs = runCommand(
            ['/usr/local/cpanel/bin/checkallsslcerts'],
            'check and renew ssl certs')
        if exit_code != 0:
            return False, self.build_result_dict(outs, errs, op_name)
        return True, self.build_result_dict(outs, errs, op_name)

def get_api_token_cpanel(self, *args: Any) -> Any:
        """Get cPanel API Token. Revoke any existing nydus-generated tokens before generating a new token."""

op_name = 'get_api_token_cpanel'
        token_prefix = 'dashboard_generated_'
        token_suffix = ShortUUID().random(length=SHORT_UUID_DEFAULT_LENGTH)
        token_name = token_prefix + token_suffix

# Retrieve and revoke any existing tokens matching the token prefix
        exit_code, outs, errs = runCommand(['/usr/local/cpanel/bin/whmapi1', '--output=jsonpretty', 'api_token_list'],
                                           'get api tokens')

if exit_code != 0:
            return False, self.build_result_dict(outs, errs, op_name)

tokens_json = json.loads(outs)
        tokens = tokens_json.get('data').get('tokens')
        for old_token_name in tokens.keys():
            if old_token_name.startswith(token_prefix):
                # Revoke token
                runCommand(['/usr/local/cpanel/bin/whmapi1', '--output=jsonpretty',
                            'api_token_revoke',
                            'token_name={old_token_name}'.format(old_token_name=old_token_name)],
                           'revoke token')

exit_code, outs, errs = runCommand(['/usr/local/cpanel/bin/whmapi1', '--output=jsonpretty', 'api_token_create',
                                            "token_name={token_name}".format(token_name=token_name)],
                                           'generate api token')
        if exit_code != 0:
            return False, self.build_result_dict(outs, errs, op_name)

token_json = json.loads(outs)
        token_data = token_json.get('data')
        if isinstance(token_data, dict):
            token = token_data.get('token')
            if token is not None:
                return self.encrypt(token).decode('utf-8')

return False, self.build_result_dict('', 'Invalid token json returned from cpanel', op_name)

class CentOSCPanel(CentOS, LinuxCPanel):
    pass

class CentOS6CPanel(CentOS6, CentOSCPanel):  # pylint: disable=too-many-ancestors
    def __init__(self, *args, **kwargs):  # pylint: disable=super-init-not-called
        raise NotImplementedError

class CentOS7CPanel(CentOS7, CentOSCPanel):  # pylint: disable=too-many-ancestors
    pass

class AlmaLinux8CPanel(AlmaLinux8, CentOSCPanel):  # pylint: disable=too-many-ancestors
    pass

class AlmaLinux9CPanel(AlmaLinux9, CentOSCPanel):  # pylint: disable=too-many-ancestors
    pass

class AlmaLinux10CPanel(AlmaLinux10, CentOSCPanel):  # pylint: disable=too-many-ancestors
    pass

class DebianCPanel(Debian, LinuxCPanel):
    def __init__(self, *args, **kwargs):  # pylint: disable=super-init-not-called
        raise NotImplementedError

class Debian8CPanel(Debian8, DebianCPanel):  # pylint: disable=too-many-ancestors
    pass

class Ubuntu1604CPanel(Ubuntu1604, DebianCPanel):  # pylint: disable=too-many-ancestors
    pass

← 返回文件管理器