Add assumeRole option

Author: yolken

cmd/topicctl/subcmd/shared.go

@@ -14,20 +14,21 @@ import (
 )
 
 type sharedOptions struct {
-	brokerAddr    string
-	clusterConfig string
-	expandEnv     bool
-	saslMechanism string
-	saslPassword  string
-	saslUsername  string
-	tlsCACert     string
-	tlsCert       string
-	tlsEnabled    bool
-	tlsKey        string
-	tlsSkipVerify bool
-	tlsServerName string
-	zkAddr        string
-	zkPrefix      string
+	brokerAddr     string
+	clusterConfig  string
+	expandEnv      bool
+	saslMechanism  string
+	saslPassword   string
+	saslUsername   string
+	saslAssumeRole string
+	tlsCACert      string
+	tlsCert        string
+	tlsEnabled     bool
+	tlsKey         string
+	tlsSkipVerify  bool
+	tlsServerName  string
+	zkAddr         string
+	zkPrefix       string
 }
 
 func (s sharedOptions) validate() error {
@@ -95,6 +96,10 @@ func (s sharedOptions) validate() error {
 			(s.saslUsername != "" || s.saslPassword != "") {
 			log.Warn("Username and password are ignored if using SASL AWS-MSK-IAM")
 		}
+
+		if saslMechanism != admin.SASLMechanismAWSMSKIAM && s.saslAssumeRole != "" {
+			log.Warn("AssumeRole is ignored unless using SASL AWS-MSK-IAM")
+		}
 	}
 
 	return err
@@ -150,10 +155,11 @@ func (s sharedOptions) getAdminClient(
 						SkipVerify: s.tlsSkipVerify,
 					},
 					SASL: admin.SASLConfig{
-						Enabled:   saslEnabled,
-						Mechanism: saslMechanism,
-						Password:  s.saslPassword,
-						Username:  s.saslUsername,
+						Enabled:    saslEnabled,
+						Mechanism:  saslMechanism,
+						Password:   s.saslPassword,
+						Username:   s.saslUsername,
+						AssumeRole: s.saslAssumeRole,
 					},
 				},
 				ReadOnly: readOnly,
@@ -211,6 +217,12 @@ func addSharedFlags(cmd *cobra.Command, options *sharedOptions) {
 		os.Getenv("TOPICCTL_SASL_USERNAME"),
 		"SASL username if using SASL; will override value set in cluster config",
 	)
+	cmd.Flags().StringVar(
+		&options.saslAssumeRole,
+		"sasl-assume-role",
+		"",
+		"Intermediate role to assume if using SASL AWS-MSK-IAM",
+	)
 	cmd.Flags().StringVar(
 		&options.tlsCACert,
 		"tls-ca-cert",

pkg/admin/connector.go

@@ -48,10 +48,11 @@ type TLSConfig struct {
 
 // SASLConfig stores the SASL-related configuration for a connection.
 type SASLConfig struct {
-	Enabled   bool
-	Mechanism SASLMechanism
-	Username  string
-	Password  string
+	Enabled    bool
+	Mechanism  SASLMechanism
+	Username   string
+	Password   string
+	AssumeRole string
 }
 
 // Connector is a wrapper around the low-level, kafka-go dialer and client.

pkg/config/cluster.go

@@ -111,6 +111,9 @@ type SASLConfig struct {
 
 	// Password is the SASL password. Ignored if mechanism is AWS-MSK-IAM.
 	Password string `json:"password"`
+
+	// Intermediate role ARN to assume. Only used if mechanism is AWS-MSK-IAM.
+	AssumeRole string `json:"assumeRole"`
 }
 
 // Validate evaluates whether the cluster config is valid.
@@ -165,6 +168,10 @@ func (c ClusterConfig) Validate() error {
 			(c.Spec.SASL.Username != "" || c.Spec.SASL.Password != "") {
 			log.Warn("Username and password are ignored if using SASL AWS-MSK-IAM")
 		}
+
+		if saslMechanism != admin.SASLMechanismAWSMSKIAM && c.Spec.SASL.AssumeRole != "" {
+			log.Warn("AssumeRole is ignored unless using SASL AWS-MSK-IAM")
+		}
 	}
 
 	return err
@@ -231,10 +238,11 @@ func (c ClusterConfig) NewAdminClient(
 						SkipVerify: c.Spec.TLS.SkipVerify,
 					},
 					SASL: admin.SASLConfig{
-						Enabled:   c.Spec.SASL.Enabled,
-						Mechanism: saslMechanism,
-						Username:  saslUsername,
-						Password:  saslPassword,
+						Enabled:    c.Spec.SASL.Enabled,
+						Mechanism:  saslMechanism,
+						Username:   saslUsername,
+						Password:   saslPassword,
+						AssumeRole: c.Spec.SASL.AssumeRole,
 					},
 				},
 				ExpectedClusterID: c.Spec.ClusterID,