编辑 transports.py - 文件管理器

✏️ 编辑：transports.py

路径：/usr/lib/python3.9/site-packages/rhn/transports.py

大小：3.95 KB · 修改：2026-07-29 13:47:30 · 权限：0644 · 只读（保存会失败）

#
# 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)