#!/opt/cloudlinux/venv/bin/python3 -sbb # coding:utf-8 # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2023 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT from __future__ import absolute_import from __future__ import division from __future__ import print_function import os import sys import json import time import getopt import subprocess from enum import Enum from clcommon.utils import is_nginx_running, is_ubuntu class State(Enum): INSTALLED = 'installed' INSTALLING = 'installing' NOT_INSTALLED = 'not_installed' class UIPackageInstaller: def __init__(self, package_name, package_path, error_log_path): self.PACKAGE_NAME = package_name self.PACKAGE_PATH = package_path self.PACKAGE_INSTALL_ERROR_LOG = error_log_path def install(self): """Initiates the installation process based on the OS type.""" if is_ubuntu(): self._install_package_apt() else: self._install_package_yum() return State.INSTALLING def _run_command(self, command): """Executes the given command and logs any output.""" with open(self.PACKAGE_INSTALL_ERROR_LOG, 'w') as logger: try: subprocess.Popen(command, stdin=open(os.devnull, 'w'), stdout=logger, stderr=logger, shell=True, executable='/bin/bash') except Exception as e: logger.write(str(e)) def _install_package_yum(self): """Installs the package using YUM.""" command = 'yum install -y {}'.format(self.PACKAGE_NAME) self._run_command(command) return State.INSTALLING def _install_package_apt(self): """Installs the package using APT.""" command = 'apt-get update -y && apt-get install -y {}'.format(self.PACKAGE_NAME) self._run_command(command) return State.INSTALLING def check_installed(self): """Checks if the package is installed, installing, or not installed.""" if self._check_status() == State.INSTALLING: return State.INSTALLING else: return State.INSTALLED if os.path.exists(self.PACKAGE_PATH) else State.NOT_INSTALLED def _check_status(self): """Checks the installation status based on the OS type.""" commands = [] if is_ubuntu(): commands = [ '/bin/ps aux | grep "[a]pt-get install -y {}"'.format(self.PACKAGE_NAME), '/bin/ps aux | grep "[a]pt-get update"' ] else: commands = ['/bin/ps aux | grep "[y]um install -y {}"'.format(self.PACKAGE_NAME)] for cmd in commands: try: subprocess.check_output(['/bin/bash', '-c', cmd], stderr=subprocess.STDOUT, text=True) return State.INSTALLING except subprocess.CalledProcessError: pass return State.NOT_INSTALLED def get_log(self): """Fetches the log from the installation error log file.""" with open(self.PACKAGE_INSTALL_ERROR_LOG, 'r') as file: return file.read() def usage(self): print("usage: {} {{-h|-i|-c}}\n".format(sys.argv[0]), "where:\n" "\t-h, --help: show help message\n" "\t-i, --install: install {} package\n".format(self.PACKAGE_NAME), "\t-c, --check: check status of {} package\n".format(self.PACKAGE_NAME), "\t-l, --log: print the log file of {} package\n".format(self.PACKAGE_NAME)) exit(2) def run(self, args): try: opts, _ = getopt.getopt(args, 'hicl', ['help', 'install', 'check', 'log']) except getopt.GetoptError: self.usage() if not opts: self.usage() result = None for o, _ in opts: if o in ('-h', '--help'): self.usage() elif o in ('-i', '--install'): result = self.install().value elif o in ('-c', '--check'): result = self.check_installed().value elif o in ('-l', '--log'): result = self.get_log() print(json.dumps({ "result": "success", "timestamp": time.time(), "response": result }))