Fixes #111

docs/reference-config.yaml

@@ -51,6 +51,7 @@ kafka:
       username: ""
       password: ""
       realm: ""
+      enableFast: true
 
 minion:
   consumerGroups:
@@ -90,7 +91,7 @@ minion:
     # infoMetric is a configuration object for the kminion_kafka_topic_info metric
     infoMetric:
       # ConfigKeys are set of strings of Topic configs that you want to have exported as part of the metric
-      configKeys: ["cleanup.policy"]
+      configKeys: [ "cleanup.policy" ]
   logDirs:
     # Enabled specifies whether log dirs shall be scraped and exported or not. This should be disabled for clusters prior
     # to version 1.0.0 as describing log dirs was not supported back then.

kafka/client_config_helper.go

@@ -74,18 +74,21 @@ func NewKgoConfig(cfg Config, logger *zap.Logger) ([]kgo.Opt, error) {
 
 		// Kerberos
 		if cfg.SASL.Mechanism == "GSSAPI" {
+			var krbClient *client.Client
+
 			kerbCfg, err := krbconfig.Load(cfg.SASL.GSSAPI.KerberosConfigPath)
 			if err != nil {
 				return nil, fmt.Errorf("failed to create kerberos config from specified config filepath: %w", err)
 			}
-			var krbClient *client.Client
+
 			switch cfg.SASL.GSSAPI.AuthType {
 			case "USER_AUTH:":
 				krbClient = client.NewWithPassword(
 					cfg.SASL.GSSAPI.Username,
 					cfg.SASL.GSSAPI.Realm,
 					cfg.SASL.GSSAPI.Password,
-					kerbCfg)
+					kerbCfg,
+					client.DisablePAFXFAST(!cfg.SASL.GSSAPI.EnableFast))
 			case "KEYTAB_AUTH":
 				ktb, err := keytab.Load(cfg.SASL.GSSAPI.KeyTabPath)
 				if err != nil {
@@ -95,7 +98,8 @@ func NewKgoConfig(cfg Config, logger *zap.Logger) ([]kgo.Opt, error) {
 					cfg.SASL.GSSAPI.Username,
 					cfg.SASL.GSSAPI.Realm,
 					ktb,
-					kerbCfg)
+					kerbCfg,
+					client.DisablePAFXFAST(!cfg.SASL.GSSAPI.EnableFast))
 			}
 			kerberosMechanism := kerberos.Auth{
 				Client:           krbClient,

kafka/config_sasl.go

@@ -25,6 +25,7 @@ type SASLConfig struct {
 func (c *SASLConfig) SetDefaults() {
 	c.Enabled = false
 	c.Mechanism = SASLMechanismPlain
+	c.GSSAPI.SetDefaults()
 }
 
 // Validate SASL config input

kafka/config_sasl_gssapi.go

@@ -9,4 +9,13 @@ type SASLGSSAPIConfig struct {
 	Username           string `koanf:"username"`
 	Password           string `koanf:"password"`
 	Realm              string `koanf:"realm"`
+
+	// EnableFAST enables FAST, which is a pre-authentication framework for Kerberos.
+	// It includes a mechanism for tunneling pre-authentication exchanges using armoured KDC messages.
+	// FAST provides increased resistance to passive password guessing attacks.
+	EnableFast bool `koanf:"enableFast"`
+}
+
+func (s *SASLGSSAPIConfig) SetDefaults() {
+	s.EnableFast = true
 }