From 30f84f50ebba412cf4194444f8e82f907a801d0d Mon Sep 17 00:00:00 2001
From: Martin Schneppenheim <weeco91@gmail.com>
Date: Wed, 1 Sep 2021 22:20:34 +0200
Subject: [PATCH] Fixes #111

---
 docs/reference-config.yaml    |  3 ++-
 kafka/client_config_helper.go | 10 +++++++---
 kafka/config_sasl.go          |  1 +
 kafka/config_sasl_gssapi.go   |  9 +++++++++
 4 files changed, 19 insertions(+), 4 deletions(-)

diff --git a/docs/reference-config.yaml b/docs/reference-config.yaml
index 672d881..501eee5 100644
--- a/docs/reference-config.yaml
+++ b/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.
diff --git a/kafka/client_config_helper.go b/kafka/client_config_helper.go
index d29f4d2..f55eaee 100644
--- a/kafka/client_config_helper.go
+++ b/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,
diff --git a/kafka/config_sasl.go b/kafka/config_sasl.go
index a541d18..522f522 100644
--- a/kafka/config_sasl.go
+++ b/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
diff --git a/kafka/config_sasl_gssapi.go b/kafka/config_sasl_gssapi.go
index e1ed021..98200be 100644
--- a/kafka/config_sasl_gssapi.go
+++ b/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
 }