fix(control-plane): fail explicitly on OAUTHBEARER without tokenProvider

Author: dmetzler

control-plane/pkg/security/oauth/token_provider.go

@@ -65,3 +65,22 @@ func NewTokenProvider(data map[string][]byte) (*TokenProvider, error) {
 		tokenIssuer: tokenIssuer,
 	}, nil
 }
+
+// errorTokenProvider is a sarama.AccessTokenProvider that always fails with a
+// descriptive error. It is used when an OAUTHBEARER secret has no tokenProvider
+// (the sasl.jaas.config / sasl.login.callback.handler.class keys are consumed by
+// the Java data plane only). Setting this instead of leaving TokenProvider nil
+// prevents a nil-pointer panic in sarama's OAUTHBEARER auth path when the Go
+// control plane attempts an admin connection, turning it into a clear reconciler
+// error instead.
+type errorTokenProvider struct{ msg string }
+
+func (e *errorTokenProvider) Token() (*sarama.AccessToken, error) {
+	return nil, fmt.Errorf("%s", e.msg)
+}
+
+// UnsupportedTokenProvider returns an AccessTokenProvider that always fails with
+// the given message. See errorTokenProvider for rationale.
+func UnsupportedTokenProvider(msg string) sarama.AccessTokenProvider {
+	return &errorTokenProvider{msg: msg}
+}

control-plane/pkg/security/secret.go

@@ -133,6 +133,11 @@ func saslConfig(protocol string, data map[string][]byte) kafka.ConfigOption {
 					return fmt.Errorf("[protocol %s] failed to create OAUTHBEARER token provider: %w", protocol, err)
 				}
 				config.Net.SASL.TokenProvider = tokenProvider
+			} else {
+				config.Net.SASL.TokenProvider = oauth.UnsupportedTokenProvider(
+					"OAUTHBEARER without tokenProvider is not supported by the Go control plane " +
+						"(sasl.jaas.config/sasl.login.callback.handler.class are consumed by the Java data plane only); " +
+						"topic and consumer group management will fail")
 			}
 			return nil
 		}

control-plane/pkg/security/secret_test.go

@@ -223,7 +223,10 @@ func TestSASLOAuthMissingTokenProvider(t *testing.T) {
 	assert.Nil(t, err)
 	assert.True(t, config.Net.SASL.Enable)
 	assert.Equal(t, sarama.SASLMechanism(sarama.SASLTypeOAuth), config.Net.SASL.Mechanism)
-	assert.Nil(t, config.Net.SASL.TokenProvider)
+	assert.NotNil(t, config.Net.SASL.TokenProvider)
+	_, tokenErr := config.Net.SASL.TokenProvider.Token()
+	assert.NotNil(t, tokenErr)
+	assert.Contains(t, tokenErr.Error(), "not supported by the Go control plane")
 }
 
 func TestSASLOAuthInvalidTokenProvider(t *testing.T) {
@@ -257,7 +260,10 @@ func TestSASLOAuthWithJavaPassthroughKeys(t *testing.T) {
 	assert.Nil(t, err)
 	assert.True(t, config.Net.SASL.Enable)
 	assert.Equal(t, sarama.SASLMechanism(sarama.SASLTypeOAuth), config.Net.SASL.Mechanism)
-	assert.Nil(t, config.Net.SASL.TokenProvider)
+	assert.NotNil(t, config.Net.SASL.TokenProvider)
+	_, tokenErr := config.Net.SASL.TokenProvider.Token()
+	assert.NotNil(t, tokenErr)
+	assert.Contains(t, tokenErr.Error(), "not supported by the Go control plane")
 	assert.True(t, config.Net.TLS.Enable)
 }