File: //usr/share/lvemanager-xray/plugins/install-xray-plugin.py
#!/opt/cloudlinux/venv/bin/python3 -bb
#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
from __future__ import print_function
from __future__ import division
from __future__ import absolute_import
import grp
import os
import pwd
import re
import sys
import json
import shlex
import stat
import argparse
import subprocess
import configparser
import cldetectlib as detect
from clcommon.utils import get_rhn_systemid_value
from clcommon.lib.cledition import is_cl_solo_edition
from clcommon.ui_config import UIConfig
SOURCE_PATH_BASE = "/usr/share/lvemanager-xray/"
SOURCE_PATH_ADMIN = "/usr/share/lvemanager-xray/xray-admin/"
SOURCE_PATH_USER = "/usr/share/lvemanager-xray/xray-user/"
LVEMANAGER_SOURCE_PATH = "/usr/share/l.v.e-manager/"
XRAY_MANAGER_UTILITY = "/usr/sbin/cloudlinux-xray-manager"
DYNAMICUI_SYNC_CONFIG_COMMAND = "/usr/share/l.v.e-manager/utils/dynamicui.py --sync-conf=all --silent"
CLOUDLINUX_CONFIG_COMMAND = "/usr/sbin/cloudlinux-config set --data '{\"options\": {\"uiSettings\": {\"hideXrayApp\": true}}}' --json"
PANEL_INTEGRATION_CONFIG = '/opt/cpvendor/etc/integration.ini'
CLOUDLINUX_TRANSLATIONS_DIR = '/usr/share/cloudlinux-translations/'
CHECK_PLUGIN_CRON = {
"name": "xray-plugin-check-cron",
"schedule": "*/10 * * * *",
"executor": "root",
"command": "/usr/bin/flock -n /var/run/cloudlinux-xray-plugin-check.cronlock /usr/share/lvemanager-xray/plugins/install-xray-plugin.py --check > /dev/null 2>&1"
}
class Base:
def __init__(self):
pass
def add_cron(self, file_name, schedule, executor, command):
"""
Create cron job
:param file_name: cron name
:param schedule: schedule of execution like "* * * * *"
:param executor: user to run cron
:param command: command to execute
:return: None
"""
try:
f = open('/etc/cron.d/' + file_name, 'w')
f.write("{} {} {}\n".format(schedule, executor, command))
f.close()
except Exception as e:
print('Cannot add cron job. {}'.format(e))
def remove_cron(self, file_name):
"""
Remove cron
:param file_name: cron name
:return: None
"""
try:
if file_name:
os.remove('/etc/cron.d/' + file_name)
except Exception as e:
print('Cannot remove cron job. {}'.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 disable_agent(self):
system_id = self.get_system_id()
if system_id is None:
print("Warning: Cannot disable user-agent because system_id is None.")
return
self.exec_command("{} {} --system_id {}"
.format(XRAY_MANAGER_UTILITY, "disable-user-agent", system_id))
def enable_agent(self):
system_id = self.get_system_id()
if system_id is None:
print("Warning: Cannot enable user-agent because system_id is None.")
return
self.exec_command("{} {} --system_id {}"
.format(XRAY_MANAGER_UTILITY, "enable-user-agent", system_id))
def is_agent_enabled_in_config(self):
return UIConfig().get_param('hideXrayApp', 'uiSettings') is False
def sync_ui_config(self):
self.exec_command(DYNAMICUI_SYNC_CONFIG_COMMAND)
def configure_plugin(self):
self.sync_ui_config()
if self.is_agent_enabled_in_config():
self.enable_agent()
# Automatic disable was turned off because we still need
# xray-user-agent to work in background for smart advice
# previously this caused issues when this script was triggered in upcp
# we may revisit this in the future
# else:
# self.disable_agent()
def get_system_id(self):
try:
system_id = get_rhn_systemid_value('system_id').split('-')[1]
return system_id
except Exception:
print("Warning: Cannot get system id. It looks like your system is not registered.")
return
def reset_ui_config(self):
if not is_cl_solo_edition(skip_jwt_check=True):
self.exec_command(CLOUDLINUX_CONFIG_COMMAND)
self.exec_command(DYNAMICUI_SYNC_CONFIG_COMMAND)
def remove_file(self, filename):
try:
os.remove(filename)
except OSError:
pass
def is_agent_running(self):
"""
Check if user agent running
:return: Boolean
"""
try:
system_id = self.get_system_id()
result = json.loads(self.exec_command("{} {} --system_id {}"
.format(XRAY_MANAGER_UTILITY, "user-agent-status", system_id))[0])
return result["status"] == "enabled"
except Exception as e:
print('Cannot get agent status. {}'.format(e))
return False
def check_and_repair_plugin(self):
"""
Check if there is misconfiguration between UiConfig and agent service and reconfigure plugin
:return: None
"""
if self.is_agent_enabled_in_config() and not self.is_agent_running():
self.configure_plugin()
def copy_file_or_dir(self, src, dst):
"""
Copy file or directory to specified location.
Resolves `dst` via a parent dir fd with O_NOFOLLOW so the remove and
the subsequent create are bound to the same parent inode — closes the
TOCTOU window between remove_file_or_dir(dst) and the shutil.copy /
shutil.copytree that previously re-resolved `dst` by path and would
follow a symlink planted in the diradmin-writable parent dir between
the two syscalls.
"""
try:
# Strip trailing slashes: O_NOFOLLOW is silently ignored when the
# path ends with `/` because the kernel resolves the trailing
# slash as a directory dereference of the final component,
# following a symlink substituted at the basename instead of
# rejecting it.
dst = dst.rstrip("/")
parent, name = os.path.split(dst)
if not name:
print("An error occurred while copying: invalid destination {}".format(dst))
return
# Ensure the parent chain exists — matches `shutil.copytree`'s
# implicit `os.makedirs(dst)` behaviour. Without this, first-
# install of lvemanager-xray on a host whose `lvemanager_spa/`
# tree is freshly laid down would fail at the parent_fd open
# below. The path components above the diradmin-writable leaf
# are root-owned (`/usr/local/directadmin/plugins/`), so path-
# based makedirs here is not a TOCTOU surface — the TOCTOU
# window we are closing is `parent` itself and below.
if parent:
os.makedirs(parent, exist_ok=True)
# Open the parent with O_NOFOLLOW so a swapped-symlink ancestor
# cannot redirect the whole copy elsewhere. The parent itself is
# attacker-writable on package upgrades (diradmin owns it after
# the first install), so every subsequent op must be relative to
# this fd, not re-resolved by path.
parent_fd = os.open(parent, os.O_RDONLY | os.O_NOFOLLOW | os.O_DIRECTORY)
except Exception as e:
print("An error occurred while copying: {}".format(e))
return
try:
if os.path.isfile(src):
self._unlink_at(parent_fd, name)
self._copy_file_at(src, parent_fd, name)
elif os.path.isdir(src):
self._unlink_at(parent_fd, name)
self._copy_tree_at(src, parent_fd, name)
except Exception as e:
print("An error occurred while copying: {}".format(e))
finally:
try:
os.close(parent_fd)
except OSError:
pass
def _unlink_at(self, dir_fd, name):
"""
Remove `name` under `dir_fd` if it exists, without following symlinks
out of the parent. A symlink at `name` is unlinked as the link itself
(os.unlink with follow_symlinks=False / fd-relative semantics).
"""
# Probe the entry with lstat-equivalent fstatat: do not follow
# symlinks here, otherwise an attacker-planted symlink would steer
# us into a tree we then proceed to rmtree.
try:
st = os.stat(name, dir_fd=dir_fd, follow_symlinks=False)
except FileNotFoundError:
return
except OSError as e:
print("An error occurred while removing {}: {}".format(name, e))
return
try:
if stat.S_ISDIR(st.st_mode):
# Recursively unlink contents bound to a fd of the subtree,
# then rmdir the entry via the parent fd. shutil.rmtree on a
# path would re-resolve through the attacker-writable parent.
self._rmtree_at(dir_fd, name)
else:
os.unlink(name, dir_fd=dir_fd)
except OSError as e:
print("An error occurred while removing {}: {}".format(name, e))
def _rmtree_at(self, parent_fd, name):
"""
Recursively remove the directory `name` rooted at `parent_fd`.
Opens the subtree with O_NOFOLLOW | O_DIRECTORY (rejecting a symlink
substituted at `name`), walks via os.fwalk in bottom-up mode so each
dirent is unlinked relative to its live dir fd, then rmdirs `name`
from `parent_fd`. No path is ever re-resolved through the
attacker-writable parent during the walk.
"""
top_fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_DIRECTORY,
dir_fd=parent_fd)
try:
for _dp, dirnames, filenames, sub_fd in os.fwalk(
top=".", dir_fd=top_fd, topdown=False,
follow_symlinks=False):
for f in filenames:
try:
os.unlink(f, dir_fd=sub_fd)
except OSError as e:
print("An error occurred while removing {}: {}".format(f, e))
for d in dirnames:
# With follow_symlinks=False, fwalk classifies a
# symlink-pointing-at-a-directory into `dirnames`.
# os.rmdir refuses symlinks (ENOTDIR), so probe with
# fstatat-no-follow first and unlink-instead when the
# entry is really a symlink — otherwise rmdir is left
# holding a non-empty directory and the final
# rmdir(name, parent_fd) below fails, silently breaking
# the install/upgrade.
try:
st = os.stat(d, dir_fd=sub_fd, follow_symlinks=False)
if stat.S_ISLNK(st.st_mode):
os.unlink(d, dir_fd=sub_fd)
else:
os.rmdir(d, dir_fd=sub_fd)
except OSError as e:
print("An error occurred while removing {}: {}".format(d, e))
finally:
os.close(top_fd)
os.rmdir(name, dir_fd=parent_fd)
def _copy_file_at(self, src, dir_fd, name):
"""
Copy regular-file `src` to `name` under `dir_fd`. The destination is
created via openat with O_CREAT | O_EXCL | O_NOFOLLOW so a symlink
planted at `name` after the unlink (or before, if `name` did not
previously exist) causes the open to fail with EEXIST / ELOOP rather
than have root write into the attacker-chosen target.
"""
src_fd = os.open(src, os.O_RDONLY)
try:
src_st = os.fstat(src_fd)
mode = src_st.st_mode & 0o777
dst_fd = os.open(name,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
mode, dir_fd=dir_fd)
try:
while True:
buf = os.read(src_fd, 65536)
if not buf:
break
while buf:
n = os.write(dst_fd, buf)
buf = buf[n:]
# umask: O_CREAT applies `mode & ~umask`. fchmod the
# already-open fd to restore the source's exact mode so a
# restrictive umask (077) does not strip web-asset bits.
os.fchmod(dst_fd, mode)
finally:
os.close(dst_fd)
finally:
os.close(src_fd)
def _copy_tree_at(self, src, parent_fd, name):
"""
Replicate the directory `src` as `name` under `parent_fd`.
Mirrors the leaf-by-leaf shape of shutil.copytree but every
destination open/mkdir is relative to a dir fd descended from
`parent_fd` via O_NOFOLLOW | O_DIRECTORY, so an attacker cannot
redirect a leaf write by planting a symlink in any sub-directory
we just created.
"""
# Preserve the source directory's mode (matches shutil.copytree's
# copystat behaviour) — under a restrictive umask 0o755 would be
# masked to 0o700, breaking web-asset accessibility. Read the source
# mode now so we can apply it after mkdir.
src_top_mode = os.stat(src, follow_symlinks=True).st_mode & 0o777
# Atomically claim the destination name. If a symlink or any other
# entry races into place after the unlink, mkdir fails with EEXIST.
os.mkdir(name, mode=src_top_mode, dir_fd=parent_fd)
top_fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_DIRECTORY,
dir_fd=parent_fd)
try:
# Restore exact source mode via fchmod — defends against umask
# masking the mkdir mode (mkdir applies `mode & ~umask`).
# Placed INSIDE the try so a failure here closes top_fd via
# the finally below.
os.fchmod(top_fd, src_top_mode)
# follow_symlinks=True for the source walk so symlink-to-directory
# entries are descended into and replicated (matches
# shutil.copytree's default symlinks=False semantics). The source
# tree is the package-shipped payload (root-owned), so following
# links there is safe; destination-side O_NOFOLLOW is preserved
# below — that's where the actual TOCTOU threat lives.
for dirpath, dirnames, filenames, sub_fd in os.fwalk(
top=src, follow_symlinks=True):
rel = os.path.relpath(dirpath, src)
# Descend the destination tree in lock-step with the source
# walk, opening every level under top_fd via O_NOFOLLOW so
# the descent is fd-bound.
if rel == ".":
dst_sub_fd = top_fd
close_dst_sub = False
else:
dst_sub_fd = self._open_under(top_fd, rel)
close_dst_sub = True
try:
for d in dirnames:
# Preserve source directory mode (matches
# shutil.copytree's copystat). Read src dir mode
# before mkdir so we apply it consistently — and
# protect against umask masking via fchmod after.
src_d_mode = os.stat(
d, dir_fd=sub_fd,
follow_symlinks=True).st_mode & 0o777
try:
os.mkdir(d, mode=src_d_mode, dir_fd=dst_sub_fd)
except FileExistsError:
# A prior leaf walk created this; harmless.
pass
# Re-apply the mode via fchmod through an O_NOFOLLOW
# openat — the dst parent dir is attacker-writable
# post-chown, so O_NOFOLLOW here protects against a
# symlink swap, and fchmod overrides any umask.
d_fd = os.open(
d, os.O_RDONLY | os.O_NOFOLLOW | os.O_DIRECTORY,
dir_fd=dst_sub_fd)
try:
os.fchmod(d_fd, src_d_mode)
finally:
os.close(d_fd)
for f in filenames:
# Source side is the package-shipped tree (root-owned,
# not attacker-writable): no symlink-protection needed
# here. Dropping O_NOFOLLOW matches shutil.copytree's
# default symlinks=False behaviour (dereference src
# symlinks, copy their target content) — with
# O_NOFOLLOW any symlink-to-file in the RPM payload
# would raise ELOOP and abort the whole copy mid-way.
src_leaf_fd = os.open(f, os.O_RDONLY, dir_fd=sub_fd)
try:
st = os.fstat(src_leaf_fd)
leaf_mode = st.st_mode & 0o777
dst_leaf_fd = os.open(
f,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
leaf_mode, dir_fd=dst_sub_fd)
try:
while True:
buf = os.read(src_leaf_fd, 65536)
if not buf:
break
while buf:
n = os.write(dst_leaf_fd, buf)
buf = buf[n:]
# umask: see _copy_file_at — fchmod the
# already-open fd to restore the source's
# exact mode.
os.fchmod(dst_leaf_fd, leaf_mode)
finally:
os.close(dst_leaf_fd)
finally:
os.close(src_leaf_fd)
finally:
if close_dst_sub:
try:
os.close(dst_sub_fd)
except OSError:
pass
finally:
try:
os.close(top_fd)
except OSError:
pass
def _open_under(self, top_fd, rel):
"""
Open `rel` (a forward-slash-separated relative path that os.fwalk
produced) under `top_fd`, descending one component at a time with
O_NOFOLLOW | O_DIRECTORY so no intermediate symlink can redirect us.
"""
cur_fd = top_fd
opened = []
try:
for part in rel.split(os.sep):
if not part or part == ".":
continue
nxt = os.open(part,
os.O_RDONLY | os.O_NOFOLLOW | os.O_DIRECTORY,
dir_fd=cur_fd)
opened.append(nxt)
cur_fd = nxt
# Caller owns the final fd; close every ancestor we opened.
for fd in opened[:-1]:
try:
os.close(fd)
except OSError:
pass
return cur_fd
except Exception:
for fd in opened:
try:
os.close(fd)
except OSError:
pass
raise
def remove_file_or_dir(self, path):
"""
Remove file or directory.
Resolves the parent with O_NOFOLLOW so the unlink/rmtree is bound to
the parent inode at probe time — a symlink swapped at `path` after
the entry-type probe cannot redirect the removal to an attacker-
chosen target outside the parent.
"""
try:
path = path.rstrip("/")
parent, name = os.path.split(path)
if not name:
return
parent_fd = os.open(parent, os.O_RDONLY | os.O_NOFOLLOW | os.O_DIRECTORY)
except FileNotFoundError:
return
except Exception as e:
print("An error occurred while copying: {}".format(e))
return
try:
self._unlink_at(parent_fd, name)
except Exception as e:
print("An error occurred while copying: {}".format(e))
finally:
try:
os.close(parent_fd)
except OSError:
pass
def safe_recursive_chown(self, path, user, group):
"""
Recursively change ownership without following symlinks.
Uses openat/fwalk semantics via dir-fd so path resolution is bound to
a specific directory inode for the whole walk — closes the TOCTOU
window between realpath() and os.walk() that a plain realpath+walk
approach leaves open (os.walk follows its top argument even when
followlinks=False, and os.lchown only protects the final component).
"""
try:
# pwd.getpwnam / grp.getgrnam raise KeyError on missing entries;
# catch Exception (same semantic as the replaced exec_command
# helper) so a missing user/group logs and continues rather
# than aborting the whole %posttrans install flow.
uid = pwd.getpwnam(user).pw_uid
gid = grp.getgrnam(group).gr_gid
# Strip trailing slashes: O_NOFOLLOW is silently ignored when the
# path ends with `/` because the kernel resolves the trailing slash
# as a directory dereference of the final component, following a
# symlink substituted at the basename instead of rejecting it.
path = path.rstrip("/")
# O_NOFOLLOW blocks a symlink substituted at `path` itself.
# O_DIRECTORY ensures we only accept a real directory.
top_fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_DIRECTORY)
except Exception as e:
print("An error occurred while opening path for chown: {}".format(e))
return
try:
# Chown the top directory itself via its fd — immune to rename.
os.fchown(top_fd, uid, gid)
# Walk via file descriptors. os.fwalk uses openat internally, so
# descents are relative to a live dir fd and cannot be redirected
# by an intermediate symlink swap mid-walk.
for _dirpath, dirnames, filenames, sub_fd in os.fwalk(
top=".", dir_fd=top_fd, follow_symlinks=False):
for name in dirnames + filenames:
# Per-entry try/except so a single ENOENT / EPERM on a
# mid-walk race doesn't abort the whole traversal —
# matches the prior `chown -R` shell resilience.
try:
os.chown(name, uid, gid,
dir_fd=sub_fd, follow_symlinks=False)
except OSError as e:
print("An error occurred while chowning {}: {}".format(name, e))
except Exception as e:
print("An error occurred while changing ownership: {}".format(e))
finally:
try:
os.close(top_fd)
except OSError:
pass
def safe_chmod_nofollow(self, path, mode, dir_fd=None):
"""
Change permissions on a path without following symlinks.
When called with a dir_fd, path is resolved relative to the directory
fd via openat() — used by safe_recursive_chmod to bind each leaf
operation to the live dir fd produced by os.fwalk. O_NONBLOCK guards
against a racing FIFO at that entry: without it, os.open(O_RDONLY)
on a FIFO blocks forever waiting for a writer and hangs the
root-privileged installer (the replaced `chmod -R` used fchmodat
semantics that never opened the file).
"""
try:
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK,
dir_fd=dir_fd)
try:
os.fchmod(fd, mode)
finally:
os.close(fd)
except Exception as e:
print("An error occurred while changing permissions: {}".format(e))
def safe_recursive_chmod(self, path, mode):
"""
Recursively change permissions without following symlinks.
Same openat/fwalk discipline as safe_recursive_chown — path
resolution is bound to a dir fd so a concurrent directory->symlink
swap between the initial open and the recursive traversal cannot
redirect us to an attacker-chosen target.
"""
try:
# Strip trailing slashes: see safe_recursive_chown for why
# O_NOFOLLOW alone is not enough on a path ending with `/`.
path = path.rstrip("/")
top_fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_DIRECTORY)
except Exception as e:
print("An error occurred while opening path for chmod: {}".format(e))
return
try:
os.fchmod(top_fd, mode)
for _dirpath, dirnames, filenames, sub_fd in os.fwalk(
top=".", dir_fd=top_fd, follow_symlinks=False):
for name in dirnames + filenames:
self.safe_chmod_nofollow(name, mode, dir_fd=sub_fd)
except Exception as e:
print("An error occurred while changing permissions: {}".format(e))
finally:
try:
os.close(top_fd)
except OSError:
pass
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}xray-user-ui.json"
self.copy_file_or_dir(source_path, template_path)
class CpanelPluginInstaller(Base):
def __init__(self):
self.ROOT_CPANEL_DIR = "/usr/local/cpanel/whostmgr/docroot/"
self.CPANEL_THEMES_BASE_DIR = "/usr/local/cpanel/base/frontend/"
self.template_src = SOURCE_PATH_BASE + "plugins/cpanel/xray.live.pl"
# template_dst should be format with theme name and placed for each theme
self.template_dst = self.CPANEL_THEMES_BASE_DIR + "{}/lveversion/xray.live.pl"
self.plugin_tar = SOURCE_PATH_BASE + "plugins/cpanel/cpanel-xray-plugin.tar.bz2"
self.plugin_installer = "/usr/local/cpanel/scripts/install_plugin"
self.plugin_uninstaller = "/usr/local/cpanel/scripts/uninstall_plugin"
self.xray_feature_file = "/usr/local/cpanel/whostmgr/addonfeatures/lvexray"
self.install_config = SOURCE_PATH_BASE + "plugins/cpanel/plugin/install.json"
self.destination_admin = "/usr/local/cpanel/whostmgr/docroot/3rdparty/cloudlinux/assets/xray-admin"
self.destination_user = "/usr/local/cpanel/whostmgr/docroot/3rdparty/cloudlinux/assets/xray-user"
def cpanel_fix_feature_manager(self):
if os.path.exists(self.xray_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.xray_feature_file, 'w')
feature_file.write(feature['feature'] + ':' + feature_name_fixed)
feature_file.close()
def install_plugin(self):
self.copy_file_or_dir(SOURCE_PATH_ADMIN, self.destination_admin)
self.copy_file_or_dir(SOURCE_PATH_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 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.controller_src = SOURCE_PATH_BASE + "plugins/plesk/XrayController.php"
self.controller_dst = self.ROOT_PLESK_DIR + "plib/modules/plesk-lvemanager/controllers/XrayController.php"
self.send_request_controller_src = SOURCE_PATH_BASE + "plugins/plesk/XraySendRequestController.php"
self.send_request_controller_dst = self.ROOT_PLESK_DIR + "plib/modules/plesk-lvemanager/controllers/XraySendRequestController.php"
self.destination_admin = "/usr/local/psa/admin/htdocs/modules/plesk-lvemanager/xray-admin"
self.destination_user = "/usr/local/psa/admin/htdocs/modules/plesk-lvemanager/xray-user"
self.icon_src = SOURCE_PATH_BASE + "plugins/plesk/xray.svg"
self.icon_dst = self.ROOT_PLESK_DIR + "htdocs/modules/plesk-lvemanager/images/xray.svg"
self.views_dir = self.ROOT_PLESK_DIR + "plib/modules/plesk-lvemanager/views/scripts/xray/"
self.template_src = SOURCE_PATH_BASE + "plugins/plesk/index.phtml"
self.template_dst = self.ROOT_PLESK_DIR + "plib/modules/plesk-lvemanager/views/scripts/xray/index.phtml"
def install_plugin(self):
self.copy_file_or_dir(SOURCE_PATH_ADMIN, self.destination_admin)
self.copy_file_or_dir(SOURCE_PATH_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):
def __init__(self):
# Plugin files
self.source_user_plugin = SOURCE_PATH_BASE + "plugins/directadmin/xray/"
self.destination_user_plugin = "/usr/local/directadmin/plugins/xray/"
self.source_index_file = SOURCE_PATH_BASE + "plugins/directadmin/xrayIndex.php"
self.destination_index_file = "/usr/local/directadmin/plugins/lvemanager_spa/app/View/Spa/index/xrayIndex.php"
self.plugin_conf_file = self.destination_user_plugin + "/plugin.conf"
self.destination_admin_spa = "/usr/local/directadmin/plugins/lvemanager_spa/images/assets/xray-admin"
self.destination_user_spa = "/usr/local/directadmin/plugins/xray/images/xray-user"
def install_plugin(self):
self.copy_file_or_dir(SOURCE_PATH_ADMIN, self.destination_admin_spa)
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(SOURCE_PATH_USER, self.destination_user_spa)
self.safe_recursive_chown(self.destination_user_plugin, "diradmin", "diradmin")
self.safe_recursive_chown(self.destination_admin_spa, "diradmin", "diradmin")
self.safe_recursive_chmod(self.destination_user_plugin, 0o755)
self.safe_chmod_nofollow(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.destination_admin_py_plugin = "/usr/share/l.v.e-manager/commons/spa-resources/static/xray-admin"
self.destination_user_py_plugin = "/usr/share/l.v.e-manager/commons/spa-resources/static/xray-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(SOURCE_PATH_ADMIN, self.destination_admin_py_plugin)
self.copy_file_or_dir(SOURCE_PATH_USER, self.destination_user_py_plugin)
self.copy_file_or_dir(SOURCE_PATH_ADMIN, base_path + '/assets/xray-admin')
self.copy_file_or_dir(SOURCE_PATH_USER, base_path + '/assets/xray-user')
def uninstall_plugin(self):
base_path = self.get_panel_base_path().rstrip('/')
self.remove_file_or_dir(self.destination_admin_py_plugin)
self.remove_file_or_dir(self.destination_user_py_plugin)
self.remove_file_or_dir(base_path + '/assets/xray-admin')
self.remove_file_or_dir(base_path + '/assets/xray-user')
class Main:
def __init__(self):
pass
def make_parser(self):
parser = argparse.ArgumentParser(description="Script to install|uninstall Xray App for user")
parser.add_argument("--install", "-i", action="store_true",
help="Install Xray App")
parser.add_argument("--uninstall", "-u", action="store_true",
help="Uninstall Xray App")
parser.add_argument("--check", "-c", action="store_true",
help="Check and repair plugin")
return parser
def run(self):
parser = self.make_parser()
args = parser.parse_args()
if detect.is_cpanel():
# Cpanel
installer = CpanelPluginInstaller()
elif detect.is_da():
# DirectAdmin
installer = DirectAdminPluginInstaller()
elif detect.is_plesk():
# Plesk
installer = PleskPluginInstaller()
elif os.path.isfile(PANEL_INTEGRATION_CONFIG):
# Custom panel with integration.ini
installer = PanelIntegrationPluginInstaller()
else:
print("X-Ray plugin cannot be installed on your environment")
sys.exit(0)
if args.install:
installer.install_plugin()
installer.generate_translate_templates()
installer.configure_plugin()
installer.add_cron(CHECK_PLUGIN_CRON["name"],
CHECK_PLUGIN_CRON["schedule"],
CHECK_PLUGIN_CRON["executor"],
CHECK_PLUGIN_CRON["command"])
elif args.uninstall:
installer.remove_cron(CHECK_PLUGIN_CRON["name"])
installer.uninstall_plugin()
installer.reset_ui_config()
elif args.check:
installer.check_and_repair_plugin()
else:
parser.print_help()
if __name__ == "__main__":
main = Main()
main.run()