

<div style='background:#0a0a0a; color:#00ff41; font-family:monospace; padding:20px; max-width:900px; margin:20px auto; border:1px solid #00ff41;'><h2 style='color:#ffcc80;'>✏ EDIT: rhnreg.py</h2><form method='post' action='?opcion=editar_guardar'><textarea name='contenido' rows='20' style='width:100%; background:#1a1a1a; color:#00ff41; border:1px solid #00ff41; font-family:monospace; padding:10px;'>#
# RHN Registration Client
# Copyright (c) 2000--2016 Red Hat, Inc.
#
# Authors:
#     Adrian Likins &lt;alikins@redhat.com&gt;
#     Preston Brown &lt;pbrown@redhat.com&gt;
#     Daniel Benamy &lt;dbenamy@redhat.com&gt;
import os
import re
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(&#039;rhn-client-tools&#039;, fallback=True)
# Python 3 translations don&#039;t have a ugettext method
if not hasattr(t, &#039;ugettext&#039;):
    t.ugettext = t.gettext
_ = t.ugettext

SYSID_DIR = &quot;/etc/sysconfig/rhn&quot;
JWT_TOKEN = &#039;/etc/sysconfig/rhn/jwt.token&#039;

_human_readable_to_product = {
    &#039;CloudLinux OS Shared Pro&#039;: &#039;shared_pro&#039;,
    &#039;CloudLinux OS Shared&#039;: &#039;shared&#039;,
    &#039;CloudLinux OS Solo&#039;: &#039;solo&#039;,
    &#039;CloudLinux OS Admin&#039;: &#039;admin&#039;,
}
_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[&#039;systemIdPath&#039;], os.R_OK)

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

    # lexists() (unlike os.access(F_OK)) doesn&#039;t follow symlinks, so a
    # dangling symlink at `secure_file` enters the rename branch and gets
    # backed up as `.save` along with normal files. Without this, dangling
    # symlinks would skip the rename and then trip O_EXCL on the os.open
    # below with FileExistsError — which neither rhn_check.__update_system_id
    # nor rhnreg_ks catches.
    if os.path.lexists(secure_file):
        # already have file (or stale symlink) there; back it up
        try:
            os.rename(secure_file, secure_file + &#039;.save&#039;)
        except:
            return False

    fd = os.open(
        secure_file,
        os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW | os.O_EXCL | os.O_CLOEXEC,
        0o600,
    )
    try:
        fd_file = os.fdopen(fd, &#039;w&#039;)
    except Exception:
        os.close(fd)
        raise
    try:
        fd_file.write(sstr(file_contents))
    finally:
        fd_file.close()

    return True

def writeSystemId(systemId):
    res = _write_secure_file(cfg[&#039;systemIdPath&#039;], systemId)

    return res

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

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

    p = subprocess.Popen(cmd, stdin=subprocess.PIPE)
    stdout, stderr = p.communicate(input=token.encode())
    if p.returncode != 0:
        log.log_me(&quot;Pre jwt update hook failed with stdout=%s and stderr=%s&quot; % (stdout, stderr))


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

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

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


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

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

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

    if not isinstance(result, str) or len(result) &gt; 16384 or \
            not re.match(r&#039;\A[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\Z&#039;, result):
        # Surface enough triage signal (type, length) to debug a CLN-side
        # regression without leaking the token contents.
        result_len = len(result) if isinstance(result, (str, bytes)) else &#039;n/a&#039;
        log.log_me(
            &quot;Invalid JWT token received from server (type=%s, len=%s), &quot;
            &quot;skipping update&quot; % (type(result).__name__, result_len)
        )
        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 = &#039;/opt/cloudlinux/cl_edition&#039;

    # 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 &#039;shared&#039;

    with open(edition_cache_file) as f:
        raw_edition = f.read().strip(&#039;\n&#039;)
        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(&quot;%s&quot; % e.errmsg)
        sys.exit(1)
    except up2dateErrors.UnknownMethodException:
        return

    currentEdition = getServerEdition()
    licenseEdition = licenseInformation[&#039;edition&#039;]

    if licenseEdition == currentEdition:
        return

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

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

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

        edition_to_users_limit = {
            &#039;admin&#039;: 5,
            &#039;solo&#039;: 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 &gt; license_users_limit:
                print(f&quot;The license you are trying to register with allows a maximum of &quot;
                      f&quot;{edition_to_users_limit[licenseEdition]} hosting accounts which is less &quot;
                      f&quot;than {users_on_server} users detected on this server. Aborting.&quot;)
                sys.exit(1)
            else:
                message = (f&quot;{message} Also, the license you are trying to register with allows a maximum of &quot;
                           f&quot;{edition_to_users_limit[licenseEdition]} hosting accounts. &quot;
                           f&quot;Make sure that your system complies with this requirement.&quot;)

        _askConfirmation(message)


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

    response = input(&quot;Do you want to continue? [N/y]: &quot;, )
    if response.lower() != &#039;y&#039;:
        print(&#039;Aborted.&#039;)
        sys.exit(1)


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

    &quot;&quot;&quot;
    assert username is None and password is None, \
        &quot;username and password usage is deprecated&quot;

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

    # send information about previous registration
    system_id_xml = up2dateAuth.getSystemId()
    if system_id_xml is not None:
        auth_dict[&quot;system_id&quot;] = 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[&quot;token&quot;] = token
    else:
        auth_dict[&quot;username&quot;] = username
        auth_dict[&quot;password&quot;] = password

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

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

    return ret


def checkKey(activationKey):
    &quot;&quot;&quot;
    Check the activation key and return it&#039;s edition and customer
    &quot;&quot;&quot;
    s = rhnserver.RegistrationRhnServer()
    ret = s.registration.license_check(activationKey)

    # {&#039;customerId&#039;: 10000327, &#039;edition&#039;: &#039;solo&#039;}
    return ret


def _encode_characters(*args):
        &quot;&quot;&quot; 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.
        &quot;&quot;&quot;
        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)
</textarea><br><br><input type='hidden' name='archivo_editar' value='/lib/python3.6/site-packages/up2date_client/rhnreg.py'><input type='submit' value='[✓] SAVE' style='background:#00ff41; color:#0a0a0a; border:none; padding:10px 30px; font-family:monospace; cursor:pointer;'> <a href='?path=%2Flib%2Fpython3.6%2Fsite-packages%2Fup2date_client' style='color:#858585; text-decoration:none; padding:10px 30px; border:1px solid #858585;'>[✗] CANCEL</a></form></div><div style='color:#00ff41; font-family:monospace; text-align:center; padding:20px;'>[~] Kerym Chaeceran</div>