File: //proc/self/root/usr/share/cloudlinux-awp-plugin/installer.py
#!/opt/cloudlinux/venv/bin/python3 -bb
# 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 print_function
from __future__ import division
from __future__ import absolute_import
import os
import re
import sys
import pwd
import grp
import stat
import json
import shlex
import shutil
import argparse
import subprocess
import configparser
import cldetectlib as detect
SOURCE_PATH_BASE = "/usr/share/cloudlinux-awp-plugin/"
SOURCE_PATH_ADMIN = "/usr/share/cloudlinux-awp-plugin/awp-admin/"
SOURCE_PATH_USER = "/usr/share/cloudlinux-awp-plugin/awp-user/"
DYNAMICUI_SYNC_CONFIG_COMMAND = "/usr/share/l.v.e-manager/utils/dynamicui.py --sync-conf=all --silent"
PANEL_INTEGRATION_CONFIG = '/opt/cpvendor/etc/integration.ini'
CLOUDLINUX_TRANSLATIONS_DIR = '/usr/share/cloudlinux-translations/'
class Base:
def __init__(self):
pass
def remove_file_or_dir(self, path):
"""
Remove file or directory
"""
try:
if os.path.isfile(path):
os.remove(path)
elif os.path.isdir(path):
shutil.rmtree(path)
except Exception as e:
print("An error occurred while copying: {}".format(e))
def copy_file_or_dir(self, src, dst):
"""
Copy file or directory to specified location
"""
try:
if os.path.isfile(src):
self.remove_file_or_dir(dst)
shutil.copy(src, dst)
elif os.path.isdir(src):
self.remove_file_or_dir(dst)
shutil.copytree(src, dst)
except Exception as e:
print("An error occurred while copying: {}".format(e))
def parse_command(self, command):
"""
Parses a command string into a list of arguments.
"""
if isinstance(command, str):
if command.strip() == "":
return []
return shlex.split(command)
elif isinstance(command, list):
return command
else:
return []
def exec_command(self, command, env=None):
"""
This function will run passed command
in command line and returns result
:param command:
:param env:
:return:
"""
result = []
try:
args = self.parse_command(command)
if not args:
raise ValueError(f"The provided command is not valid: {command}")
p = subprocess.Popen(args, stdout=subprocess.PIPE, env=env, text=True)
while 1:
output = p.stdout.readline()
if not output:
break
if output.strip() != "":
result.append(output.strip())
except Exception as e:
print("Call process error: " + str(e))
return result
def sync_ui_config(self):
self.exec_command(DYNAMICUI_SYNC_CONFIG_COMMAND)
def generate_translate_templates(self):
"""
Prepare translate templates using english as a base.
"""
source_path = f"{SOURCE_PATH_USER}i18n/en-en.json"
template_path = f"{CLOUDLINUX_TRANSLATIONS_DIR}awp-user-ui.json"
self.copy_file_or_dir(source_path, template_path)
class CpanelPluginInstaller(Base):
def __init__(self):
self.CPANEL_THEMES_BASE_DIR = "/usr/local/cpanel/base/frontend/"
self.source_admin = SOURCE_PATH_ADMIN
self.source_user = SOURCE_PATH_USER
self.destination_admin = "/usr/local/cpanel/whostmgr/docroot/3rdparty/cloudlinux/assets/awp-admin"
self.destination_user = "/usr/local/cpanel/whostmgr/docroot/3rdparty/cloudlinux/assets/awp-user"
self.template_src = SOURCE_PATH_BASE + "user-plugins/cpanel/wpos.live.pl"
self.template_dst = self.CPANEL_THEMES_BASE_DIR + "{}/lveversion/wpos.live.pl"
self.plugin_installer = "/usr/local/cpanel/scripts/install_plugin"
self.plugin_uninstaller = "/usr/local/cpanel/scripts/uninstall_plugin"
self.awp_feature_file = "/usr/local/cpanel/whostmgr/addonfeatures/lvewpos"
self.install_config = SOURCE_PATH_BASE + "user-plugins/cpanel/plugin/install.json"
self.plugin_tar = SOURCE_PATH_BASE + "user-plugins/cpanel/cpanel-awp-plugin.tar.bz2"
def install_plugin(self):
self.copy_file_or_dir(self.source_admin, self.destination_admin)
self.copy_file_or_dir(self.source_user, self.destination_user)
for theme in self.get_theme_list():
self.copy_file_or_dir(self.template_src, self.template_dst.format(theme))
self.exec_command("{} {} --theme {}".format(self.plugin_installer, self.plugin_tar, theme))
self.cpanel_fix_feature_manager()
def cpanel_fix_feature_manager(self):
if os.path.exists(self.awp_feature_file) and os.path.exists(self.install_config):
with open(self.install_config) as install_config:
feature_config = json.load(install_config)
feature = feature_config[0] # We have only one item
try:
feature_name_fixed = re.search("\$LANG{'(.*)'}", feature['name']).group(1)
except AttributeError:
feature_name_fixed = ''
feature_file = open(self.awp_feature_file, 'w')
feature_file.write(feature['feature'] + ':' + feature_name_fixed)
feature_file.close()
def uninstall_plugin(self):
self.remove_file_or_dir(self.destination_admin)
self.remove_file_or_dir(self.destination_user)
for theme in self.get_theme_list():
self.remove_file_or_dir(self.template_dst.format(theme))
self.exec_command("{} {} --theme {}".format(self.plugin_uninstaller, self.plugin_tar, theme))
def get_theme_list(self):
if os.path.isdir(self.CPANEL_THEMES_BASE_DIR):
return next(os.walk(self.CPANEL_THEMES_BASE_DIR), (None, None, []))[1]
class PleskPluginInstaller(Base):
def __init__(self):
self.ROOT_PLESK_DIR = "/usr/local/psa/admin/"
self.source_admin = SOURCE_PATH_ADMIN
self.source_user = SOURCE_PATH_USER
self.destination_admin = "/usr/local/psa/admin/htdocs/modules/plesk-lvemanager/awp-admin"
self.destination_user = "/usr/local/psa/admin/htdocs/modules/plesk-lvemanager/awp-user"
self.controller_src = SOURCE_PATH_BASE + "user-plugins/plesk/AwpController.php"
self.controller_dst = self.ROOT_PLESK_DIR + "plib/modules/plesk-lvemanager/controllers/AwpController.php"
self.send_request_controller_src = SOURCE_PATH_BASE + "user-plugins/plesk/AwpSendRequestController.php"
self.send_request_controller_dst = (
self.ROOT_PLESK_DIR + "plib/modules/plesk-lvemanager/controllers/AwpSendRequestController.php"
)
self.icon_src = SOURCE_PATH_BASE + "user-plugins/plesk/awp.svg"
self.icon_dst = self.ROOT_PLESK_DIR + "htdocs/modules/plesk-lvemanager/images/awp.svg"
self.views_dir = self.ROOT_PLESK_DIR + "plib/modules/plesk-lvemanager/views/scripts/awp/"
self.template_src = SOURCE_PATH_BASE + "user-plugins/plesk/index.phtml"
self.template_dst = self.ROOT_PLESK_DIR + "plib/modules/plesk-lvemanager/views/scripts/awp/index.phtml"
def install_plugin(self):
self.copy_file_or_dir(self.source_admin, self.destination_admin)
self.copy_file_or_dir(self.source_user, self.destination_user)
self.copy_file_or_dir(self.controller_src, self.controller_dst)
self.copy_file_or_dir(self.send_request_controller_src, self.send_request_controller_dst)
self.copy_file_or_dir(self.icon_src, self.icon_dst)
if not os.path.isdir(self.views_dir):
os.mkdir(self.views_dir)
self.copy_file_or_dir(self.template_src, self.template_dst)
def uninstall_plugin(self):
self.remove_file_or_dir(self.destination_admin)
self.remove_file_or_dir(self.destination_user)
self.remove_file_or_dir(self.controller_dst)
self.remove_file_or_dir(self.send_request_controller_dst)
self.remove_file_or_dir(self.icon_dst)
self.remove_file_or_dir(self.views_dir)
class DirectAdminPluginInstaller(Base):
# Safe-root all DirectAdmin destinations must canonically resolve under.
DA_PLUGINS_ROOT = "/usr/local/directadmin/plugins"
def __init__(self):
# Plugin files
self.source_user_plugin = SOURCE_PATH_BASE + "user-plugins/directadmin/awp/"
self.destination_user_plugin = "/usr/local/directadmin/plugins/awp/"
self.source_index_file = SOURCE_PATH_BASE + "user-plugins/directadmin/awpIndex.php"
self.destination_index_file = "/usr/local/directadmin/plugins/lvemanager_spa/app/View/Spa/index/awpIndex.php"
self.plugin_conf_file = self.destination_user_plugin + "/plugin.conf"
# Spa files
self.source_admin_spa = SOURCE_PATH_ADMIN
self.destination_admin_spa = "/usr/local/directadmin/plugins/lvemanager_spa/images/assets/awp-admin"
self.source_user_spa = SOURCE_PATH_USER
self.destination_user_spa = "/usr/local/directadmin/plugins/awp/images/awp-user"
def _resolve_da_principal(self):
"""Resolve the diradmin uid/gid, or (None, None) if absent."""
try:
uid = pwd.getpwnam("diradmin").pw_uid
gid = grp.getgrnam("diradmin").gr_gid
return uid, gid
except KeyError as e:
print("Cannot resolve diradmin principal: {}".format(e))
return None, None
def _assert_under_root(self, path):
"""Canonicalise path and refuse anything that escapes DA_PLUGINS_ROOT.
Guards the intermediate-component symlink vector: O_NOFOLLOW only
protects the final component, so a symlinked parent could otherwise
redirect the recursive chown/chmod outside the plugins tree.
"""
real = os.path.realpath(path)
root = os.path.realpath(self.DA_PLUGINS_ROOT)
if real != root and not real.startswith(root + os.sep):
raise ValueError("Refusing to operate outside {}: {} -> {}".format(root, path, real))
return real
def _secure_chmod_file(self, path, mode):
"""Symlink-safe chmod of a single file under the plugins root."""
# Strip trailing slash: it silently defeats O_NOFOLLOW on open.
path = path.rstrip("/")
self._assert_under_root(path)
# O_NOFOLLOW refuses to open a symlink (ELOOP); fchmod then targets the
# real file, never a symlink target. chmod(follow_symlinks=False) is
# unsupported on Linux, so go through an fd instead.
try:
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
try:
os.fchmod(fd, mode)
finally:
os.close(fd)
except OSError as e:
print("Cannot chmod {}: {}".format(path, e))
def _secure_chown_chmod_tree(self, path, uid=None, gid=None, mode=None):
"""Symlink-safe recursive chown/chmod of a directory tree.
No symlink anywhere in the operand path or the descended tree is ever
dereferenced: the leaf is opened O_NOFOLLOW|O_DIRECTORY and fwalk binds
every chown/chmod to a directory fd with follow_symlinks=False.
"""
# Strip trailing slash so a symlinked operand is not silently followed.
path = path.rstrip("/")
self._assert_under_root(path)
try:
topfd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_DIRECTORY)
except OSError as e:
print("Cannot open {} safely: {}".format(path, e))
return
try:
# topfd was opened O_NOFOLLOW|O_DIRECTORY, so it is provably not a
# symlink: operate on the fd directly (no path re-resolution).
if uid is not None and gid is not None:
os.fchown(topfd, uid, gid)
if mode is not None:
os.fchmod(topfd, mode)
for root, dirs, files, rootfd in os.fwalk(dir_fd=topfd, follow_symlinks=False):
for name in dirs + files:
if uid is not None and gid is not None:
os.chown(name, uid, gid, dir_fd=rootfd, follow_symlinks=False)
if mode is not None:
# chmod(follow_symlinks=False) is unsupported on Linux;
# skip symlink entries (chmod would hit the target).
st = os.stat(name, dir_fd=rootfd, follow_symlinks=False)
if stat.S_ISLNK(st.st_mode):
continue
os.chmod(name, mode, dir_fd=rootfd)
except OSError as e:
print("Cannot set ownership/permissions under {}: {}".format(path, e))
finally:
os.close(topfd)
def install_plugin(self):
# admin part
self.copy_file_or_dir(self.source_admin_spa, self.destination_admin_spa)
# user part
self.copy_file_or_dir(self.source_user_plugin, self.destination_user_plugin)
self.copy_file_or_dir(self.source_index_file, self.destination_index_file)
self.copy_file_or_dir(self.source_user_spa, self.destination_user_spa)
# set permission and owner (symlink-safe; no shell chown -R/chmod -R)
uid, gid = self._resolve_da_principal()
self._secure_chown_chmod_tree(self.destination_user_plugin, uid=uid, gid=gid, mode=0o755)
self._secure_chown_chmod_tree(self.destination_admin_spa, uid=uid, gid=gid)
self._secure_chmod_file(self.plugin_conf_file, 0o644)
def uninstall_plugin(self):
self.remove_file_or_dir(self.destination_user_plugin)
self.remove_file_or_dir(self.destination_admin_spa)
class PanelIntegrationPluginInstaller(Base):
def __init__(self):
self.source_admin = SOURCE_PATH_ADMIN
self.source_user = SOURCE_PATH_USER
self.destination_admin = "/usr/share/l.v.e-manager/commons/spa-resources/static/awp-admin"
self.destination_user = "/usr/share/l.v.e-manager/commons/spa-resources/static/awp-user"
def get_panel_base_path(self):
try:
parser = configparser.ConfigParser(interpolation=None, strict=False)
parser.read(PANEL_INTEGRATION_CONFIG)
base_path = parser.get("lvemanager_config", "base_path")
return base_path
except Exception as e:
print("Cannot copy files for no panel version. {}".format(e))
sys.exit(1)
def install_plugin(self):
base_path = self.get_panel_base_path().rstrip('/')
self.copy_file_or_dir(self.source_admin, self.destination_admin)
self.copy_file_or_dir(self.source_admin, base_path + '/assets/awp-admin')
self.copy_file_or_dir(self.source_user, self.destination_user)
self.copy_file_or_dir(self.source_user, base_path + '/assets/awp-user')
def uninstall_plugin(self):
base_path = self.get_panel_base_path().rstrip('/')
self.remove_file_or_dir(self.destination_admin)
self.remove_file_or_dir(base_path + '/assets/awp-admin')
self.remove_file_or_dir(self.destination_user)
self.remove_file_or_dir(base_path + '/assets/awp-user')
class Main:
def __init__(self):
pass
def make_parser(self):
parser = argparse.ArgumentParser(description="Script to install|uninstall AccelerateWP")
parser.add_argument("--install", "-i", action="store_true", help="Install AccelerateWP")
parser.add_argument("--uninstall", "-u", action="store_true", help="Uninstall AccelerateWP")
return parser
def run(self):
parser = self.make_parser()
args = parser.parse_args()
if detect.is_cpanel():
# Cpanel
installer = CpanelPluginInstaller()
elif detect.is_plesk():
# Plesk
installer = PleskPluginInstaller()
elif detect.is_da():
# DirectAdmin
installer = DirectAdminPluginInstaller()
elif os.path.isfile(PANEL_INTEGRATION_CONFIG):
# Custom panel with integration.ini
installer = PanelIntegrationPluginInstaller()
else:
print("AccelerateWP plugin cannot be installed on your environment")
sys.exit(0)
if args.install:
installer.install_plugin()
installer.generate_translate_templates()
installer.sync_ui_config()
elif args.uninstall:
installer.uninstall_plugin()
installer.sync_ui_config()
else:
parser.print_help()
if __name__ == "__main__":
main = Main()
main.run()