def test_getExtension_with_duplicated_extensions(self):
        client_hello = ClientHello().create((3, 3), bytearray(1), bytearray(0),
                [], extensions=[TLSExtension().create(0, bytearray(0)),
                                SNIExtension().create(b'localhost')])

        with self.assertRaises(TLSInternalError):
            client_hello.getExtension(0)

    def generate(self, state):
        if self.version is None:
            self.version = state.client_version
        if self.random:
            state.client_random = self.random
        if self.session_id is None:
            self.session_id = state.session_id
        if not state.client_random:
            state.client_random = bytearray(32)

        extensions = None
        if self.extensions is not None:
            extensions = self._generate_extensions(state)

        clnt_hello = ClientHello().create(self.version,
                                          state.client_random,
                                          self.session_id,
                                          self.ciphers,
                                          extensions=extensions)
        clnt_hello.compression_methods = self.compression
        state.client_version = self.version

        self.msg = clnt_hello

        return clnt_hello

    def test_getExtension(self):
        client_hello = ClientHello().create((3, 3), bytearray(1), bytearray(0),
                [], extensions=[TLSExtension().create(0, bytearray(0))])

        ext = client_hello.getExtension(1)

        self.assertIsNone(ext)

    def test_parse_with_TACK_extension(self):
        p = Parser(bytearray(
            # we don't include the type of message as it is handled by the
            # hello protocol parser
            #b'x01' +             # type of message - client_hello
            b'\x00'*2 + b'\x2c' + # length - 44 bytes
            b'\x01\x01' +         # protocol version - arbitrary (invalid)
            b'\x00'*32 +          # client random
            b'\x00' +             # session ID length
            b'\x00'*2 +           # cipher suites length
            b'\x00' +             # compression methods length
            b'\x00\x04' +         # extensions length - 4 bytes
            b'\xf3\x00' +         # extension type - TACK (62208)
            b'\x00\x00'           # extension length - 0 bytes
            ))
        client_hello = ClientHello()
        client_hello = client_hello.parse(p)

        self.assertEqual((1,1), client_hello.client_version)
        self.assertEqual(bytearray(32), client_hello.random)
        self.assertEqual(bytearray(0), client_hello.session_id)
        self.assertEqual([], client_hello.cipher_suites)
        self.assertEqual([], client_hello.compression_methods)
        self.assertEqual(True, client_hello.tack)
        tack = TLSExtension().create(62208, bytearray(0))
        self.assertEqual([tack], client_hello.extensions)

    def __call__(self, hostname):
        """Generate a client hello object, use hostname in SNI extension."""
        # SNI is special in that we don't want to send it if it is empty
        if self.extensions:
            sni = next((x for x in self.extensions
                        if isinstance(x, SNIExtension)),
                       None)
            if sni:
                if hostname is not None:
                    if sni.serverNames is None:
                        sni.serverNames = []
                    sni.hostNames = [hostname]
                else:
                    # but if we were not provided with a host name, we want
                    # to remove empty extension
                    if sni.serverNames is None:
                        self.extensions = [x for x in self.extensions
                                           if not isinstance(x, SNIExtension)]

        if self.random:
            rand = self.random
        else:
            # we're not doing any crypto with it, just need "something"
            # TODO: place unix time at the beginning
            rand = numberToByteArray(random.getrandbits(256), 32)

        ch = ClientHello(self.ssl2).create(self.version, rand, self.session_id,
                                           self.ciphers,
                                           extensions=self.extensions)
        ch.compression_methods = self.compression_methods
        for cb in self.callbacks:
            ch = cb(ch)
        return ch

    def test_process_with_not_matching_signature_algorithms(self):
        exp = ExpectServerKeyExchange(valid_sig_algs=[(HashAlgorithm.sha256,
                                                       SignatureAlgorithm.rsa)])

        state = ConnectionState()
        state.cipher = CipherSuite.TLS_DHE_RSA_WITH_AES_128_CBC_SHA

        cert = Certificate(CertificateType.x509).\
                create(X509CertChain([X509().parse(srv_raw_certificate)]))

        private_key = parsePEMKey(srv_raw_key, private=True)
        client_hello = ClientHello()
        client_hello.client_version = (3, 3)
        client_hello.random = bytearray(32)
        state.client_random = client_hello.random
        state.handshake_messages.append(client_hello)
        server_hello = ServerHello()
        server_hello.server_version = (3, 3)
        server_hello.random = bytearray(32)
        state.server_random = server_hello.random
        # server hello is not necessary for the test to work
        #state.handshake_messages.append(server_hello)
        state.handshake_messages.append(cert)
        srv_key_exchange = DHE_RSAKeyExchange(\
                CipherSuite.TLS_DHE_RSA_WITH_AES_128_CBC_SHA,
                client_hello,
                server_hello,
                private_key)

        msg = srv_key_exchange.makeServerKeyExchange('sha1')

        with self.assertRaises(TLSIllegalParameterException):
            exp.process(state, msg)

    def test_update_binders_with_ticket(self):
        clientHello = ClientHello()
        clientHello.create((3, 3), bytearray(32), bytearray(0), [0])
        identities = [PskIdentity().create(bytearray(b'\x00ticket\x00ident'),
                                           123)]
        binders = [bytearray(48)]
        psk_ext = PreSharedKeyExtension().create(identities, binders)
        clientHello.extensions = [psk_ext]

        ticket = NewSessionTicket().create(3600,  # ticket lifetime
                                           123,   # age_add
                                           bytearray(b'\xc0' * 48),  # nonce
                                           bytearray(b'\x00ticket\x00ident'),
                                           [])
        hh = HandshakeHashes()
        resum_master_secret = bytearray(b'\x01' * 48)

        HandshakeHelpers.update_binders(clientHello, hh, [], [ticket],
                                        resum_master_secret)

        self.assertIsInstance(clientHello.extensions[-1],
                              PreSharedKeyExtension)

        ch_ext = clientHello.extensions[-1]
        self.assertEqual(ch_ext.identities, identities)
        self.assertEqual(ch_ext.binders,
                         [bytearray(b'<\x03\xcd\xd5\xce\xaeo\x8d\xc6\x8c\xe3'
                                    b'\xe3\xbc\xa2h\xdcm0+\xa7\xbe\xf7\x9ca-'
                                    b'\xcc\x0c\xdb\xb2ZtE\x1e:\xe2\xc4\xb8'
                                    b'\x1bd\x10wN\x8a\xb0\x90\x7f\xb1F')])

    def test_update_binders_default_prf(self):
        """
        Verify that configurations that don't specify the associated hash
        explicitly still work correctly (as the TLS 1.3 standard mandates
        that SHA-256 is used by default)
        """
        clientHello = ClientHello()
        clientHello.create((3, 3), bytearray(32), bytearray(0), [0])
        identities = [PskIdentity().create(bytearray(b'test'), 0)]
        binders = [bytearray(32)]
        psk_ext = PreSharedKeyExtension().create(identities, binders)
        clientHello.extensions = [psk_ext]

        hh = HandshakeHashes()

        pskConfigs = [(b'test', b'\x00\x12\x13')]

        HandshakeHelpers.update_binders(clientHello, hh, pskConfigs)

        self.assertIsInstance(clientHello.extensions[-1],
                              PreSharedKeyExtension)
        ch_ext = clientHello.extensions[-1]
        self.assertEqual(ch_ext.identities, identities)
        self.assertEqual(ch_ext.binders,
                         [bytearray(b'wOl\xbe\x9b\xca\xa4\xf3tS\x08M\ta\xa2t'
                                    b'\xa5lYF\xb7\x01F{M\xab\x85R\xa3'
                                    b'\xf3\x11^')])

    def test_parse(self):
        p = Parser(bytearray(
            # we don't include the type of message as it is handled by the
            # hello protocol parser
            #b'x01' +             # type of message - client_hello
            b'\x00'*2 + b'\x26' + # length - 38 bytes
            b'\x01\x01' +         # protocol version - arbitrary (invalid)
            b'\x00'*32 +          # client random
            b'\x00' +             # session ID length
            b'\x00'*2 +           # cipher suites length
            b'\x00'               # compression methods length
            ))
        client_hello = ClientHello()
        client_hello = client_hello.parse(p)

        self.assertEqual((1,1), client_hello.client_version)
        self.assertEqual(bytearray(32), client_hello.random)
        self.assertEqual(bytearray(0), client_hello.session_id)
        self.assertEqual([], client_hello.cipher_suites)
        self.assertEqual([], client_hello.compression_methods)
        self.assertEqual(bytearray(0), client_hello.server_name)
        # XXX not sent
        self.assertEqual([0], client_hello.certificate_types)
        self.assertEqual(False, client_hello.supports_npn)
        self.assertEqual(False, client_hello.tack)
        self.assertEqual(None, client_hello.srp_username)
        self.assertEqual(None, client_hello.extensions)

    def test_update_binders_wrong_last_ext(self):
        """
        PSK binders mandate that the PSK extension be the very last extension
        in client hello (as it's necessary to truncate the body of the hello
        up to the PSK extension and calculate hash over it)
        check if the updater will abort if the passed in message has
        PSK extension that is not last
        """
        clientHello = ClientHello()
        clientHello.create((3, 3), bytearray(32), bytearray(0), [0])
        identities = [PskIdentity().create(bytearray(b'test'), 0)]
        binders = [bytearray(32)]
        psk_ext = PreSharedKeyExtension().create(identities, binders)
        sni_ext = SNIExtension().create(b'example.com')

        clientHello.extensions = [psk_ext, sni_ext]

        hh = HandshakeHashes()

        pskConfigs = [(b'test', b'\x00\x12\x13')]

        with self.assertRaises(ValueError) as e:
            HandshakeHelpers.update_binders(clientHello, hh, pskConfigs)

        self.assertIn('Last extension', str(e.exception))

    def test_getExtension_with_present_id(self):
        client_hello = ClientHello().create((3, 3), bytearray(1), bytearray(0),
                [], extensions=[TLSExtension().create(0, bytearray(0))])

        ext = client_hello.getExtension(0)

        self.assertEqual(ext, TLSExtension().create(0, bytearray(0)))

    def test_write_with_certificate_types(self):

        # note that ClienHello is "clever" and doesn't send the extension
        # if only x509 certificate type is present, so we pass two values
        client_hello = ClientHello().create((3,1),
                bytearray(b'\x00'*31 + b'\xff'), bytearray(0),
                [], certificate_types=[
                    CertificateType.x509, CertificateType.openpgp])

        self.assertEqual(list(bytearray(
                b'\x01' +               # type of message - client_hello
                b'\x00'*2 + b'\x30' +   # length - 48 bytes
                b'\x03\x01' +           # protocol version (TLS 1.0)
                b'\x00'*31 + b'\xff' +  # client random
                b'\x00' +               # session ID length
                b'\x00\x00' +           # cipher suites length
                b'\x01' +               # compression methods length
                b'\x00' +               # supported method - NULL
                b'\x00\x07' +           # extensions length
                b'\x00\x09' +           # cert_type extension value (9)
                b'\x00\x03' +           # size of the extension
                b'\x02' +               # length of supported types
                b'\x00' +               # type - X.509
                b'\x01'                 # type - OpenPGP
                )), list(client_hello.write()))

    def test_parse_with_SRP_extension(self):
        p = Parser(bytearray(
            # we don't include the type of message as it is handled by the
            # hello protocol parser
            #b'x01' +             # type of message - client_hello
            b'\x00'*2 + b'\x35' + # length - 53 bytes
            b'\x01\x01' +         # protocol version - arbitrary (invalid)
            b'\x00'*32 +          # client random
            b'\x00' +             # session ID length
            b'\x00'*2 +           # cipher suites length
            b'\x00' +             # compression methods length
            b'\x00\x0d' +         # extensions length - 13 bytes
            b'\x00\x0c' +         # extension type - SRP (12)
            b'\x00\x09' +         # extension length - 9 bytes
            b'\x08' +             # length of name - 8 bytes
            b'username'           # UTF-8 encoding of "username" :)
            ))
        client_hello = ClientHello()
        client_hello = client_hello.parse(p)

        self.assertEqual((1,1), client_hello.client_version)
        self.assertEqual(bytearray(32), client_hello.random)
        self.assertEqual(bytearray(0), client_hello.session_id)
        self.assertEqual([], client_hello.cipher_suites)
        self.assertEqual([], client_hello.compression_methods)
        self.assertEqual(bytearray(b'username'), client_hello.srp_username)
        srp = SRPExtension().create(bytearray(b'username'))
        self.assertEqual([srp], client_hello.extensions)

def filter(packetNo, data, source, target):
    bytes = stringToBytes(data)
    if packetNo == 0 and 'Client2Server' in str(source):
        p = Parser(bytes[5:])
        p.get(1)
        clientHello = ClientHello()
        clientHello.parse(p)
        print bcolors.OKGREEN + "Client supports TLS version: %s" % \
            str(clientHello.client_version)
        print "Client supports ciphersuites: %s" % \
            str([CIPHER_MAP.get(i,i) for i in clientHello.cipher_suites]) \
            + bcolors.ENDC
    elif packetNo == 0 and 'Client2Server' not in str(source):
        p = Parser(bytes[5:])
        p.get(1)
        serverHello = ServerHello()
        serverHello.parse(p)
        print bcolors.OKGREEN + "Server selected TLS version: %s" % \
            str(serverHello.server_version)
        print "Server selected ciphersuite: %s" % \
            str(CIPHER_MAP.get(serverHello.cipher_suite,
                               serverHello.cipher_suite)) + bcolors.ENDC

    target.write(data)        
    return data

    def test_parse_with_cert_type_extension(self):
        p = Parser(bytearray(
            # we don't include the type of message as it is handled by the
            # hello protocol parser
            #b'x01' +             # type of message - client_hello
            b'\x00'*2 + b'\x2f' + # length - 47 bytes
            b'\x01\x01' +         # protocol version - arbitrary (invalid)
            b'\x00'*32 +          # client random
            b'\x00' +             # session ID length
            b'\x00'*2 +           # cipher suites length
            b'\x00' +             # compression methods length
            b'\x00\x07' +         # extensions length - 7 bytes
            b'\x00\x09' +         # extension type - cert_types (9)
            b'\x00\x03' +         # extension length - 3 bytes
            b'\x02' +             # length of array - 2 bytes
            b'\x00' +             # type - x509 (0)
            b'\x01'               # type - opengpg (1)
            ))
        client_hello = ClientHello()
        client_hello = client_hello.parse(p)

        self.assertEqual((1,1), client_hello.client_version)
        self.assertEqual(bytearray(32), client_hello.random)
        self.assertEqual(bytearray(0), client_hello.session_id)
        self.assertEqual([], client_hello.cipher_suites)
        self.assertEqual([], client_hello.compression_methods)
        self.assertEqual([0,1], client_hello.certificate_types)
        cert_types = ClientCertTypeExtension().create([0,1])
        self.assertEqual([cert_types], client_hello.extensions)

    def test_parse_with_SNI_extension(self):
        p = Parser(bytearray(
            # we don't include the type of message as it is handled by the
            # hello protocol parser
            #b'x01' +             # type of message - client_hello
            b'\x00'*2 + b'\x3c' + # length - 60 bytes
            b'\x01\x01' +         # protocol version - arbitrary (invalid)
            b'\x00'*32 +          # client random
            b'\x00' +             # session ID length
            b'\x00'*2 +           # cipher suites length
            b'\x00' +             # compression methods length
            b'\x00\x14' +         # extensions length - 20 bytes
            b'\x00\x00' +         # extension type - SNI (0)
            b'\x00\x10' +         # extension length - 16 bytes
            b'\x00\x0e' +         # length of array - 14 bytes
            b'\x00' +             # type of entry - host_name (0)
            b'\x00\x0b' +         # length of name - 11 bytes
            # UTF-8 encoding of example.com
            b'\x65\x78\x61\x6d\x70\x6c\x65\x2e\x63\x6f\x6d'
            ))
        client_hello = ClientHello()
        client_hello = client_hello.parse(p)

        self.assertEqual((1,1), client_hello.client_version)
        self.assertEqual(bytearray(32), client_hello.random)
        self.assertEqual(bytearray(0), client_hello.session_id)
        self.assertEqual([], client_hello.cipher_suites)
        self.assertEqual([], client_hello.compression_methods)
        self.assertEqual(bytearray(b'example.com'), client_hello.server_name)
        sni = SNIExtension().create(bytearray(b'example.com'))
        self.assertEqual([sni], client_hello.extensions)

    def test_parse_with_SSLv2_client_hello(self):
        parser = Parser(bytearray(
            # length and type is handled by hello protocol parser
            #b'\x80\x2e' +           # length - 46 bytes
            #b'\x01' +               # message type - client hello
            b'\x00\x02' +           # version - SSLv2
            b'\x00\x15' +           # cipher spec length - 21 bytes
            b'\x00\x00' +           # session ID length - 0 bytes
            b'\x00\x10' +           # challange length - 16 bytes
            b'\x07\x00\xc0' +       # cipher - SSL2_DES_192_EDE3_CBC_WITH_MD5
            b'\x05\x00\x80' +       # cipher - SSL2_IDEA_128_CBC_WITH_MD5
            b'\x03\x00\x80' +       # cipher - SSL2_RC2_CBC_128_CBC_WITH_MD5
            b'\x01\x00\x80' +       # cipher - SSL2_RC4_128_WITH_MD5
            b'\x06\x00\x40' +       # cipher - SSL2_DES_64_CBC_WITH_MD5
            b'\x04\x00\x80' +       # cipher - SSL2_RC2_CBC_128_CBC_WITH_MD5
            b'\x02\x00\x80' +       # cipher - SSL2_RC4_128_EXPORT40_WITH_MD5
            b'\x01' * 16            # challenge
            ))
        client_hello = ClientHello(ssl2=True)

        client_hello = client_hello.parse(parser)

        # XXX the value on the wire is LSB, but should be interpreted MSB for
        # SSL2
        self.assertEqual((0, 2), client_hello.client_version)
        self.assertEqual(bytearray(0), client_hello.session_id)
        self.assertEqual([458944, 327808, 196736, 65664, 393280, 262272,
                          131200],
                         client_hello.cipher_suites)
        self.assertEqual(bytearray(b'\x00'*16 + b'\x01'*16),
                         client_hello.random)
        self.assertEqual([0], client_hello.compression_methods)

    def test_update_binders_with_missing_secret(self):
        clientHello = ClientHello()
        psk = PreSharedKeyExtension()
        clientHello.extensions = [psk]
        hh = HandshakeHashes()

        with self.assertRaises(ValueError):
            HandshakeHelpers.update_binders(clientHello, hh, [], [None])

    def test_create(self):
        client_hello = ClientHello()
        client_hello.create((3,0), bytearray(32), bytearray(0), \
                [])

        self.assertEqual((3,0), client_hello.client_version)
        self.assertEqual(bytearray(32), client_hello.random)
        self.assertEqual(bytearray(0), client_hello.session_id)
        self.assertEqual([], client_hello.cipher_suites)
        self.assertEqual([0], client_hello.compression_methods)

    def test_server_name_other_than_dns_name(self):
        client_hello = ClientHello().create((3, 3), bytearray(1), bytearray(0),
                [])

        sni_ext = SNIExtension().create(server_names=[\
                SNIExtension.ServerName(1, b'test')])

        client_hello.extensions = [sni_ext]

        self.assertEqual(client_hello.server_name, bytearray(0))