File Manager

✏️ /

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

Batal

⬅ site-packages 13 item

Nama	Tipe	Ukuran	Diubah	Aksi
📁__pycache__	dir	—	2025-02-27 15:06	📂 Buka
📁actions	dir	—	2026-03-21 15:47	📂 Buka
🐍SSL.py	py	10.8 KB	2022-12-07 13:15	✏️ 👁️
🐍SmartIO.py	py	2.1 KB	2018-02-19 08:18	✏️ 👁️
🐍UserDictCase.py	py	3 KB	2018-02-19 08:18	✏️ 👁️
🐍__init__.py	py	132 B	2018-02-19 08:18	✏️ 👁️
🐍connections.py	py	9.5 KB	2022-12-07 13:15	✏️ 👁️
🐍i18n.py	py	1.7 KB	2018-02-19 08:18	✏️ 👁️
🐍nonblocking.py	py	2.4 KB	2018-02-19 08:18	✏️ 👁️
🐍rhnLockfile.py	py	3.3 KB	2018-02-19 08:18	✏️ 👁️
🐍rpclib.py	py	23.7 KB	2022-12-07 13:15	✏️ 👁️
🐍tb.py	py	836 B	2018-02-19 08:18	✏️ 👁️
🐍transports.py	py	31.2 KB	2022-12-07 13:15	✏️ 👁️

✏️ Rename