✏️ Editing: nginx_utils.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 utilities for ea-nginx cache disabling during tracing
"""
import json
import logging
import os.path
import subprocess
from dataclasses import dataclass
from enum import Enum
from typing import Optional

from xray import gettext as _

from .constants import nginx_cache_stat_storage, allow_disable_nginx_cache
from .exceptions import XRayError
from .utils import dbm_storage

logger = logging.getLogger('nginx_utils')

class Cache(Enum):
    """Cache statuses enum"""
    ENABLED = 1
    DISABLED = 0

@dataclass
class SavedState:
    """"""
    initial: Cache
    active_tasks: int = 0

@dataclass
class NginxUserCache:
    """
    Class which implements reading/writing the user's cache status
    from the local storage
    """
    username: str

@property
    def current_status(self) -> Cache:
        """
        Detect if cache is enabled or disabled for user
        using cPanel's guidelines
        https://docs.cpanel.net/knowledge-base/web-services/nginx-with-reverse-proxy/#third-party-integration
        """
        global_file = '/etc/nginx/ea-nginx/cache.json'
        user_file = f'/var/cpanel/userdata/{self.username}/nginx-cache.json'
        if not os.path.isfile(global_file):
            # this means that ea-nginx is not installed
            # thus cache status can be safety interpreted as DISABLED
            # as we do nothing in such a case
            return Cache.DISABLED

if os.path.isfile(user_file):
            # try to find cache setting in user's file
            user_setting = self.get_cache_status_from_file(user_file)
            if user_setting is not None:
                return Cache(user_setting)

global_setting = self.get_cache_status_from_file(global_file)
        if global_setting is not None:
            return Cache(global_setting)
        # cache is enabled by default
        return Cache.ENABLED

@property
    def is_enabled(self) -> bool:
        """Check if cache is enabled for user"""
        return self.current_status is Cache.ENABLED

@property
    def saved_state(self) -> SavedState:
        """
        Get saved state (initial cache setting and active tasks count)
        of nginx cache manipulations
        """
        try:
            with dbm_storage(nginx_cache_stat_storage,
                             is_shelve=True) as _nginx_cache_stat:
                return _nginx_cache_stat[self.username]
        except RuntimeError as e:
            logger.warning(f'Failed to get saved state of nginx cache with %s',
                           e)
        except KeyError:
            logger.info(f'No nginx cache entry found for {self.username}')
        # if failed to find entry in file or error occurred
        # consider initial state as DISABLED with no active tasks
        return SavedState(Cache.DISABLED)

@saved_state.setter
    def saved_state(self, value: SavedState) -> None:
        """
        Save state of nginx cache manipulations -- initial cache setting
        and active tasks count
        """
        try:
            with dbm_storage(nginx_cache_stat_storage,
                             is_shelve=True) as _nginx_cache_stat:
                _nginx_cache_stat[self.username] = value
        except RuntimeError as e:
            logger.warning('Failed to save state of nginx cache with %s', e)

@saved_state.deleter
    def saved_state(self) -> None:
        """
        Delete the record of state of nginx cache manipulations --
        initial cache setting and active tasks count
        """
        try:
            with dbm_storage(nginx_cache_stat_storage,
                             is_shelve=True) as _nginx_cache_stat:
                del _nginx_cache_stat[self.username]
        except RuntimeError as e:
            logger.warning('Failed to delete state of nginx cache with %s', e)

def set(self, enum_member: Cache) -> None:
        """Enable or disable cache for user"""
        cmd = ['/usr/local/cpanel/scripts/ea-nginx', 'cache',
               self.username, f'--enabled={enum_member.value}']
        try:
            subprocess.run(cmd, check=True, text=True,
                           capture_output=True)
        except subprocess.CalledProcessError as e:
            logger.info(
                'Failed command %s details: exitcode %i, stdout %s, stderr %s',
                e.args, e.returncode, e.stdout, e.stderr)
            raise XRayError(str(e), flag='warning',
                            extra={'out': e.stdout, 'err': e.stderr})
        except (OSError, ValueError, subprocess.SubprocessError) as e:
            raise XRayError(
                _('Failed to set nginx cache as {} with {}'.format(str(enum_member.name), str(e))),
                flag='warning')

@staticmethod
    def get_cache_status_from_file(filename: str) -> Optional[bool]:
        """Get the 'enabled' key from given file"""
        try:
            with open(filename) as _f:
                try:
                    data = json.load(_f)
                except json.JSONDecodeError as e:
                    logger.error('Invalid JSON from %s: %s',
                                 filename, str(e),
                                 extra={'contents': _f.read()})
                else:
                    return data.get('enabled')
        except (IOError, OSError) as e:
            logger.error('Failed to read file %s with %s',
                         filename, str(e))

def disable(self) -> None:
        """
        If cache is enabled for user, disable it.
        Save retrieved cache status.
        For ENABLED initial status increment number of active tasks.
        Skip disabling cache if it is disallowed in constants file
        """
        if not allow_disable_nginx_cache:
            # terminate operation if cache disabling is disallowed
            return

_state = self.saved_state

if self.is_enabled:
            try:
                self.set(Cache.DISABLED)
            except XRayError:
                pass
            else:
                _state.initial = Cache.ENABLED

if _state.initial is Cache.ENABLED:
            _state.active_tasks += 1

self.saved_state = _state

def restore(self) -> None:
        """
        For ENABLED initial status decrement number of active tasks.
        Restore the cache state according to saved status:
        - enable if it was enabled and number of active tasks <= 0,
        - do nothing if it was enabled, but number of active tasks > 0,
        - do nothing if it was disabled.
        Clear the record if enable has taken place.
        Skip operation if cache disabling is disallowed in constants file
        """
        _state = self.saved_state

if _state.initial is Cache.ENABLED:
            _state.active_tasks -= 1

if _state.active_tasks <= 0:
                try:
                    self.set(Cache.ENABLED)
                except XRayError:
                    pass
                del self.saved_state
                return

self.saved_state = _state

🔐 Sid Gifari File Manager Pro