EVOLUTION-MANAGER

Edit File: libcagefs.py

# coding:utf-8

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
from __future__ import print_function
from __future__ import division
from __future__ import absolute_import
import json
import os
import shutil
import subprocess
import sys
from past.builtins import basestring  # noqa

class CageFs(object):
    CAGEFS_SKELETON = "/usr/share/cagefs-skeleton/bin"
    CAGEFS_RESULTFILE = "/var/log/cagefs-update.log"
    CAGEFS_RESULTFILEUPD = "/var/log/cagefs-update.log"

COMMAND = '/usr/sbin/cagefsctl'

def status(self):
        try:
            status = self.run(['--cagefs-status']).strip()
        except OSError:
            status = 'not installed'
        mode = self.check_cagefs_mode()
        has_logs = os.path.exists(self.CAGEFS_RESULTFILE) or \
            os.path.exists(self.CAGEFS_RESULTFILEUPD)
        return {
            'skeleton': self.check_skeleteon(),
            'status': status,
            'mode': mode,
            'hasLogs': has_logs,
            'process': self.check_cagefsctl_exists()
        }

def init(self):
        self.run_background(['--init', '--do-not-ask', '--silent'])
        return self.check_cagefsctl_exists()

def update(self):
        self.run_background(['--force-update', '--do-not-ask', '--silent'])
        return self.check_cagefsctl_exists()

def check_skeleteon(self):
        return os.path.isdir(self.CAGEFS_SKELETON)

def run(self, params):
        if not params:
            params = ['--cagefs-status']
        params.insert(0, self.COMMAND)
        try:
            response = subprocess.check_output(params, stderr=subprocess.STDOUT,
                                               shell=False, text=True)
        except subprocess.CalledProcessError:
            response = 'disabled'
        return response

def run_background(self, params):
        command = '{} {}'.format(self.COMMAND, ' '.join(params))
        FNULL = open(os.devnull, 'w')
        subprocess.Popen(command,
                         stdin=FNULL,
                         stdout=FNULL,
                         stderr=FNULL,
                         shell=True,
                         executable='/bin/bash')

@staticmethod
    def check_cagefsctl_exists():
        command = ['/bin/bash',
                   '-c',
                   '/bin/ps aux | /bin/grep cagefsctl']
        response = subprocess.check_output(command, stderr=subprocess.STDOUT, text=True)
        for row in response.split("\n"):
            if '--init' in row or '--reinit' in row:
                return 'init'
            elif '--update' in row or '--force-update' in row:
                return 'update'
        return ''

def check_cagefs_mode(self):
        try:
            result = self.run(['--display-user-mode'])
        except OSError:
            return 'not installed'
        if 'CageFS is disabled' in result:
            return 'disabled'
        elif 'Mode: Enable All' in result:
            return 'enabled'
        else:
            return 'disabled'

def change_status(self, action, users):
        """
        Changes user's CageFS status. Supports performing
        action on multiple users.
        :param action: This can be either enable or disable
        :param users: list of users in the form of the array
        or the string separated by space to which action will be performed
        :return: Run class's run function and return result
        """
        if isinstance(users, basestring):
            users = users.split(' ')
        underscore_appended = '--' + action
        request_params = [underscore_appended] + users
        return self.run(request_params)

def _get_filename(self, operation):
        """
        Check file exists and return path
        :param operation: This can be either init or update
        :return: file path
        """
        log_file_name = self.CAGEFS_RESULTFILE if operation == 'init' else self.CAGEFS_RESULTFILEUPD
        if not os.path.isfile(log_file_name):
            self.exit_with_error("File {{filename}} does not exists", {'filename': log_file_name})
        if not os.access(log_file_name, os.R_OK):
            self.exit_with_error("File {{filename}} not available for reading", {'filename': log_file_name})
        return log_file_name

def get_log(self, operation):
        """
        Retrieves log files from the log file
        and returns to the UI
        :param operation: This can be either init or update
        and depending on the operation type corresponding log
        file will be used. If the currently cageFs is run with update
        or init flag then only last 100 lines of the log file will be returned
        :return: Iterate the lines by attaching number line and return result
        """
        log_file_name = self._get_filename(operation)
        in_progress = True if self.check_cagefsctl_exists() != '' else False
        result_log = []
        if os.path.exists(log_file_name):
            with open(log_file_name) as f:
                lines = f.readlines()
                result_log = lines[(-100 if in_progress else 0):]
        if not in_progress:
            result_log.append('Init complete' if operation == 'init' else 'Update complete')
        return result_log

def cagefs_log_data(self, operation):
        """
        Check log file exists and return filename to wrapper to seed
        """
        filepath = self._get_filename(operation)
        with open(filepath) as log_file:
            shutil.copyfileobj(log_file, sys.stdout, 20*1024)
        sys.exit(0)

def download_log(self, operation):
        """
        Check file exists and return information for downloading
        :param operation:
        :return:
        """
        log_file_name = self._get_filename(operation)
        print(json.dumps({
            "result": "file",
            "filepath": log_file_name,
            "filesize": os.path.getsize(log_file_name)
        }))
        sys.exit(0)

def exit_with_error(self, error_string, context):
        """
        Exit script with error for user
        :param error_string: error_text
        :return:
        """
        print(json.dumps({"result": error_string, 'context': context}))
        exit(1)