File Manager

X7ROOT File Manager

Current Path: /lib/python3.6/site-packages/up2date_client

lib / python3.6 / site-packages / up2date_client /

?? ..
?? __init__.py(0 B)
?? __pycache__
?? clpwd.py(3.56 KB)
?? config.py(13.26 KB)
?? hardware.py(5.09 KB)
?? pkgplatform.py(309 B)
?? rhncli.py(7.46 KB)
?? rhnreg.py(10.52 KB)
?? rhnserver.py(8.54 KB)
?? rpcServer.py(11.1 KB)
?? statistics.py(1.27 KB)
?? up2dateAuth.py(10.49 KB)
?? up2dateErrors.py(7.99 KB)
?? up2dateLog.py(2.06 KB)
?? up2dateUtils.py(1.73 KB)

Editing: rhnreg.py

#
# RHN Registration Client
# Copyright (c) 2000--2016 Red Hat, Inc.
#
# Authors:
#     Adrian Likins <alikins@redhat.com>
#     Preston Brown <pbrown@redhat.com>
#     Daniel Benamy <dbenamy@redhat.com>
import os
import sys

from up2date_client import up2dateUtils
from up2date_client import up2dateErrors
from up2date_client import up2dateAuth
from up2date_client import rhnserver
from up2date_client import up2dateLog
from up2date_client import statistics
from up2date_client.pkgplatform import getPlatform
from rhn.i18n import ustr, sstr

try: # python2
    import urlparse
    import xmlrpclib
    from types import ListType, TupleType, StringType, UnicodeType, DictType, DictionaryType
except ImportError: # python3
    import urllib.parse as urlparse
    import xmlrpc.client as xmlrpclib
    ListType = list
    TupleType = tuple
    StringType = bytes
    UnicodeType = str
    DictType = dict
    DictionaryType = dict
    long = int

import gettext
t = gettext.translation('rhn-client-tools', fallback=True)
# Python 3 translations don't have a ugettext method
if not hasattr(t, 'ugettext'):
    t.ugettext = t.gettext
_ = t.ugettext

SYSID_DIR = "/etc/sysconfig/rhn"
JWT_TOKEN = '/etc/sysconfig/rhn/jwt.token'

_human_readable_to_product = {
    'CloudLinux OS Shared Pro': 'shared_pro',
    'CloudLinux OS Shared': 'shared',
    'CloudLinux OS Solo': 'solo',
    'CloudLinux OS Admin': 'admin',
}
_product_to_human_readable = {
    v: k for k, v in _human_readable_to_product.items()}

from up2date_client import config
cfg = config.initUp2dateConfig()
log = up2dateLog.initLog()

def registered():
    return os.access(cfg['systemIdPath'], os.R_OK)

def _write_secure_file(secure_file, file_contents):
    """ Write a file to disk that is not readable by other users. """
    dir_name = os.path.dirname(secure_file)
    if not os.access(dir_name, os.W_OK):
        return False

if os.access(secure_file, os.F_OK):
        # already have file there; let's back it up
        try:
            os.rename(secure_file, secure_file + '.save')
        except:
            return False

fd = os.open(secure_file, os.O_WRONLY | os.O_CREAT, int('0600', 8))
    fd_file = os.fdopen(fd, 'w')
    try:
        fd_file.write(sstr(file_contents))
    finally:
        fd_file.close()

return True

def writeSystemId(systemId):
    res = _write_secure_file(cfg['systemIdPath'], systemId)

return res

def _execute_pre_jwt_update_hook(token: str, allowTransition: bool):
    """
    Execute binary file which we use as hook for jwt token updates
    """
    import subprocess, os
    # during cldeploy we don't have python or any other packages, so just exit
    if not os.path.exists('/opt/cloudlinux/venv/bin/python3'):
        return

cmd = ['/usr/sbin/cl-pre-jwt-update', '--new-token', token]
    if allowTransition:
        cmd.append('--allow-transition')

p = subprocess.Popen(cmd)
    stdout, stderr = p.communicate()
    if p.returncode != 0:
        log.log_me("Pre jwt update hook failed with stdout=%s and stderr=%s" % (stdout, stderr))

def _execute_post_jwt_update_hook(allowTransition: bool):
    """
    Execute binary file which we use as hook for jwt token updates
    """
    import subprocess, os
    # during cldeploy we don't have python or any other packages, so just exit
    if not os.path.exists('/opt/cloudlinux/venv/bin/python3'):
        return

cmd = ['/usr/sbin/cl-post-jwt-update']
    if allowTransition:
        cmd.append('--allow-transition')

p = subprocess.Popen(cmd)
    stdout, stderr = p.communicate()
    if p.returncode != 0:
        log.log_me("Post jwt update hook failed with stdout=%s and stderr=%s" % (stdout, stderr))

def getAndWriteJWTTokenToFile(systemId, allowTransition = False):
    """
    Get a JWT token from CLN and save it to the file
    :param systemId: content of file `/etc/sysconfig/rhn/systemid`
    :return: None
    """

xmlrpm_server = rhnserver.RhnServer()
    try:
        result = xmlrpm_server.up2date.getJWTToken(systemId)

except up2dateErrors.UnknownMethodException:
        # if CLN doesn't have this method we do nothing
        return
    except (up2dateErrors.AuthenticationTicketError,
            up2dateErrors.RhnUuidUniquenessError,
            up2dateErrors.CommunicationError,
            up2dateErrors.AuthenticationOrAccountCreationError):
        log.log_exception(*sys.exc_info())
        return

_execute_pre_jwt_update_hook(result, allowTransition)
    _write_secure_file(JWT_TOKEN, result)
    _execute_post_jwt_update_hook(allowTransition)

def getServerEdition(human_readable: bool = False):
    edition_cache_file = '/opt/cloudlinux/cl_edition'

# edition cache does not exist only in the case when
    # we did not register server yet
    if not os.path.exists(edition_cache_file):
        return 'shared'

with open(edition_cache_file) as f:
        raw_edition = f.read().strip('\n')
        if human_readable:
            return raw_edition
        return _human_readable_to_product[raw_edition]

def checkLicenseKey(activationKey, strictEdition, silentMigration):
    try:
        licenseInformation = checkKey(activationKey)
    except up2dateErrors.CommunicationError as e:
        print("%s" % e.errmsg)
        sys.exit(1)
    except up2dateErrors.UnknownMethodException:
        return

currentEdition = getServerEdition()
    licenseEdition = licenseInformation['edition']

if licenseEdition == currentEdition:
        return

if strictEdition:
        print(
            "WARNING: Automatic registration in yum transactions is "
            "only available when edition matches the provided license. "
            "Your current edition is {current_edition} and your license is {new_edition}.".format(
                current_edition=_product_to_human_readable[currentEdition],
                new_edition=_product_to_human_readable[licenseEdition]
            )
        )
        print("Run clnreg_ks manually to complete registration.")
        sys.exit(1)

if not silentMigration:
        if not sys.stdin.isatty():
            print('Error: interactive input required for edition migration, but tool '
                  'is running in non-interactive mode. Please try running the tool again '
                  'in interactive shell or add `--migrate-silently` flag to accept all'
                  'questions and perform the edition migration silently.')
            exit(1)

message = (f"{_product_to_human_readable[currentEdition]} edition installed on your server "
                   f"does not match license you are trying to register server with: "
                   f"{_product_to_human_readable[licenseEdition]}. Migration is required. "
                   f"You may lose access to the services which are not supported by the new edition.")

edition_to_users_limit = {
            'admin': 5,
            'solo': 1
        }
        license_users_limit = edition_to_users_limit.get(licenseEdition)
        if license_users_limit is not None:
            users_on_server = statistics.count_server_users()

if users_on_server > license_users_limit:
                print(f"The license you are trying to register with allows a maximum of "
                      f"{edition_to_users_limit[licenseEdition]} hosting accounts which is less "
                      f"than {users_on_server} users detected on this server. Aborting.")
                sys.exit(1)
            else:
                message = (f"{message} Also, the license you are trying to register with allows a maximum of "
                           f"{edition_to_users_limit[licenseEdition]} hosting accounts. "
                           f"Make sure that your system complies with this requirement.")

_askConfirmation(message)

def _askConfirmation(confirmationMessage: str):
    """
    Prints message and makes sure that client is ready for edition migration.
    """
    print(confirmationMessage)

response = input("Do you want to continue? [N/y]: ", )
    if response.lower() != 'y':
        print('Aborted.')
        sys.exit(1)

def registerSystem(username = None, password = None,
                   profileName = None,
                   token = None, other = None, edition=None):
    """Wrapper for the old xmlrpc to register a system. Activates subscriptions
    if a reg num is given.

"""
    assert username is None and password is None, \
        "username and password usage is deprecated"

auth_dict = { "profile_name" : profileName,
                  "os_release" : up2dateUtils.getVersion(),
                  "release_name" : up2dateUtils.getOSRelease(),
                  "architecture" : up2dateUtils.getArch()
                }

# send information about previous registration
    system_id_xml = up2dateAuth.getSystemId()
    if system_id_xml is not None:
        auth_dict["system_id"] = system_id_xml

# dict of other bits to send
    if other:
        for (key, item) in other.items():
            auth_dict[key] = item
    if token:
        auth_dict["token"] = token
    else:
        auth_dict["username"] = username
        auth_dict["password"] = password

if edition is not None:
        auth_dict['edition'] = edition
    else:
        auth_dict['edition'] = 'solo' \
            if os.path.exists('/etc/cloudlinux-edition-solo') \
            else 'admin' if os.path.exists('/etc/cloudlinux-edition-admin') else 'shared'

s = rhnserver.RegistrationRhnServer()
    ret = s.registration.new_system(auth_dict)

return ret

def checkKey(activationKey):
    """
    Check the activation key and return it's edition and customer
    """
    s = rhnserver.RegistrationRhnServer()
    ret = s.registration.license_check(activationKey)

# {'customerId': 10000327, 'edition': 'solo'}
    return ret

def _encode_characters(*args):
        """ All the data we gathered from dmi, bios, gudev are in utf-8,
            we need to convert characters beyond ord(127) - e.g \xae to unicode.
        """
        result=[]
        for item in args:
            item_type = type(item)
            if item_type == StringType:
                item = ustr(item)
            elif item_type == TupleType:
                item = tuple(_encode_characters(i) for i in item)
            elif item_type == ListType:
                item = [_encode_characters(i) for i in item]
            elif item_type == DictType or item_type == DictionaryType:
                item = dict([(_encode_characters(name, val)) for name, val in item.items()])
            # else: numbers or UnicodeType - are safe
            result.append(item)
        if len(result) == 1:
            return result[0]
        else:
            return tuple(result)

X7ROOT File Manager

Editing: rhnreg.py

Upload File

Create Folder