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

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

# -*- 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 a class with basic apply action for WPOS modules
"""
import json
import logging
import subprocess
from functools import wraps
from typing import Callable, Optional, Tuple

from xray.internal.user_plugin_utils import get_xray_exec_user
from xray.internal.utils import build_clwpos_user_cmd
from xray import gettext as _
from clcommon.clwpos_lib import get_locale_from_envars

from ..exceptions import ApplyAdviceException, RollbackAdviceException
from ..progress import SmartAdviceProgress

logger = logging.getLogger('roc_advice')

def inc_total_for_admin(func: Callable) -> Callable:
    """
    Increment total stages in admin mode:
     account 'allow' command as an extra stage.
    Applies to get_progress method of WPOSModuleApply
    """

@wraps(func)
    def wrapper(*args, **kwargs):
        """
        Wraps func
        """
        data = func(*args, **kwargs)
        if get_xray_exec_user() is None:
            # admin mode
            try:
                if data.total_stages:
                    data.total_stages += 1
            except AttributeError:
                return data
        return data

return wrapper

class WPOSModuleApply:
    """Basic advice on WPOS module"""

module_name = 'base_apply'
    suite = 'base_suite'

short_description: str
    detailed_description: str
    apply_advice_button_text = Optional[str]
    upgrade_to_apply_button_text = Optional[str]
    email_view_advice_text = Optional[str]
    email_subject = Optional[str]

is_premium_feature: bool

def _build_admin_cmd(self, args: dict) -> list:
        """Build subprocess argv for clwpos-admin invocation"""
        return ['/usr/bin/clwpos-admin', 'set-suite', '--allowed',
                '--suites=' + self.suite, '--users=' + args['username']]

def action_cmds(self, action, args: dict) -> tuple:

if action == 'apply':
            admin_cmd = self._build_admin_cmd(args)
        else:
            admin_cmd = None

ignore_errors = args.get('ignore_errors', '')
        clwpos_base = ['--domain=' + args['domain'],
                       '--wp-path=' + args['website'],
                       '--feature=' + self.module_name]

if action == 'apply':
            clwpos_args = ['enable'] + clwpos_base
            if ignore_errors:
                clwpos_args.append(ignore_errors)
            user_cmd = build_clwpos_user_cmd(args['username'], clwpos_args)
        elif action == 'rollback':
            user_cmd = build_clwpos_user_cmd(args['username'], ['disable'] + clwpos_base)
        else:
            raise ValueError(_('Unsupported action with advice, passed action: %s') % str(action))
        if get_xray_exec_user() is None and admin_cmd is not None:
            # admin mode
            return admin_cmd, user_cmd
        else:
            # user mode
            return (user_cmd, )

@staticmethod
    def _build_progress_cmd(username: str) -> list:
        """Build subprocess argv for get-progress"""
        return build_clwpos_user_cmd(username, ['get-progress'])

@property
    def progress_fields(self) -> tuple:
        """Expected fields of progress"""
        return 'total_stages', 'completed_stages'

@inc_total_for_admin
    def get_progress(self, as_user: str) -> SmartAdviceProgress:
        """
        Get progress for currently executed command of given user
        """
        try:
            _, result = self._exec_external(
                self._build_progress_cmd(as_user))
        except (ApplyAdviceException, ValueError):
            return SmartAdviceProgress()
        stages_data = {k: v for k, v in result.items() if k in self.progress_fields}
        return SmartAdviceProgress(**stages_data)

def apply(self, **kwargs) -> tuple:
        """
        Apply actions for object cache advice
        """
        return self.apply_actions(kwargs)

def rollback(self, **kwargs) -> tuple:
        """
        Rollback actions for <feature> advice
        """
        return self.rollback_actions(kwargs)

def apply_actions(self, data: dict) -> tuple:
        """
        Run commands in the given sequence of actions
        """
        data['ignore_errors'] = '--ignore-errors' if data.pop('ignore_errors',
                                                              False) else ''
        applied, result = False, ''
        try:
            cmds = self.action_cmds('apply', data)
        except ValueError as e:
            return False, str(e)
        for cmd in cmds:
            try:
                result, _ = self._exec_external(cmd)
            except ApplyAdviceException as e:
                result = str(e)
                break
        else:
            applied = True

return applied, result

def rollback_actions(self, data: dict) -> tuple:
        rollback, result = False, ''
        try:
            cmds = self.action_cmds('rollback', data)
        except ValueError as e:
            return False, str(e)
        for cmd in cmds:
            try:
                result, _ = self._exec_external(cmd)
            except RollbackAdviceException as e:
                result = str(e)
                break
        else:
            rollback = True

return rollback, result

def _exec_external(self, argv: list) -> Optional[str]:
        """
        Execute external command and return its output or error message
        """
        logger.debug('Attempt to run: %s', argv)

try:
            p = subprocess.run(argv, capture_output=True,
                               text=True, check=True,
                               env={'CL_WPOS_WAIT_CHILD_PROCESS': '1',
                                    'LANG': get_locale_from_envars()})
        except subprocess.CalledProcessError as e:
            logger.debug('non-zero returncode: %s; %s', e.stdout, e.stderr)
            raise ApplyAdviceException(e.stdout) from e
        except (OSError, ValueError, subprocess.SubprocessError) as e:
            logger.debug('cmd failed: %s', str(e))
            raise ApplyAdviceException(str(e)) from e
        else:
            logger.debug('Success! %s', p.stdout)
            return self._postprocess_exec_external(p.stdout)

@staticmethod
    def _postprocess_exec_external(output: str) -> Optional[Tuple[str, dict]]:
        """Check if 'result' == 'success', raise exception otherwise"""
        try:
            response = json.loads(output)
            if response['result'] != 'success':
                logger.debug('bad result from utility (result != success)')
                raise ApplyAdviceException(output)
        except (json.JSONDecodeError, KeyError) as e:
            logger.debug('malformed result from utility')
            raise ApplyAdviceException(output) from e
        else:
            return output, response

[+] __pycache__
[-] imgopt.py [edit]
[-] som.py [edit]
[-] wpos_base.py [edit]
[-] roc.py [edit]
[-] cdn.py [edit]
[-] __init__.py [edit]
[+] ..
[-] cpcss.py [edit]