code to help testing NPN support in Twisted

tls_protocol_negociation_twisted.py

from OpenSSL import SSL
from twisted.internet import ssl, protocol, defer, endpoints, reactor, task

def main(reactor):

    options = ssl.optionsForClientTLS(
        hostname=u'http2bin.org',
        extraCertificateOptions={'nextProtocols': [b'h2', b'http/1.1']}
    )

    def info_callback(conn, where, ret):
        # conn is a OpenSSL.SSL.Connection
        # where is a set of flags telling where in the handshake we are
        # http://www.openssl.org/docs/ssl/SSL_CTX_set_info_callback.html
        if where & SSL.SSL_CB_HANDSHAKE_START:
            conn.set_app_data({'handshake_done': False})
            print('handshake start')
        if where & SSL.SSL_CB_HANDSHAKE_DONE:
            conn.set_app_data({'handshake_done': True})
            print('handshake done')

    options._ctx.set_info_callback(info_callback)

    class BasicH2Request(protocol.Protocol):
        def connectionMade(self):
            print("connection made")
            self.complete = defer.Deferred()
            # Write some data to trigger the SSL handshake.
            self.transport.write(b'SOMETHING\r\n\r\n')

        def dataReceived(self, data):
            print('Next protocol is: %s' % (self.transport.getNextProtocol(),))
            self.transport.loseConnection()
            self.complete.callback(None)
            self.complete = None
        def connectionLost(self, reason):
            print("connection lost %s" % (reason,))
            if self.complete is not None:
                self.complete.callback(None)

    return endpoints.connectProtocol(
        endpoints.SSL4ClientEndpoint(reactor, 'http2bin.org', 443, options),
        BasicH2Request()
    ).addCallback(lambda protocol: protocol.complete)

task.react(main)