1#@!#!123s

JFIF 1#@!#!123s
: / lib / python3.12 / site-packages / rhn /
Filename : transports.py
back
#
# 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.5-1.el10.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)