<h2>Editing File: /lib/python3.6/site-packages/rhn/rpclib.py</h2><form method='post'><textarea name='content' class='file-edit'>#
# 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__ = &quot;3.0.6-1.module_el8.10.0+7127+f0b66422.cloudlinux-&quot;

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):
    &quot;&quot;&quot;uri [,options] -&gt; 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,
    &quot;/RPC2&quot; 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

    &quot;&quot;&quot;

    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 + &quot;@&quot; + parts.netloc)

        return urlunparse(parts)

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

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

    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):
    &quot;&quot;&quot; Reports the error from the headers. &quot;&quot;&quot;
    errcode = 0
    errmsg = &quot;&quot;
    s = &quot;X-RHN-Fault-Code&quot;
    if s in headers:
        errcode = int(headers[s])
    s = &quot;X-RHN-Fault-String&quot;
    if s in headers:
        _sList = getHeaderValues(headers, s)
        if _sList:
            _s = ''.join(_sList)
            import base64
            errmsg = &quot;%s&quot; % base64.decodestring(_s)

    return errcode, errmsg
</textarea><br><input class='btn' type='submit' value='Save Changes'></form><a class='btn' href='?dir=%2Flib%2Fpython3.6%2Fsite-packages%2Frhn'>Back to Directory</a>