| .. | |||||
| __pycache__ | |||||
| actions | |||||
| SSL.py | |||||
| SmartIO.py | |||||
| UserDictCase.py | |||||
| __init__.py | |||||
| connections.py | |||||
| i18n.py | |||||
| nonblocking.py | |||||
| rhnLockfile.py | |||||
| rpclib.py | |||||
| tb.py | |||||
| transports.py |
| Server IP : 84.32.84.139 / Your IP : 216.73.216.37 Web Server : LiteSpeed System : Linux sg-nme-web1517.main-hosting.eu 5.14.0-611.16.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Mon Dec 22 03:40:39 EST 2025 x86_64 User : u323805470 ( 323805470) PHP Version : 7.0.33 Disable Function : system, exec, shell_exec, passthru, mysql_list_dbs, ini_alter, dl, symlink, link, chgrp, leak, popen, apache_child_terminate, virtual, mb_send_mail MySQL : OFF | cURL : ON | WGET : ON | Perl : OFF | Python : OFF | Sudo : OFF | Pkexec : OFF Directory : /lib/python3.9/site-packages/rhn/ |
Upload File : |
#
# Helper transport objects
#
# Copyright (c) 2002--2016 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>
# Transport objects
import os
import sys
import time
from rhn import connections
from rhn.i18n import sstr, bstr
from rhn.SmartIO import SmartIO
from rhn.UserDictCase import UserDictCase
try: # python2
import xmlrpclib
from types import IntType, StringType, ListType
except ImportError: # python3
import xmlrpc.client as xmlrpclib
IntType = int
StringType = bytes
ListType = list
__version__ = "2.8.62.8.6-8.el98.el9"
# XXX
COMPRESS_LEVEL = 6
# Exceptions
class NotProcessed(Exception):
pass
class Transport(xmlrpclib.Transport):
user_agent = "rhn.rpclib.py/%s" % __version__
def __init__(self, transfer=0, encoding=0, refreshCallback=None,
progressCallback=None, use_datetime=None, timeout=None):
self._use_builtin_types = False
self._transport_flags = {'transfer' : 0, 'encoding' : 0}
self.set_transport_flags(transfer=transfer, encoding=encoding)
self._headers = UserDictCase()
self.verbose = 0
self.connection = None
self.method = "POST"
self._lang = None
self.refreshCallback = refreshCallback
self.progressCallback = progressCallback
self.bufferSize = 16384
self.headers_in = None
self.response_status = None
self.response_reason = None
self._redirected = None
self._use_datetime = use_datetime
self.timeout = timeout
# set the progress callback
def set_progress_callback(self, progressCallback, bufferSize=16384):
self.progressCallback = progressCallback
self.bufferSize = bufferSize
# set the refresh callback
def set_refresh_callback(self, refreshCallback):
self.refreshCallback = refreshCallback
# set the buffer size
# The bigger this is, the faster the read is, but the more seldom is the
# progress callback called
def set_buffer_size(self, bufferSize):
if bufferSize is None:
# No buffer size specified; go with 16k
bufferSize = 16384
self.bufferSize = bufferSize
# set the request method
def set_method(self, method):
if method not in ("GET", "POST"):
raise IOError("Unknown request method %s" % method)
self.method = method
# reset the transport options
def set_transport_flags(self, transfer=None, encoding=None, **kwargs):
# For backwards compatibility, we keep transfer and encoding as
# positional parameters (they could come in as kwargs easily)
self._transport_flags.update(kwargs)
if transfer is not None:
self._transport_flags['transfer'] = transfer
if encoding is not None:
self._transport_flags['encoding'] = encoding
self.validate_transport_flags()
def get_transport_flags(self):
return self._transport_flags.copy()
def validate_transport_flags(self):
# Transfer and encoding are guaranteed to be there
transfer = self._transport_flags.get('transfer')
transfer = lookupTransfer(transfer, strict=1)
self._transport_flags['transfer'] = transfer
encoding = self._transport_flags.get('encoding')
encoding = lookupEncoding(encoding, strict=1)
self._transport_flags['encoding'] = encoding
# Add arbitrary additional headers.
def set_header(self, name, arg):
if type(arg) in [ type([]), type(()) ]:
# Multivalued header
self._headers[name] = [str(a) for a in arg]
else:
self._headers[name] = str(arg)
def add_header(self, name, arg):
if name in self._headers:
vlist = self._headers[name]
if not isinstance(vlist, ListType):
vlist = [ vlist ]
else:
vlist = self._headers[name] = []
vlist.append(str(arg))
def clear_headers(self):
self._headers.clear()
def get_connection(self, host):
if self.verbose:
print("Connecting via http to %s" % (host, ))
if self.timeout:
return connections.HTTPConnection(host, timeout=self.timeout)
else:
return connections.HTTPConnection(host)
def request(self, host, handler, request_body, verbose=0):
# issue XML-RPC request
# XXX: automatically compute how to send depending on how much data
# you want to send
# XXX Deal with HTTP/1.1 if necessary
self.verbose = verbose
# implement BASIC HTTP AUTHENTICATION
host, extra_headers, x509 = self.get_host_info(host)
if not extra_headers:
extra_headers = []
# Establish the connection
connection = self.get_connection(host)
# Setting the user agent. Only interesting for SSL tunnels, in any
# other case the general headers are good enough.
connection.set_user_agent(self.user_agent)
if self.verbose:
connection.set_debuglevel(self.verbose - 1)
# Get the output object to push data with
req = Output(connection=connection, method=self.method)
req.set_transport_flags(**self._transport_flags)
# Add the extra headers
req.set_header('User-Agent', self.user_agent)
for header, value in list(self._headers.items()) + extra_headers:
# Output.set_header correctly deals with multivalued headers now
req.set_header(header, value)
# Content-Type
req.set_header("Content-Type", "text/xml")
req.process(request_body)
# Host and Content-Length are set by HTTP*Connection
for h in ['Content-Length', 'Host']:
req.clear_header(h)
headers, fd = req.send_http(host, handler)
if self.verbose:
print("Incoming headers:")
for header, value in headers.items():
print("\t%s : %s" % (header, value))
if fd.status in (301, 302):
self._redirected = headers["Location"]
self.response_status = fd.status
return None
# Save the headers
self.headers_in = headers
self.response_status = fd.status
self.response_reason = fd.reason
return self._process_response(fd, connection)
def _process_response(self, fd, connection):
# Now use the Input class in case we get an enhanced response
resp = Input(self.headers_in, progressCallback=self.progressCallback,
bufferSize=self.bufferSize)
fd = resp.decode(fd)
if isinstance(fd, InputStream):
# When the File object goes out of scope, so will the InputStream;
# that will eventually call the connection's close() method and
# cleanly reap it
f = File(fd.fd, fd.length, fd.name, bufferSize=self.bufferSize,
progressCallback=self.progressCallback)
# Set the File's close method to the connection's
# Note that calling the HTTPResponse's close() is not enough,
# since the main socket would remain open, and this is
# particularily bad with SSL
f.close = connection.close
return f
# We can safely close the connection now; if we had an
# application/octet/stream (for which Input.read passes the original
# socket object), Input.decode would return an InputStream,
# so we wouldn't reach this point
connection.close()
return self.parse_response(fd)
# Give back the new URL if redirected
def redirected(self):
return self._redirected
# Rewrite parse_response to provide refresh callbacks
def parse_response(self, f):
# read response from input file, and parse it
p, u = self.getparser()
while 1:
response = f.read(1024)
if not response:
break
if self.refreshCallback:
self.refreshCallback()
if self.verbose:
print("body:", repr(response))
p.feed(response)
f.close()
p.close()
return u.close()
def setlang(self, lang):
self._lang = lang
class SafeTransport(Transport):
def __init__(self, transfer=0, encoding=0, refreshCallback=None,
progressCallback=None, trusted_certs=None, timeout=None):
Transport.__init__(self, transfer, encoding,
refreshCallback=refreshCallback, progressCallback=progressCallback,
timeout=timeout)
self.trusted_certs = []
for certfile in (trusted_certs or []):
self.add_trusted_cert(certfile)
def add_trusted_cert(self, certfile):
if not os.access(certfile, os.R_OK):
raise ValueError("Certificate file %s is not accessible" % certfile)
self.trusted_certs.append(certfile)
def get_connection(self, host):
# implement BASIC HTTP AUTHENTICATION
host, extra_headers, x509 = self.get_host_info(host)
if self.verbose:
print("Connecting via https to %s" % (host, ))
if self.timeout:
return connections.HTTPSConnection(host,
trusted_certs=self.trusted_certs, timeout=self.timeout)
else:
return connections.HTTPSConnection(host,
trusted_certs=self.trusted_certs)
class ProxyTransport(Transport):
def __init__(self, proxy, proxyUsername=None, proxyPassword=None,
transfer=0, encoding=0, refreshCallback=None, progressCallback=None,
timeout=None):
Transport.__init__(self, transfer, encoding,
refreshCallback=refreshCallback, progressCallback=progressCallback,
timeout=timeout)
self._proxy = proxy
self._proxy_username = proxyUsername
self._proxy_password = proxyPassword
def get_connection(self, host):
if self.verbose:
print("Connecting via http to %s proxy %s, username %s, pass %s" % (
host, self._proxy, self._proxy_username, self._proxy_password))
if self.timeout:
return connections.HTTPProxyConnection(self._proxy, host,
username=self._proxy_username, password=self._proxy_password,
timeout=self.timeout)
else:
return connections.HTTPProxyConnection(self._proxy, host,
username=self._proxy_username, password=self._proxy_password)
class SafeProxyTransport(ProxyTransport):
def __init__(self, proxy, proxyUsername=None, proxyPassword=None,
transfer=0, encoding=0, refreshCallback=None,
progressCallback=None, trusted_certs=None, timeout=None):
ProxyTransport.__init__(self, proxy,
proxyUsername=proxyUsername, proxyPassword=proxyPassword,
transfer=transfer, encoding=encoding,
refreshCallback=refreshCallback,
progressCallback=progressCallback,
timeout=timeout)
self.trusted_certs = []
for certfile in (trusted_certs or []):
self.add_trusted_cert(certfile)
def add_trusted_cert(self, certfile):
if not os.access(certfile, os.R_OK):
raise ValueError("Certificate file %s is not accessible" % certfile)
self.trusted_certs.append(certfile)
def get_connection(self, host):
if self.verbose:
print("Connecting via https to %s proxy %s, username %s, pass %s" % (
host, self._proxy, self._proxy_username, self._proxy_password))
if self.timeout:
return connections.HTTPSProxyConnection(self._proxy, host,
username=self._proxy_username, password=self._proxy_password,
trusted_certs=self.trusted_certs, timeout=self.timeout)
else:
return connections.HTTPSProxyConnection(self._proxy, host,
username=self._proxy_username, password=self._proxy_password,
trusted_certs=self.trusted_certs)
# ============================================================================
# Extended capabilities for transport
#
# We allow for the following possible headers:
#
# Content-Transfer-Encoding:
# This header tells us how the POST data is encoded in what we read.
# If it is not set, we assume plain text that can be passed along
# without any other modification. If set, valid values are:
# - binary : straight binary data
# - base64 : will pass through base64 decoder to get the binary data
#
# Content-Encoding:
# This header tells us what should we do with the binary data obtained
# after acting on the Content-Transfer-Encoding header. Valid values:
# - x-gzip : will need to pass through GNU gunzip-like to get plain
# text out
# - x-zlib : this denotes the Python's own zlib bindings which are a
# datastream based on gzip, but not quite
# - x-gpg : will need to pass through GPG to get out the text we want
# ============================================================================
# Input class to automate reading the posting from the network
# Having to work with environment variables blows, though
class Input:
def __init__(self, headers=None, progressCallback=None, bufferSize=1024,
max_mem_size=16384):
self.transfer = None
self.encoding = None
self.type = None
self.length = 0
self.lang = "C"
self.name = ""
self.progressCallback = progressCallback
self.bufferSize = bufferSize
self.max_mem_size = max_mem_size
if not headers:
# we need to get them from environment
if "HTTP_CONTENT_TRANSFER_ENCODING" in os.environ:
self.transfer = os.environ["HTTP_CONTENT_TRANSFER_ENCODING"].lower()
if "HTTP_CONTENT_ENCODING" in os.environ:
self.encoding = os.environ["HTTP_CONTENT_ENCODING"].lower()
if "CONTENT-TYPE" in os.environ:
self.type = os.environ["CONTENT-TYPE"].lower()
if "CONTENT_LENGTH" in os.environ:
self.length = int(os.environ["CONTENT_LENGTH"])
if "HTTP_ACCEPT_LANGUAGE" in os.environ:
self.lang = os.environ["HTTP_ACCEPT_LANGUAGE"]
if "HTTP_X_PACKAGE_FILENAME" in os.environ:
self.name = os.environ["HTTP_X_PACKAGE_FILENAME"]
else:
# The stupid httplib screws up the headers from the HTTP repsonse
# and converts them to lowercase. This means that we have to
# convert to lowercase all the dictionary keys in case somebody calls
# us with sane values --gaftonc (actually mimetools is the culprit)
for header in headers.keys():
value = headers[header]
h = header.lower()
if h == "content-length":
try:
self.length = int(value)
except ValueError:
self.length = 0
elif h == "content-transfer-encoding":
# RFC 2045 #6.1: case insensitive
self.transfer = value.lower()
elif h == "content-encoding":
# RFC 2616 #3.5: case insensitive
self.encoding = value.lower()
elif h == "content-type":
# RFC 2616 #3.7: case insensitive
self.type = value.lower()
elif h == "accept-language":
# RFC 2616 #3.10: case insensitive
self.lang = value.lower()
elif h == "x-package-filename":
self.name = value
self.io = None
def read(self, fd = sys.stdin):
# The octet-streams are passed right back
if self.type == "application/octet-stream":
return
if self.length:
# Read exactly the amount of data we were told
self.io = _smart_read(fd, self.length,
bufferSize=self.bufferSize,
progressCallback=self.progressCallback,
max_mem_size=self.max_mem_size)
else:
# Oh well, no clue; read until EOF (hopefully)
self.io = _smart_total_read(fd)
if not self.transfer or self.transfer == "binary":
return
elif self.transfer == "base64":
import base64
old_io = self.io
old_io.seek(0, 0)
self.io = SmartIO(max_mem_size=self.max_mem_size)
base64.decode(old_io, self.io)
else:
raise NotImplementedError(self.transfer)
def decode(self, fd = sys.stdin):
# The octet-stream data are passed right back
if self.type == "application/octet-stream":
return InputStream(fd, self.length, self.name, close=fd.close)
if not self.io:
self.read(fd)
# At this point self.io exists (the only case when self.read() does
# not initialize self.io is when content-type is
# "application/octet-stream" - and we already dealt with that case
# We can now close the file descriptor
if hasattr(fd, "close"):
fd.close()
# Now we have the binary goo
if not self.encoding or self.encoding == "__plain":
# all is fine.
pass
elif self.encoding in ("x-zlib", "deflate"):
import zlib
obj = zlib.decompressobj()
self.io.seek(0, 0)
data = obj.decompress(self.io.read()) + obj.flush()
del obj
self.length = len(data)
self.io = SmartIO(max_mem_size=self.max_mem_size)
self.io.write(data)
elif self.encoding in ("x-gzip", "gzip"):
import gzip
self.io.seek(0, 0)
gz = gzip.GzipFile(mode="rb", compresslevel = COMPRESS_LEVEL,
fileobj=self.io)
data = gz.read()
self.length = len(data)
self.io = SmartIO(max_mem_size=self.max_mem_size)
self.io.write(data)
elif self.encoding == "x-gpg":
# XXX: should be written
raise NotImplementedError(self.transfer, self.encoding)
else:
raise NotImplementedError(self.transfer, self.encoding)
# Play nicely and rewind the file descriptor
self.io.seek(0, 0)
return self.io
def getlang(self):
return self.lang
# Utility functions
def _smart_total_read(fd, bufferSize=1024, max_mem_size=16384):
"""
Tries to read data from the supplied stream, and puts the results into a
StmartIO object. The data will be in memory or in a temporary file,
depending on how much it's been read
Returns a SmartIO object
"""
io = SmartIO(max_mem_size=max_mem_size)
while 1:
chunk = fd.read(bufferSize)
if not chunk:
# EOF reached
break
io.write(chunk)
return io
def _smart_read(fd, amt, bufferSize=1024, progressCallback=None,
max_mem_size=16384):
# Reads amt bytes from fd, or until the end of file, whichever
# occurs first
# The function will read in memory if the amout to be read is smaller than
# max_mem_size, or to a temporary file otherwise
#
# Unlike read(), _smart_read tries to return exactly the requested amount
# (whereas read will return _up_to_ that amount). Reads from sockets will
# usually reaturn less data, or the read can be interrupted
#
# Inspired by Greg Stein's httplib.py (the standard in python 2.x)
#
# support for progress callbacks added
startTime = time.time()
lastTime = startTime
buf = SmartIO(max_mem_size=max_mem_size)
origsize = amt
while amt > 0:
curTime = time.time()
l = min(bufferSize, amt)
chunk = fd.read(l)
# read guarantees that len(chunk) <= l
l = len(chunk)
if not l:
# Oops. Most likely EOF
break
# And since the original l was smaller than amt, we know amt >= 0
amt = amt - l
buf.write(chunk)
if progressCallback is None:
# No progress callback, so don't do fancy computations
continue
# We update the progress callback if:
# we haven't updated it for more than a secord, or
# it's the last read (amt == 0)
if curTime - lastTime >= 1 or amt == 0:
lastTime = curTime
# use float() so that we force float division in the next step
bytesRead = float(origsize - amt)
# if amt == 0, on a fast machine it is possible to have
# curTime - lastTime == 0, so add an epsilon to prevent a division
# by zero
speed = bytesRead / ((curTime - startTime) + .000001)
if origsize == 0:
secs = 0
else:
# speed != 0 because bytesRead > 0
# (if bytesRead == 0 then origsize == amt, which means a read
# of 0 length; but that's impossible since we already checked
# that l is non-null
secs = amt / speed
progressCallback(bytesRead, origsize, speed, secs)
# Now rewind the SmartIO
buf.seek(0, 0)
return buf
class InputStream:
def __init__(self, fd, length, name = "<unknown>", close=None):
self.fd = fd
self.length = int(length)
self.name = name
# Close function
self.close = close
def __repr__(self):
return "Input data is a stream of %d bytes for file %s.\n" % (self.length, self.name)
# ============================================================================
# Output class that will be used to build the temporary output string
class BaseOutput:
# DEFINES for instances use
# Content-Encoding
ENCODE_NONE = 0
ENCODE_GZIP = 1
ENCODE_ZLIB = 2
ENCODE_GPG = 3
# Content-Transfer-Encoding
TRANSFER_NONE = 0
TRANSFER_BINARY = 1
TRANSFER_BASE64 = 2
# Mappings to make things easy
encodings = [
[None, "__plain"], # ENCODE_NONE
["x-gzip", "gzip"], # ENCODE_GZIP
["x-zlib", "deflate"], # ENCODE_ZLIB
["x-gpg"], # ENCODE_GPG
]
transfers = [
None, # TRANSFER_NONE
"binary", # TRANSFRE_BINARY
"base64", # TRANSFER_BASE64
]
def __init__(self, transfer=0, encoding=0, connection=None, method="POST"):
# Assumes connection is an instance of HTTPConnection
if connection:
if not isinstance(connection, connections.HTTPConnection):
raise Exception("Expected an HTTPConnection type object")
self.method = method
# Store the connection
self._connection = connection
self.data = None
self.headers = UserDictCase()
self.encoding = 0
self.transfer = 0
self.transport_flags = {}
# for authenticated proxies
self.username = None
self.password = None
# Fields to keep the information about the server
self._host = None
self._handler = None
self._http_type = None
self._protocol = None
# Initialize self.transfer and self.encoding
self.set_transport_flags(transfer=transfer, encoding=encoding)
# internal flags
self.__processed = 0
def set_header(self, name, arg):
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.
self.headers[name] = ','.join(map(str, arg))
else:
self.headers[name] = str(arg)
def clear_header(self, name):
if name in self.headers:
del self.headers[name]
def process(self, data):
# Assume straight text/xml
self.data = data
# Content-Encoding header
if self.encoding == self.ENCODE_GZIP:
import gzip
encoding_name = self.encodings[self.ENCODE_GZIP][0]
self.set_header("Content-Encoding", encoding_name)
f = SmartIO(force_mem=1)
gz = gzip.GzipFile(mode="wb", compresslevel=COMPRESS_LEVEL,
fileobj = f)
if sys.version_info[0] == 3:
gz.write(bstr(data))
else:
gz.write(sstr(data))
gz.close()
self.data = f.getvalue()
f.close()
elif self.encoding == self.ENCODE_ZLIB:
import zlib
encoding_name = self.encodings[self.ENCODE_ZLIB][0]
self.set_header("Content-Encoding", encoding_name)
obj = zlib.compressobj(COMPRESS_LEVEL)
self.data = obj.compress(data) + obj.flush()
elif self.encoding == self.ENCODE_GPG:
# XXX: fix me.
raise NotImplementedError(self.transfer, self.encoding)
encoding_name = self.encodings[self.ENCODE_GPG][0]
self.set_header("Content-Encoding", encoding_name)
# Content-Transfer-Encoding header
if self.transfer == self.TRANSFER_BINARY:
transfer_name = self.transfers[self.TRANSFER_BINARY]
self.set_header("Content-Transfer-Encoding", transfer_name)
self.set_header("Content-Type", "application/binary")
elif self.transfer == self.TRANSFER_BASE64:
import base64
transfer_name = self.transfers[self.TRANSFER_BASE64]
self.set_header("Content-Transfer-Encoding", transfer_name)
self.set_header("Content-Type", "text/base64")
self.data = base64.encodestring(self.data)
self.set_header("Content-Length", len(self.data))
rpc_version = __version__
if len(__version__.split()) > 1:
rpc_version = __version__.split()[1]
# other headers
self.set_header("X-Transport-Info",
'Extended Capabilities Transport (C) Red Hat, Inc (version %s)' %
rpc_version)
self.__processed = 1
# reset the transport options
def set_transport_flags(self, transfer=0, encoding=0, **kwargs):
self.transfer = transfer
self.encoding = encoding
self.transport_flags.update(kwargs)
def send_http(self, host, handler="/RPC2"):
if not self.__processed:
raise NotProcessed
self._host = host
if self._connection is None:
raise Exception("No connection object found")
self._connection.connect()
# wrap self data into binary object, otherwise HTTPConnection.request
# will encode it as ISO-8859-1 https://docs.python.org/3/library/http.client.html#httpconnection-objects
self._connection.request(self.method, handler, body=bstr(self.data), headers=self.headers)
response = self._connection.getresponse()
if not self.response_acceptable(response):
raise xmlrpclib.ProtocolError("%s %s" %
(self._host, handler),
response.status, response.reason, response.msg)
# A response object has read() and close() methods, so we can safely
# pass the whole object back
return response.msg, response
def response_acceptable(self, response):
"""Returns true if the response is acceptable"""
if response.status == 200:
return 1
if response.status in (301, 302):
return 1
if response.status != 206:
return 0
# If the flag is not set, it's unacceptable
if not self.transport_flags.get('allow_partial_content'):
return 0
if response.msg['Content-Type'] != 'application/octet-stream':
# Don't allow anything else to be requested as a range, it could
# break the XML parser
return 0
return 1
def close(self):
if self._connection:
self._connection.close()
self._connection = None
def lookupTransfer(transfer, strict=0):
"""Given a string or numeric representation of a transfer, return the
transfer code"""
if transfer is None:
# Plain
return 0
if isinstance(transfer, IntType) and 0 <= transfer < len(Output.transfers):
return transfer
if isinstance(transfer, StringType):
for i in range(len(Output.transfers)):
if Output.transfers[i] == transfer.lower():
return i
if strict:
raise ValueError("Unsupported transfer %s" % transfer)
# Return default
return 0
def lookupEncoding(encoding, strict=0):
"""Given a string or numeric representation of an encoding, return the
encoding code"""
if encoding is None:
# Plain
return 0
if isinstance(encoding, IntType) and 0 <= encoding < len(Output.encodings):
return encoding
if isinstance(encoding, StringType):
for i in range(len(Output.encodings)):
if encoding.lower() in Output.encodings[i]:
return i
if strict:
raise ValueError("Unsupported encoding %s" % encoding)
# Return default
return 0
Output = BaseOutput
# File object
class File:
def __init__(self, file_obj, length = 0, name = None,
progressCallback=None, bufferSize=16384):
self.length = length
self.file_obj = file_obj
self.close = file_obj.close
self.bufferSize=bufferSize
self.name = ""
if name:
self.name = name[name.rfind("/")+1:]
self.progressCallback = progressCallback
def __len__(self):
return self.length
def read(self, amt=None):
# If they want to read everything, use _smart_read
if amt is None:
fd = self._get_file()
return fd.read()
return self.file_obj.read(amt)
def read_to_file(self, file):
"""Copies the contents of this File object into another file
object"""
fd = self._get_file()
while 1:
buf = fd.read(self.bufferSize)
if not buf:
break
if sys.version_info[0] == 3:
file.write(bstr(buf))
else:
file.write(sstr(buf))
return file
def _get_file(self):
"""Read everything into a temporary file and call the progress
callbacks if the file length is defined, or just reads till EOF"""
if self.length:
io = _smart_read(self.file_obj, self.length,
bufferSize=self.bufferSize,
progressCallback=self.progressCallback)
io.seek(0, 0)
else:
# Read everuthing - no callbacks involved
io = _smart_total_read(self.file_obj, bufferSize=self.bufferSize)
io.seek(0, 0)
return io
def __del__(self):
if self.close:
self.close()
self.close = None
| .. | |||||
| __pycache__ | |||||
| actions | |||||
| SSL.py | |||||
| SmartIO.py | |||||
| UserDictCase.py | |||||
| __init__.py | |||||
| connections.py | |||||
| i18n.py | |||||
| nonblocking.py | |||||
| rhnLockfile.py | |||||
| rpclib.py | |||||
| tb.py | |||||
| transports.py |