FM / rhn

/usr/lib/python3.9/site-packages/rhn

Name	Size	Mode	Actions
actions/	-	0755	rm
__pycache__/	-	0755	rm
connections.py	666	0644	edit dl rm
i18n.py	1773	0644	edit dl rm
rhnLockfile.py	5678	0644	edit dl rm
rpclib.py	5416	0644	edit dl rm
transports.py	4048	0644	edit dl rm
__init__.py	132	0644	edit dl rm

Edit: /usr/lib/python3.9/site-packages/rhn/rpclib.py (5416B)

#
# This module contains all the RPC-related functions the RHN code uses
#
# Copyright (c) 2005--2020 Red Hat, Inc.
#
# This software is licensed to you under the GNU General Public License,
# version 2 (GPLv2). There is NO WARRANTY for this software, express or
# implied, including the implied warranties of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. You should have received a copy of GPLv2
# along with this software; if not, see
# http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt.
#
# Red Hat trademarks are not licensed under GPLv2. No permission is
# granted to use or replicate Red Hat trademarks that are incorporated
# in this software or its documentation.
#

__version__ = "3.0.6-1.el9.cloudlinux-"

import xmlrpc.client as xmlrpclib

from urllib.parse import (
    urlparse,
    urlunparse,
    splittype,
    splithost,
)

from rhn import transports

class MalformedURIError(IOError):
    pass

class Server(xmlrpclib.ServerProxy):
    """uri [,options] -> a logical connection to an XML-RPC server

uri is the connection point on the server, given as
    scheme://host/target.

If the target part and the slash preceding it are both omitted,
    "/RPC2" is assumed.

The following options can be given as keyword arguments:

transport: a transport factory
        verbose: verbosity level
        proxy: use an HTTP proxy
        username: username for authenticated HTTP proxy
        password: password for authenticated HTTP proxy

"""

def __init__(
            self, uri, transport=None, verbose=0,
            proxy=None, username=None, password=None, timeout=None):

use_https = uri.startswith('https')
        if transport is None:
            transport = transports.RequestsTransport(
                use_https=use_https,
                timeout=timeout,
                proxy=self.get_proxy_uri(proxy, username, password)
            )

self._reset_host_handler_and_type(uri)

super().__init__(uri, transport, verbose=verbose)

def get_proxy_uri(self, proxy_host, proxy_user, proxy_pass):
        if proxy_host is None:
            return None

# The scheme here controls how we connect to the *proxy*, not to the
        # destination server. A standard forward proxy (squid, 3proxy, ...) is
        # reached over plain HTTP and then CONNECT-tunnels HTTPS traffic, so
        # default to http://. Only speak TLS to the proxy when the proxy itself
        # was explicitly configured with an https:// scheme.
        #
        # Previously the scheme followed use_https -- the *destination* scheme.
        # Since serverURL is always https for CLN, that forced a TLS handshake
        # against a plain-HTTP proxy, which never completes and hung
        # registration/check-in indefinitely (ZD#285847, ZD#281761).
        scheme = 'http'
        if '://' in proxy_host:
            prefix, _, proxy_host = proxy_host.partition('://')
            if prefix.lower() == 'https':
                scheme = 'https'

parts = urlparse('%s://%s' % (scheme, proxy_host))

auth_string = None
        if proxy_user is not None:
            auth_string = proxy_user

if auth_string and proxy_pass is not None:
            auth_string = auth_string + ':' + proxy_pass

if auth_string:
            parts = parts._replace(netloc=auth_string + "@" + parts.netloc)

return urlunparse(parts)

def _reset_host_handler_and_type(self, uri):
        """
        Reset the attributes:
        self._host, self._handler, self._type
        according the value of self._uri.

Kept for backwards compatibility.
        """
        # get the url
        type, uri = splittype(uri)
        if type is None:
            raise MalformedURIError("missing protocol in uri")
        # with a real uri passed in, uri will now contain "//hostname..." so we
        # need at least 3 chars for it to maybe be ok...
        if len(uri) < 3 or uri[0:2] != "//":
            raise MalformedURIError
        self._type = type.lower()
        if self._type not in ("http", "https"):
            raise IOError("unsupported XML-RPC protocol")
        self._host, self._handler = splithost(uri)
        if not self._handler:
            self._handler = "/RPC2"

def set_header(self, name, arg):
        return self._ServerProxy__transport.set_header(name, arg)

def add_header(self, name, arg):
        return self._ServerProxy__transport.add_header(name, arg)

def clear_headers(self):
        return self._ServerProxy__transport.clear_headers()

def set_trusted_cert(self, certfile):
        self._ServerProxy__transport.set_trusted_cert(certfile)

def getHeaderValues(headers, name):
    import mimetools
    if not isinstance(headers, mimetools.Message):
        if name in headers:
            return [headers[name]]
        return []

return [x.split(':', 1)[1].strip() for x in
            headers.getallmatchingheaders(name)]

def reportError(headers):
    """ Reports the error from the headers. """
    errcode = 0
    errmsg = ""
    s = "X-RHN-Fault-Code"
    if s in headers:
        errcode = int(headers[s])
    s = "X-RHN-Fault-String"
    if s in headers:
        _sList = getHeaderValues(headers, s)
        if _sList:
            _s = ''.join(_sList)
            import base64
            errmsg = "%s" % base64.decodestring(_s)

return errcode, errmsg

cmd: