Edit: /opt/cloudlinux/venv/bin/clcpapi (5409B)
#!/opt/cloudlinux/venv/bin/python3 -bb
# -*- coding: utf-8 -*-
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
"""
CloudLinux control panel API command line interface
"""
import argparse
import os
import pwd
from clcommon import cpapi
from clcommon.cloutput import prepare_data_json, prepare_attention_json
from clcommon.cpapi import cpapiexceptions
def cpapi_parser():
parser = argparse.ArgumentParser(description="%(prog)s - CloudLinux control panel API command line interface")
parser.add_argument('-j', '--json', action='store_true', default=False, help='Output in json format')
subparsers = parser.add_subparsers(dest='command')
docroot_parser = subparsers.add_parser('docroot')
docroot_parser.add_argument('domain', help='Return document root for domain')
userdomains_parser = subparsers.add_parser('userdomains')
userdomains_parser.add_argument('userdomains', help='Return domain and document root pairs for control panel user')
dblogin_cplogin_pairs_parser = subparsers.add_parser("dblogin_cplogin_pairs")
dblogin_cplogin_pairs_parser.add_argument("--cplogin_lst", nargs="*",
help="Return mapping for this system users")
dblogin_cplogin_pairs_parser.add_argument("--with_system_users", action="store_true", default=False,
help="Show system users in output")
reseller_users = subparsers.add_parser("reseller-users",
help="Show reseller users")
reseller_users.add_argument(
'resellername', type=str, nargs='?',
help='Reseller name')
return parser
def flat_plain(data, sep=' '):
return '\n'.join([sep.join(tuple_) for tuple_ in data])
def _enforce_authorization(namespace):
"""
clcpapi is world-executable (mode 755, not setuid) so it runs with the
invoking user's privileges. root and panel admins keep unrestricted
access; any other local caller may only read metadata it owns. Rejections
raise CPAPIException (clear error, non-zero exit, no data leaked); the
db->cp-user mapping is narrowed to the caller instead.
"""
if os.getuid() == 0:
return
try:
caller = pwd.getpwuid(os.getuid()).pw_name
except KeyError as exc:
# UID has no passwd entry (LDAP gap / orphaned UID): fail closed
raise cpapiexceptions.NoPanelUser(
f"Caller UID {os.getuid()} has no passwd entry; access denied") from exc
if cpapi.is_admin(caller):
return
if namespace.command == 'docroot':
if cpapi.domain_owner(namespace.domain) != caller:
raise cpapiexceptions.NoDomain(
f"Can't obtain document root for domain '{namespace.domain}'")
elif namespace.command == 'userdomains':
if namespace.userdomains != caller:
raise cpapiexceptions.NoPanelUser(
f"Not authorized to list domains of user '{namespace.userdomains}'")
elif namespace.command == 'reseller-users':
if namespace.resellername != caller:
raise cpapiexceptions.NoPanelUser(
f"Not authorized to list users of reseller '{namespace.resellername}'")
elif namespace.command == 'dblogin_cplogin_pairs':
# restrict the mapping dump to the caller's own login
namespace.cplogin_lst = [caller]
def main():
parser = cpapi_parser()
namespace = parser.parse_args()
output_ = None
exit_code = 0
try:
_enforce_authorization(namespace)
if namespace.command == 'docroot':
domain, user = cpapi.docroot(namespace.domain)
if namespace.json:
output_ = prepare_data_json({'domain': domain, 'user': user})
else:
output_ = domain + ' ' + user
elif namespace.command == 'userdomains':
domain_docroot_pairs = cpapi.userdomains(namespace.userdomains)
if namespace.json:
data = [{'domain': domain_docroot[0],
'docroot': domain_docroot[1]}
for domain_docroot in domain_docroot_pairs]
output_ = prepare_data_json(data)
else:
output_ = flat_plain(domain_docroot_pairs)
elif namespace.command == 'dblogin_cplogin_pairs':
db_mapping = cpapi.dblogin_cplogin_pairs(namespace.cplogin_lst,
namespace.with_system_users)
if namespace.json:
output_ = prepare_data_json(db_mapping)
else:
output_ = flat_plain(db_mapping)
elif namespace.command == 'reseller-users':
users = cpapi.reseller_users(namespace.resellername)
if namespace.json:
output_ = prepare_data_json(users)
else:
output_ = '\n'.join(users)
except cpapiexceptions.CPAPIException as e_:
exit_code = 1
if namespace.json:
output_ = prepare_attention_json(str(e_))
else:
output_ = 'ERROR: ' + str(e_)
return output_, exit_code
if __name__ == '__main__':
import sys
output_, exit_code = main()
if output_:
print(output_)
if exit_code:
sys.exit(exit_code)