✏️ Editing: cli_api.py Read Only

# -*- 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 X Ray Smart Advice local utility main class
"""
import json
import logging
import os
import hashlib
import pwd
import shlex
import subprocess
from contextlib import contextmanager
from dataclasses import asdict, dataclass
from typing import Any, Tuple, Optional, List, Dict
from enum import Enum

from xray.internal.clwpos_safe_imports import LicenseApproveStatus

from xray.internal.clwpos_safe_imports import (
    is_feature_allowed,
    is_subscription_pending,
    get_subscription_upgrade_url,
    get_license_approve_status,
    approve_license_agreement,
    get_license_agreement_text,
    get_subscriptions_info,
    get_user_auth_key,
    is_smart_advice_notifications_disabled_server_wide,
    is_smart_advice_reminders_disabled_server_wide,
    is_smart_advice_wordpress_plugin_disabled_server_wide
)
from clcommon.cpapi import (
    docroot,
    is_panel_feature_supported,
    get_user_emails_list,
    panel_awp_link,
    getCPName,
    userdomains
)
from clcommon.clwpos_lib import is_wp_path
from clcommon.clpwd import drop_privileges
from clcommon.const import Feature
from xray import gettext as _
from .advice_types import get_advice_instance
from .progress import SmartAdviceProgress
from ..apiclient import get_client, api_client
from ..internal.constants import (
    advice_pending_storage,
    advice_processed_storage,
    advice_list_cache,
    advice_reason_max_len
)
from ..internal.user_limits import get_lve_limits, get_lve_usage
from ..internal.exceptions import XRayError, SmartAdvicePluginError
from ..internal.user_plugin_utils import user_mode_advice_verification, username_verification, get_xray_exec_user
from ..internal.utils import timestamp, safe_move, get_user_php_version, filelock
from ..analytics.utils import report_usage_action_or_error
from xray.adviser.clwpos_get import ClWposGetter
from xray.smart_advice_plugin_helpers import (
    get_mu_directory,
    create_mu_plugins_dir_if_not_exist
)
from xray.smart_advice_plugin_manager import plugin_installed, prepare_system_analytics_data
from xray.analytics import report_analytics
from xray.imunify_manager import ImunifyManager, ADV_TYPE
from xray.internal.utils import validate_system_user

advice_cache_separator = ';'

# Server-wide Smart Advice disable flag (uses existing flags mechanism)
SMART_ADVICE_DISABLED_FLAG = '/opt/cloudlinux/flags/enabled-flags.d/xray-smart-advice-disabled.flag'

def _sa_disabled() -> bool:
    return os.path.exists(SMART_ADVICE_DISABLED_FLAG)

class AdviceActions(Enum):
    APPLY = 'apply'
    ROLLBACK = 'rollback'

@dataclass
class SmartAdviceOptions:
    panel_type: str
    panel_url: str
    panel_emails: str
    upgrade_url: str
    upgrade_url_cdn: str
    subscription: dict
    notifications: dict

class SmartAdviceUtil:
    """Main Smart Advice local utility class"""

def __init__(self):
        self.logger = logging.getLogger('smart_advice')
        # check existence of pending and apply storage and create it if missing
        for stor in (advice_pending_storage, advice_processed_storage):
            self.create_dir(stor)
        # initialize Adviser API client
        adviser_client_object = get_client('adviser')
        self.adviser_client: api_client.SmartAdviceAPIClient = adviser_client_object()
        self.imunify_manager = ImunifyManager()

@staticmethod
    def create_dir(dpath: str) -> None:
        """Create dir if missing"""
        try:
            os.mkdir(dpath)
        except FileExistsError:
            if not os.path.isdir(dpath):
                raise

@staticmethod
    def response(**kwargs) -> str:
        """
        Create JSON response message with result field == success
        and given keyword arguments in other fields
        :return: json packed string
        """
        initial = {'result': 'success', 'timestamp': timestamp()}
        if kwargs:
            initial.update(kwargs)
        return json.dumps(dict(sorted(initial.items())))

@staticmethod
    def _apply_datafile(a_id: int) -> str:
        """Per-advice file with results of apply"""
        # Validate a_id is a proper integer to prevent path traversal
        a_id = int(a_id)
        return f'{advice_processed_storage}/{a_id}'

def _apply_results(self, a_id: int) -> Optional[Any]:
        """Retrieve data stored in per-advice file with results of apply"""
        datafile = self._apply_datafile(a_id)
        # Open with O_NOFOLLOW to atomically reject symlinks (no TOCTOU gap)
        try:
            fd = os.open(datafile, os.O_RDONLY | os.O_NOFOLLOW)
        except FileNotFoundError:
            return None
        except OSError:
            # O_NOFOLLOW on a symlink raises ELOOP; other errors also handled
            try:
                os.unlink(datafile)
            except OSError:
                pass
            return None
        try:
            _f = os.fdopen(fd, 'r')
        except OSError:
            # os.fdopen failed before taking ownership — close fd manually
            try:
                os.close(fd)
            except OSError:
                pass
            data = None
        else:
            try:
                with _f:
                    data = json.load(_f)
            except (json.JSONDecodeError, OSError):
                data = None
        finally:
            try:
                os.unlink(datafile)
            except OSError:
                pass
        return data

@staticmethod
    def _pending_flag(a_id: int) -> str:
        """Per-advice pending flag"""
        # Validate a_id is a proper integer to prevent path traversal
        a_id = int(a_id)
        return f'{advice_pending_storage}/{a_id}'

def is_advice_pending(self, advice_id: int) -> bool:
        """Is advice in pending state"""
        return os.path.exists(self._pending_flag(advice_id))

def _progress_file(self, a_id: int) -> str:
        """Per-advice progress storage"""
        return self._pending_flag(a_id) + '.progress'

def _get_user_awp_link(self, username):
        try:
            return panel_awp_link(username)
        except Exception:
            self.logger.exception('Error while getting user login link')
            return ''

@user_mode_advice_verification
    def get_detailed_advice(self, advice_id: int) -> Tuple[dict, object]:
        """
        Get advice details from API along with an
        appropriate Advice object by obtained advice type
        """
        response = self.adviser_client.advice_details(advice_id)
        return response, get_advice_instance(response['advice']['type'])

@staticmethod
    def dump_to_file(dst: str, data: Any, as_json: bool = False) -> None:
        """Dump data inside given dst using .bkp file and then move"""
        _tmp_dst = dst + '.bkp'
        with open(_tmp_dst, 'w') as _f:
            if as_json:
                try:
                    json.dump(data, _f)
                except TypeError:
                    _f.write(data)
            else:
                _f.write(data)
        safe_move(_tmp_dst, dst)

def progress(self, advice_id: int, current: SmartAdviceProgress) -> SmartAdviceProgress:
        """
        Smart Advice apply own progress,
        based on current and previous results of 'clwpos get-progress'.
        Returns the maximum progress values among current and previous.
        Previous (latest) is stored inside a pending file-flag.
        """

@contextmanager
        def resolve_progress() -> SmartAdviceProgress:
            """
            Get progress stored in file and return the maximum one among
            current and stored results
            """
            self.logger.debug('Current progress value: %s', current)
            progress_dst = self._progress_file(advice_id)
            try:
                # read stored result
                with open(progress_dst) as prev:
                    prev_stages = SmartAdviceProgress(**json.load(prev))
            except (OSError, json.JSONDecodeError) as e:
                self.logger.debug('Error during reading stored value: %s',
                                  str(e))
                # or set a dummy one
                prev_stages = SmartAdviceProgress()

self.logger.debug('Stored progress value: %s', prev_stages)
            yield current if current > prev_stages else prev_stages

if self.is_advice_pending(advice_id) and current > prev_stages:
                # update stored result only if pending file exists
                self.logger.debug('Updating stored progress: %s', current)
                self.dump_to_file(progress_dst, asdict(current),
                                  as_json=True)

return resolve_progress()

def get_current_status(self, _id: int, advice_obj: object,
                           advice_data: dict) -> Tuple[str, dict]:
        """
        Resolve if advice is in pending status.
        Progress is retrieved for pending advice only.
        Dummy result (0, 0) is returned for other advice states.
        """
        if self.is_advice_pending(_id):
            status = 'pending'
            with self.progress(_id, advice_obj.get_progress(
                    advice_data['metadata']['username'])) as p:
                stages = p
        else:
            status = advice_data['advice']['status']
            stages = SmartAdviceProgress()
        return status, asdict(stages)

def advice_list(self, extended=False) -> str:
        """Load validated advice list and update it before returning"""

# If SA is disabled server-wide, return an empty list without any side effects
        if _sa_disabled():
            return self.response(data=[])

api_response = self.adviser_client.advice_list()

response_advice_list = self.prepare_advices_response(api_response, extended=extended)

# cl-smart-advice plugin syncing should not break whole command
        try:
            exec_user = get_xray_exec_user()
            if exec_user:
                # Sync only for the current user and their sites with at least one non-outdated advice
                # or where MU plugin is already installed (to uninstall it if needed)
                all_user_sites = set()
                sites_with_active_advices = set()

for item in api_response:
                    meta = item.get('metadata', {})
                    advice = item.get('advice', {})
                    if meta.get('username') != exec_user:
                        continue
                    key = (meta.get('domain'), meta.get('website'))
                    all_user_sites.add(key)
                    if advice.get('status') != 'outdated':
                        sites_with_active_advices.add(key)

# Add sites where MU plugin is already installed
                sites_with_plugin = set()
                for domain, website in (all_user_sites - sites_with_active_advices):
                    full_website_path = docroot(domain)[0] + website
                    mu_plugin_dir = get_mu_directory(exec_user, full_website_path)
                    if mu_plugin_dir and plugin_installed(mu_plugin_dir):
                        sites_with_plugin.add((domain, website))

sites_to_sync = sites_with_active_advices | sites_with_plugin

for domain, website in sites_to_sync:
                    self._sync_advice(exec_user, domain, website)
            # If exec_user is None — do nothing (admin mode).
        except Exception:
            self.logger.exception('Unable to sync cl-smart-advice plugin during getting list of advices')

return self.response(data=response_advice_list)

@username_verification
    def get_site_statuses(self, username) -> str:
        api_response = self.adviser_client.site_info(username)
        site_statuses_data = {}
        for website in api_response:
            domain = website['domain']
            if domain not in site_statuses_data:
                site_statuses_data[domain] = []
            site_statuses_data[domain].append({
                'website': website['website'],
                'urls_scanned': len(website['urls']),
                'advices': self.prepare_advices_response(website['advices'])
            })
        result = [{'domain': domain, 'websites': websites} for domain, websites in site_statuses_data.items()]
        return self.response(data=result)

def prepare_advices_response(self, from_api, status_from_microservice=False, extended=False):
        advices = from_api.copy()

visible_advices = []
        for item in advices:
            # TODO: would be great to rewrite this part and avoid
            #       changing array inplace, instead create new one with proper typing
            advice_itself = item['advice']
            advice_type = advice_itself['type']

# It's only for smart advice wp plugin
            if advice_type.startswith(ADV_TYPE):
                visible_advices.append(item)
                continue

advice_instance = get_advice_instance(advice_type)
            advice_itself['description'] = advice_instance.short_description

# when we need real response from microservice - only before sending json to wp plugin
            if status_from_microservice:
                status = advice_itself['status']
                stages = {'total_stages': 0, 'completed_stages': 0}
            else:
                status, stages = self.get_current_status(advice_itself['id'],
                                                         advice_instance, item)
            if extended:
                advice_itself['detailed_description'] = advice_instance.detailed_description

advice_itself['status'] = status
            advice_itself['is_premium'] = advice_instance.is_premium_feature
            advice_itself['module_name'] = advice_instance.module_name

advice_itself['license_status'] = get_license_approve_status(
                advice_instance.module_name, item['metadata']['username']).name

if advice_itself['is_premium']:
                subscription_status = self._get_subscription_status(
                    advice_instance.module_name, item['metadata']['username'])
                advice_itself['subscription'] = dict(
                    status=subscription_status,
                    upgrade_url=get_subscription_upgrade_url(
                        advice_instance.module_name, item['metadata']['username'])
                )
            advice_itself.update(stages)

visible_advices.append(item)
        return visible_advices

def filter_plugin_advices(self, username, website, domain, advice_list):
        return [advice for advice in advice_list
                if advice.get('metadata', {}).get('domain') == domain
                and advice.get('metadata', {}).get('username') == username
                and advice.get('metadata', {}).get('website') == website
                and advice.get('advice', {}).get('status') != 'outdated']

def _get_subscription_status(self, module_name, username):
        """
        Determines current subscription status based on feature status.
        """
        subscription_status = 'no'
        if is_feature_allowed(module_name, username):
            subscription_status = 'active'
        if is_subscription_pending(module_name, username):
            subscription_status = 'pending'
        return subscription_status

def advice_details(self, advice_id: int) -> str:
        """Load validated advice details and update it before returning"""
        api_response, advice_instance = self.get_detailed_advice(advice_id)

advice_itself = api_response['advice']

"""
        The microservice responds with field names: title and description.
        We must assign:
        [local advice] detailed_description = [microservice] description
        [local advice] description          = [microservice] title
        """

# Until we won't rename detailed_description to description
        if advice_itself['description']:
            advice_itself['detailed_description'] = advice_itself['description']
        else:
            advice_itself['detailed_description'] = advice_instance.detailed_description

# Until we won't rename description to title
        if advice_itself['title']:
            advice_itself['description'] = advice_itself['title']
        else:
            advice_itself['description'] = advice_instance.short_description
        del advice_itself['title']

return self.response(data=api_response)

def manage_advice(self, action, advice_id: int, ignore_errors: bool = False,
                      async_mode: bool = False, source: str = 'ACCELERATE_WP',
                      reason: str = None, accept_terms: bool = False,
                      analytics_data: str = None) -> Optional[str]:

open(self._pending_flag(advice_id), 'w').close()

if async_mode:
            # put itself to background in async mode
            # and return current status
            child = os.fork()
            if child:
                return self.manage_advice_status(advice_id)
        output = self._exec_advice_managing(action, advice_id, ignore_errors, async_mode,
                                            source, reason, accept_terms)
        if analytics_data is not None:
            report_usage_action_or_error(analytics_data, advice_id, source, output, action)
        if async_mode:
            # retrun no output in async mode
            output = None
        return output

def _exec_advice_managing(self, action: str, advice_id: int, ignore_errors: bool = False,
                              async_mode: bool = False, source: str = 'ACCELERATE_WP',
                              reason: str = None, accept_terms: bool = False) -> Optional[str]:
        """Execute managing advice with passed action: apply/rollback"""
        try:
            api_response, advice_instance = self.get_detailed_advice(advice_id)

if action == AdviceActions.APPLY.value:
                if accept_terms:
                    approve_license_agreement(
                        advice_instance.module_name, api_response['metadata']['username'])

if get_license_approve_status(
                        advice_instance.module_name,
                        api_response['metadata']['username']) == LicenseApproveStatus.NOT_APPROVED:
                    return json.dumps({
                        'result': 'LICENCE_TERMS_APPROVE_REQUIRED',
                        'text': _('License approve required to use this feature. '
                                  'Open AccelerateWP plugin in your control panel, apply advice and '
                                  'accept terms and conditions to proceed with installation.')
                    })
                action_result, output = advice_instance.apply(**api_response['metadata'],
                                                              ignore_errors=ignore_errors)
                if action_result:
                    self.logger.debug('Applied successfully')
                    self.adviser_client.update_advice(advice_id=advice_id,
                                                      status='applied',
                                                      source=source)

try:
                        self._sync_advice(api_response['metadata']['username'],
                                          api_response['metadata']['domain'],
                                          api_response['metadata']['website'])
                    except Exception:
                        self.logger.exception('Error while syncing cl-smart-advice plugin on advice apply')

elif action == AdviceActions.ROLLBACK.value:
                action_result, output = advice_instance.rollback(**api_response['metadata'])
                if action_result:
                    self.logger.debug('Rollback successfully')
                    if reason:
                        self.adviser_client.update_advice(advice_id=advice_id,
                                                          status='review',
                                                          source=source,
                                                          reason=reason[:advice_reason_max_len])
                    else:
                        self.adviser_client.update_advice(advice_id=advice_id,
                                                          status='review',
                                                          source=source)

finally:
            if os.path.isfile(self._pending_flag(advice_id)):
                os.unlink(self._pending_flag(advice_id))

if async_mode:
            # in async mode write result to a special file
            self.dump_to_file(self._apply_datafile(advice_id),
                              output)
            # return output for analytics will be cleared after reporting
            return output
        else:
            # in sync mode return output
            return output

def advice_counters(self) -> str:
        """Return advice counters for a server"""
        try:
            api_response = self.adviser_client.advice_list()
        except XRayError:
            # bad API responses (500 for example)
            # or JWT token failed check (CL Shared for example)
            result = dict.fromkeys(['total', 'applied'], None)
        else:
            result = dict(
                total=len(api_response),
                applied=len([advice for advice in api_response if
                             advice['advice']['status'] == 'applied'])
            )
        return self.response(data=result)

def manage_advice_status(self, advice_id: int) -> str:
        """
        Return current status and progress of managing particular advice.
        """
        api_response, advice_instance = self.get_detailed_advice(advice_id)
        status, stages = self.get_current_status(advice_id, advice_instance,
                                                 api_response)

subscription_status = self._get_subscription_status(
            advice_instance.module_name, api_response['metadata']['username'])

if status != 'pending':
            # include result of apply/rollback for non-pending advice only
            result = self._apply_results(advice_id)
            # drop stored progress
            _progress = self._progress_file(advice_id)
            if os.path.isfile(_progress):
                os.unlink(_progress)

if result:
                return self.response(
                    status=status,
                    subscription=dict(
                        object_cache=subscription_status
                    ),
                    upgrade_url=get_subscription_upgrade_url(
                        advice_instance.module_name,
                        api_response['metadata']['username']),
                    **stages,
                    data=result)

return self.response(
            status=status,
            subscription=dict(
                object_cache=subscription_status
            ),
            upgrade_url=get_subscription_upgrade_url(
                advice_instance.module_name,
                api_response['metadata']['username']),
            **stages
        )

def update_advices_metadata(self):
        """
        Updates incompatibilities only, but could be extended to more magic
        how to make microservice know that users/domains/websites no longer exist
        """
        if _sa_disabled():
            return self.response(data=_('Smart Advice is disabled server-wide'))
        current_advices = self.adviser_client.advice_list(filtered=False, show_all=True)
        getter = ClWposGetter()

user_domain_pairs = {}

for advice_object in current_advices:
            current_adv_metadata = advice_object.get('metadata')
            advice_data = advice_object.get('advice') or {}

if not current_adv_metadata:
                self.logger.error('Malformed advice metadata, does not have metadata key: %s',
                                  str(advice_object))
                continue

# Skip outdated advices entirely for metadata update
            if advice_data.get('status') == 'outdated':
                continue

username, domain = current_adv_metadata.get('username'), current_adv_metadata.get('domain')

if not all([domain, username]):
                self.logger.error('Malformed advice metadata, does not have required username or domain field: %s',
                                  str(current_adv_metadata))
                continue

user_domain_pairs[username] = domain

# If there are no valid (non-outdated) pairs, return early
        if not user_domain_pairs:
            return self.response(data=_('No valid (non-outdated) advices to update'))

for user, user_domain in user_domain_pairs.items():
            try:
                updated_metadata = getter.get_updated_extended_metadata(user, user_domain, current_advices)

if not updated_metadata:
                    logging.warning('Updated metadata for user: %s will not be sent', str(user))
                    continue

getter.send(updated_metadata)
            except Exception as e:
                logging.error('Failed to update advice metadata with Smart Advice: %s',
                              e,
                              extra={
                                  'fingerprint': e,
                              })
                continue

return self.response(data=_('Advices metadata update finished'))

def _sync_advice(self, username, domain, website):
        current_advices = self.adviser_client.advice_list()

# Get IM360 advice cache
        im360_advice_list = self.imunify_manager.im360_advice(cache_only=True)
        current_advices.extend(im360_advice_list)

advice_hash_map = self.get_advice_hash_map(current_advices)
        try:
            filtered_advice_list = self.prepare_advices_response(current_advices,
                                                                 status_from_microservice=True, extended=True)
            filtered_advice_list = self.filter_plugin_advices(username, website, domain, filtered_advice_list)
            self._run_smart_advice_script(username, website, domain, filtered_advice_list)
        except Exception:
            self.logger.exception('Smart Advice plugin sync failed for website: %s', str(website))
            return

cached_advices_by_site = self._read_advices_cache()
        key = f'{domain}{advice_cache_separator}{username}{advice_cache_separator}{website}'
        if key not in cached_advices_by_site:
            cached_advices_by_site[key] = {}
        cached_advices_by_site[key] = advice_hash_map[key]
        self._update_advices_cache(cached_advices_by_site)

def _get_cdn_usage_statistics(self, username):
        from xray.adviser.awp_provision_api import AWPProvisionAPI
        acc_id = get_user_auth_key(username)
        cdn_usage = AWPProvisionAPI().awp_client_api.get_usage(acc_id)
        return cdn_usage

@username_verification
    def analytics_report(self, username, feature, source, event, advice_id=None,
                         journey_id=None, user_hash=None, variant_id=None):
        analytics_data = prepare_system_analytics_data(username, user_hash, journey_id)
        report_analytics(data=analytics_data,
                         advice_id=advice_id,
                         source=source.lower(),
                         event=event,
                         feature=feature,
                         variant_id=variant_id)
        return self.response(data=_('Analytics sent'))

@username_verification
    def get_options(self, username):
        """
        gets options:
        "panel_type": "cpanel",
        "panel_url": "https:/cpanel-foo.com/", # or https://plesk-bar.com/
        "panel_emails": "root@hosting.com,manager@hostring.com,foo@bar.com",
        "upgrade_url": "https://...",
        "upgrade_url_cdn": "https://...",
        "subscription": dict,
        "notifications": dict,
        """
        try:
            # [("domain.com"),("/path/to/docroot")]
            domain = userdomains(username)[0][0]
            email = get_user_emails_list(username, domain)
        except Exception:
            self.logger.exception('Unable to get user emails')
            email = ''
        try:
            subscription = get_subscriptions_info(username)
        except Exception:
            self.logger.exception('Cannot obtain subscription info')
            subscription = {}

try:
            im360_advice_notification = self.imunify_manager.get_im360_settings().get('advice_email_notification')
        except Exception:
            self.logger.exception('Cannot obtain im360 advice notification setting')
            im360_advice_notification = False

try:
            protection_status = 'active' if self.imunify_manager.get_im360_protection_status(username) is True else 'no'
        except Exception:
            self.logger.exception('Cannot obtain im360 protection status')
            protection_status = 'no'

subscription['protection'] = {'status': protection_status}

notifications = dict(
            email_status="disabled_server" if is_smart_advice_notifications_disabled_server_wide() is True else "enabled",
            reminders_status="disabled_server" if is_smart_advice_reminders_disabled_server_wide() is True else "enabled",
            imunify_email_status="enabled" if im360_advice_notification is True else "disabled_server",
        )

return self.response(**asdict(
            SmartAdviceOptions(panel_type=getCPName(),
                               panel_url=self._get_user_awp_link(username),
                               panel_emails=email,
                               upgrade_url=get_subscription_upgrade_url('object_cache', username),
                               upgrade_url_cdn=get_subscription_upgrade_url('cdn', username),
                               subscription=subscription,
                               notifications=notifications)
        ))

@username_verification
    def get_limits(self, username):
        """
        Retrieves the LVE limits for a specified user.
        {
            "lve_cpu": {
                "limit": int
            },
            "lve_ep": {
                "limit": int
            },
            "lve_pmem": {
                "limit": int
            },
            "lve_iops": {
                "limit": int
            },
            "lve_io": {
                "limit": int
            },
            "lve_nproc": {
                "limit": int
            },
        }
        or
        {
            "error": str
        }
        """
        result = get_lve_limits(username)
        return json.dumps(result, indent=2)

@username_verification
    def get_usage(self, username):
        """
        Retrieves the current usage for a specified user.
        {
            "lve_cpu": {
                "usage": int
            },
            "lve_ep": {
                "usage": int
            },
            "lve_pmem": {
                "usage": int
            },
            "lve_iops": {
                "usage": int
            },
            "lve_io": {
                "usage": int
            },
            "lve_nproc": {
                "usage": int
            },
        }
        or
        {
            "error": str
        }
        """
        result = get_lve_usage(username)
        return json.dumps(result, indent=2)

def sync_advices_wordpress_plugin(self, current_advices=None, im360_cached=True):
        if _sa_disabled():
            return self.response(data=_('Smart Advice is disabled server-wide'))

if not current_advices:
            current_advices = self.adviser_client.advice_list()

# Get IM360 advice fresh data by cron, or return cache
        im360_advice_list = self.imunify_manager.im360_advice(cache_only=im360_cached)
        current_advices.extend(im360_advice_list)

new_hash_cache_by_site = self.get_advice_hash_map(current_advices)
        existing_hash_cache_by_site = self._read_advices_cache()

if not new_hash_cache_by_site and not existing_hash_cache_by_site:
            return self.response(data='No advices found on the server')

ClWposGetter().save_clwpos_info()
        updated_cache = self._sync_sa_plugin(
            new_hash_cache_by_site,
            existing_hash_cache_by_site,
            current_advices_json=self.prepare_advices_response(current_advices, extended=True)
        )
        self._update_advices_cache(updated_cache)
        return self.response(data='Plugin installed')

def uninstall_wordpress_plugins(self):
        """
        Uninstalls all Smart Advice WordPress plugins on server.
        """
        existing_hash_cache_by_site = self._read_advices_cache()

if not existing_hash_cache_by_site:
            return self.response(data=_('No plugins found on the server'))

ClWposGetter().save_clwpos_info()

for key in existing_hash_cache_by_site:
            (domain, username, website) = key.split(advice_cache_separator)

try:
                self._run_smart_advice_script(username, website, domain, advices_json=None)
            except Exception:
                self.logger.exception('Smart advice plugin removal failed for website: %s', str(website))
                continue

self._update_advices_cache(new_cached_advices={})
        return self.response(data=_('Plugin uninstalled'))

def _sync_sa_plugin(self, new_hash_advice_cache, existing_hash_advice_cache, current_advices_json=None):
        logging.info('Synchronization of smart advice plugin begin')

updated_advices = {}

for key, id_status_hash in new_hash_advice_cache.items():
            (domain, username, website) = key.split(advice_cache_separator)
            if not self._should_sync_website(domain, username, website, id_status_hash,
                                             existing_hash_advice_cache.get(key)):
                logging.debug('Website %s does not need update because cache hash matches', key)
                updated_advices[key] = id_status_hash
                continue

# keep only current user/website advices
            if current_advices_json:
                filtered_advice_list = self.filter_plugin_advices(username, website, domain, current_advices_json)
            else:
                filtered_advice_list = None

# Early exit: if there are no relevant advices and MU plugin is not installed, skip heavy calls
            full_website_path = docroot(domain)[0] + website
            mu_plugin_dir = get_mu_directory(username, full_website_path)
            if not filtered_advice_list and mu_plugin_dir and not plugin_installed(mu_plugin_dir):
                updated_advices[key] = id_status_hash
                continue

try:
                self._run_smart_advice_script(username, website, domain, filtered_advice_list)
            except Exception:
                self.logger.exception('Smart advice plugin sync failed for website: %s', str(website))
                continue

updated_advices[key] = id_status_hash

# keep tracking of requests that no longer exist in new cache
        orphan_hash_records = set(existing_hash_advice_cache) - set(new_hash_advice_cache)
        if orphan_hash_records:
            # loop over orphan records and remove plugins on those websites
            logging.info('Found outdated records in cache, '
                         'removing wordpress plugins for those websites')
            for key in orphan_hash_records:
                (domain, username, website) = key.split(advice_cache_separator)

logging.info('Removing outdated cache record %s', key)

try:
                    self._run_smart_advice_script(username, website, domain, advices_json=None)
                except Exception:
                    self.logger.exception('Smart advice plugin sync failed for website: %s', str(website))
                    continue
        return updated_advices

@staticmethod
    def _update_advices_cache(new_cached_advices: dict):
        with open(f'{advice_list_cache}.lock', 'a+') as lock_fd, \
                filelock(lock_fd), \
                open(advice_list_cache, 'w') as f:
            logging.debug('Updating advice hash map cache with values %s', new_cached_advices)
            json.dump(new_cached_advices, f, indent=2)

def _read_advices_cache(self) -> dict:
        cached_advices_by_site = {}
        if os.path.exists(advice_list_cache):
            try:
                with open(advice_list_cache) as f:
                    cached_advices_by_site = json.load(f)
            except Exception:
                self.logger.exception('Unable to read advices cache json')
                cached_advices_by_site = {}
        return cached_advices_by_site

@staticmethod
    def _should_sync_website(domain, username, website, current_hash, cached_hash):
        if current_hash != cached_hash:
            return True

full_website_path = docroot(domain)[0] + website
        mu_plugin_dir = get_mu_directory(username, full_website_path)

if not plugin_installed(mu_plugin_dir):
            return True

return False

@staticmethod
    def make_hash(id_status_pairs):
        h = hashlib.md5()
        for advice_id, status in id_status_pairs:
            h.update(str(advice_id).encode())
            h.update(status.encode())
        return h.hexdigest()

def get_advice_hash_map(self, advices) -> Dict[str, str]:
        """
        Prepare fingerprints for each of the advices generated,
        grouped by the website that advice is assigned to.

Result example:
        {
          "demo-10-194-0-9.traefik.me;demowp;/": "f43c7bcdccc28fc1ac92b2384f35f9dd"
        }

Key is combination of domain, usernane and webiste path and
        value is hash against list of advices and their current statuses.
        """
        websites = {}
        for advice in advices:
            metadata = advice['metadata']
            advice_data = advice['advice']

key = f'{metadata["domain"]}{advice_cache_separator}{metadata["username"]}' \
                  f'{advice_cache_separator}{metadata["website"]}'
            if key not in websites:
                websites[key] = []
            websites[key].append((advice_data['id'], advice_data['status']))

return {
            key: self.make_hash(tuple(sorted(id_status_pairs, key=lambda item: str(item[0]))))
            for key, id_status_pairs in websites.items()
        }

@staticmethod
    def _save_custom_php(path, content):
        """
        For easy mocking
        """
        fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW, 0o600)
        with os.fdopen(fd, 'w') as f:
            f.write(content)

def _prepare_php_for_plesk(self, username, domain):
        clwpos_dir = f'{pwd.getpwnam(username).pw_dir}/.clwpos'
        custom_php = '.php-ini'
        custom_php_full = os.path.join(clwpos_dir, custom_php)
        extra_paths = [
            '/opt/alt/php-xray/',
            '/opt/cloudlinux-site-optimization-module/',
            '/opt/cloudlinux/'
        ]
        updated_setting = None

plesk_result = subprocess.run(
            ['/usr/sbin/plesk', 'bin', 'site', '--show-php-settings', domain],
            text=True, capture_output=True)
        if plesk_result.returncode != 0:
            self.logger.error('Plesk --show-php-settings failed for %s: %s',
                              domain, plesk_result.stderr)
            return
        open_basedir_setting = subprocess.run(
            ['/usr/bin/grep', 'open_basedir'],
            input=plesk_result.stdout,
            text=True, capture_output=True)

open_basedir_setting_sanitized = open_basedir_setting.stdout.replace('open_basedir =', '').replace(
            'open_basedir=', '').strip()
        self.logger.info('Current open_basedir settings: %s', str(open_basedir_setting.stdout))

# empty string means no restrictions; no need to add the extra paths
        if len(open_basedir_setting_sanitized) == 0:
            updated_setting = 'open_basedir ='
        # non-empty string and not "none"
        elif len(open_basedir_setting_sanitized) > 0 and open_basedir_setting_sanitized != 'none':
            # if 'none' is in the list of values obtained by splitting the string
            if 'none' in open_basedir_setting_sanitized.split(':'):
                updated_setting = 'open_basedir = none'
            elif open_basedir_setting_sanitized.startswith(':'):
                updated_setting = 'open_basedir = {WEBSPACEROOT}{/}{:}{TMP}{/}:' + ":".join(extra_paths)
            else:
                # collect missing extra paths
                missing_paths = []
                for extra_path in extra_paths:
                    # extra path not present
                    if extra_path not in open_basedir_setting.stdout:
                        missing_paths.append(extra_path)
                # some extra paths are missing, settings update is required
                if missing_paths:
                    updated_setting = f'open_basedir = {open_basedir_setting_sanitized}:{":".join(missing_paths)}'

if updated_setting:
            self.logger.info('Updating open_basedir setting: %s for domain: %s', updated_setting, domain)
            # Both folder for custom PHP ini file and the file itself should be owned by appropriate user. We don't want
            # to leave files owned by root in user's home directory.
            with drop_privileges(username):
                if not os.path.exists(custom_php_full):
                    if not os.path.isdir(clwpos_dir):
                        os.mkdir(clwpos_dir, mode=0o700)
                self._save_custom_php(custom_php_full, updated_setting)

result = subprocess.run(['/usr/sbin/plesk', 'bin', 'site', '--update-php-settings', domain, '-settings',
                                     custom_php_full],
                                    text=True, capture_output=True)
            if result.returncode != 0:
                self.logger.error('Plesk --update-php-settings failed for %s: %s',
                                  domain, result.stderr)
                return
            self.logger.info('Plesk settings update result: %s', str(result.stdout))
            self.logger.error('open_basedir setting updated for domain "%s" from "%s" to "%s"',
                              str(domain),
                              str(open_basedir_setting.stdout),
                              str(updated_setting))

def _uninstall_wordpress_plugin(self, username, website, domain):
        """
        Uninstall WordPress SmartAdvice plugin.
        """
        try:
            json_string = self.response(data=[])
            attrs = self._prepare_smart_advice_script_args(username, website, domain, json_string, 'true')
            command = self._get_smart_advice_script_cmd(username, attrs)
        except ValueError:
            # fixme: why?
            return

logging.debug('Uninstalling Smart Advice plugin for website %s with advices: %s',
                      website, json_string)
        result = subprocess.run(command, text=True, capture_output=True)
        if result.returncode != 0:
            self.logger.error('Smart advice plugin uninstalling failed with exit code: %s, stdout: %s, stderr: %s',
                              str(result.returncode),
                              str(result.stdout),
                              str(result.stderr),
                              extra={
                                  'fingerprint': f'smart-advice-plugin-sh-returncode-{result.returncode}',
                              })
            raise SmartAdvicePluginError(_('Smart Advice plugin uninstalling failed'))

def _install_wordpress_plugin(self, username, website, domain, advices_json):
        """
        Install plugin and sync advice
        plugin creation is part of main process -> to handle quota exceeded
        """
        try:
            json_string = self.response(data=advices_json)
            attrs = self._prepare_smart_advice_script_args(username, website, domain, json_string, 'false')
            command = self._get_smart_advice_script_cmd(username, attrs)
        except ValueError:
            return

logging.info('Updating Smart Advice plugin for website %s with advices: %s',
                     website, json_string)
        result = subprocess.run(command, text=True, capture_output=True)
        if result.returncode != 0:
            self.logger.error('Smart advice plugin failed with exit code: %s, stdout: %s, stderr: %s',
                              str(result.returncode),
                              str(result.stdout),
                              str(result.stderr),
                              extra={
                                  'fingerprint': f'smart-advice-plugin-sh-returncode-{result.returncode}',
                              })
            raise SmartAdvicePluginError(_('Smart Advice plugin installation failed'))
        self.logger.info('Smart Advice plugin is installed successfully: output=%s',
                         str(result.stdout))

def _should_wordpress_plugin_installed(self):
        is_smart_advice_disabled = is_smart_advice_wordpress_plugin_disabled_server_wide()
        is_im360_mu_disabled = self.imunify_manager.is_im360_mu_plugin_disabled_server_wide()
        self.logger.info("Checking whether to install MU plugin: "
                         "is_smart_advice_disabled=%s, "
                         "is_im360_mu_disabled=%s",
                         str(is_smart_advice_disabled),
                         str(is_im360_mu_disabled))
        return not is_smart_advice_disabled or not is_im360_mu_disabled

def _run_smart_advice_script(self, username, website, domain, advices_json=None):
        def _filter_by_advice_type(advices_data):
            if not advices_data:
                return advices_data

filtered_data = advices_data
            if is_smart_advice_wordpress_plugin_disabled_server_wide():
                # leave only im360 advices
                filtered_data = [item for item in filtered_data if item.get("advice", {}).get("type") == ADV_TYPE]
                self.logger.info("AccelerateWP advices are filtered, advices=%s", str(filtered_data))

if self.imunify_manager.is_im360_mu_plugin_disabled_server_wide():
                # leave only non-im360 advices
                filtered_data = [item for item in filtered_data if item.get("advice", {}).get("type") != ADV_TYPE]
                self.logger.info("IM360 advices are filtered, advices=%s", str(filtered_data))
            self.logger.info("Filtered advices for Smart Advice plugin, advices=%s", str(filtered_data))
            return filtered_data

filtered = _filter_by_advice_type(advices_json)
        # Uninstall when none or empty list
        if not filtered:
            self._uninstall_wordpress_plugin(username, website, domain)
        # install only if wordpress plugin feature is enabled by hoster
        elif self._should_wordpress_plugin_installed():
            self.logger.info("Smart Advice plugin installation is allowed, advices list=%s",
                             str(advices_json))
            self._install_wordpress_plugin(username, website, domain, filtered)

def _prepare_smart_advice_script_args(self, username, website, domain: List[str], advices_json: str,
                                          uninstall='false') -> List[str]:
        panel = getCPName()
        script_name = 'run.py'
        script_path = f'/opt/alt/php-xray/php/smart-advice-plugin/{script_name}'
        panel_user_awp_link = self._get_user_awp_link(username)

self.logger.info('Got panel login link: %s', panel_user_awp_link)

try:
            email = get_user_emails_list(username, domain)
        except Exception:
            self.logger.exception('Unable to get user emails')
            email = ''

full_website_path = docroot(domain)[0] + website

if not is_wp_path(full_website_path):
            self.logger.info('[Smart Advice Plugin]: wordpress: %s does not exist, skipped',
                             full_website_path)
            raise ValueError(_('Non-existing wordpress site'))

php_info = get_user_php_version(username)

for php_data in php_info:
            if php_data.get('vhost') == domain:
                php_binary_path = php_data.get('php_binary')
                break
        else:
            self.logger.error('Php data for domain %s was not found in %s', domain, str(php_info))
            raise SmartAdvicePluginError(_('Unable to get php version for domain %s') % domain)

if not php_binary_path:
            self.logger.error('Php version for user: %s was not obtained',
                              username)
            raise SmartAdvicePluginError(_('Php binary was not identified for domain %s') % domain)

if panel == 'Plesk':
            try:
                self._prepare_php_for_plesk(username, domain)
            except Exception:
                self.logger.exception('Setting up php for Plesk for SA plugin failed')

mu_plugin_dir = get_mu_directory(username, full_website_path)

if not mu_plugin_dir:
            raise SmartAdvicePluginError(_('Unable to detect MU plugin directory'))

# mu_plugins dir creation is part of main process -> to handle quota exceeded
        create_mu_plugins_dir_if_not_exist(username, mu_plugin_dir)

# ATTENTION: this order is important in further code, add new atts with caution
        res = [
            script_path,
            php_binary_path,
            full_website_path,
            mu_plugin_dir,
            advices_json,
            email,
            panel_user_awp_link,
            uninstall,
        ]
        return res

@staticmethod
    def _get_smart_advice_script_cmd(username: str, attrs: List[str]) -> List[str]:
        validate_system_user(username)
        if is_panel_feature_supported(Feature.CAGEFS):
            res = ['/sbin/cagefs_enter_user', username]
            res.extend(attrs)
        else:
            # Use shlex.quote on all arguments to prevent shell injection
            command_str = ' '.join(shlex.quote(str(a)) for a in attrs)
            res = ['sudo', '-u', username, '-s', '/bin/bash', '-c', command_str]
        return res

def get_agreement_text(self, feature):
        """
        Method with handles work with agreements.
        """
        user_context = get_xray_exec_user()
        if not user_context:
            raise SmartAdvicePluginError(_('This command can only be executed as user'))

return json.dumps({
            'result': 'success',
            'timestamp': timestamp(),
            'text': get_license_agreement_text(feature, user_context)
        })

def create_subscription(self, advice_id):
        """
        Proxy method to AWP which creates subscription
        listening for the status of module and automatic advice appy.
        """
        user_context = get_xray_exec_user()
        if not user_context:
            raise SmartAdvicePluginError(_('This command can only be executed as user'))

resp, advice_instance = self.get_detailed_advice(advice_id)

result = subprocess.run([
            'cloudlinux-awp-user',
            '--user', str(user_context),
            'subscription',
            '--listen',
            '--advice-id', str(advice_id),
            '--feature', advice_instance.module_name
        ], text=True, capture_output=True)

if result.returncode != 0:
            self.logger.error('Create subscription smart advice plugin failed with exit code: '
                              '%s, stdout: %s, stderr: %s',
                              str(result.returncode),
                              str(result.stdout),
                              str(result.stderr),
                              extra={
                                  'fingerprint': f'smart-advice-plugin-subscription-returncode-{result.returncode}',
                              })
            raise SmartAdvicePluginError(_('cloudlinux-awp-user call failed %s') % result.stderr)

🔐 Sid Gifari File Manager Pro