← Back
Editing: rpclib.py
# # 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.3-1.module_el8.10.0+7020+fd2ced8d.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, use_https) ) self._reset_host_handler_and_type(uri) super().__init__(uri, transport, verbose=verbose) def get_proxy_uri(self, proxy_host, proxy_user, proxy_pass, use_https): if proxy_host is None: return None parts = urlparse('http://' + 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) if use_https: parts = parts._replace(scheme='https') else: parts = parts._replace(scheme='http') 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
Save File
Cancel