Python OpenSSL.SSL 模块，SENT_SHUTDOWN 实例源码

def test_shutdown(self):
        """
        `Connection.shutdown` performs an SSL-level connection shutdown.
        """
        server, client = loopback()
        assert not server.shutdown()
        assert server.get_shutdown() == SENT_SHUTDOWN
        with pytest.raises(ZeroReturnError):
            client.recv(1024)
        assert client.get_shutdown() == RECEIVED_SHUTDOWN
        client.shutdown()
        assert client.get_shutdown() == (SENT_SHUTDOWN | RECEIVED_SHUTDOWN)
        with pytest.raises(ZeroReturnError):
            server.recv(1024)
        assert server.get_shutdown() == (SENT_SHUTDOWN | RECEIVED_SHUTDOWN)

def test_shutdown(self):
        """
        :py:obj:`Connection.shutdown` performs an SSL-level connection shutdown.
        """
        server, client = self._loopback()
        self.assertFalse(server.shutdown())
        self.assertEquals(server.get_shutdown(), SENT_SHUTDOWN)
        self.assertRaises(ZeroReturnError, client.recv, 1024)
        self.assertEquals(client.get_shutdown(), RECEIVED_SHUTDOWN)
        client.shutdown()
        self.assertEquals(client.get_shutdown(), SENT_SHUTDOWN|RECEIVED_SHUTDOWN)
        self.assertRaises(ZeroReturnError, server.recv, 1024)
        self.assertEquals(server.get_shutdown(), SENT_SHUTDOWN|RECEIVED_SHUTDOWN)

def test_shutdown(self):
        """
        :py:obj:`Connection.shutdown` performs an SSL-level connection shutdown.
        """
        server, client = self._loopback()
        self.assertFalse(server.shutdown())
        self.assertEquals(server.get_shutdown(), SENT_SHUTDOWN)
        self.assertRaises(ZeroReturnError, client.recv, 1024)
        self.assertEquals(client.get_shutdown(), RECEIVED_SHUTDOWN)
        client.shutdown()
        self.assertEquals(client.get_shutdown(), SENT_SHUTDOWN|RECEIVED_SHUTDOWN)
        self.assertRaises(ZeroReturnError, server.recv, 1024)
        self.assertEquals(server.get_shutdown(), SENT_SHUTDOWN|RECEIVED_SHUTDOWN)

def test_shutdown(self):
        """
        L{Connection.shutdown} performs an SSL-level connection shutdown.
        """
        server, client = self._loopback()
        self.assertFalse(server.shutdown())
        self.assertEquals(server.get_shutdown(), SENT_SHUTDOWN)
        self.assertRaises(ZeroReturnError, client.recv, 1024)
        self.assertEquals(client.get_shutdown(), RECEIVED_SHUTDOWN)
        client.shutdown()
        self.assertEquals(client.get_shutdown(), SENT_SHUTDOWN|RECEIVED_SHUTDOWN)
        self.assertRaises(ZeroReturnError, server.recv, 1024)
        self.assertEquals(server.get_shutdown(), SENT_SHUTDOWN|RECEIVED_SHUTDOWN)

def _sendCloseAlert(self):
        # Okay, *THIS* is a bit complicated.

        # Basically, the issue is, OpenSSL seems to not actually return
        # errors from SSL_shutdown. Therefore, the only way to
        # determine if the close notification has been sent is by 
        # SSL_shutdown returning "done". However, it will not claim it's
        # done until it's both sent *and* received a shutdown notification.

        # I don't actually want to wait for a received shutdown
        # notification, though, so, I have to set RECEIVED_SHUTDOWN
        # before calling shutdown. Then, it'll return True once it's
        # *SENT* the shutdown.

        # However, RECEIVED_SHUTDOWN can't be left set, because then
        # reads will fail, breaking half close.

        # Also, since shutdown doesn't report errors, an empty write call is
        # done first, to try to detect if the connection has gone away.
        # (*NOT* an SSL_write call, because that fails once you've called
        # shutdown)
        try:
            os.write(self.socket.fileno(), '')
        except OSError, se:
            if se.args[0] in (EINTR, EWOULDBLOCK, ENOBUFS):
                return 0
            # Write error, socket gone
            return main.CONNECTION_LOST

        try:
            if hasattr(self.socket, 'set_shutdown'):
                laststate = self.socket.get_shutdown()
                self.socket.set_shutdown(laststate | SSL.RECEIVED_SHUTDOWN)
                done = self.socket.shutdown()
                if not (laststate & SSL.RECEIVED_SHUTDOWN):
                    self.socket.set_shutdown(SSL.SENT_SHUTDOWN)
            else:
                #warnings.warn("SSL connection shutdown possibly unreliable, "
                #              "please upgrade to ver 0.XX", category=UserWarning)
                self.socket.shutdown()
                done = True
        except SSL.Error, e:
            return e

        if done:
            self.stopWriting()
            # Note that this is tested for by identity below.
            return main.CONNECTION_DONE
        else:
            self.startWriting()
            return None

def _sendCloseAlert(self):
        # Okay, *THIS* is a bit complicated.

        # Basically, the issue is, OpenSSL seems to not actually return
        # errors from SSL_shutdown. Therefore, the only way to
        # determine if the close notification has been sent is by 
        # SSL_shutdown returning "done". However, it will not claim it's
        # done until it's both sent *and* received a shutdown notification.

        # I don't actually want to wait for a received shutdown
        # notification, though, so, I have to set RECEIVED_SHUTDOWN
        # before calling shutdown. Then, it'll return True once it's
        # *SENT* the shutdown.

        # However, RECEIVED_SHUTDOWN can't be left set, because then
        # reads will fail, breaking half close.

        # Also, since shutdown doesn't report errors, an empty write call is
        # done first, to try to detect if the connection has gone away.
        # (*NOT* an SSL_write call, because that fails once you've called
        # shutdown)
        try:
            os.write(self.socket.fileno(), '')
        except OSError, se:
            if se.args[0] in (EINTR, EWOULDBLOCK, ENOBUFS):
                return 0
            # Write error, socket gone
            return main.CONNECTION_LOST

        try:
            if hasattr(self.socket, 'set_shutdown'):
                laststate = self.socket.get_shutdown()
                self.socket.set_shutdown(laststate | SSL.RECEIVED_SHUTDOWN)
                done = self.socket.shutdown()
                if not (laststate & SSL.RECEIVED_SHUTDOWN):
                    self.socket.set_shutdown(SSL.SENT_SHUTDOWN)
            else:
                #warnings.warn("SSL connection shutdown possibly unreliable, "
                #              "please upgrade to ver 0.XX", category=UserWarning)
                self.socket.shutdown()
                done = True
        except SSL.Error, e:
            return e

        if done:
            self.stopWriting()
            # Note that this is tested for by identity below.
            return main.CONNECTION_DONE
        else:
            self.startWriting()
            return None