SSL socket wrapping a socket.socket. Instances returned by SSLContext.wrap_socket().
| 693 | |
| 694 | |
| 695 | class SSLSocket(socket): |
| 696 | """SSL socket wrapping a socket.socket. Instances returned by SSLContext.wrap_socket().""" |
| 697 | |
| 698 | def __init__(self, *args, **kwargs): |
| 699 | raise TypeError( |
| 700 | f"{self.__class__.__name__} does not have a public constructor. " |
| 701 | f"Instances are returned by SSLContext.wrap_socket()." |
| 702 | ) |
| 703 | |
| 704 | @classmethod |
| 705 | def _create( |
| 706 | cls, |
| 707 | sock, |
| 708 | server_side=False, |
| 709 | do_handshake_on_connect=True, |
| 710 | suppress_ragged_eofs=True, |
| 711 | server_hostname=None, |
| 712 | context=None, |
| 713 | session=None, |
| 714 | ): |
| 715 | import _socket |
| 716 | |
| 717 | ssl_sock = cls.__new__(cls) |
| 718 | |
| 719 | # Initialize the C-level _socket.socket with explicit family/type/proto |
| 720 | # to avoid getsockopt() auto-detection which fails on Emscripten's |
| 721 | # NodeSockFS file descriptors. |
| 722 | fd = sock.fileno() |
| 723 | _socket.socket.__init__(ssl_sock, sock.family, sock.type, sock.proto, fd) |
| 724 | ssl_sock._io_refs = 0 |
| 725 | ssl_sock._closed = False |
| 726 | |
| 727 | # Detach the original socket so it doesn't close the fd on __del__ |
| 728 | sock.detach() |
| 729 | |
| 730 | ssl_sock._sslobj = None |
| 731 | ssl_sock._context = context |
| 732 | ssl_sock._server_hostname = server_hostname |
| 733 | ssl_sock._server_side = server_side |
| 734 | ssl_sock._connected = True |
| 735 | ssl_sock._suppress_ragged_eofs = suppress_ragged_eofs |
| 736 | ssl_sock._session = session |
| 737 | |
| 738 | if do_handshake_on_connect: |
| 739 | ssl_sock.do_handshake() |
| 740 | return ssl_sock |
| 741 | |
| 742 | context = property(lambda self: getattr(self, "_context", None)) |
| 743 | server_hostname = property(lambda self: getattr(self, "_server_hostname", None)) |
| 744 | session = property(lambda self: getattr(self, "_session", None)) |
| 745 | session_reused = property(lambda self: None) |
| 746 | |
| 747 | def dup(self): |
| 748 | raise NotImplementedError("Can't dup() %s instances" % self.__class__.__name__) |
| 749 | |
| 750 | def read(self, len=1024, buffer=None): |
| 751 | if buffer is not None: |
| 752 | # `len` shadows the builtin here, so read straight into the buffer |
nothing calls this directly
no outgoing calls
no test coverage detected
searching dependent graphs…