Feat: Support client JWT auth method for kafka oidc (#4057)

Signed-off-by: Albert Callarisa <[email protected]>

Author: acroca

bindings/kafka/metadata.yaml

@@ -33,7 +33,7 @@ builtinAuthenticationProfiles:
         type: string
         required: false
         description: |
-          This maintains backwards compatibility with existing fields. 
+          This maintains backwards compatibility with existing fields.
           It will be deprecated as of Dapr 1.17. Use 'accessKey' instead.
           If both fields are set, then 'accessKey' value will be used.
           AWS access key associated with an IAM account.
@@ -43,7 +43,7 @@ builtinAuthenticationProfiles:
         required: false
         sensitive: true
         description: |
-          This maintains backwards compatibility with existing fields. 
+          This maintains backwards compatibility with existing fields.
           It will be deprecated as of Dapr 1.17. Use 'secretKey' instead.
           If both fields are set, then 'secretKey' value will be used.
           The secret key associated with the access key.
@@ -52,7 +52,7 @@ builtinAuthenticationProfiles:
         type: string
         sensitive: true
         description: |
-          This maintains backwards compatibility with existing fields. 
+          This maintains backwards compatibility with existing fields.
           It will be deprecated as of Dapr 1.17. Use 'sessionToken' instead.
           If both fields are set, then 'sessionToken' value will be used.
           AWS session token to use. A session token is only required if you are using temporary security credentials.
@@ -61,15 +61,15 @@ builtinAuthenticationProfiles:
         type: string
         required: false
         description: |
-          This maintains backwards compatibility with existing fields. 
+          This maintains backwards compatibility with existing fields.
           It will be deprecated as of Dapr 1.17. Use 'assumeRoleArn' instead.
           If both fields are set, then 'assumeRoleArn' value will be used.
           IAM role that has access to MSK. This is another option to authenticate with MSK aside from the AWS Credentials.
         example: '"arn:aws:iam::123456789:role/mskRole"'
       - name: awsStsSessionName
         type: string
         description: |
-          This maintains backwards compatibility with existing fields. 
+          This maintains backwards compatibility with existing fields.
           It will be deprecated as of Dapr 1.17. Use 'sessionName' instead.
           If both fields are set, then 'sessionName' value will be used.
           Represents the session name for assuming a role.
@@ -78,7 +78,7 @@ builtinAuthenticationProfiles:
 authenticationProfiles:
   - title: "OIDC Authentication"
     description: |
-      Authenticate using OpenID Connect.
+      Authenticate using OpenID Connect providing a client secret.
     metadata:
       - name: authType
         type: string
@@ -121,6 +121,72 @@ authenticationProfiles:
         example: |
           {"cluster":"kafka","poolid":"kafkapool"}
         type: string
+  - title: "OIDC Private Key JWT Authentication"
+    description: |
+      Authenticate using OpenID Connect providing a client certificate and private key.
+    metadata:
+      - name: authType
+        type: string
+        required: true
+        description: |
+          Authentication type.
+          This must be set to "oidc_private_key_jwt" for this authentication profile.
+        example: '"oidc_private_key_jwt"'
+        allowedValues:
+          - "oidc_private_key_jwt"
+      - name: oidcTokenEndpoint
+        type: string
+        required: true
+        description: |
+          URL of the OAuth2 identity provider access token endpoint.
+        example: '"https://identity.example.com/v1/token"'
+      - name: oidcClientID
+        description: |
+          The OAuth2 client ID that has been provisioned in the identity provider.
+        example: '"my-client-id"'
+        type: string
+        required: true
+      - name: oidcClientAssertionCert
+        type: string
+        required: true
+        description: |
+          PEM-encoded X.509 certificate used to advertise the client certificate in the x5c header.
+        example: |
+          -----BEGIN CERTIFICATE-----\n...
+      - name: oidcClientAssertionKey
+        type: string
+        required: true
+        sensitive: true
+        description: |
+          PEM-encoded private key used to sign the client certificate.
+        example: |
+          -----BEGIN PRIVATE KEY-----\n...
+      - name: oidcResource
+        type: string
+        required: false
+        description: |
+          Optional OAuth2 resource (audience) parameter to include in the token request when required by the identity provider.
+        example: '"api://kafka"'
+      - name: oidcAudience
+        type: string
+        required: false
+        description: |
+          Overrides the JWT client assertion audience (aud). If not set, the component uses the
+          issuer derived from the token endpoint URL when available; otherwise, it falls back to the token URL.
+        example: '"http://<idp-host>/realms/local"'
+      - name: oidcScopes
+        type: string
+        description: |
+          Comma-delimited list of OAuth2/OIDC scopes to request with the access token.
+          Although not required, this field is recommended.
+        example: '"openid,kafka-prod"'
+        default: '"openid"'
+      - name: oidcExtensions
+        description: |
+          String containing a JSON-encoded dictionary of OAuth2/OIDC extensions to request with the access token.
+        example: |
+          {"cluster":"kafka","poolid":"kafkapool"}
+        type: string
   - title: "SASL Authentication"
     description: |
       Authenticate using SASL.
@@ -348,7 +414,7 @@ metadata:
     type: bool
     required: false
     description: |
-      Enables URL escaping of the message header values. 
+      Enables URL escaping of the message header values.
       It allows sending headers with special characters that are usually not allowed in HTTP headers.
     example: "true"
     default: "false"

common/component/kafka/auth.go

@@ -88,3 +88,20 @@ func updateOidcAuthInfo(config *sarama.Config, metadata *KafkaMetadata) error {
 
 	return nil
 }
+
+func updateOidcPrivateKeyJWTAuthInfo(config *sarama.Config, metadata *KafkaMetadata) error {
+	tokenProvider := metadata.getOAuthTokenSourcePrivateKeyJWT()
+
+	if metadata.TLSCaCert != "" {
+		err := tokenProvider.addCa(metadata.TLSCaCert)
+		if err != nil {
+			return fmt.Errorf("kafka: error setting oauth client trusted CA: %w", err)
+		}
+	}
+
+	config.Net.SASL.Enable = true
+	config.Net.SASL.Mechanism = sarama.SASLTypeOAuth
+	config.Net.SASL.TokenProvider = tokenProvider
+
+	return nil
+}

common/component/kafka/kafka.go

@@ -181,6 +181,12 @@ func (k *Kafka) Init(ctx context.Context, metadata map[string]string) error {
 		if err != nil {
 			return err
 		}
+	case oidcPrivateKeyJWTAuthType:
+		k.logger.Info("Configuring SASL OAuth2/OIDC authentication with private key JWT")
+		err = updateOidcPrivateKeyJWTAuthInfo(config, meta)
+		if err != nil {
+			return err
+		}
 	case passwordAuthType:
 		k.logger.Info("Configuring SASL Password authentication")
 		k.saslUsername = meta.SaslUsername

common/component/kafka/metadata.go

@@ -42,6 +42,7 @@ const (
 	authType                                 = "authType"
 	passwordAuthType                         = "password"
 	oidcAuthType                             = "oidc"
+	oidcPrivateKeyJWTAuthType                = "oidc_private_key_jwt"
 	mtlsAuthType                             = "mtls"
 	awsIAMAuthType                           = "awsiam"
 	noAuthType                               = "none"
@@ -65,36 +66,40 @@ const (
 )
 
 type KafkaMetadata struct {
-	Brokers                string              `mapstructure:"brokers"`
-	internalBrokers        []string            `mapstructure:"-"`
-	ConsumerGroup          string              `mapstructure:"consumerGroup"`
-	ClientID               string              `mapstructure:"clientId"`
-	AuthType               string              `mapstructure:"authType"`
-	SaslUsername           string              `mapstructure:"saslUsername"`
-	SaslPassword           string              `mapstructure:"saslPassword"`
-	SaslMechanism          string              `mapstructure:"saslMechanism"`
-	InitialOffset          string              `mapstructure:"initialOffset"`
-	internalInitialOffset  int64               `mapstructure:"-"`
-	MaxMessageBytes        int                 `mapstructure:"maxMessageBytes"`
-	OidcTokenEndpoint      string              `mapstructure:"oidcTokenEndpoint"`
-	OidcClientID           string              `mapstructure:"oidcClientID"`
-	OidcClientSecret       string              `mapstructure:"oidcClientSecret"`
-	OidcScopes             string              `mapstructure:"oidcScopes"`
-	OidcExtensions         string              `mapstructure:"oidcExtensions"`
-	internalOidcScopes     []string            `mapstructure:"-"`
-	TLSDisable             bool                `mapstructure:"disableTls"`
-	TLSSkipVerify          bool                `mapstructure:"skipVerify"`
-	TLSCaCert              string              `mapstructure:"caCert"`
-	TLSClientCert          string              `mapstructure:"clientCert"`
-	TLSClientKey           string              `mapstructure:"clientKey"`
-	ConsumeRetryEnabled    bool                `mapstructure:"consumeRetryEnabled"`
-	ConsumeRetryInterval   time.Duration       `mapstructure:"consumeRetryInterval"`
-	HeartbeatInterval      time.Duration       `mapstructure:"heartbeatInterval"`
-	SessionTimeout         time.Duration       `mapstructure:"sessionTimeout"`
-	Version                string              `mapstructure:"version"`
-	EscapeHeaders          bool                `mapstructure:"escapeHeaders"`
-	internalVersion        sarama.KafkaVersion `mapstructure:"-"`
-	internalOidcExtensions map[string]string   `mapstructure:"-"`
+	Brokers                 string              `mapstructure:"brokers"`
+	internalBrokers         []string            `mapstructure:"-"`
+	ConsumerGroup           string              `mapstructure:"consumerGroup"`
+	ClientID                string              `mapstructure:"clientId"`
+	AuthType                string              `mapstructure:"authType"`
+	SaslUsername            string              `mapstructure:"saslUsername"`
+	SaslPassword            string              `mapstructure:"saslPassword"`
+	SaslMechanism           string              `mapstructure:"saslMechanism"`
+	InitialOffset           string              `mapstructure:"initialOffset"`
+	internalInitialOffset   int64               `mapstructure:"-"`
+	MaxMessageBytes         int                 `mapstructure:"maxMessageBytes"`
+	OidcTokenEndpoint       string              `mapstructure:"oidcTokenEndpoint"`
+	OidcClientID            string              `mapstructure:"oidcClientID"`
+	OidcClientSecret        string              `mapstructure:"oidcClientSecret"`
+	OidcScopes              string              `mapstructure:"oidcScopes"`
+	OidcExtensions          string              `mapstructure:"oidcExtensions"`
+	OidcClientAssertionCert string              `mapstructure:"oidcClientAssertionCert"`
+	OidcClientAssertionKey  string              `mapstructure:"oidcClientAssertionKey"`
+	OidcResource            string              `mapstructure:"oidcResource"`
+	OidcAudience            string              `mapstructure:"oidcAudience"`
+	internalOidcScopes      []string            `mapstructure:"-"`
+	TLSDisable              bool                `mapstructure:"disableTls"`
+	TLSSkipVerify           bool                `mapstructure:"skipVerify"`
+	TLSCaCert               string              `mapstructure:"caCert"`
+	TLSClientCert           string              `mapstructure:"clientCert"`
+	TLSClientKey            string              `mapstructure:"clientKey"`
+	ConsumeRetryEnabled     bool                `mapstructure:"consumeRetryEnabled"`
+	ConsumeRetryInterval    time.Duration       `mapstructure:"consumeRetryInterval"`
+	HeartbeatInterval       time.Duration       `mapstructure:"heartbeatInterval"`
+	SessionTimeout          time.Duration       `mapstructure:"sessionTimeout"`
+	Version                 string              `mapstructure:"version"`
+	EscapeHeaders           bool                `mapstructure:"escapeHeaders"`
+	internalVersion         sarama.KafkaVersion `mapstructure:"-"`
+	internalOidcExtensions  map[string]string   `mapstructure:"-"`
 
 	// configs for kafka client
 	ClientConnectionTopicMetadataRefreshInterval time.Duration `mapstructure:"clientConnectionTopicMetadataRefreshInterval"`
@@ -251,6 +256,32 @@ func (k *Kafka) getKafkaMetadata(meta map[string]string) (*KafkaMetadata, error)
 			}
 		}
 		k.logger.Debug("Configuring SASL token authentication via OIDC.")
+	case oidcPrivateKeyJWTAuthType:
+		if m.OidcTokenEndpoint == "" {
+			return nil, errors.New("kafka error: missing OIDC Token Endpoint for authType 'oidc_private_key_jwt'")
+		}
+		if m.OidcClientID == "" {
+			return nil, errors.New("kafka error: missing OIDC Client ID for authType 'oidc_private_key_jwt'")
+		}
+		if m.OidcClientAssertionCert == "" {
+			return nil, errors.New("kafka error: missing OIDC Client Assertion Cert for authType 'oidc_private_key_jwt'")
+		}
+		if m.OidcClientAssertionKey == "" {
+			return nil, errors.New("kafka error: missing OIDC Client Assertion Key for authType 'oidc_private_key_jwt'")
+		}
+		if m.OidcScopes != "" {
+			m.internalOidcScopes = strings.Split(m.OidcScopes, ",")
+		} else {
+			k.logger.Warn("Warning: no OIDC scopes specified, using default 'openid' scope only. This is a security risk for token reuse.")
+			m.internalOidcScopes = []string{"openid"}
+		}
+		if m.OidcExtensions != "" {
+			err = json.Unmarshal([]byte(m.OidcExtensions), &m.internalOidcExtensions)
+			if err != nil || len(m.internalOidcExtensions) < 1 {
+				return nil, errors.New("kafka error: improper OIDC Extensions format for authType 'oidc_private_key_jwt'")
+			}
+		}
+		k.logger.Debug("Configuring SASL token authentication via OIDC with private_key_jwt.")
 	case mtlsAuthType:
 		if m.TLSClientCert != "" {
 			if !isValidPEM(m.TLSClientCert) {

common/component/kafka/metadata_test.go

@@ -216,7 +216,7 @@ func TestMissingSaslValuesOnUpgrade(t *testing.T) {
 	require.Equal(t, fmt.Sprintf("kafka error: missing SASL Username for authType '%s'", passwordAuthType), err.Error())
 }
 
-func TestMissingOidcValues(t *testing.T) {
+func TestMissingOidcClientSecretValues(t *testing.T) {
 	k := getKafka()
 	m := map[string]string{"brokers": "akfak.com:9092", "authType": oidcAuthType}
 	meta, err := k.getKafkaMetadata(m)
@@ -243,6 +243,38 @@ func TestMissingOidcValues(t *testing.T) {
 	require.Contains(t, meta.internalOidcScopes, "openid")
 }
 
+func TestMissingOidcPrivateKeyJwtValues(t *testing.T) {
+	k := getKafka()
+	m := map[string]string{"brokers": "akfak.com:9092", "authType": oidcPrivateKeyJWTAuthType}
+	meta, err := k.getKafkaMetadata(m)
+	require.Error(t, err)
+	require.Nil(t, meta)
+	require.Equal(t, "kafka error: missing OIDC Token Endpoint for authType 'oidc_private_key_jwt'", err.Error())
+
+	m["oidcTokenEndpoint"] = "https://sassa.fra/"
+	meta, err = k.getKafkaMetadata(m)
+	require.Error(t, err)
+	require.Nil(t, meta)
+	require.Equal(t, "kafka error: missing OIDC Client ID for authType 'oidc_private_key_jwt'", err.Error())
+
+	m["oidcClientID"] = "sassafras"
+	meta, err = k.getKafkaMetadata(m)
+	require.Error(t, err)
+	require.Nil(t, meta)
+	require.Equal(t, "kafka error: missing OIDC Client Assertion Cert for authType 'oidc_private_key_jwt'", err.Error())
+
+	m["oidcClientAssertionCert"] = "sassapass"
+	meta, err = k.getKafkaMetadata(m)
+	require.Error(t, err)
+	require.Nil(t, meta)
+	require.Equal(t, "kafka error: missing OIDC Client Assertion Key for authType 'oidc_private_key_jwt'", err.Error())
+
+	m["oidcClientAssertionKey"] = "sassapass"
+	meta, err = k.getKafkaMetadata(m)
+	require.NoError(t, err)
+	require.Contains(t, meta.internalOidcScopes, "openid")
+}
+
 func TestPresentSaslValues(t *testing.T) {
 	k := getKafka()
 	m := map[string]string{

common/component/kafka/sasl_oauthbearer.go

@@ -17,12 +17,13 @@ import (
 	ctx "context"
 	"crypto/tls"
 	"crypto/x509"
-	"encoding/pem"
 	"errors"
 	"fmt"
 	"net/http"
 	"time"
 
+	"github.com/dapr/kit/crypto/pem"
+
 	"github.com/IBM/sarama"
 	"golang.org/x/oauth2"
 	ccred "golang.org/x/oauth2/clientcredentials"
@@ -51,26 +52,21 @@ func (m KafkaMetadata) getOAuthTokenSource() *OAuthTokenSource {
 	}
 }
 
-var tokenRequestTimeout, _ = time.ParseDuration("30s")
-
 func (ts *OAuthTokenSource) addCa(caPem string) error {
 	pemBytes := []byte(caPem)
 
-	block, _ := pem.Decode(pemBytes)
-
-	if block == nil || block.Type != "CERTIFICATE" {
-		return errors.New("PEM data not valid or not of a valid type (CERTIFICATE)")
-	}
-
-	caCert, err := x509.ParseCertificate(block.Bytes)
+	caCerts, err := pem.DecodePEMCertificates(pemBytes)
 	if err != nil {
 		return fmt.Errorf("error parsing PEM certificate: %w", err)
 	}
+	if len(caCerts) > 1 {
+		return fmt.Errorf("expected 1 certificate, got %d", len(caCerts))
+	}
 
 	if ts.trustedCas == nil {
 		ts.trustedCas = make([]*x509.Certificate, 0)
 	}
-	ts.trustedCas = append(ts.trustedCas, caCert)
+	ts.trustedCas = append(ts.trustedCas, caCerts[0])
 
 	return nil
 }
@@ -113,9 +109,15 @@ func (ts *OAuthTokenSource) Token() (*sarama.AccessToken, error) {
 		return nil, errors.New("cannot generate token, OAuthTokenSource not fully configured")
 	}
 
-	oidcCfg := ccred.Config{ClientID: ts.ClientID, ClientSecret: ts.ClientSecret, Scopes: ts.Scopes, TokenURL: ts.TokenEndpoint.TokenURL, AuthStyle: ts.TokenEndpoint.AuthStyle}
+	oidcCfg := ccred.Config{
+		ClientID:     ts.ClientID,
+		ClientSecret: ts.ClientSecret,
+		Scopes:       ts.Scopes,
+		TokenURL:     ts.TokenEndpoint.TokenURL,
+		AuthStyle:    ts.TokenEndpoint.AuthStyle,
+	}
 
-	timeoutCtx, cancel := ctx.WithTimeout(ctx.TODO(), tokenRequestTimeout)
+	timeoutCtx, cancel := ctx.WithTimeout(ctx.TODO(), 30*time.Second)
 	defer cancel()
 
 	ts.configureClient()