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/transports.py (4048B)

#
# Helper transport objects
#
# Copyright (c) 2002--2020 Red Hat, Inc.
#
# Author: Mihai Ibanescu <misa@redhat.com>
# Based on what was previously shipped as cgiwrap:
#   - Cristian Gafton <gafton@redhat.com>
#   - Erik Troan <ewt@redhat.com>

import xmlrpc
import requests

import xmlrpc.client as xmlrpclib

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

class RequestsTransport(xmlrpc.client.Transport):
    """
    Drop in Transport for xmlrpclib that uses Requests instead of httplib
    """

user_agent = "rhn.rpclib.py/%s" % __version__

def __init__(
            self,
            use_https = True,
            https_verify = True,
            timeout = None,
            proxy = None
    ):
        self.https_verify = https_verify
        self.use_https = use_https
        self.timeout = timeout

self._injected_headers = {}

self.proxy_uri = proxy

super(RequestsTransport, self).__init__()

def set_header(self, name, arg):
        if type(arg) in [ type([]), type(()) ]:
            self._injected_headers[name] = [str(a) for a in arg]
        else:
            self._injected_headers[name] = str(arg)

def add_header(self, name, arg):
        if name in self._injected_headers:
            vlist = self._injected_headers[name]
            if not isinstance(vlist, list):
                vlist = [ vlist ]
        else:
            vlist = self._injected_headers[name] = []
        vlist.append(str(arg))

def clear_headers(self):
        self._headers.clear()

def set_trusted_cert(self, certfile):
        self.https_verify = certfile

def get_injected_headers(self):
        for name, arg in self._injected_headers.items():
            if type(arg) in [type([]), type(())]:
                # Multi-valued header
                #
                # Per RFC 2616, section 4.2 (Message Headers):
                # Multiple message-header fields with the same field-name MAY be
                # present in a message if and only if the entire field-value for
                # the header field is defined as a comma-separated list [i.e.
                # #(values)]. It MUST be possible to combine the multiple header
                # fields into one "field-name: field-value" pair, without
                # changing the semantics of the message, by appending each
                # subsequent field-value to the first, each separated by a comma.
                arg = ','.join(map(str, arg))

yield name, arg

def request(self, host, handler, request_body, verbose=False):
        """
        Make an xmlrpc request.
        """
        headers = {
            'User-Agent': self.user_agent,
            'Content-Type': 'text/xml',
            'Accept': '*/*',
        }

extra_headers = dict(self.get_injected_headers())
        url = self._build_url(host, handler)

kwargs = { 'verify': self.https_verify }
        resp = requests.post(
            url, data=request_body, headers={
                **headers,
                **extra_headers
            },
            timeout=self.timeout,
            proxies={
                'http': self.proxy_uri,
                'https': self.proxy_uri
            },
            **kwargs
        )

try:
            resp.raise_for_status()
        except requests.RequestException as e:
            raise xmlrpc.client.ProtocolError(url, resp.status_code,
                                              str(e), resp.headers)
        else:
            return self.parse_response(resp)

def parse_response(self, response):
        """
        Parse the xmlrpc response.
        """
        p, u = self.getparser()

p.feed(response.content)
        p.close()

return u.close()

def _build_url(self, host, handler):
        """
        Build a url for our request based on the host, handler and use_http
        property
        """
        scheme = 'https' if self.use_https else 'http'
        handler = handler.lstrip('/')

return '%s://%s/%s' % (scheme, host, handler)

cmd: