Cluster Security: Add Service Account authentication to the Kafka Agent

cluster-operator/src/main/java/io/strimzi/operator/cluster/auth/RequestedServiceAccountAuthIdentity.java

@@ -60,4 +60,40 @@ public Map<String, String> kafkaClientProperties() {
 
         return config;
     }
+
+    /**
+     * Returns the namespace of the service account
+     *
+     * @return  The namespace of the service account
+     */
+    public String namespace() {
+        return namespace;
+    }
+
+    /**
+     * Returns the service account name
+     *
+     * @return  The service account name
+     */
+    public String serviceAccountName() {
+        return serviceAccountName;
+    }
+
+    /**
+     * Returns the audience for the token
+     *
+     * @return  The token audience
+     */
+    public String audience()    {
+        return audience;
+    }
+
+    /**
+     * Returns the expiration time for the token in seconds
+     *
+     * @return  The token expiration time in seconds
+     */
+    public long expirationSeconds()    {
+        return expirationSeconds;
+    }
 }

cluster-operator/src/main/java/io/strimzi/operator/cluster/model/KafkaAgentConfigurationBuilder.java

@@ -4,8 +4,10 @@
  */
 package io.strimzi.operator.cluster.model;
 
+import io.strimzi.api.kafka.model.kafka.KafkaResources;
 import io.strimzi.operator.cluster.model.clustersecurity.kafka.KafkaClusterSecurityContext;
 import io.strimzi.operator.cluster.model.clustersecurity.kafka.MtlsAuthenticationConfiguration;
+import io.strimzi.operator.cluster.model.clustersecurity.kafka.ServiceAccountAuthenticationConfiguration;
 import io.strimzi.operator.cluster.model.clustersecurity.kafka.TlsEncryptionConfiguration;
 import io.strimzi.operator.common.Reconciliation;
 
@@ -48,10 +50,17 @@ public KafkaAgentConfigurationBuilder withSecurity(KafkaClusterSecurityContext s
         if (securityContext.encryption() instanceof TlsEncryptionConfiguration)   {
             writer.println("namespace=" + reconciliation.namespace());
             writer.println("sslKeyStoreSecretName=" + node.podName());
+        }
 
-            if (securityContext.authentication() instanceof MtlsAuthenticationConfiguration) {
-                writer.println("sslTrustStoreSecretName=" + reconciliation.name() + "-cluster-ca-cert");
-            }
+        if (securityContext.authentication() instanceof MtlsAuthenticationConfiguration) {
+            // The Security Context validated that mTLS is used only when TLS is used. We do not need to validate it again.
+            writer.println("sslTrustStoreSecretName=" + reconciliation.name() + "-cluster-ca-cert");
+        } else if (securityContext.authentication() instanceof ServiceAccountAuthenticationConfiguration saAuthentication) {
+            writer.println("tokenIssuer=" + ServiceAccountAuthenticationConfiguration.ISSUER);
+            writer.println("tokenJwksUri=" + ServiceAccountAuthenticationConfiguration.JWKS_URI);
+            writer.println("tokenJwksCaPath=" + "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt");
+            writer.println("tokenAudience=" + saAuthentication.audience());
+            writer.println("tokenAllowedUsers=system:serviceaccount:%s:%s".formatted(reconciliation.namespace(), KafkaResources.clusterOperatorServiceAccount(reconciliation.name())));
         }
 
         return this;

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/DefaultKafkaAgentClientProvider.java

@@ -4,20 +4,27 @@
  */
 package io.strimzi.operator.cluster.operator.resource;
 
+import io.fabric8.kubernetes.client.KubernetesClient;
 import io.strimzi.operator.common.Reconciliation;
 import io.strimzi.operator.common.auth.Identity;
 
 /**
  * Class to provide the real KafkaAgentClient which connects to actual Kafka Agent
  */
 public class DefaultKafkaAgentClientProvider implements KafkaAgentClientProvider {
+    private final KubernetesClient kubernetesClient;
+
     /**
      * Constructor
+     *
+     * @param kubernetesClient  Kubernetes client to interact with the Kubernetes API
      */
-    public DefaultKafkaAgentClientProvider() { }
+    public DefaultKafkaAgentClientProvider(KubernetesClient kubernetesClient) {
+        this.kubernetesClient = kubernetesClient;
+    }
 
     @Override
     public KafkaAgentClient createKafkaAgentClient(Reconciliation reconciliation, Identity identity) {
-        return new KafkaAgentClient(reconciliation, reconciliation.name(), reconciliation.namespace(), identity);
+        return new KafkaAgentClient(reconciliation, reconciliation.name(), reconciliation.namespace(), identity, kubernetesClient);
     }
 }

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/KafkaAgentClient.java

@@ -6,7 +6,11 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import io.fabric8.kubernetes.api.model.authentication.TokenRequest;
+import io.fabric8.kubernetes.api.model.authentication.TokenRequestBuilder;
+import io.fabric8.kubernetes.client.KubernetesClient;
 import io.strimzi.api.kafka.model.kafka.KafkaResources;
+import io.strimzi.operator.cluster.auth.RequestedServiceAccountAuthIdentity;
 import io.strimzi.operator.cluster.model.DnsNameGenerator;
 import io.strimzi.operator.common.Reconciliation;
 import io.strimzi.operator.common.ReconciliationLogger;
@@ -26,6 +30,7 @@
 import java.net.http.HttpResponse;
 import java.security.GeneralSecurityException;
 import java.time.Duration;
+import java.time.Instant;
 
 /**
  * Creates HTTP client and interacts with Kafka Agent's REST endpoint
@@ -41,25 +46,34 @@ public class KafkaAgentClient {
     // executor indefinitely. The Kafka Agent only serves a small broker-state JSON, so 10 seconds is well above
     // the expected response time on a healthy broker yet small enough to keep the roller responsive.
     private static final Duration HTTP_REQUEST_TIMEOUT = Duration.ofSeconds(10);
+    // Fraction of the token lifetime after which the token is renewed.
+    private static final double TOKEN_RENEWAL_THRESHOLD = 0.8;
+
     private final String namespace;
     private final Reconciliation reconciliation;
     private final String cluster;
     private final Identity identity;
+    private final KubernetesClient kubernetesClient;
     private final HttpClient httpClient;
 
+    private String cachedToken;
+    private long cachedTokenRefreshAt;
+
     /**
      * Constructor
      *
      * @param reconciliation    Reconciliation marker
-     * @param cluster   Cluster name
-     * @param namespace Cluster namespace
-     * @param identity Trust set and identity for authentication for connecting to the Kafka cluster
+     * @param cluster           Cluster name
+     * @param namespace         Cluster namespace
+     * @param identity          Trust set and identity for authentication for connecting to the Kafka cluster
+     * @param kubernetesClient  Kubernetes client used to get the token for the service account
      */
-    public KafkaAgentClient(Reconciliation reconciliation, String cluster, String namespace, Identity identity) {
+    public KafkaAgentClient(Reconciliation reconciliation, String cluster, String namespace, Identity identity, KubernetesClient kubernetesClient) {
         this.reconciliation = reconciliation;
         this.cluster = cluster;
         this.namespace = namespace;
         this.identity = identity;
+        this.kubernetesClient = kubernetesClient;
         this.httpClient = createHttpClient();
     }
 
@@ -106,13 +120,16 @@ private HttpClient createHttpClient() {
 
     String doGet(URI uri) {
         try {
-            HttpRequest req = HttpRequest.newBuilder()
+            HttpRequest.Builder reqBuilder = HttpRequest.newBuilder()
                     .uri(uri)
                     .timeout(HTTP_REQUEST_TIMEOUT)
-                    .GET()
-                    .build();
+                    .GET();
+
+            if (identity.authIdentity() instanceof RequestedServiceAccountAuthIdentity authIdentity) {
+                reqBuilder.header("Authorization", "Bearer " + currentToken(authIdentity));
+            }
 
-            var response = httpClient.send(req, HttpResponse.BodyHandlers.ofString());
+            var response = httpClient.send(reqBuilder.build(), HttpResponse.BodyHandlers.ofString());
             if (response.statusCode() != 200) {
                 throw new RuntimeException("Unexpected HTTP status code: " + response.statusCode());
             }
@@ -145,4 +162,41 @@ public BrokerState getBrokerState(String podName) {
         }
         return brokerstate;
     }
+
+    /**
+     * Returns a valid Service Account token for the per-cluster cluster-operator SA, minting a fresh one via the
+     * Kubernetes TokenRequest API when no cached token is available, or when the cached one should be refreshed.
+     *
+     * @param authIdentity  The RequestedServiceAccountAuthIdentity containing the authentication details
+     *
+     * @return  JWT token string suitable for use in an HTTP Authorization Bearer header
+     */
+    synchronized String currentToken(RequestedServiceAccountAuthIdentity authIdentity) {
+        // If we do not have the token yet or it should be refreshed, we get the new token from Kube API
+        if (cachedToken == null || System.currentTimeMillis() >= cachedTokenRefreshAt) {
+            TokenRequest request = new TokenRequestBuilder()
+                    .withNewSpec()
+                        .withAudiences(authIdentity.audience())
+                        .withExpirationSeconds(authIdentity.expirationSeconds())
+                    .endSpec()
+                    .build();
+            TokenRequest response = kubernetesClient.serviceAccounts()
+                    .inNamespace(authIdentity.namespace())
+                    .withName(authIdentity.serviceAccountName())
+                    .tokenRequest(request);
+
+            if (response == null || response.getStatus() == null || response.getStatus().getToken() == null) {
+                throw new RuntimeException("Kubernetes API did not return a token for ServiceAccount " + authIdentity.namespace() + "/" + authIdentity.serviceAccountName());
+            }
+
+            cachedToken = response.getStatus().getToken();
+
+            // The token refresh time is computed from its expiration time to refresh it before it is expired.
+            long now = System.currentTimeMillis();
+            long expiresAt = Instant.parse(response.getStatus().getExpirationTimestamp()).toEpochMilli();
+            cachedTokenRefreshAt = now + (long) (TOKEN_RENEWAL_THRESHOLD * (expiresAt - now));
+        }
+
+        return cachedToken;
+    }
 }

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/ResourceOperatorSupplier.java

@@ -245,7 +245,7 @@ public ResourceOperatorSupplier(Executor asyncExecutor, KubernetesClient client,
         this(asyncExecutor,
             client,
             new DefaultAdminClientProvider(),
-            new DefaultKafkaAgentClientProvider(),
+            new DefaultKafkaAgentClientProvider(client),
             metricsProvider,
             pfa,
             new KubernetesRestartEventPublisher(client, operatorName),

cluster-operator/src/test/java/io/strimzi/operator/cluster/auth/RequestedServiceAccountAuthIdentityTest.java

@@ -27,4 +27,14 @@ public void testClientConfiguration() {
         assertThat(authIdentity.isSasl(), is(true));
         assertThat(authIdentity.kafkaClientProperties(), is(expectedClientProperties));
     }
+
+    @Test
+    public void testTokenRequestDetails() {
+        RequestedServiceAccountAuthIdentity authIdentity = new RequestedServiceAccountAuthIdentity(Reconciliation.DUMMY_RECONCILIATION, "strimzi.io/kafka/namespace/name", 1800L);
+
+        assertThat(authIdentity.namespace(), is("namespace"));
+        assertThat(authIdentity.serviceAccountName(), is("name-cluster-operator"));
+        assertThat(authIdentity.audience(), is("strimzi.io/kafka/namespace/name"));
+        assertThat(authIdentity.expirationSeconds(), is(1800L));
+    }
 }

cluster-operator/src/test/java/io/strimzi/operator/cluster/model/KafkaAgentConfigurationBuilderTest.java

@@ -4,6 +4,9 @@
  */
 package io.strimzi.operator.cluster.model;
 
+import io.strimzi.api.kafka.model.kafka.clustersecurity.ClusterSecurityAuthenticationBuilder;
+import io.strimzi.api.kafka.model.kafka.clustersecurity.ClusterSecurityAuthenticationType;
+import io.strimzi.operator.cluster.model.clustersecurity.kafka.AuthenticationConfiguration;
 import io.strimzi.operator.cluster.model.clustersecurity.kafka.KafkaClusterSecurityContext;
 import io.strimzi.operator.cluster.model.clustersecurity.kafka.NoneAuthenticationConfiguration;
 import io.strimzi.operator.cluster.model.clustersecurity.kafka.NoneEncryptionConfiguration;
@@ -48,6 +51,46 @@ public void testTlsWithoutAuthentication()  {
         ));
     }
 
+    @Test
+    public void testTlsAndServiceAccountAuthentication()  {
+        KafkaClusterSecurityContext securityContext = mock(KafkaClusterSecurityContext.class);
+        when(securityContext.encryption()).thenReturn(new TlsEncryptionConfiguration());
+        when(securityContext.authentication()).thenReturn(AuthenticationConfiguration.fromCrd("namespace", "name", new ClusterSecurityAuthenticationBuilder().withType(ClusterSecurityAuthenticationType.SERVICE_ACCOUNT).build()));
+
+        String configuration = new KafkaAgentConfigurationBuilder(Reconciliation.DUMMY_RECONCILIATION, NODE_REF)
+                .withSecurity(securityContext)
+                .build();
+
+        assertThat(configuration, isEquivalent(
+                "namespace=namespace",
+                "sslKeyStoreSecretName=my-cluster-kafka-2",
+                "tokenIssuer=https://kubernetes.default.svc.cluster.local",
+                "tokenJwksUri=https://kubernetes.default.svc.cluster.local/openid/v1/jwks",
+                "tokenJwksCaPath=/var/run/secrets/kubernetes.io/serviceaccount/ca.crt",
+                "tokenAudience=strimzi.io/kafka/namespace/name",
+                "tokenAllowedUsers=system:serviceaccount:namespace:name-cluster-operator"
+        ));
+    }
+
+    @Test
+    public void testServiceAccountAuthenticationWithoutTls()  {
+        KafkaClusterSecurityContext securityContext = mock(KafkaClusterSecurityContext.class);
+        when(securityContext.encryption()).thenReturn(new NoneEncryptionConfiguration());
+        when(securityContext.authentication()).thenReturn(AuthenticationConfiguration.fromCrd("namespace", "name", new ClusterSecurityAuthenticationBuilder().withType(ClusterSecurityAuthenticationType.SERVICE_ACCOUNT).build()));
+
+        String configuration = new KafkaAgentConfigurationBuilder(Reconciliation.DUMMY_RECONCILIATION, NODE_REF)
+                .withSecurity(securityContext)
+                .build();
+
+        assertThat(configuration, isEquivalent(
+                "tokenIssuer=https://kubernetes.default.svc.cluster.local",
+                "tokenJwksUri=https://kubernetes.default.svc.cluster.local/openid/v1/jwks",
+                "tokenJwksCaPath=/var/run/secrets/kubernetes.io/serviceaccount/ca.crt",
+                "tokenAudience=strimzi.io/kafka/namespace/name",
+                "tokenAllowedUsers=system:serviceaccount:namespace:name-cluster-operator"
+        ));
+    }
+
     @Test
     public void testWithoutTlsOrAuthentication()  {
         KafkaClusterSecurityContext securityContext = mock(KafkaClusterSecurityContext.class);

cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/ConnectorMockTest.java

@@ -190,7 +190,7 @@ public void beforeEach(TestInfo testInfo, VertxTestContext testContext) {
         ResourceOperatorSupplier ros = new ResourceOperatorSupplier(VertxUtil.asExecutor(vertx.createSharedWorkerExecutor("kubernetes-ops-pool")),
                 client,
                 new DefaultAdminClientProvider(),
-                new DefaultKafkaAgentClientProvider(),
+                new DefaultKafkaAgentClientProvider(null),
                 metricsProvider,
                 pfa);
 

cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/KafkaConnectorIT.java

@@ -158,7 +158,7 @@ public void testConnectorNotUpdatedWhenConfigUnchanged(VertxTestContext context)
         ResourceOperatorSupplier ros = new ResourceOperatorSupplier(VertxUtil.asExecutor(vertx.createSharedWorkerExecutor("kubernetes-ops-pool")),
                 client,
                 new DefaultAdminClientProvider(),
-                new DefaultKafkaAgentClientProvider(),
+                new DefaultKafkaAgentClientProvider(null),
                 metrics,
                 pfa
         );
@@ -225,7 +225,7 @@ public void     testConnectorResourceNotReadyWhenConnectorFailed(VertxTestContex
         ResourceOperatorSupplier ros = new ResourceOperatorSupplier(VertxUtil.asExecutor(vertx.createSharedWorkerExecutor("kubernetes-ops-pool")),
                 client,
                 new DefaultAdminClientProvider(),
-                new DefaultKafkaAgentClientProvider(),
+                new DefaultKafkaAgentClientProvider(null),
                 metrics,
                 pfa
         );
@@ -271,7 +271,7 @@ public void testConnectorResourceNotReadyWhenTaskFailed(VertxTestContext context
         ResourceOperatorSupplier ros = new ResourceOperatorSupplier(VertxUtil.asExecutor(vertx.createSharedWorkerExecutor("kubernetes-ops-pool")),
                 client,
                 new DefaultAdminClientProvider(),
-                new DefaultKafkaAgentClientProvider(),
+                new DefaultKafkaAgentClientProvider(null),
                 metrics,
                 pfa
         );
@@ -328,7 +328,7 @@ public void testConnectorIsAutoRestarted(VertxTestContext context) {
         ResourceOperatorSupplier ros = new ResourceOperatorSupplier(VertxUtil.asExecutor(vertx.createSharedWorkerExecutor("kubernetes-ops-pool")),
                 client,
                 new DefaultAdminClientProvider(),
-                new DefaultKafkaAgentClientProvider(),
+                new DefaultKafkaAgentClientProvider(null),
             metrics,
                 pfa
         );
@@ -374,7 +374,7 @@ public void testTaskIsAutoRestarted(VertxTestContext context) {
         ResourceOperatorSupplier ros = new ResourceOperatorSupplier(VertxUtil.asExecutor(vertx.createSharedWorkerExecutor("kubernetes-ops-pool")),
                 client,
                 new DefaultAdminClientProvider(),
-                new DefaultKafkaAgentClientProvider(),
+                new DefaultKafkaAgentClientProvider(null),
             metrics,
                 pfa
         );

cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/KafkaMirrorMaker2AssemblyOperatorMockTest.java

@@ -158,7 +158,7 @@ public void beforeEach(TestInfo testInfo) {
         supplier = new ResourceOperatorSupplier(VertxUtil.asExecutor(vertx.createSharedWorkerExecutor("kubernetes-ops-pool")),
                 client,
                 new DefaultAdminClientProvider(),
-                new DefaultKafkaAgentClientProvider(),
+                new DefaultKafkaAgentClientProvider(null),
                 ResourceUtils.metricsProvider(),
                 PFA);
         podSetController = new StrimziPodSetController(namespace, Labels.EMPTY, supplier.kafkaOperator, supplier.connectOperator, supplier.mirrorMaker2Operator, supplier.strimziPodSetOperator, supplier.podOperations, supplier.metricsProvider, Integer.parseInt(ClusterOperatorConfig.POD_SET_CONTROLLER_WORK_QUEUE_SIZE.defaultValue()));