Add kid header to kafka jwt auth

bindings/kafka/metadata.yaml

@@ -187,6 +187,12 @@ authenticationProfiles:
         example: |
           {"cluster":"kafka","poolid":"kafkapool"}
         type: string
+      - name: oidcKid
+        type: string
+        required: false
+        description: |
+          The JWT key ID (kid) to use for the client assertion.
+        example: '"1234567890"'
   - title: "SASL Authentication"
     description: |
       Authenticate using SASL.

common/component/kafka/metadata.go

@@ -86,6 +86,7 @@ type KafkaMetadata struct {
 	OidcClientAssertionKey  string              `mapstructure:"oidcClientAssertionKey"`
 	OidcResource            string              `mapstructure:"oidcResource"`
 	OidcAudience            string              `mapstructure:"oidcAudience"`
+	OidcKid                 string              `mapstructure:"oidcKid"`
 	internalOidcScopes      []string            `mapstructure:"-"`
 	TLSDisable              bool                `mapstructure:"disableTls"`
 	TLSSkipVerify           bool                `mapstructure:"skipVerify"`

common/component/kafka/metadata_test.go

@@ -273,6 +273,11 @@ func TestMissingOidcPrivateKeyJwtValues(t *testing.T) {
 	meta, err = k.getKafkaMetadata(m)
 	require.NoError(t, err)
 	require.Contains(t, meta.internalOidcScopes, "openid")
+
+	m["oidcKid"] = "1234567890"
+	meta, err = k.getKafkaMetadata(m)
+	require.NoError(t, err)
+	require.Equal(t, "1234567890", meta.OidcKid)
 }
 
 func TestPresentSaslValues(t *testing.T) {

common/component/kafka/sasl_oauthbearer_private_key_jwt.go

@@ -31,6 +31,7 @@ import (
 	"github.com/IBM/sarama"
 	"github.com/google/uuid"
 	"github.com/lestrrat-go/jwx/v2/jwa"
+	"github.com/lestrrat-go/jwx/v2/jws"
 	"github.com/lestrrat-go/jwx/v2/jwt"
 	"golang.org/x/oauth2"
 )
@@ -50,6 +51,7 @@ type OAuthTokenSourcePrivateKeyJWT struct {
 	ClientAssertionKey  string
 	Resource            string
 	Audience            string
+	Kid                 string
 }
 
 type tokenResponse struct {
@@ -70,6 +72,7 @@ func (m KafkaMetadata) getOAuthTokenSourcePrivateKeyJWT() *OAuthTokenSourcePriva
 		ClientAssertionKey:  m.OidcClientAssertionKey,
 		Resource:            m.OidcResource,
 		Audience:            m.OidcAudience,
+		Kid:                 m.OidcKid,
 	}
 }
 
@@ -166,7 +169,15 @@ func (ts *OAuthTokenSourcePrivateKeyJWT) Token() (*sarama.AccessToken, error) {
 		return nil, fmt.Errorf("failed to build token: %w", err)
 	}
 
-	assertion, err := jwt.Sign(token, jwt.WithKey(jwa.RS256, rsaKey))
+	var signOptions []jwt.Option
+	if ts.Kid != "" {
+		headers := jws.NewHeaders()
+		if err := headers.Set("kid", ts.Kid); err != nil {
+			return nil, fmt.Errorf("error setting JWT kid header: %w", err)
+		}
+		signOptions = append(signOptions, jws.WithProtectedHeaders(headers))
+	}
+	assertion, err := jwt.Sign(token, jwt.WithKey(jwa.RS256, rsaKey, signOptions...))
 	if err != nil {
 		return nil, fmt.Errorf("error signing client assertion: %w", err)
 	}

pubsub/kafka/metadata.yaml

@@ -181,6 +181,12 @@ authenticationProfiles:
         example: |
           {"cluster":"kafka","poolid":"kafkapool"}
         type: string
+      - name: oidcKid
+        type: string
+        required: false
+        description: |
+          The JWT key ID (kid) to use for the client assertion.
+        example: '"1234567890"'
   - title: "SASL Authentication"
     description: |
       Authenticate using SASL.

tests/certification/pubsub/kafka/components/auth_oidc_certs/kafka.yaml

@@ -28,6 +28,10 @@ spec:
     value: "openid"
   - name: disableTls
     value: "true"
+  - name: oidcKid
+    secretKeyRef:
+      name: OIDC_CLIENT_KID
+      key:  OIDC_CLIENT_KID
 
 auth:
   secretStore: envvar-secret-store

tests/certification/pubsub/kafka/data/realm-export.json

@@ -8,8 +8,8 @@
       "secret": "dapr-kafka-secret",
       "clientAuthenticatorType": "client-jwt",
       "attributes": {
-        "token.endpoint.auth.signing.alg": "RS256",
-        "jwt.credential.certificate": "${OIDC_CLIENT_ASSERTION_CERT_ONELINE}"
+         "use.jwks.url": "true",
+          "jwks.url": "http://jwks:80/jwks.json"
       },
       "enabled": true,
       "protocol": "openid-connect",

tests/certification/pubsub/kafka/docker-compose.auth.yml

@@ -60,3 +60,14 @@ services:
       - "8080:8080"
     volumes:
       - ./data/realm-export.json:/opt/keycloak/data/import/local-realm.json
+
+  jwks:
+    image: nginx:1.27-alpine
+    hostname: jwks
+    container_name: kafka_jwks
+    environment:
+      - OIDC_CLIENT_KID
+      - OIDC_CLIENT_JWK_N
+      - OIDC_CLIENT_JWK_E
+    command: >
+      /bin/sh -c 'printf "{\n  \"keys\": [\n    {\n      \"kty\": \"RSA\",\n      \"use\": \"sig\",\n      \"alg\": \"RS256\",\n      \"kid\": \"%s\",\n      \"n\": \"%s\",\n      \"e\": \"%s\"\n    }\n  ]\n}\n" "$$OIDC_CLIENT_KID" "$$OIDC_CLIENT_JWK_N" "$$OIDC_CLIENT_JWK_E" > /usr/share/nginx/html/jwks.json && exec nginx -g "daemon off;"'

tests/certification/pubsub/kafka/kafka_test.go

@@ -19,6 +19,7 @@ import (
 	"crypto/rsa"
 	"crypto/x509"
 	"crypto/x509/pkix"
+	"encoding/base64"
 	"encoding/json"
 	"encoding/pem"
 	"fmt"
@@ -522,6 +523,12 @@ func TestKafkaAuth(t *testing.T) {
 	os.Setenv("OIDC_CLIENT_ASSERTION_KEY", string(keyPEM))
 	os.Setenv("OIDC_CLIENT_ASSERTION_CERT_ONELINE", strings.ReplaceAll(string(certPEM), "\n", "\\n"))
 
+	modulus := key.PublicKey.N.Bytes()
+	os.Setenv("OIDC_CLIENT_JWK_N", base64.RawURLEncoding.EncodeToString(modulus))
+	exponent := big.NewInt(int64(key.PublicKey.E)).Bytes()
+	os.Setenv("OIDC_CLIENT_JWK_E", base64.RawURLEncoding.EncodeToString(exponent))
+	os.Setenv("OIDC_CLIENT_KID", uuid.New().String())
+
 	flow.New(t, "kafka authentication").
 		Step(dockercompose.Run(clusterNameAuth, dockerComposeYAMLAuth)).
 		Step("wait for broker sockets",