✏️ Editing: daemon_subscription_handler.py Read Only

import logging
import os
import pwd
import re
import socket
import subprocess
import time
from dataclasses import dataclass
from typing import Optional, Dict, List

from clcommon.clpwd import drop_user_privileges

from clwpos import socket_utils
from clwpos.utils import run_in_cagefs_if_needed

_ADVICE_ID_RE = re.compile(r'^[1-9][0-9]*\Z')

@dataclass
class EnableFeatureTask:
    uid: int

# some arguments that we need when
    # we enable module manually
    domain: str
    wp_path: str

# the features list that we would like to enable
    feature: list

# if true, appends --ignore-errors to enable command
    ignore_errors: bool

# time when we give up and
    # think that waiting is unsuccessful
    timeout_timestamp: int

# optional argument which indicates that
    # we are applying advice
    advice_id: Optional[str] = None

# pid of the process which took the task in work
    pid: Optional[int] = None

class PendingSubscriptionWatcher:
    """
    Listen to set-suite and user events in order to automatically
    enable module when suite is purchased.
    """

# max time that we wait for the feature to be allowed
    _FEATURE_STATUS_CHECK_TIMEOUT = 180

_MAX_PENDING_TASKS_PER_UID = 10
    _MAX_WP_PATH_LENGTH = 256

def __init__(self, logger=None):
        # list of pending tasks to automatically enable
        # module when suite is allowed
        self._pending_enable_tasks: Dict[int, List[EnableFeatureTask]] = dict()

self._logger = logger or logging.getLogger(__name__)

def _cleanup_pending_tasks(self):
        """
        Cleanup list of pending tasks by removing outdated.
        """
        for uid, pending_tasks in list(self._pending_enable_tasks.items()):
            for pending_task in pending_tasks:
                if pending_task.timeout_timestamp >= time.time():
                    continue
                if pending_task.pid:
                    try:
                        os.kill(pending_task.pid, 0)
                        continue
                    except OSError:
                        # process is dead
                        pass
                pending_tasks.remove(pending_task)
            if not pending_tasks:
                self._pending_enable_tasks.pop(uid)
        self._logger.info('Cleanup of pending tasks. Still active tasks %s', self._pending_enable_tasks)

def _run_and_log_results(self, args):
        self._logger.info('Running %s', ' '.join(args))
        try:
            output = run_in_cagefs_if_needed(args, check=True)
            self._logger.info('Command succeded with output: \n`%s`', output)
        except subprocess.CalledProcessError as e:
            self._logger.exception('Unable to activate feature in background. '
                                   'Stdout is %s. Stderr is %s', e.stdout, e.stderr)
        except Exception:
            self._logger.exception('Unable to activate feature in background')
        self._logger.debug('Finished %s', ' '.join(args))

def _run_enable_command(self, task):
        if task.advice_id:
            self._run_and_log_results([
                '/opt/alt/php-xray/cl-smart-advice-user', 'apply',
                '--advice_id', str(task.advice_id)
            ])
        elif task.domain:
            for f in task.feature:
                self._logger.info('activate feature in background: feature=%s', f)
                cmd = [
                    'cloudlinux-awp-user', 'enable',
                    '--feature', f,
                    '--domain', task.domain,
                    '--wp-path', task.wp_path
                ]
                if task.ignore_errors:
                    cmd.append('--ignore-errors')
                self._run_and_log_results(cmd)
        else:
            logging.info('Task does not have any advice or domain specified, skipping')

def suite_allowed_callback(self, client_socket_obj: socket.socket,
                               daemon_socket: socket.socket = None):
        self._cleanup_pending_tasks()

for uid, tasks in list(self._pending_enable_tasks.items()):
            # Iterate over a copy of the task list so each completed task can be
            # removed without mutating the list the for-loop walks. Whole-uid
            # del was unsafe when multiple tasks are pending for a single uid
            # (bugbot 9e736ea9 on MR !39).
            for task in list(tasks):
                fp = os.fork()
                if fp:
                    self._logger.info('background process forked: pid=%d', fp)
                    task.pid = fp
                    os.waitpid(fp, 0)
                    self._logger.info('background process finished: pid=%d', fp)
                    tasks.remove(task)
                else:
                    try:
                        if daemon_socket is not None:
                            daemon_socket.close()
                        client_socket_obj.close()
                        from clwpos.feature_suites import get_allowed_modules
                        # drop privileges forever.. or at least till the end of process
                        drop_user_privileges(pwd.getpwuid(task.uid).pw_name,
                                             effective_or_real=True, set_env=True)
                        allowed_features = get_allowed_modules(task.uid)

disallowed = [f for f in task.feature if f not in allowed_features]
                        for f in disallowed:
                            self._logger.info('unable to activate feature in background: feature=%s is not allowed', f)
                        task.feature = [f for f in task.feature if f in allowed_features]
                        if not task.feature:
                            os._exit(0)
                            return  # unreachable in production; keeps control flow explicit for test mocks

self._run_enable_command(task)
                    finally:
                        os._exit(0)
            if uid in self._pending_enable_tasks and not self._pending_enable_tasks[uid]:
                del self._pending_enable_tasks[uid]

response: dict = {
            "result": "success"
        }
        socket_utils.send_dict_to_socket_connection_and_close(client_socket_obj, response)

def add_pending_upgrade_task(self, client_socket_obj: socket.socket, user_request: dict, uid: int):
        self._cleanup_pending_tasks()

from clcommon.cpapi import get_main_username_by_uid, userdomains
        from clwpos.optimization_features.features import ALL_OPTIMIZATION_FEATURES

known_features = {f.to_interface_name() for f in ALL_OPTIMIZATION_FEATURES}
        requested_features = user_request['feature'].split(',')
        for f in requested_features:
            if f not in known_features:
                response: dict = {"result": "error", "context": "unknown feature: %s" % f}
                socket_utils.send_dict_to_socket_connection_and_close(client_socket_obj, response)
                return

advice_id = user_request.get('advice_id')
        if advice_id is not None:
            advice_id = str(advice_id)
            if not _ADVICE_ID_RE.match(advice_id):
                response: dict = {"result": "error", "context": "invalid advice_id format"}
                socket_utils.send_dict_to_socket_connection_and_close(client_socket_obj, response)
                return

wp_path = user_request['wp_path']
        if wp_path is not None and len(wp_path) > self._MAX_WP_PATH_LENGTH:
            response: dict = {"result": "error", "context": "wp_path is too long"}
            socket_utils.send_dict_to_socket_connection_and_close(client_socket_obj, response)
            return
        if wp_path and '..' in wp_path.split('/'):
            response: dict = {"result": "error", "context": "invalid wp_path: must not contain '..' components"}
            socket_utils.send_dict_to_socket_connection_and_close(client_socket_obj, response)
            return

domain = user_request['domain']
        # Mirror the CLI's domain/wp_path coupling (wpos_user.py:444-447): both
        # must be set together. Without this, an IPC client that bypasses the
        # CLI can queue a task with truthy domain + wp_path=None, which crashes
        # _run_enable_command on `' '.join([..., '--wp-path', None])` and silently
        # fails feature activation.
        if bool(domain) != (wp_path is not None):
            response: dict = {"result": "error", "context": "domain and wp_path must be set together"}
            socket_utils.send_dict_to_socket_connection_and_close(client_socket_obj, response)
            return
        if domain:
            username = get_main_username_by_uid(uid)
            user_domains = {d for d, _ in userdomains(username)}
            if domain not in user_domains:
                response: dict = {"result": "error", "context": "domain does not belong to the user"}
                socket_utils.send_dict_to_socket_connection_and_close(client_socket_obj, response)
                return

if uid not in self._pending_enable_tasks:
            self._pending_enable_tasks[uid] = []

pending = self._pending_enable_tasks[uid]

for existing in pending:
            if (existing.domain == domain
                    and existing.wp_path == wp_path
                    and set(existing.feature) == set(requested_features)
                    and existing.advice_id == advice_id
                    and existing.ignore_errors == user_request['ignore_errors']):
                response: dict = {"result": "success"}
                socket_utils.send_dict_to_socket_connection_and_close(client_socket_obj, response)
                return

if len(pending) >= self._MAX_PENDING_TASKS_PER_UID:
            response: dict = {"result": "error", "context": "too many pending tasks"}
            socket_utils.send_dict_to_socket_connection_and_close(client_socket_obj, response)
            return

self._pending_enable_tasks[uid].append(EnableFeatureTask(
            uid=uid,
            domain=domain,
            wp_path=wp_path,
            feature=requested_features,
            ignore_errors=user_request['ignore_errors'],
            advice_id=advice_id,
            timeout_timestamp=int(time.time() + self._FEATURE_STATUS_CHECK_TIMEOUT)
        ))

self._logger.info('Successfully added pending upgrade subscription task %s', self._pending_enable_tasks[uid])

response: dict = {
            "result": "success"
        }
        socket_utils.send_dict_to_socket_connection_and_close(client_socket_obj, response)

def get_upgrade_task_status(self, client_socket_obj: socket.socket, uid: int, feature: str):
        self._cleanup_pending_tasks()
        pending_tasks = self._pending_enable_tasks.get(uid, [])

response: dict = {
            "result": "success",
            "pending": bool(any(feature in pending_task.feature for pending_task in pending_tasks))
        }
        socket_utils.send_dict_to_socket_connection_and_close(client_socket_obj, response)

🔐 Sid Gifari File Manager Pro