sasl: validate non-empty user/pass/token

This avoids sending empty credentials to the user. It is difficult to
validate this up front because all mechanisms are designed for
hot-reloading credentials, but we can validate at the time just before
we connect and issue a request.

Closes #472.

pkg/sasl/oauth/oauth.go

@@ -52,6 +52,9 @@ func (fn oauth) Authenticate(ctx context.Context, _ string) (sasl.Session, []byt
 	if err != nil {
 		return nil, nil, err
 	}
+	if auth.Token == "" {
+		return nil, nil, errors.New("OAUTHBEARER token must be non-empty")
+	}
 
 	// We sort extensions for consistency, but it is not required.
 	type kv struct {

pkg/sasl/plain/plain.go

@@ -3,6 +3,7 @@ package plain
 
 import (
 	"context"
+	"errors"
 
 	"github.com/twmb/franz-go/pkg/sasl"
 )
@@ -46,6 +47,9 @@ func (fn plain) Authenticate(ctx context.Context, _ string) (sasl.Session, []byt
 	if err != nil {
 		return nil, nil, err
 	}
+	if auth.User == "" || auth.Pass == "" {
+		return nil, nil, errors.New("PLAIN user and pass must be non-empty")
+	}
 	return session{}, []byte(auth.Zid + "\x00" + auth.User + "\x00" + auth.Pass), nil
 }
 

pkg/sasl/sasl.go

@@ -1,4 +1,4 @@
-// Package sasl specifies interfaces that any sasl authentication must provide
+// Package sasl specifies interfaces that any SASL authentication must provide
 // to interop with Kafka SASL.
 package sasl
 
@@ -32,7 +32,7 @@ type Mechanism interface {
 	Authenticate(ctx context.Context, host string) (Session, []byte, error)
 }
 
-// ClosingMechanism is an optional interface for sasl mechanism's. Implementing
+// ClosingMechanism is an optional interface for SASL mechanisms. Implementing
 // this interface signals that the mechanism should be closed if it will never
 // be used again.
 type ClosingMechanism interface {

pkg/sasl/scram/scram.go

@@ -105,6 +105,9 @@ func (s scram) Authenticate(ctx context.Context, _ string) (sasl.Session, []byte
 	if err != nil {
 		return nil, nil, err
 	}
+	if auth.User == "" || auth.Pass == "" {
+		return nil, nil, errors.New(s.name + " user and pass must be non-empty")
+	}
 	if len(auth.Nonce) == 0 {
 		buf := make([]byte, 20)
 		if _, err = rand.Read(buf); err != nil {