Server : ApacheSystem : Linux host44.registrar-servers.com 4.18.0-513.18.1.lve.2.el8.x86_64 #1 SMP Sat Mar 30 15:36:11 UTC 2024 x86_64 User : vapecompany ( 2719) PHP Version : 7.4.33 Disable Function : NONE Directory : /opt/cloudlinux/venv/lib64/python3.11/site-packages/clselect/ |
Upload File : |
# -*- 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 absolute_import
from __future__ import division
import getopt
import glob
import os
import shutil
import sys
import traceback
from future.utils import iteritems
from . import clpassenger
from . import clselectctl
from . import utils
from .clselectprint import clprint
from .clselectexcept import ClSelectExcept, BaseClSelectException
from .clselectruby import environments, extensions, interpreters
from clcommon.cpapi import userdomains, docroot, CP_NAME
from clcommon.cpapi.cpapiexceptions import NoDomain, NotSupported
INTERPRETER = 'ruby'
def usage():
print(' -v | --version VERSION : Specify alternative version')
print(' -u | --user USERNAME : Username')
print(' --domain DOMAIN : Domain (or subdomain); users main domain as default')
print(' -l | --list : List alternatives for interpreter')
print(' -G | --list-extensions : List global set of packages')
print(' -K | --list-extensions-version : List version set of packages')
print(' -s | --user-summary : List user summary of webapps')
print(' -y | --create-webapp : Create user webapp')
print(' -n | --destroy-webapp : Destroy user webapp')
print(' -f | --relocate-webapp : Change webapp directory (files must be moved manually)')
print(' -F | --transit-webapp : Change webapp domain/alias')
print(' -Z | --restart-webapp : Restart webapp')
print(' -c | --user-current : Show currently selected alternative for user webapp')
print(' -b | --set-user-current : Set alternative as user webapp default')
print(' -e | --enable-user-extensions GEMS : Install comma-separated list of gems for user webapp')
print(' -d | --disable-user-extensions GEMS : Uninstall comma-separated list of gems for user webapp')
print(" : Use '-' (minus) for all gems")
print(' -r | --replace-user-extensions GEMS : Update comma-separated list of gems for user webapp')
print(" : Use '-' (minus) for all gems")
print(' -g | --list-user-extensions : List installed gems for user webapp')
print(' -p | --print-summary : If specified along with setting an alternative prints user summary')
print(' -j | --json : Print data as JSON')
def _create_environment(user, directory, version, env_name=None):
prefix = _get_prefix(user, directory)
if not env_name:
env_name = version
environment = environments.Environment(env_name, user, prefix)
if not environment.exists():
try:
interpreter = interpreters.interpreters(key='version')[version]
except KeyError:
raise ClSelectExcept.NoSuchAlternativeVersion(version)
environment.create(interpreter)
return environment
def _get_environment(user, directory, app_summary=None):
prefix = _get_prefix(user, directory)
if app_summary is None:
user_summary = clpassenger.summary(user)
app_summary = user_summary.get(directory)
if not app_summary:
raise ClSelectExcept.NoSuchApplication('No such application (or application not configured) "%s"' % directory)
binary = app_summary['binary']
env_name = os.path.basename(os.path.dirname(os.path.dirname(binary)))
environment = environments.Environment(env_name, user, prefix)
return environment
def _get_prefix(user, directory):
_, rel_dir = utils.get_abs_rel(user, directory)
return os.path.join(environments.DEFAULT_PREFIX,
clselectctl.get_prefix(rel_dir))
def create(user, directory, alias, version=None, env_name=None, doc_root=None):
if version is None:
raise ClSelectExcept.WrongData('Not passed version as argument')
user = clselectctl.get_user(user)
clselectctl.check_directory(directory)
alias = clselectctl.get_alias(alias)
environment = _create_environment(user, directory, version, env_name)
binary = environment.interpreter().binary
clpassenger.configure(user, directory, alias, INTERPRETER, binary, doc_root=doc_root)
clpassenger.restart(user, directory)
def current(user, directory, version=None, env_name=None):
user = clselectctl.get_user(user)
old_environment = _get_environment(user, directory)
if not version:
return {old_environment.name: old_environment.interpreter().as_dict()}
old_extensions = set(old_environment.extensions())
new_environment = _create_environment(user, directory, version, env_name)
new_extensions = set(new_environment.extensions())
for extension in new_extensions - old_extensions:
try:
new_environment.extension_uninstall(extension)
except ClSelectExcept.ExternalProgramFailed:
pass
for extension in old_extensions - new_extensions:
try:
new_environment.extension_install(extension)
except ClSelectExcept.ExternalProgramFailed:
pass
app_summary = clpassenger.summary(user)[directory]
alias = app_summary['alias']
doc_root = app_summary['docroot']
htaccess_path = app_summary['htaccess']
binary = new_environment.interpreter().binary
clpassenger._unconfigure(htaccess=htaccess_path)
clpassenger.configure(user, directory, alias, INTERPRETER, binary, doc_root=doc_root)
clpassenger.restart(user, directory)
def destroy(user, directory):
user = clselectctl.get_user(user)
prefix = _get_environment(user, directory).prefix
abs_dir, _ = utils.get_abs_rel(user, prefix)
try:
shutil.rmtree(abs_dir)
except OSError:
pass
clpassenger.unconfigure(user, directory)
try:
clpassenger.restart(user, directory)
except ClSelectExcept.MissingApprootDirectory:
pass
def install(user, directory, extension):
user = clselectctl.get_user(user)
environment = _get_environment(user, directory)
environment.extension_install(extension)
def list_extensions(user, directory):
user = clselectctl.get_user(user)
environment = _get_environment(user, directory)
return environment.extensions()
def relocate(user, old_directory, new_directory, fmt):
if '/' in new_directory:
raise ClSelectExcept.WrongData('You cannot move appication to subdir')
user = clselectctl.get_user(user)
clselectctl.check_directory(new_directory)
old_abs, old_rel = utils.get_abs_rel(user, old_directory)
new_abs, new_rel = utils.get_abs_rel(user, new_directory)
old_user_summary = clpassenger.summary(user)
if new_directory in old_user_summary:
raise ClSelectExcept.WebAppError("Specified directory already used by '%s'" % new_abs)
if old_directory not in old_user_summary:
raise ClSelectExcept.WrongData("No such application (or application not configured) \"%s\"" % old_directory)
htaccess_path = old_user_summary[old_directory]['htaccess']
doc_root = old_user_summary[old_directory]['docroot']
alias = old_user_summary[old_directory]['alias']
env_name = _get_environment(user, old_directory).name
old_prefix = clselectctl.get_prefix(old_rel)
new_prefix = clselectctl.get_prefix(new_rel)
_old_envs, _ = utils.get_abs_rel(user, os.path.join(
environments.DEFAULT_PREFIX, old_prefix))
old_envs = os.path.join(_old_envs, '')
_new_envs, _ = utils.get_abs_rel(user, os.path.join(
environments.DEFAULT_PREFIX, new_prefix))
new_envs = os.path.join(_new_envs, '')
old_prompt = '(' + old_rel + ':'
new_prompt = '(' + new_rel + ':'
shutil.move(old_envs, new_envs)
for activate in glob.glob(os.path.join(new_envs, '*', 'bin', '*')):
old_activate = utils.file_read(activate)
if old_prompt in old_activate:
new_activate = old_activate.replace(old_prompt, new_prompt)
utils.file_write(activate, new_activate, 'w')
if not os.path.exists(new_abs):
os.rename(old_abs, new_abs)
prefix = _get_prefix(user, new_directory)
environment = environments.Environment(env_name, user, prefix)
binary = environment.interpreter().binary
clpassenger._unconfigure(htaccess=htaccess_path)
clpassenger.configure(user, new_directory, alias, INTERPRETER, binary, doc_root=doc_root)
clpassenger.restart(user, new_directory)
def restart(user, directory):
user = clselectctl.get_user(user)
apps_summary = clpassenger.summary(user)
if directory not in apps_summary:
raise ClSelectExcept.WrongData("No such application (or application not configured) \"%s\"" % directory)
clpassenger.restart(user, directory)
def summary(user):
user = clselectctl.get_user(user)
summ = {}
for directory, data in iteritems(clpassenger.summary(user)):
if data['interpreter'] != INTERPRETER:
continue
environment = _get_environment(user, directory, data).as_deepdict()
summ[directory] = {
'domain': data['domain'],
'alias': data['alias'],
'environment': environment['name'],
'interpreter': environment['interpreter'],
'extensions': environment['extensions'],
}
# add only list with additions domains
if "domains" in data and len(data["domains"]) > 1:
summ[directory]["domains"] = data["domains"]
return summ
def transit(user, directory, alias, doc_root=None):
user = clselectctl.get_user(user)
apps_summary = clpassenger.summary(user)
if directory not in apps_summary:
raise ClSelectExcept.WrongData("No such application (or application not configured) \"%s\"" % directory)
old_app_summary = apps_summary[directory]
old_alias = old_app_summary['alias']
old_doc_root = old_app_summary['docroot']
new_alias = clselectctl.get_alias(alias)
environment = _get_environment(user, directory)
binary = environment.interpreter().binary
clpassenger.configure(user, directory, new_alias, INTERPRETER, binary, True, 'transit', doc_root=doc_root)
clpassenger.move(user, directory, old_alias, new_alias, old_doc_root=old_doc_root, new_doc_root=doc_root)
clpassenger.restart(user, directory)
def uninstall(user, directory, extension):
user = clselectctl.get_user(user)
environment = _get_environment(user, directory)
environment.extension_uninstall(extension)
def update(user, directory, extension):
user = clselectctl.get_user(user)
environment = _get_environment(user, directory)
environment.extension_update(extension)
def main():
try:
opts, args = getopt.getopt(
sys.argv[1:],
'hi:v:u:lGsynfFZcbe:d:r:gpjK:', [
'help',
'interpreter=',
'version=',
'user=',
'domain=',
'list',
'list-extensions',
'user-summary',
'create-webapp',
'destroy-webapp',
'relocate-webapp',
'transit-webapp',
'restart-webapp',
'user-current',
'set-user-current',
'enable-user-extensions=',
'disable-user-extensions=',
'replace-user-extensions=',
'list-user-extensions',
'print-summary',
'json',
'list-extensions-version='
])
except getopt.GetoptError as err:
sys.stderr.write(str(err))
usage()
sys.exit(1)
ext_list = ''
fmt = 'text'
print_summary = False
user = None
domain = None
action = ''
version = None
if not opts:
usage()
sys.exit(1)
for o, a in opts:
if o in ('-i', '--interpreter'):
pass
elif o in ('-l', '--list'):
action = 'list'
elif o in ('-y', '--create-webapp'):
action = 'create-webapp'
elif o in ('-n', '--destroy-webapp'):
action = 'destroy-webapp'
elif o in ('-f', '--relocate-webapp'):
action = 'relocate-webapp'
elif o in ('-F', '--transit-webapp'):
action = 'transit-webapp'
elif o in ('-Z', '--restart-webapp'):
action = 'restart-webapp'
elif o in ('-c', '--user-current'):
action = 'user-current'
elif o in ('-b', '--set-user-current'):
action = 'set-user-current'
elif o in ('-g', '--list-user-extensions'):
action = 'list-user-extensions'
elif o in ('-e', '--enable-user-extensions'):
action = 'enable-user-extensions'
ext_list = a
elif o in ('-s', '--user-summary'):
action = 'user-summary'
elif o in ('-j', '--json'):
fmt = 'json'
elif o in ('-r', '--replace-user-extensions'):
action = 'replace-user-extensions'
ext_list = a
elif o in ('-d', '--disable-user-extensions'):
action = 'disable-user-extensions'
ext_list = a
elif o in ('-v', '--version'):
version = a
elif o in ('-G', '--list-extensions'):
action = 'list-extensions'
elif o in ('-u', '--user'):
user = a
elif o == '--domain':
domain = a
elif o in ('-p', '--print-summary'):
print_summary = True
elif o in ('-K', '--list-extensions-version'):
action = 'list-extensions-version'
ext_list = a
else:
sys.stderr.write('unhandled option')
sys.exit(1)
if action == '':
sys.stderr.write('ERROR:you must provide option for interpreter ruby')
sys.exit(1)
if action in ('create-webapp', 'destroy-webapp', 'relocate-webapp',
'transit-webapp', 'restart-webapp',
'enable-user-extensions', 'list-user-extensions',
'replace-user-extensions', 'disable-user-extensions',
):
if not args:
sys.stderr.write('webapp must be specified')
sys.exit(3)
if domain:
try:
doc_root, user_ = docroot(domain)
except NoDomain:
clprint.print_diag(fmt, {'status': 'ERROR', 'message': 'No such domain: "%s"' % domain})
sys.exit(1)
except NotSupported:
clprint.print_diag(fmt, {'status': 'ERROR', 'message': 'Ruby selector not supported for %s' % CP_NAME})
sys.exit(1)
if not user: # we can obtain user name for domain
user = user_
elif user != user_:
clprint.print_diag(fmt, {'status': 'ERROR', 'message': 'domain %s is not owned by the user %s' % (domain, user)})
sys.exit(1)
elif user and not domain:
try:
domain_list = userdomains(user)
except NotSupported:
clprint.print_diag(fmt, {'status': 'ERROR', 'message': 'Ruby selector not supported for %s' % CP_NAME})
sys.exit(1)
_, doc_root = domain_list[0] # get document root for main domain
if ext_list == '-':
if action == 'enable-user-extensions':
sys.stderr.write('installlation of all extensions is not possible')
sys.exit(4)
elif ext_list:
_exts = [_f for _f in ext_list.split(',') if _f]
try:
error = 0
ok = 0
result = {}
if action == 'list-extensions':
result = extensions.ExtensionInfo().list_extensions_cached()
elif action == 'list-extensions-version':
result = extensions.ExtensionInfo().list_extensions_version(_exts)
ext_list = ''
elif action == 'list':
result = interpreters.interpreters_dict('version')
elif action == 'user-summary':
result = summary(user)
elif action == 'create-webapp':
create(user, args[0], args[1], version, doc_root=doc_root)
elif action == 'destroy-webapp':
destroy(user, args[0])
elif action == 'relocate-webapp':
relocate(user, args[0], args[1], fmt)
elif action == 'transit-webapp':
new_doc_root = None
if domain: # remove app to new domain if domain present
new_doc_root = doc_root
transit(user, args[0], args[1], doc_root=new_doc_root)
elif action == 'restart-webapp':
restart(user, args[0])
elif action == 'user-current':
result = current(user, args[0])
elif action == 'set-user-current':
current(user, args[0], version)
elif action == 'list-user-extensions':
result = list_extensions(user, args[0])
else:
alias = args[0]
if ext_list == '-':
_exts = list_extensions(user, alias)
for extension in _exts:
try:
if action == 'enable-user-extensions':
install(user, alias, extension)
elif action == 'disable-user-extensions':
uninstall(user, alias, extension)
elif action == 'replace-user-extensions':
update(user, alias, extension)
result.update({extension: {'status': 'OK'}})
ok += 1
except (ValueError, ClSelectExcept.ExternalProgramFailed) as err:
result.update(
{extension: {'status': 'ERROR', 'message': str(err)}})
error += 1
except BaseClSelectException as err:
clprint.print_diag(fmt, {'status': 'ERROR', 'message': str(err)})
sys.exit(1)
except Exception as err:
msg = traceback.format_exc()
clprint.print_diag(fmt, {'status': 'ERROR', 'message': msg})
sys.exit(1)
if not result and print_summary:
result = summary(user)
if error and ok:
status = 'PARTIAL'
exit_status = 2
elif error:
status = 'ERROR'
exit_status = 5
else:
if ext_list and ok < 2:
if print_summary:
clprint.print_data(fmt, summary(user))
else:
clprint.print_data(fmt, {})
else:
clprint.print_data(fmt, result)
sys.exit(0)
message = '\n'.join(
'%s: %s' % (k, v.get('message', v.get('status', '')))
for k, v in iteritems(result))
clprint.print_diag(fmt, {'status': status, 'message': message})
sys.exit(exit_status)