# 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' # Allowlist of cagefsctl flags accepted from external callers. # The raw-passthrough branch in cloudlinux_cli.py:cagefsctl falls into # CageFs.run with caller-supplied argv when ``method`` is absent; this # set restricts that branch to flags the admin SPA legitimately needs. # Internal callers in this module use --cagefs-status / --display-user-mode # / --enable / --disable (via change_status), all listed here. The admin # SPA cagefs-plugin component sends --enable-cagefs / --disable-cagefs / # --toggle-mode through cageFsRun(); those server-wide toggles must # remain reachable, so they are part of the allowlist too. _ALLOWED_FLAGS = frozenset(( '--cagefs-status', '--display-user-mode', '--enable', '--disable', '--enable-cagefs', '--disable-cagefs', '--toggle-mode', '--remount-all', '--force-update', '--update-etc', '--list-enabled', '--list-disabled', )) 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'] # Allowlist-gate the first token (the cagefsctl flag); subsequent # tokens are positional args (e.g. usernames for --enable/--disable) # and have already been validated upstream by check_param_value. first = params[0] if not first.startswith('--') or first not in self._ALLOWED_FLAGS: raise ValueError('cagefsctl flag not allowed: {}'.format(first)) params = [self.COMMAND] + list(params) 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 exist", {'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)