Method test_psk

Lib/test/test_ssl.py:4473–4537

(self)

    @requires_tls_version('TLSv1_2')
    @unittest.skipUnless(ssl.HAS_PSK, 'TLS-PSK disabled on this OpenSSL build')
    def test_psk(self):
        psk = bytes.fromhex('deadbeef')

        client_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        client_context.check_hostname = False
        client_context.verify_mode = ssl.CERT_NONE
        client_context.maximum_version = ssl.TLSVersion.TLSv1_2
        client_context.set_ciphers('PSK')
        client_context.set_psk_client_callback(lambda hint: (None, psk))

        server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        server_context.maximum_version = ssl.TLSVersion.TLSv1_2
        server_context.set_ciphers('PSK')
        server_context.set_psk_server_callback(lambda identity: psk)

        # correct PSK should connect
        server = ThreadedEchoServer(context=server_context)
        with server:
            with client_context.wrap_socket(socket.socket()) as s:
                s.connect((HOST, server.port))

        # incorrect PSK should fail
        incorrect_psk = bytes.fromhex('cafebabe')
        client_context.set_psk_client_callback(lambda hint: (None, incorrect_psk))
        server = ThreadedEchoServer(context=server_context)
        with server:
            with client_context.wrap_socket(socket.socket()) as s:
                with self.assertRaises(ssl.SSLError):
                    s.connect((HOST, server.port))

        # identity_hint and client_identity should be sent to the other side
        identity_hint = 'identity-hint'
        client_identity = 'client-identity'

        def client_callback(hint):
            self.assertEqual(hint, identity_hint)
            return client_identity, psk

        def server_callback(identity):
            self.assertEqual(identity, client_identity)
            return psk

        client_context.set_psk_client_callback(client_callback)
        server_context.set_psk_server_callback(server_callback, identity_hint)
        server = ThreadedEchoServer(context=server_context)
        with server:
            with client_context.wrap_socket(socket.socket()) as s:
                s.connect((HOST, server.port))

        # adding client callback to server or vice versa raises an exception
        with self.assertRaisesRegex(ssl.SSLError, 'Cannot add PSK server callback'):
            client_context.set_psk_server_callback(server_callback, identity_hint)
        with self.assertRaisesRegex(ssl.SSLError, 'Cannot add PSK client callback'):
            server_context.set_psk_client_callback(client_callback)

        # test with UTF-8 identities
        identity_hint = '身份暗示'  # Translation: "Identity hint"
        client_identity = '客户身份'  # Translation: "Customer identity"