404

[ Avaa Bypassed ]




Upload:

Command:

elspacio@3.147.45.159: ~ $
import socket


class TransportSocket:

    """A socket-like wrapper for exposing real transport sockets.

    These objects can be safely returned by APIs like
    `transport.get_extra_info('socket')`.  All potentially disruptive
    operations (like "socket.close()") are banned.
    """

    __slots__ = ('_sock',)

    def __init__(self, sock: socket.socket):
        self._sock = sock

    @property
    def family(self):
        return self._sock.family

    @property
    def type(self):
        return self._sock.type

    @property
    def proto(self):
        return self._sock.proto

    def __repr__(self):
        s = (
            f"<asyncio.TransportSocket fd={self.fileno()}, "
            f"family={self.family!s}, type={self.type!s}, "
            f"proto={self.proto}"
        )

        if self.fileno() != -1:
            try:
                laddr = self.getsockname()
                if laddr:
                    s = f"{s}, laddr={laddr}"
            except socket.error:
                pass
            try:
                raddr = self.getpeername()
                if raddr:
                    s = f"{s}, raddr={raddr}"
            except socket.error:
                pass

        return f"{s}>"

    def __getstate__(self):
        raise TypeError("Cannot serialize asyncio.TransportSocket object")

    def fileno(self):
        return self._sock.fileno()

    def dup(self):
        return self._sock.dup()

    def get_inheritable(self):
        return self._sock.get_inheritable()

    def shutdown(self, how):
        # asyncio doesn't currently provide a high-level transport API
        # to shutdown the connection.
        self._sock.shutdown(how)

    def getsockopt(self, *args, **kwargs):
        return self._sock.getsockopt(*args, **kwargs)

    def setsockopt(self, *args, **kwargs):
        self._sock.setsockopt(*args, **kwargs)

    def getpeername(self):
        return self._sock.getpeername()

    def getsockname(self):
        return self._sock.getsockname()

    def getsockbyname(self):
        return self._sock.getsockbyname()

    def settimeout(self, value):
        if value == 0:
            return
        raise ValueError(
            'settimeout(): only 0 timeout is allowed on transport sockets')

    def gettimeout(self):
        return 0

    def setblocking(self, flag):
        if not flag:
            return
        raise ValueError(
            'setblocking(): transport sockets cannot be blocking')

Filemanager

Name Type Size Permission Actions
__pycache__ Folder 0755
__init__.py File 1.16 KB 0644
__main__.py File 3.3 KB 0644
base_events.py File 73.62 KB 0644
base_futures.py File 1.96 KB 0644
base_subprocess.py File 8.66 KB 0644
base_tasks.py File 2.58 KB 0644
constants.py File 1.29 KB 0644
coroutines.py File 3.32 KB 0644
events.py File 27.97 KB 0644
exceptions.py File 1.71 KB 0644
format_helpers.py File 2.35 KB 0644
futures.py File 13.88 KB 0644
locks.py File 18.57 KB 0644
log.py File 124 B 0644
mixins.py File 481 B 0644
proactor_events.py File 32.48 KB 0644
protocols.py File 6.79 KB 0644
queues.py File 7.79 KB 0644
runners.py File 6.68 KB 0644
selector_events.py File 44.34 KB 0644
sslproto.py File 31 KB 0644
staggered.py File 5.85 KB 0644
streams.py File 26.86 KB 0644
subprocess.py File 7.5 KB 0644
taskgroups.py File 8.27 KB 0644
tasks.py File 33.63 KB 0644
threads.py File 790 B 0644
timeouts.py File 5.2 KB 0644
transports.py File 10.47 KB 0644
trsock.py File 2.42 KB 0644
unix_events.py File 50.7 KB 0644
windows_events.py File 33.88 KB 0644
windows_utils.py File 4.94 KB 0644