[#3514] Add support for AWS MSK IAM authentication using SASL OAUTHBEARER

control-plane/config/eventing-kafka-broker/100-source/100-kafka-source.yaml

@@ -567,7 +567,7 @@ spec:
                                   type: boolean
                               x-kubernetes-map-type: atomic
                         type:
-                          description: Type of saslType, defaults to plain (vs SCRAM-SHA-512 or SCRAM-SHA-256)
+                          description: Type of saslType, defaults to plain (vs SCRAM-SHA-512 or SCRAM-SHA-256 or OAUTHBEARER)
                           type: object
                           properties:
                             secretKeyRef:

control-plane/pkg/apis/bindings/v1/kafka_lifecycle.go

@@ -92,22 +92,41 @@ func (kfb *KafkaBinding) Do(ctx context.Context, ps *duckv1.WithPod) {
 			spec.InitContainers[i].Env = append(spec.InitContainers[i].Env, corev1.EnvVar{
 				Name:  "KAFKA_NET_SASL_ENABLE",
 				Value: "true",
-			}, corev1.EnvVar{
-				Name: "KAFKA_NET_SASL_USER",
-				ValueFrom: &corev1.EnvVarSource{
-					SecretKeyRef: kfb.Spec.Net.SASL.User.SecretKeyRef,
-				},
-			}, corev1.EnvVar{
-				Name: "KAFKA_NET_SASL_PASSWORD",
-				ValueFrom: &corev1.EnvVarSource{
-					SecretKeyRef: kfb.Spec.Net.SASL.Password.SecretKeyRef,
-				},
 			}, corev1.EnvVar{
 				Name: "KAFKA_NET_SASL_TYPE",
 				ValueFrom: &corev1.EnvVarSource{
 					SecretKeyRef: kfb.Spec.Net.SASL.Type.SecretKeyRef,
 				},
 			})
+			if kfb.Spec.Net.SASL.User.SecretKeyRef != nil {
+				spec.InitContainers[i].Env = append(spec.InitContainers[i].Env, corev1.EnvVar{
+					Name: "KAFKA_NET_SASL_USER",
+					ValueFrom: &corev1.EnvVarSource{
+						SecretKeyRef: kfb.Spec.Net.SASL.User.SecretKeyRef,
+					},
+				}, corev1.EnvVar{
+					Name: "KAFKA_NET_SASL_PASSWORD",
+					ValueFrom: &corev1.EnvVarSource{
+						SecretKeyRef: kfb.Spec.Net.SASL.Password.SecretKeyRef,
+					},
+				})
+			}
+			if kfb.Spec.Net.SASL.TokenProvider.SecretKeyRef != nil {
+				spec.Containers[i].Env = append(spec.Containers[i].Env, corev1.EnvVar{
+					Name: "KAFKA_NET_SASL_TOKEN_PROVIDER",
+					ValueFrom: &corev1.EnvVarSource{
+						SecretKeyRef: kfb.Spec.Net.SASL.TokenProvider.SecretKeyRef,
+					},
+				})
+			}
+			if kfb.Spec.Net.SASL.RoleARN.SecretKeyRef != nil {
+				spec.Containers[i].Env = append(spec.Containers[i].Env, corev1.EnvVar{
+					Name: "KAFKA_NET_SASL_ROLE_ARN",
+					ValueFrom: &corev1.EnvVarSource{
+						SecretKeyRef: kfb.Spec.Net.SASL.RoleARN.SecretKeyRef,
+					},
+				})
+			}
 		}
 		if kfb.Spec.Net.TLS.Enable {
 			spec.InitContainers[i].Env = append(spec.InitContainers[i].Env, corev1.EnvVar{
@@ -142,22 +161,41 @@ func (kfb *KafkaBinding) Do(ctx context.Context, ps *duckv1.WithPod) {
 			spec.Containers[i].Env = append(spec.Containers[i].Env, corev1.EnvVar{
 				Name:  "KAFKA_NET_SASL_ENABLE",
 				Value: "true",
-			}, corev1.EnvVar{
-				Name: "KAFKA_NET_SASL_USER",
-				ValueFrom: &corev1.EnvVarSource{
-					SecretKeyRef: kfb.Spec.Net.SASL.User.SecretKeyRef,
-				},
-			}, corev1.EnvVar{
-				Name: "KAFKA_NET_SASL_PASSWORD",
-				ValueFrom: &corev1.EnvVarSource{
-					SecretKeyRef: kfb.Spec.Net.SASL.Password.SecretKeyRef,
-				},
 			}, corev1.EnvVar{
 				Name: "KAFKA_NET_SASL_TYPE",
 				ValueFrom: &corev1.EnvVarSource{
 					SecretKeyRef: kfb.Spec.Net.SASL.Type.SecretKeyRef,
 				},
 			})
+			if kfb.Spec.Net.SASL.User.SecretKeyRef != nil {
+				spec.Containers[i].Env = append(spec.Containers[i].Env, corev1.EnvVar{
+					Name: "KAFKA_NET_SASL_USER",
+					ValueFrom: &corev1.EnvVarSource{
+						SecretKeyRef: kfb.Spec.Net.SASL.User.SecretKeyRef,
+					},
+				}, corev1.EnvVar{
+					Name: "KAFKA_NET_SASL_PASSWORD",
+					ValueFrom: &corev1.EnvVarSource{
+						SecretKeyRef: kfb.Spec.Net.SASL.Password.SecretKeyRef,
+					},
+				})
+			}
+			if kfb.Spec.Net.SASL.TokenProvider.SecretKeyRef != nil {
+				spec.Containers[i].Env = append(spec.Containers[i].Env, corev1.EnvVar{
+					Name: "KAFKA_NET_SASL_TOKEN_PROVIDER",
+					ValueFrom: &corev1.EnvVarSource{
+						SecretKeyRef: kfb.Spec.Net.SASL.TokenProvider.SecretKeyRef,
+					},
+				})
+			}
+			if kfb.Spec.Net.SASL.RoleARN.SecretKeyRef != nil {
+				spec.Containers[i].Env = append(spec.Containers[i].Env, corev1.EnvVar{
+					Name: "KAFKA_NET_SASL_ROLE_ARN",
+					ValueFrom: &corev1.EnvVarSource{
+						SecretKeyRef: kfb.Spec.Net.SASL.RoleARN.SecretKeyRef,
+					},
+				})
+			}
 		}
 		if kfb.Spec.Net.TLS.Enable {
 			spec.Containers[i].Env = append(spec.Containers[i].Env, corev1.EnvVar{

control-plane/pkg/apis/bindings/v1/kafka_types.go

@@ -57,7 +57,17 @@ type KafkaSASLSpec struct {
 	// +optional
 	Password SecretValueFromSource `json:"password,omitempty"`
 
-	// Type of saslType, defaults to plain (vs SCRAM-SHA-512 or SCRAM-SHA-256)
+	// RoleARN is the Kubernetes secret containing the ARN of the IAM role to assume.
+	// Only used if saslType is OAUTHBEARER and tokenProvider is MSKRoleAccessTokenProvider.
+	// +optional
+	RoleARN SecretValueFromSource `json:"roleARN,omitempty"`
+
+	// Token Provider is the Kubernetes secret containing the OAUTHBEARER
+	// token provider function. Only used if saslType is OAUTHBEARER.
+	// +optional
+	TokenProvider SecretValueFromSource `json:"tokenProvider,omitempty"`
+
+	// Type of saslType, defaults to plain (vs SCRAM-SHA-512 or SCRAM-SHA-256 or OAUTHBEARER).
 	// +optional
 	Type SecretValueFromSource `json:"type,omitempty"`
 }

control-plane/pkg/reconciler/consumergroup/auth.go

@@ -89,6 +89,8 @@ func hasNetSpecAuthConfig(auth *kafkainternals.Auth) bool {
 			auth.NetSpec.TLS.Key.SecretKeyRef != nil ||
 			auth.NetSpec.SASL.User.SecretKeyRef != nil ||
 			auth.NetSpec.SASL.Password.SecretKeyRef != nil ||
+			auth.NetSpec.SASL.RoleARN.SecretKeyRef != nil ||
+			auth.NetSpec.SASL.TokenProvider.SecretKeyRef != nil ||
 			auth.NetSpec.SASL.Type.SecretKeyRef != nil)
 }
 

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

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2025 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package oauth
+
+import (
+	"os"
+)
+
+const (
+	saslTokenProviderKey = "tokenProvider"
+	saslRoleARNKey       = "roleARN"
+	saslAWSRegion        = "awsRegion"
+
+	// Environment variable names
+	awsRegionEnvVar        = "AWS_REGION"
+	awsDefaultRegionEnvVar = "AWS_DEFAULT_REGION"
+
+	// Default AWS region
+	defaultAWSRegion = "us-east-1"
+
+	// Token provider types
+	mskAccessTokenProvider     = "MSKAccessTokenProvider"
+	mskRoleAccessTokenProvider = "MSKRoleAccessTokenProvider"
+
+	// MSK IAM Auth constants
+	knativeEventingUserAgent = "knative-eventing"
+)
+
+func getAWSRegion(data map[string][]byte) string {
+	if region, ok := data[saslAWSRegion]; ok && len(region) > 0 {
+		return string(region)
+	}
+
+	awsRegion := os.Getenv(awsRegionEnvVar)
+	if awsRegion != "" {
+		return awsRegion
+	}
+
+	// Fallback to AWS_DEFAULT_REGION if AWS_REGION is not set
+	awsRegion = os.Getenv(awsDefaultRegionEnvVar)
+	if awsRegion != "" {
+		return awsRegion
+	}
+
+	return defaultAWSRegion
+}

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

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2025 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package oauth
+
+import (
+	"context"
+
+	"github.com/aws/aws-msk-iam-sasl-signer-go/signer"
+)
+
+// mskAccessTokenIssuer implements TokenIssuer for the MSK access token
+type mskAccessTokenIssuer struct {
+	region string
+}
+
+func (m *mskAccessTokenIssuer) IssueToken(ctx context.Context) (string, error) {
+	token, _, err := signer.GenerateAuthToken(ctx, m.region)
+	return token, err
+}
+
+func NewMSKAccessTokenIssuer(data map[string][]byte) (*mskAccessTokenIssuer, error) {
+	region := getAWSRegion(data)
+	return &mskAccessTokenIssuer{
+		region: region,
+	}, nil
+}