attempt to detect improper TLS config (#843)

* attempt to detect improper TLS config

when an io.ErrUnexpectedEOF is returned when reading
a correlation ID, attempt to detect if the messages
from the broker was a TLS Alert Message and decorate
the returned error

* update TLS info in README.md

* simplify tls version check

Co-authored-by: Achille <[email protected]>

Author: rhansen2

README.md

@@ -480,6 +480,8 @@ longer the case and import of the compression packages are now no-ops._
 ## TLS Support
 
 For a bare bones Conn type or in the Reader/Writer configs you can specify a dialer option for TLS support. If the TLS field is nil, it will not connect with TLS.
+*Note:* Connecting to a Kafka cluster with TLS enabled without configuring TLS on the Conn/Reader/Writer can manifest in opaque io.ErrUnexpectedEOF errors.
+
 
 ### Connection
 
@@ -512,6 +514,8 @@ r := kafka.NewReader(kafka.ReaderConfig{
 
 ### Writer
 
+Using `kafka.NewWriter`
+
 ```go
 dialer := &kafka.Dialer{
     Timeout:   10 * time.Second,
@@ -527,6 +531,20 @@ w := kafka.NewWriter(kafka.WriterConfig{
 })
 ```
 
+Direct Writer creation
+
+```go
+w := kafka.Writer{
+        Addr: kafka.TCP("localhost:9093"),
+	    Topic:   "topic-A",
+	    Balancer: &kafka.Hash{},
+        Transport: &kafka.Transport{
+            TLS: &tls.Config{},
+        },
+    }
+
+```
+
 ## SASL Support
 
 You can specify an option on the `Dialer` to use SASL authentication. The `Dialer` can be used directly to open a `Conn` or it can be passed to a `Reader` or `Writer` via their respective configs. If the `SASLMechanism` field is `nil`, it will not authenticate with SASL.

protocol/response.go

@@ -1,6 +1,9 @@
 package protocol
 
 import (
+	"crypto/tls"
+	"encoding/binary"
+	"errors"
 	"fmt"
 	"io"
 )
@@ -35,6 +38,25 @@ func ReadResponse(r io.Reader, apiKey ApiKey, apiVersion int16) (correlationID i
 
 	d.remain = int(size)
 	correlationID = d.readInt32()
+	if err = d.err; err != nil {
+		if errors.Is(err, io.ErrUnexpectedEOF) {
+			// If a Writer/Reader is configured without TLS and connects
+			// to a broker expecting TLS the only message we return to the
+			// caller is io.ErrUnexpetedEOF which is opaque. This section
+			// tries to determine if that's what has happened.
+			// We first deconstruct the initial 4 bytes of the message
+			// from the size which was read earlier.
+			// Next, we examine those bytes to see if they looks like a TLS
+			// error message. If they do we wrap the io.ErrUnexpectedEOF
+			// with some context.
+			if looksLikeUnexpectedTLS(size) {
+				err = fmt.Errorf("%w: broker appears to be expecting TLS", io.ErrUnexpectedEOF)
+			}
+			return
+		}
+		err = dontExpectEOF(err)
+		return
+	}
 
 	res := &t.responses[apiVersion-minVersion]
 
@@ -109,3 +131,21 @@ func WriteResponse(w io.Writer, apiVersion int16, correlationID int32, msg Messa
 
 	return err
 }
+
+const (
+	tlsAlertByte byte = 0x15
+)
+
+// looksLikeUnexpectedTLS returns true if the size passed in resemble
+// the TLS alert message that is returned to a client which sends
+// an invalid ClientHello message.
+func looksLikeUnexpectedTLS(size int32) bool {
+	var sizeBytes [4]byte
+	binary.BigEndian.PutUint32(sizeBytes[:], uint32(size))
+
+	if sizeBytes[0] != tlsAlertByte {
+		return false
+	}
+	version := int(sizeBytes[1])<<8 | int(sizeBytes[2])
+	return version <= tls.VersionTLS13 && version >= tls.VersionTLS10
+}

protocol/response_test.go

@@ -0,0 +1,28 @@
+package protocol
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"strings"
+	"testing"
+)
+
+func TestReadResponseUnexpectedTLSDetection(t *testing.T) {
+	var buf bytes.Buffer
+
+	buf.Write([]byte{tlsAlertByte, 0x03, 0x03, 10, 0, 0, 0})
+
+	correlationID, _, err := ReadResponse(&buf, ApiVersions, 0)
+	if !errors.Is(err, io.ErrUnexpectedEOF) {
+		t.Fatalf("expected an io.ErrUnexpectedEOF from ReadResponse got %v", err)
+	}
+
+	if !strings.Contains(err.Error(), "broker appears to be expecting TLS") {
+		t.Fatalf("expected error messae to contain %s got %s", "broker appears to be expecting TLS", err.Error())
+	}
+
+	if correlationID != 0 {
+		t.Fatalf("expected correlationID of 0 got %d", correlationID)
+	}
+}