brod_gssapi fails to connect to working Kafka instance with Kerberos authetnication

[kjellwinblad]

How to reproduce

1. Checkout PR Add example to make it easy to test brod_gssapi #6

2. cd brod_gssapi/example

3. ./up

4. Check that the Kafka instance is functional and that SASL GSSAPI (Kerberos) authentication works by running ./test_send_receive.sh

5. Run the example brod_gssapi project in example/brod_client running docker-compose up brod_client

6. You will now see that the Erlang code crashes with the Error message

   brod_client  | ===> Verifying dependencies...
   brod_client  | ===> Upgrading brod_gssapi (from {path,"../../",{mtime,<<"2022-06-03T12:23:57Z">>}})
   brod_client  | ===> Cleaning out example...
   brod_client  | ===> Verifying dependencies...
   brod_client  | ===> Upgrading brod_gssapi (from {path,"../../",{mtime,<<"2022-06-03T12:23:57Z">>}})
   brod_client  | ===> Analyzing applications...
   brod_client  | ===> Compiling brod_gssapi
   brod_client  | ===> Compiling brod
   brod_client  | ===> Analyzing applications...
   brod_client  | ===> Compiling example
   brod_client  | Args: []
   brod_client  | {going_to_connect,[{"kafka.kerberos-demo.local",9093}]}
   brod_client  | =ERROR REPORT==== 3-Jun-2022::13:35:33.385727 ===
   brod_client  | ** Generic server client1 terminating
   brod_client  | ** Last message in was init
   brod_client  | ** When Server state == {state,client1,
   brod_client  |                          [{"kafka.kerberos-demo.local",9093}],
   brod_client  |                          undefined,[],undefined,undefined,
   brod_client  |                          [{sasl,
   brod_client  |                            {callback,brod_gssapi_v1,
   brod_client  |                             {gssapi,<<"/var/lib/secret/rig.key">>,
   brod_client  |                              <<"rig@TEST.CONFLUENT.IO">>}}}],
   brod_client  |                          client1}
   brod_client  | ** Reason for termination ==
   brod_client  | ** {[{{"kafka.kerberos-demo.local",9093},
   brod_client  |       {{{kpro_req,#Ref<0.3212240052.812122119.53473>,sasl_authenticate,1,
   brod_client  |                   false,
   brod_client  |                   [<<"����">>]},
   brod_client  |         closed},
   brod_client  |        [{kpro_lib,send_and_recv_raw,4,
   brod_client  |                   [{file,"/opt/brod_gssapi/example/brod_client/_build/default/lib/kafka_protocol/src/kpro_lib.erl"},
   brod_client  |                    {line,70}]},
   brod_client  |         {kpro_lib,send_and_recv,5,
   brod_client  |                   [{file,"/opt/brod_gssapi/example/brod_client/_build/default/lib/kafka_protocol/src/kpro_lib.erl"},
   brod_client  |                    {line,81}]},
   brod_client  |         {brod_gssapi_v1,send_sasl_token,6,
   brod_client  |                         [{file,"/opt/brod_gssapi/example/brod_client/_build/default/lib/brod_gssapi/src/brod_gssapi_v1.erl"},
   brod_client  |                          {line,129}]},
   brod_client  |         {brod_gssapi_v1,'-sasl_recv/7-fun-0-',7,
   brod_client  |                         [{file,"/opt/brod_gssapi/example/brod_client/_build/default/lib/brod_gssapi/src/brod_gssapi_v1.erl"},
   brod_client  |                          {line,78}]},
   brod_client  |         {brod_gssapi_v1,do_while,1,
   brod_client  |                         [{file,"/opt/brod_gssapi/example/brod_client/_build/default/lib/brod_gssapi/src/brod_gssapi_v1.erl"},
   brod_client  |                          {line,113}]},
   brod_client  |         {brod_gssapi_v1,sasl_recv,7,
   brod_client  |                         [{file,"/opt/brod_gssapi/example/brod_client/_build/default/lib/brod_gssapi/src/brod_gssapi_v1.erl"},
   brod_client  |                          {line,84}]},
   brod_client  |         {kpro_sasl,auth,7,
   brod_client  |                    [{file,"/opt/brod_gssapi/example/brod_client/_build/default/lib/kafka_protocol/src/kpro_sasl.erl"},
   brod_client  |                     {line,38}]},
   brod_client  |         {kpro_connection,init_connection,2,
   brod_client  |                          [{file,"/opt/brod_gssapi/example/brod_client/_build/default/lib/kafka_protocol/src/kpro_connection.erl"},
   brod_client  |                           {line,240}]}]}}],
   brod_client  |     [{brod_client,ensure_metadata_connection,1,
   brod_client  |                   [{file,"/opt/brod_gssapi/example/brod_client/_build/default/lib/brod/src/brod_client.erl"},
   brod_client  |                    {line,570}]},
   brod_client  |      {brod_client,handle_info,2,
   brod_client  |                   [{file,"/opt/brod_gssapi/example/brod_client/_build/default/lib/brod/src/brod_client.erl"},
   brod_client  |                    {line,298}]},
   brod_client  |      {gen_server,try_dispatch,4,[{file,"gen_server.erl"},{line,695}]},
   brod_client  |      {gen_server,handle_msg,6,[{file,"gen_server.erl"},{line,771}]},
   brod_client  |      {proc_lib,init_p_do_apply,3,[{file,"proc_lib.erl"},{line,226}]}]}
   brod_client  |
   brod_client  | {"init terminating in do_boot",{{badmatch,{error,{client_down,[{{"kafka.kerberos-demo.local",9093},{{{kpro_req,#Ref<0.3212240052.812122119.53473>,sasl_authenticate,1,false,[<<255,255,255,255>>]},closed},[{kpro_lib,send_and_recv_raw,4,[{file,"/opt/brod_gssapi/example/brod_client/_build/default/lib/kafka_protocol/src/kpro_lib.erl"},{line,70}]},{kpro_lib,send_and_recv,5,[{file,"/opt/brod_gssapi/example/brod_client/_build/default/lib/kafka_protocol/src/kpro_lib.erl"},{line,81}]},{brod_gssapi_v1,send_sasl_token,6,[{file,"/opt/brod_gssapi/example/brod_client/_build/default/lib/brod_gssapi/src/brod_gssapi_v1.erl"},{line,129}]},{brod_gssapi_v1,'-sasl_recv/7-fun-0-',7,[{file,"/opt/brod_gssapi/example/brod_client/_build/default/lib/brod_gssapi/src/brod_gssapi_v1.erl"},{line,78}]},{brod_gssapi_v1,do_while,1,[{file,"/opt/brod_gssapi/example/brod_client/_build/default/lib/brod_gssapi/src/brod_gssapi_v1.erl"},{line,113}]},{brod_gssapi_v1,sasl_recv,7,[{file,"/opt/brod_gssapi/example/brod_client/_build/default/lib/brod_gssapi/src/brod_gssapi_v1.erl"},{line,84}]},{kpro_sasl,auth,7,[{file,"/opt/brod_gssapi/example/brod_client/_build/default/lib/kafka_protocol/src/kpro_sasl.erl"},{line,38}]},{kpro_connection,init_connection,2,[{file,"/opt/brod_gssapi/example/brod_client/_build/default/lib/kafka_protocol/src/kpro_connection.erl"},{line,240}]}]}}]}}},[{example,main,1,[{file,"/opt/brod_gssapi/example/brod_client/src/example.erl"},{line,22}]},{erl_eval,do_apply,6,[{file,"erl_eval.erl"},{line,685}]},{init,start_it,1,[]},{init,start_em,1,[]},{init,do_boot,3,[]}]}}
   brod_client  | init terminating in do_boot ({{badmatch,{error,{client_down,[_]}}},[{example,main,1,[{_},{_}]},{erl_eval,do_apply,6,[{_},{_}]},{init,start_it,1,[]},{init,start_em,1,[]},{init,do_boot,3,[]}]})
   brod_client  |
   brod_client  | Crash dump is being written to: erl_crash.dump...done
   brod_client exited with code 1
   
   

7. The Kafka logs with authentication debugging turned on can be seen by running docker-compose logs kafka. The relevant part can be seen here:

   kafka  | [2022-06-03 13:35:33,367] DEBUG Accepted connection from /192.168.160.6:55466 on /192.168.160.4:9093 and assigned it to processor 2, sendBufferSize [actual|requested]: [102400|102400] recvBufferSize [actual|requested]: [102400|102400] (kafka.network.Acceptor)
   kafka  | [2022-06-03 13:35:33,367] DEBUG Processor 2 listening to new connection from /192.168.160.6:55466 (kafka.network.Processor)
   kafka  | [2022-06-03 13:35:33,367] DEBUG connections.max.reauth.ms for mechanism=GSSAPI: 0 (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
   kafka  | [2022-06-03 13:35:33,370] DEBUG Set SASL server state to HANDSHAKE_OR_VERSIONS_REQUEST during authentication (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
   kafka  | [2022-06-03 13:35:33,370] DEBUG Handling Kafka request API_VERSIONS during authentication (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
   kafka  | [2022-06-03 13:35:33,371] DEBUG Set SASL server state to HANDSHAKE_REQUEST during authentication (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
   kafka  | [2022-06-03 13:35:33,380] DEBUG Handling Kafka request SASL_HANDSHAKE during authentication (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
   kafka  | [2022-06-03 13:35:33,380] DEBUG Using SASL mechanism 'GSSAPI' provided by client (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
   kafka  | [2022-06-03 13:35:33,380] DEBUG Creating SaslServer for kafka/kafka.kerberos-demo.local@TEST.CONFLUENT.IO with mechanism GSSAPI (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
   kafka  | Found KeyTab /var/lib/secret/kafka.key for kafka/kafka.kerberos-demo.local@TEST.CONFLUENT.IO
   kafka  | Found ticket for kafka/kafka.kerberos-demo.local@TEST.CONFLUENT.IO to go to krbtgt/TEST.CONFLUENT.IO@TEST.CONFLUENT.IO expiring on Sat Jun 04 13:34:18 UTC 2022
   kafka  | [2022-06-03 13:35:33,381] DEBUG Set SASL server state to AUTHENTICATE during authentication (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
   kafka  | Entered Krb5Context.acceptSecContext with state=STATE_NEW
   kafka  | Looking for keys for: kafka/kafka.kerberos-demo.local@TEST.CONFLUENT.IO
   kafka  | Added key: 17version: 1
   kafka  | Added key: 18version: 1
   kafka  | >>> EType: sun.security.krb5.internal.crypto.Aes256CtsHmacSha1EType
   kafka  | Using builtin default etypes for permitted_enctypes
   kafka  | default etypes for permitted_enctypes: 18 17 20 19 16 23.
   kafka  | >>> EType: sun.security.krb5.internal.crypto.Aes256CtsHmacSha1EType
   kafka  | MemoryCache: add 1654263333/378345/44366FF2746B2515F667914D7E95B8E60BE6CC8C732918E76AE309D45B3AAFB5/rig@TEST.CONFLUENT.IO to rig@TEST.CONFLUENT.IO|kafka/kafka.kerberos-demo.local@TEST.CONFLUENT.IO
   kafka  | >>> KrbApReq: authenticate succeed.
   kafka  | Krb5Context setting peerSeqNumber to: 539147621
   kafka  | >>> EType: sun.security.krb5.internal.crypto.Aes256CtsHmacSha1EType
   kafka  | Krb5Context setting mySeqNumber to: 839975820
   kafka  | [2022-06-03 13:35:33,384] DEBUG Failed during authentication: Error getting request for apiKey: SASL_AUTHENTICATE, apiVersion: 1, connectionId: 192.168.160.4:9093-192.168.160.6:55466-9, listenerName: ListenerName(SASL_PLAINTEXT), principal: User:ANONYMOUS (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
   kafka  | [2022-06-03 13:35:33,384] WARN [SocketServer listenerType=ZK_BROKER, nodeId=0] Unexpected error from /192.168.160.6; closing connection (org.apache.kafka.common.network.Selector)
   kafka  | org.apache.kafka.common.errors.InvalidRequestException: Error getting request for apiKey: SASL_AUTHENTICATE, apiVersion: 1, connectionId: 192.168.160.4:9093-192.168.160.6:55466-9, listenerName: ListenerName(SASL_PLAINTEXT), principal: User:ANONYMOUS
   kafka  | Caused by: java.lang.RuntimeException: non-nullable field authBytes was serialized as null
   kafka  | 	at org.apache.kafka.common.message.SaslAuthenticateRequestData.read(SaslAuthenticateRequestData.java:103)
   kafka  | 	at org.apache.kafka.common.message.SaslAuthenticateRequestData.<init>(SaslAuthenticateRequestData.java:71)
   kafka  | 	at org.apache.kafka.common.requests.SaslAuthenticateRequest.parse(SaslAuthenticateRequest.java:78)
   kafka  | 	at org.apache.kafka.common.requests.AbstractRequest.doParseRequest(AbstractRequest.java:230)
   kafka  | 	at org.apache.kafka.common.requests.AbstractRequest.parseRequest(AbstractRequest.java:152)
   kafka  | 	at org.apache.kafka.common.requests.RequestContext.parseRequest(RequestContext.java:95)
   kafka  | 	at org.apache.kafka.common.security.authenticator.SaslServerAuthenticator.handleSaslToken(SaslServerAuthenticator.java:430)
   kafka  | 	at org.apache.kafka.common.security.authenticator.SaslServerAuthenticator.authenticate(SaslServerAuthenticator.java:280)
   kafka  | 	at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:181)
   kafka  | 	at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:543)
   kafka  | 	at org.apache.kafka.common.network.Selector.poll(Selector.java:481)
   kafka  | 	at kafka.network.Processor.poll(SocketServer.scala:989)
   kafka  | 	at kafka.network.Processor.run(SocketServer.scala:892)
   kafka  | 	at java.base/java.lang.Thread.run(Thread.java:829)
   

This might be related to problems described here: kafka4beam/brod#383

@starbelly , @vikas15bhardwaj do you have any ideas about what the problem might be and how to fix it?

[kjellwinblad]

Fixed. See #6