From e38c8c80d964a034a2cffa0e09a6e91c7e06c02d Mon Sep 17 00:00:00 2001 From: Jakub Scholz Date: Sun, 30 Aug 2026 21:20:36 +0200 Subject: [PATCH] Cluster Security: Add Service Account authentication to the Kafka Agent Signed-off-by: Jakub Scholz --- .../RequestedServiceAccountAuthIdentity.java | 36 ++ .../model/KafkaAgentConfigurationBuilder.java | 15 +- .../DefaultKafkaAgentClientProvider.java | 11 +- .../operator/resource/KafkaAgentClient.java | 70 +++- .../resource/ResourceOperatorSupplier.java | 2 +- ...questedServiceAccountAuthIdentityTest.java | 10 + .../KafkaAgentConfigurationBuilderTest.java | 43 +++ .../operator/assembly/ConnectorMockTest.java | 2 +- .../operator/assembly/KafkaConnectorIT.java | 10 +- ...aMirrorMaker2AssemblyOperatorMockTest.java | 2 +- .../resource/KafkaAgentClientTest.java | 197 +++++++++- .../operator/resource/KafkaRollerTest.java | 56 +-- kafka-agent/pom.xml | 7 + .../io/strimzi/kafka/agent/KafkaAgent.java | 22 +- .../ServiceAccountAuthenticationHandler.java | 195 ++++++++++ .../strimzi/kafka/agent/KafkaAgentTest.java | 88 +++++ ...rviceAccountAuthenticationHandlerTest.java | 363 ++++++++++++++++++ pom.xml | 5 + 18 files changed, 1079 insertions(+), 55 deletions(-) create mode 100644 kafka-agent/src/main/java/io/strimzi/kafka/agent/ServiceAccountAuthenticationHandler.java create mode 100644 kafka-agent/src/test/java/io/strimzi/kafka/agent/ServiceAccountAuthenticationHandlerTest.java diff --git a/cluster-operator/src/main/java/io/strimzi/operator/cluster/auth/RequestedServiceAccountAuthIdentity.java b/cluster-operator/src/main/java/io/strimzi/operator/cluster/auth/RequestedServiceAccountAuthIdentity.java index 3e6eae00128..32362bd1608 100644 --- a/cluster-operator/src/main/java/io/strimzi/operator/cluster/auth/RequestedServiceAccountAuthIdentity.java +++ b/cluster-operator/src/main/java/io/strimzi/operator/cluster/auth/RequestedServiceAccountAuthIdentity.java @@ -60,4 +60,40 @@ public Map 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; + } } diff --git a/cluster-operator/src/main/java/io/strimzi/operator/cluster/model/KafkaAgentConfigurationBuilder.java b/cluster-operator/src/main/java/io/strimzi/operator/cluster/model/KafkaAgentConfigurationBuilder.java index fa9d1ba11cf..4d65dee6eef 100644 --- a/cluster-operator/src/main/java/io/strimzi/operator/cluster/model/KafkaAgentConfigurationBuilder.java +++ b/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; diff --git a/cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/DefaultKafkaAgentClientProvider.java b/cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/DefaultKafkaAgentClientProvider.java index 4d13c475b3a..12e0963b62b 100644 --- a/cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/DefaultKafkaAgentClientProvider.java +++ b/cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/DefaultKafkaAgentClientProvider.java @@ -4,6 +4,7 @@ */ 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; @@ -11,13 +12,19 @@ * 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); } } diff --git a/cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/KafkaAgentClient.java b/cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/KafkaAgentClient.java index 37d2fb8fd58..90c03d2e0ce 100644 --- a/cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/KafkaAgentClient.java +++ b/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; + } } diff --git a/cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/ResourceOperatorSupplier.java b/cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/ResourceOperatorSupplier.java index 0ec1197db46..395c65ccf6a 100644 --- a/cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/ResourceOperatorSupplier.java +++ b/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), diff --git a/cluster-operator/src/test/java/io/strimzi/operator/cluster/auth/RequestedServiceAccountAuthIdentityTest.java b/cluster-operator/src/test/java/io/strimzi/operator/cluster/auth/RequestedServiceAccountAuthIdentityTest.java index da84fdbaa08..ae03c609674 100644 --- a/cluster-operator/src/test/java/io/strimzi/operator/cluster/auth/RequestedServiceAccountAuthIdentityTest.java +++ b/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)); + } } diff --git a/cluster-operator/src/test/java/io/strimzi/operator/cluster/model/KafkaAgentConfigurationBuilderTest.java b/cluster-operator/src/test/java/io/strimzi/operator/cluster/model/KafkaAgentConfigurationBuilderTest.java index 021451f7ea6..378edffb7ba 100644 --- a/cluster-operator/src/test/java/io/strimzi/operator/cluster/model/KafkaAgentConfigurationBuilderTest.java +++ b/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); diff --git a/cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/ConnectorMockTest.java b/cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/ConnectorMockTest.java index 735d5182385..b765ea23666 100644 --- a/cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/ConnectorMockTest.java +++ b/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); diff --git a/cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/KafkaConnectorIT.java b/cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/KafkaConnectorIT.java index f195df9e9bf..5d93c6c17e3 100644 --- a/cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/KafkaConnectorIT.java +++ b/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 ); diff --git a/cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/KafkaMirrorMaker2AssemblyOperatorMockTest.java b/cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/KafkaMirrorMaker2AssemblyOperatorMockTest.java index a50eba5766e..68da61207f3 100644 --- a/cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/KafkaMirrorMaker2AssemblyOperatorMockTest.java +++ b/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())); diff --git a/cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/resource/KafkaAgentClientTest.java b/cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/resource/KafkaAgentClientTest.java index ee4cce013f1..410e5ae4d3f 100644 --- a/cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/resource/KafkaAgentClientTest.java +++ b/cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/resource/KafkaAgentClientTest.java @@ -4,22 +4,112 @@ */ package io.strimzi.operator.cluster.operator.resource; +import com.sun.net.httpserver.HttpServer; +import io.fabric8.kubernetes.api.model.ServiceAccount; +import io.fabric8.kubernetes.api.model.ServiceAccountList; +import io.fabric8.kubernetes.api.model.authentication.TokenRequest; +import io.fabric8.kubernetes.api.model.authentication.TokenRequestBuilder; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.dsl.MixedOperation; +import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation; +import io.fabric8.kubernetes.client.dsl.ServiceAccountResource; +import io.strimzi.operator.cluster.auth.RequestedServiceAccountAuthIdentity; import io.strimzi.operator.common.Reconciliation; import io.strimzi.operator.common.auth.Identity; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.List; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.nullValue; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; public class KafkaAgentClientTest { private static final Reconciliation RECONCILIATION = new Reconciliation("test", "kafka", "namespace", "my-cluster"); + private static final String AUDIENCE = "strimzi.io/kafka/namespace/my-cluster"; + private static final long EXPIRATION_SECONDS = 3600L; + + private HttpServer httpServer; + private String receivedAuthorization; + + @AfterEach + public void tearDown() { + if (httpServer != null) { + httpServer.stop(0); + } + } + + private static RequestedServiceAccountAuthIdentity authIdentity() { + return new RequestedServiceAccountAuthIdentity(RECONCILIATION, AUDIENCE, EXPIRATION_SECONDS); + } + + private static TokenRequest tokenRequestResponse(String token, Instant expiration) { + return new TokenRequestBuilder() + .withNewStatus() + .withToken(token) + .withExpirationTimestamp(expiration != null ? expiration.toString() : null) + .endStatus() + .build(); + } + + @SuppressWarnings("unchecked") + private static KubernetesClient mockKubernetesClient(ServiceAccountResource serviceAccountResource) { + NonNamespaceOperation namespacedOp = mock(NonNamespaceOperation.class); + when(namespacedOp.withName("my-cluster-cluster-operator")).thenReturn(serviceAccountResource); + + MixedOperation op = mock(MixedOperation.class); + when(op.inNamespace("namespace")).thenReturn(namespacedOp); + + KubernetesClient client = mock(KubernetesClient.class); + when(client.serviceAccounts()).thenReturn(op); + + return client; + } + + /** + * Starts a plain HTTP server which records the Authorization header of the received request and returns the + * broker state JSON. + * + * @return URI of the endpoint served by the server + */ + private URI startHttpServer() throws IOException { + httpServer = HttpServer.create(new InetSocketAddress("localhost", 0), 0); + httpServer.createContext("/v1/broker-state/", exchange -> { + receivedAuthorization = exchange.getRequestHeaders().getFirst("Authorization"); + + byte[] body = "{\"brokerState\":3}".getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(200, body.length); + + try (OutputStream out = exchange.getResponseBody()) { + out.write(body); + } + }); + httpServer.start(); + + return URI.create("http://localhost:" + httpServer.getAddress().getPort() + "/v1/broker-state/"); + } @Test public void testBrokerInRecoveryState() { - KafkaAgentClient kafkaAgentClient = spy(new KafkaAgentClient(RECONCILIATION, "my-cluster", "namespace", new Identity(null, null))); + KafkaAgentClient kafkaAgentClient = spy(new KafkaAgentClient(RECONCILIATION, "my-cluster", "namespace", new Identity(null, null), null)); doAnswer(invocation -> "{\"brokerState\":2,\"recoveryState\":{\"remainingLogsToRecover\":10,\"remainingSegmentsToRecover\":100}}").when(kafkaAgentClient).doGet(any()); BrokerState actual = kafkaAgentClient.getBrokerState("mypod"); assertTrue(actual.isBrokerInRecovery(), "broker is not in log recovery as expected"); @@ -29,7 +119,7 @@ public void testBrokerInRecoveryState() { @Test public void testBrokerInRunningState() { - KafkaAgentClient kafkaAgentClient = spy(new KafkaAgentClient(RECONCILIATION, "my-cluster", "namespace", new Identity(null, null))); + KafkaAgentClient kafkaAgentClient = spy(new KafkaAgentClient(RECONCILIATION, "my-cluster", "namespace", new Identity(null, null), null)); doAnswer(invocation -> "{\"brokerState\":3}").when(kafkaAgentClient).doGet(any()); BrokerState actual = kafkaAgentClient.getBrokerState("mypod"); @@ -40,7 +130,7 @@ public void testBrokerInRunningState() { @Test public void testInvalidJsonResponse() { - KafkaAgentClient kafkaAgentClient = spy(new KafkaAgentClient(RECONCILIATION, "my-cluster", "namespace", new Identity(null, null))); + KafkaAgentClient kafkaAgentClient = spy(new KafkaAgentClient(RECONCILIATION, "my-cluster", "namespace", new Identity(null, null), null)); doAnswer(invocation -> "&\"brokerState\":3&").when(kafkaAgentClient).doGet(any()); BrokerState actual = kafkaAgentClient.getBrokerState("mypod"); @@ -51,7 +141,7 @@ public void testInvalidJsonResponse() { @Test public void testErrorResponse() { - KafkaAgentClient kafkaAgentClient = spy(new KafkaAgentClient(RECONCILIATION, "my-cluster", "namespace", new Identity(null, null))); + KafkaAgentClient kafkaAgentClient = spy(new KafkaAgentClient(RECONCILIATION, "my-cluster", "namespace", new Identity(null, null), null)); doAnswer(invocation -> { throw new RuntimeException("Test failure"); }).when(kafkaAgentClient).doGet(any()); @@ -61,4 +151,103 @@ public void testErrorResponse() { assertEquals(0, actual.remainingLogsToRecover()); assertEquals(0, actual.remainingSegmentsToRecover()); } + + @Test + public void testTokenIsRequestedForTheClusterOperatorServiceAccount() { + ServiceAccountResource serviceAccountResource = mock(ServiceAccountResource.class); + when(serviceAccountResource.tokenRequest(any())).thenReturn(tokenRequestResponse("my-token", Instant.now().plusSeconds(EXPIRATION_SECONDS))); + + KafkaAgentClient kafkaAgentClient = new KafkaAgentClient(RECONCILIATION, "my-cluster", "namespace", new Identity(null, authIdentity()), mockKubernetesClient(serviceAccountResource)); + + assertThat(kafkaAgentClient.currentToken(authIdentity()), is("my-token")); + + ArgumentCaptor requestCaptor = ArgumentCaptor.forClass(TokenRequest.class); + verify(serviceAccountResource).tokenRequest(requestCaptor.capture()); + assertThat(requestCaptor.getValue().getSpec().getAudiences(), is(List.of(AUDIENCE))); + assertThat(requestCaptor.getValue().getSpec().getExpirationSeconds(), is(EXPIRATION_SECONDS)); + } + + @Test + public void testTokenIsCachedAndReused() { + ServiceAccountResource serviceAccountResource = mock(ServiceAccountResource.class); + when(serviceAccountResource.tokenRequest(any())).thenReturn(tokenRequestResponse("my-token", Instant.now().plusSeconds(EXPIRATION_SECONDS))); + + KafkaAgentClient kafkaAgentClient = new KafkaAgentClient(RECONCILIATION, "my-cluster", "namespace", new Identity(null, authIdentity()), mockKubernetesClient(serviceAccountResource)); + + assertThat(kafkaAgentClient.currentToken(authIdentity()), is("my-token")); + assertThat(kafkaAgentClient.currentToken(authIdentity()), is("my-token")); + + verify(serviceAccountResource, times(1)).tokenRequest(any()); + } + + @Test + public void testTokenIsCachedWhenKubernetesShortensTheRequestedExpiration() { + ServiceAccountResource serviceAccountResource = mock(ServiceAccountResource.class); + // The Kubernetes API can issue the token with a shorter validity than requested. The renewal has to follow the + // returned expiration time and not the requested one, otherwise the token would be renewed on every call. + when(serviceAccountResource.tokenRequest(any())).thenReturn(tokenRequestResponse("my-token", Instant.now().plusSeconds(60))); + + KafkaAgentClient kafkaAgentClient = new KafkaAgentClient(RECONCILIATION, "my-cluster", "namespace", new Identity(null, authIdentity()), mockKubernetesClient(serviceAccountResource)); + + assertThat(kafkaAgentClient.currentToken(authIdentity()), is("my-token")); + assertThat(kafkaAgentClient.currentToken(authIdentity()), is("my-token")); + + verify(serviceAccountResource, times(1)).tokenRequest(any()); + } + + @Test + public void testTokenIsRenewedWhenItIsDueForRefresh() { + ServiceAccountResource serviceAccountResource = mock(ServiceAccountResource.class); + when(serviceAccountResource.tokenRequest(any())) + // The first token is already past its refresh time and should not be reused + .thenReturn(tokenRequestResponse("my-old-token", Instant.now().minusSeconds(EXPIRATION_SECONDS))) + .thenReturn(tokenRequestResponse("my-new-token", Instant.now().plusSeconds(EXPIRATION_SECONDS))); + + KafkaAgentClient kafkaAgentClient = new KafkaAgentClient(RECONCILIATION, "my-cluster", "namespace", new Identity(null, authIdentity()), mockKubernetesClient(serviceAccountResource)); + + assertThat(kafkaAgentClient.currentToken(authIdentity()), is("my-old-token")); + assertThat(kafkaAgentClient.currentToken(authIdentity()), is("my-new-token")); + + verify(serviceAccountResource, times(2)).tokenRequest(any()); + } + + @Test + public void testFailsWhenKubernetesApiReturnsNoToken() { + ServiceAccountResource serviceAccountResource = mock(ServiceAccountResource.class); + when(serviceAccountResource.tokenRequest(any())).thenReturn(tokenRequestResponse(null, Instant.now().plusSeconds(EXPIRATION_SECONDS))); + + KafkaAgentClient kafkaAgentClient = new KafkaAgentClient(RECONCILIATION, "my-cluster", "namespace", new Identity(null, authIdentity()), mockKubernetesClient(serviceAccountResource)); + + RuntimeException e = assertThrows(RuntimeException.class, () -> kafkaAgentClient.currentToken(authIdentity())); + assertThat(e.getMessage(), is("Kubernetes API did not return a token for ServiceAccount namespace/my-cluster-cluster-operator")); + } + + @Test + public void testFailsWhenKubernetesApiReturnsNoResponse() { + ServiceAccountResource serviceAccountResource = mock(ServiceAccountResource.class); + when(serviceAccountResource.tokenRequest(any())).thenReturn(null); + + KafkaAgentClient kafkaAgentClient = new KafkaAgentClient(RECONCILIATION, "my-cluster", "namespace", new Identity(null, authIdentity()), mockKubernetesClient(serviceAccountResource)); + + assertThrows(RuntimeException.class, () -> kafkaAgentClient.currentToken(authIdentity())); + } + + @Test + public void testRequestIsSentWithTheServiceAccountToken() throws IOException { + ServiceAccountResource serviceAccountResource = mock(ServiceAccountResource.class); + when(serviceAccountResource.tokenRequest(any())).thenReturn(tokenRequestResponse("my-token", Instant.now().plusSeconds(EXPIRATION_SECONDS))); + + KafkaAgentClient kafkaAgentClient = new KafkaAgentClient(RECONCILIATION, "my-cluster", "namespace", new Identity(null, authIdentity()), mockKubernetesClient(serviceAccountResource)); + + assertThat(kafkaAgentClient.doGet(startHttpServer()), is("{\"brokerState\":3}")); + assertThat(receivedAuthorization, is("Bearer my-token")); + } + + @Test + public void testRequestIsSentWithoutTokenWhenServiceAccountAuthenticationIsNotUsed() throws IOException { + KafkaAgentClient kafkaAgentClient = new KafkaAgentClient(RECONCILIATION, "my-cluster", "namespace", new Identity(null, null), null); + + assertThat(kafkaAgentClient.doGet(startHttpServer()), is("{\"brokerState\":3}")); + assertThat(receivedAuthorization, is(nullValue())); + } } diff --git a/cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/resource/KafkaRollerTest.java b/cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/resource/KafkaRollerTest.java index 1d5a13d7468..7537a18dc3c 100644 --- a/cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/resource/KafkaRollerTest.java +++ b/cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/resource/KafkaRollerTest.java @@ -252,7 +252,7 @@ public void testRollWithDisconnectedBrokerPodNames() { bootstrapBrokers -> bootstrapBrokers != null && bootstrapBrokers.stream().map(NodeRef::nodeId).toList().equals(singletonList(1)) ? new RuntimeException("Test Exception") : null, null, noException(), noException(), brokerId -> CompletableFuture.completedFuture(true), - new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, 30); + new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, 30); // The algorithm should carry on rolling the pods (errors are logged), // because we never find the controller we get ascending order doSuccessfulRollingRestart(kafkaRoller, @@ -268,7 +268,7 @@ public void testRollHandlesErrorWhenOpeningBrokerAdminClient() { bootstrapBrokers -> bootstrapBrokers != null && bootstrapBrokers.stream().map(NodeRef::nodeId).toList().equals(singletonList(1)) ? new RuntimeException("Test Exception") : null, null, noException(), noException(), brokerId -> CompletableFuture.completedFuture(true), - new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, -1); + new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, -1); // The algorithm should carry on rolling the pods (errors are logged), // because we force restart the node if we can't create an admin client doSuccessfulRollingRestart(kafkaRoller, @@ -283,7 +283,7 @@ public void testRollHandlesErrorWhenOpeningCombinedAdminClient() { bootstrapBrokers -> bootstrapBrokers != null && bootstrapBrokers.stream().map(NodeRef::nodeId).toList().equals(singletonList(1)) ? new RuntimeException("Test Exception") : null, null, noException(), noException(), brokerId -> CompletableFuture.completedFuture(true), - new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, -1); + new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, -1); // The algorithm should carry on rolling the pods (errors are logged), // because we force restart the node if we can't create an admin client doSuccessfulRollingRestart(kafkaRoller, @@ -298,7 +298,7 @@ public void testHandlesErrorWhenOpenControllerAdminClient() { bootstrapBrokers -> bootstrapBrokers != null && bootstrapBrokers.stream().map(NodeRef::nodeId).toList().equals(singletonList(1)) ? new RuntimeException("Test Exception") : null, null, noException(), noException(), brokerId -> CompletableFuture.completedFuture(true), - new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, 4); + new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, 4); // The algorithm should carry on rolling the pods (errors are logged), // because we force restart the node if we can't create an admin client doSuccessfulRollingRestart(kafkaRoller, @@ -315,7 +315,7 @@ public void testRollHandlesErrorWhenGettingActiveControllerFromController() { noException(), null, podId -> podId == nonActiveController ? new RuntimeException("Test Exception") : null, noException(), brokerId -> CompletableFuture.completedFuture(true), - new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, activeController); + new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, activeController); doSuccessfulRollingRestart(kafkaRoller, asList(0, 1, 2, 3, 4), asList(0, 3, 4, 1, 2)); @@ -329,7 +329,7 @@ public void testRollHandlesErrorWhenGettingActiveControllerFromActiveController( noException(), null, podId -> podId == activeController ? new RuntimeException("Test Exception") : null, noException(), brokerId -> CompletableFuture.completedFuture(true), - new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, activeController); + new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, activeController); doSuccessfulRollingRestart(kafkaRoller, asList(0, 1, 2, 3, 4), asList(0, 1, 3, 4, 2)); @@ -342,7 +342,7 @@ public void testRollHandlesErrorWhenClosingBrokerAdminClient() { noException(), new RuntimeException("Test Exception"), noException(), noException(), brokerId -> CompletableFuture.completedFuture(true), - new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, -1); + new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, -1); // The algorithm should carry on rolling the pods (errors are logged) doSuccessfulRollingRestart(kafkaRoller, asList(0, 1, 2, 3, 4), @@ -356,7 +356,7 @@ public void testRollHandlesErrorWhenClosingControllerAdminClient() { noException(), new RuntimeException("Test Exception"), noException(), noException(), brokerId -> CompletableFuture.completedFuture(true), - new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, 3); + new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, 3); // The algorithm should carry on rolling the pods (errors are logged), // because we did the active controller last order doSuccessfulRollingRestart(kafkaRoller, @@ -373,7 +373,7 @@ public void testBrokerNotInitiallyRollable() { brokerId -> brokerId == 1 ? CompletableFuture.completedFuture(count.getAndDecrement() == 0) : CompletableFuture.completedFuture(true), - new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, -1); + new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, -1); doSuccessfulRollingRestart(kafkaRoller, asList(0, 1, 2, 3, 4), asList(0, 2, 3, 4, 1)); @@ -388,7 +388,7 @@ public void testCombinedNotInitiallyRollable() { noException(), null, noException(), noException(), combinedId -> combinedId == 1 ? CompletableFuture.completedFuture(count.getAndDecrement() < 0) : CompletableFuture.completedFuture(true), - new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, 2); + new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, 2); doSuccessfulRollingRestart(kafkaRoller, asList(0, 1, 2, 3, 4), asList(0, 3, 4, 1, 2)); @@ -402,7 +402,7 @@ public void testControllerNotInitiallyRollable() { noException(), null, noException(), noException(), controllerId -> (controllerId == 1) ? CompletableFuture.completedFuture(count.getAndDecrement() == 0) : CompletableFuture.completedFuture(true), - new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, 2); + new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, 2); doSuccessfulRollingRestart(kafkaRoller, asList(0, 1, 2, 3, 4), asList(0, 3, 4, 1, 2)); @@ -416,7 +416,7 @@ public void testBrokerNeverRollable() { brokerId -> brokerId == 1 ? CompletableFuture.completedFuture(false) : CompletableFuture.completedFuture(true), - new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, -1); + new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, -1); doFailingRollingRestart(kafkaRoller, asList(0, 1, 2, 3, 4), KafkaRoller.UnforceableProblem.class, "Pod c-kafka-1 cannot be updated right now.", @@ -425,7 +425,7 @@ public void testBrokerNeverRollable() { kafkaRoller = new TestingKafkaRoller(addPodNames(REPLICAS, 0, 0), podOps, noException(), null, noException(), noException(), brokerId -> CompletableFuture.completedFuture(brokerId != 1), - new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, -1); + new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, -1); clearRestarted(); doFailingRollingRestart(kafkaRoller, singletonList(1), @@ -442,7 +442,7 @@ public void testCombinedNeverRollable() { brokerId -> brokerId == 1 ? CompletableFuture.completedFuture(false) : CompletableFuture.completedFuture(true), - new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, -1); + new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, -1); doFailingRollingRestart(kafkaRoller, asList(0, 1, 2, 3, 4), KafkaRoller.UnforceableProblem.class, "Pod c-kafka-1 cannot be updated right now.", @@ -452,7 +452,7 @@ public void testCombinedNeverRollable() { kafkaRoller = new TestingKafkaRoller(addPodNames(REPLICAS, 0, 0), podOps, noException(), null, noException(), noException(), brokerId -> CompletableFuture.completedFuture(brokerId != 1), - new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, -1); + new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, -1); clearRestarted(); doFailingRollingRestart(kafkaRoller, singletonList(1), @@ -470,7 +470,7 @@ public void testControllerNeverRollable() { brokerId -> brokerId == 2 ? CompletableFuture.completedFuture(false) : CompletableFuture.completedFuture(true), - new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, 2); + new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, 2); doFailingRollingRestart(kafkaRoller, asList(0, 1, 2, 3, 4), KafkaRoller.UnforceableProblem.class, "Pod c-kafka-2 cannot be updated right now.", @@ -481,7 +481,7 @@ public void testControllerNeverRollable() { podOps, noException(), null, noException(), noException(), brokerId -> CompletableFuture.completedFuture(brokerId != 2), - new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, 2); + new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, 2); doFailingRollingRestart(kafkaRoller, singletonList(2), KafkaRoller.UnforceableProblem.class, "Pod c-kafka-2 cannot be updated right now.", @@ -494,7 +494,7 @@ public void testRollHandlesErrorWhenGettingBrokerConfig() { TestingKafkaRoller kafkaRoller = new TestingKafkaRoller(addPodNames(REPLICAS, 0, 0), podOps, noException(), null, noException(), podId -> podId == 1 ? new KafkaRoller.ForceableProblem("could not get config exception") : null, - brokerId -> CompletableFuture.completedFuture(true), new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, -1); + brokerId -> CompletableFuture.completedFuture(true), new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, -1); // The algorithm should carry on rolling the pods doSuccessfulRollingRestart(kafkaRoller, asList(0, 1, 2, 3, 4), @@ -507,7 +507,7 @@ public void testRollHandlesErrorWhenGettingCombinedConfig() { TestingKafkaRoller kafkaRoller = new TestingKafkaRoller(addPodNames(0, REPLICAS, 0), podOps, noException(), null, noException(), podId -> podId == 1 ? new KafkaRoller.ForceableProblem("could not get config exception") : null, - brokerId -> CompletableFuture.completedFuture(true), new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, -1); + brokerId -> CompletableFuture.completedFuture(true), new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, -1); // The algorithm should carry on rolling the pods doSuccessfulRollingRestart(kafkaRoller, asList(0, 1, 2, 3, 4), @@ -520,7 +520,7 @@ public void testRollHandlesErrorWhenGettingControllerConfig() { TestingKafkaRoller kafkaRoller = new TestingKafkaRoller(addPodNames(0, 0, REPLICAS), podOps, noException(), null, noException(), podId -> podId == 1 ? new KafkaRoller.ForceableProblem("could not get config exception") : null, - brokerId -> CompletableFuture.completedFuture(true), new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, -1); + brokerId -> CompletableFuture.completedFuture(true), new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, -1); // The algorithm should carry on rolling the pods doSuccessfulRollingRestart(kafkaRoller, asList(0, 1, 2, 3, 4), @@ -534,7 +534,7 @@ public void testRollHandlesErrorWhenGettingConfigFromController() { TestingKafkaRoller kafkaRoller = new TestingKafkaRoller(addPodNames(0, 0, REPLICAS), podOps, noException(), null, noException(), podId -> podId == controller ? new KafkaRoller.ForceableProblem("could not get config exception") : null, - brokerId -> CompletableFuture.completedFuture(true), new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, controller); + brokerId -> CompletableFuture.completedFuture(true), new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, controller); // The algorithm should carry on rolling the pods doSuccessfulRollingRestart(kafkaRoller, asList(0, 1, 2, 3, 4), @@ -548,7 +548,7 @@ public void testSuccessfulAlteringConfigNotRoll() { TestingKafkaRoller kafkaRoller = new TestingKafkaRoller(addPodNames(3, 3, 3), podOps, noException(), null, noException(), noException(), - brokerId -> CompletableFuture.completedFuture(true), new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, -1); + brokerId -> CompletableFuture.completedFuture(true), new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, -1); // The algorithm should not carry on rolling the pods doSuccessfulConfigUpdate(kafkaRoller, emptyList()); @@ -664,7 +664,7 @@ public void testDynamicUpdateThrowsForceableProblemOnFailure() { public void testSuccessfulRollingControllers() { TestingKafkaRoller kafkaRoller = new TestingKafkaRoller(addPodNames(0, 3, 3), mockPodOps(podId -> CompletableFuture.completedFuture(null)), noException(), null, noException(), noException(), - brokerId -> CompletableFuture.completedFuture(true), new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, -1); + brokerId -> CompletableFuture.completedFuture(true), new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, -1); doSuccessfulRollingRestart(kafkaRoller, asList(0, 1, 2, 3, 4, 5), asList(0, 1, 2, 3, 4, 5)); @@ -674,7 +674,7 @@ public void testSuccessfulRollingControllers() { public void testControllerNoQuorum() { TestingKafkaRoller kafkaRoller = new TestingKafkaRoller(addPodNames(0, 0, 3), mockPodOps(podId -> CompletableFuture.completedFuture(null)), noException(), null, noException(), noException(), - brokerId -> CompletableFuture.completedFuture(false), new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, -1); + brokerId -> CompletableFuture.completedFuture(false), new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, -1); doFailingRollingRestart(kafkaRoller, asList(0, 1, 2), KafkaRoller.UnforceableProblem.class, "Pod c-kafka-0 cannot be updated right now.", @@ -688,7 +688,7 @@ public void testControllerAndOneMoreNeverRollable() { podOps, noException(), null, noException(), noException(), brokerId -> brokerId == 2 || brokerId == 3 ? CompletableFuture.completedFuture(false) : CompletableFuture.completedFuture(true), - new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, 2); + new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, 2); doFailingRollingRestart(kafkaRoller, asList(0, 1, 2, 3, 4), KafkaRoller.ForceableProblem.class, "Pod c-kafka-2 is the active controller and there are other pods to verify first", @@ -828,7 +828,7 @@ public void testFailWhenPodIsReadyThrows() { public void testRollWithAllRoles() { TestingKafkaRoller kafkaRoller = new TestingKafkaRoller(addPodNames(3, 3, 3), mockPodOps(podId -> CompletableFuture.completedFuture(null)), noException(), null, noException(), noException(), - brokerId -> CompletableFuture.completedFuture(true), new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, 6); + brokerId -> CompletableFuture.completedFuture(true), new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, 6); doSuccessfulRollingRestart(kafkaRoller, asList(0, 1, 2, 3, 4, 5, 6, 7, 8), asList(3, 4, 5, 7, 8, 6, 0, 1, 2)); @@ -856,7 +856,7 @@ public void testRollUnreadyPodFirstAllRoles() { }); TestingKafkaRoller kafkaRoller = new TestingKafkaRoller(addPodNames(3, 3, 3), podOps, noException(), null, noException(), noException(), - brokerId -> CompletableFuture.completedFuture(true), new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, -1); + brokerId -> CompletableFuture.completedFuture(true), new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, -1); doSuccessfulRollingRestart(kafkaRoller, asList(0, 1, 2, 3, 4, 5, 6, 7, 8), // brokers, combined, controllers asList(7, 4, 3, 5, 6, 8, 1, 0, 2)); //Rolls in order: unready controllers, ready controllers, unready brokers, ready brokers @@ -889,7 +889,7 @@ private TestingKafkaRoller rollerWithActiveController(PodOperator podOps, Set CompletableFuture.completedFuture(true), - new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(), false, null, activeController); + new DefaultAdminClientProvider(), new DefaultKafkaAgentClientProvider(null), false, null, activeController); } private void doSuccessfulConfigUpdate(TestingKafkaRoller kafkaRoller, diff --git a/kafka-agent/pom.xml b/kafka-agent/pom.xml index 1700005a238..58ae00ce790 100644 --- a/kafka-agent/pom.xml +++ b/kafka-agent/pom.xml @@ -63,6 +63,13 @@ jackson-databind provided + + + io.strimzi + kafka-oauth-common + provided + org.junit.jupiter junit-jupiter-api diff --git a/kafka-agent/src/main/java/io/strimzi/kafka/agent/KafkaAgent.java b/kafka-agent/src/main/java/io/strimzi/kafka/agent/KafkaAgent.java index 1014bf23210..bf4a8097937 100644 --- a/kafka-agent/src/main/java/io/strimzi/kafka/agent/KafkaAgent.java +++ b/kafka-agent/src/main/java/io/strimzi/kafka/agent/KafkaAgent.java @@ -59,6 +59,12 @@ *
Returns HTTP code 204 if broker state is RUNNING(3). Otherwise returns non successful HTTP code. *
* + *

+ * The endpoints are exposed on two connectors. The external connector is used by the operator and can optionally use + * TLS encryption and require the clients to authenticate with a TLS client certificate or with a Kubernetes Service + * Account token. The internal connector is bound to localhost and is used by the health checks of this Pod without any + * encryption or authentication. + *

*/ public class KafkaAgent { private static final Logger LOGGER = LoggerFactory.getLogger(KafkaAgent.class); @@ -66,6 +72,8 @@ public class KafkaAgent { private static final String READINESS_ENDPOINT_PATH = "/v1/ready"; private static final int EXTERNAL_HTTP_PORT = 8443; private static final int INTERNAL_HTTP_PORT = 8080; + private static final String EXTERNAL_CONNECTOR_NAME = "external"; + private static final String INTERNAL_CONNECTOR_NAME = "internal"; private static final long GRACEFUL_SHUTDOWN_TIMEOUT_MS = 30 * 1000; private static final byte BROKER_RUNNING_STATE = 3; private static final byte BROKER_RECOVERY_STATE = 2; @@ -171,7 +179,8 @@ private boolean isRemainingSegmentsToRecover(MetricName name) { ServerConnector externalConnector = createExternalHttpConnector(server); // Internal connector is used within the Pod only for health checks - ServerConnector internalConnector = new ServerConnector(server); + ServerConnector internalConnector = new ServerConnector(server); + internalConnector.setName(INTERNAL_CONNECTOR_NAME); internalConnector.setHost("localhost"); // Should not be exposed outside the Pod. So we use localhost only here. internalConnector.setPort(INTERNAL_HTTP_PORT); @@ -181,8 +190,16 @@ private boolean isRemainingSegmentsToRecover(MetricName name) { ContextHandler readinessContext = new ContextHandler(READINESS_ENDPOINT_PATH); readinessContext.setHandler(getReadinessHandler()); + Handler handler = new ContextHandlerCollection(brokerStateContext, readinessContext); + + if (config.get("tokenIssuer") != null) { + // Service Account authentication is used => the requests arriving through the external connector have to be + // authenticated with a valid Kubernetes Service Account token + handler = new ServiceAccountAuthenticationHandler(handler, EXTERNAL_CONNECTOR_NAME, config); + } + server.setConnectors(new Connector[] {externalConnector, internalConnector}); - server.setHandler(new ContextHandlerCollection(brokerStateContext, readinessContext)); + server.setHandler(handler); server.setStopTimeout(GRACEFUL_SHUTDOWN_TIMEOUT_MS); server.setStopAtShutdown(true); @@ -203,6 +220,7 @@ private ServerConnector createExternalHttpConnector(Server server) throws Genera httpConnector = new ServerConnector(server); } + httpConnector.setName(EXTERNAL_CONNECTOR_NAME); httpConnector.setHost("0.0.0.0"); httpConnector.setPort(EXTERNAL_HTTP_PORT); diff --git a/kafka-agent/src/main/java/io/strimzi/kafka/agent/ServiceAccountAuthenticationHandler.java b/kafka-agent/src/main/java/io/strimzi/kafka/agent/ServiceAccountAuthenticationHandler.java new file mode 100644 index 00000000000..aa20e8ba506 --- /dev/null +++ b/kafka-agent/src/main/java/io/strimzi/kafka/agent/ServiceAccountAuthenticationHandler.java @@ -0,0 +1,195 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.kafka.agent; + +import io.strimzi.kafka.oauth.common.FileBasedTokenProvider; +import io.strimzi.kafka.oauth.common.PrincipalExtractor; +import io.strimzi.kafka.oauth.common.SSLUtil; +import io.strimzi.kafka.oauth.common.TokenInfo; +import io.strimzi.kafka.oauth.validator.JWTSignatureValidator; +import io.strimzi.kafka.oauth.validator.TokenValidator; +import jakarta.servlet.http.HttpServletResponse; +import org.eclipse.jetty.http.HttpHeader; +import org.eclipse.jetty.server.Handler; +import org.eclipse.jetty.server.Request; +import org.eclipse.jetty.server.Response; +import org.eclipse.jetty.server.handler.ConditionalHandler; +import org.eclipse.jetty.util.Callback; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +/** + * Handler that authenticates the requests using Kubernetes Service Account tokens. The tokens are expected in the + * Authorization header as a Bearer token. They are validated against the public keys downloaded from the JWKS endpoint + * of the Kubernetes API server. This uses the same code from the Strimzi OAuth library and the same validation rules as + * the Kafka brokers use for the Kafka protocol connections. + * + * Only the requests arriving through the external connector are authenticated. The requests arriving through the + * internal connector are passed through because the internal connector is bound to localhost and is used + * only by the health checks of this Pod. (Jetty does not allow us to configure the handler only on one of the + * connectors.) + */ +class ServiceAccountAuthenticationHandler extends ConditionalHandler.ElseNext { + private static final Logger LOGGER = LoggerFactory.getLogger(ServiceAccountAuthenticationHandler.class); + + private static final String BEARER_PREFIX = "Bearer "; + // The regular Service Account token that is used to authenticate against the JWKS endpoint of the Kubernetes API server. + private static final String KUBERNETES_TOKEN_PATH = "/var/run/secrets/kubernetes.io/serviceaccount/token"; + // The various timing configurations related to the JWKS keys used to verify the tokens + private static final int JWKS_REFRESH_SECONDS = 300; + private static final int JWKS_REFRESH_MIN_PAUSE_SECONDS = 1; + private static final int JWKS_EXPIRY_SECONDS = 360; + private static final int CONNECT_TIMEOUT_SECONDS = 10; + private static final int READ_TIMEOUT_SECONDS = 10; + + private final TokenValidator validator; + private final Set allowedUsers; + + /** + * Constructor of the ServiceAccountAuthenticationHandler + * + * @param handler Handler to which the authenticated requests are passed + * @param connectorName Name of the connector on which the requests have to be authenticated + * @param config Map with Kafka Agent configurations + */ + ServiceAccountAuthenticationHandler(Handler handler, String connectorName, Map config) { + this(handler, + connectorName, + createValidator(config.get("tokenJwksUri"), config.get("tokenJwksCaPath"), config.get("tokenIssuer"), config.get("tokenAudience"), config.getOrDefault("tokenPath", KUBERNETES_TOKEN_PATH)), + parseAllowedUsers(config.get("tokenAllowedUsers"))); + } + + /** + * Constructor of the ServiceAccountAuthenticationHandler + * + * @param handler Handler to which the authenticated requests are passed + * @param connectorName Name of the connector on which the requests have to be authenticated + * @param validator Validator used to validate the tokens + * @param allowedUsers Users which are allowed to access the endpoints + */ + /* test */ ServiceAccountAuthenticationHandler(Handler handler, String connectorName, TokenValidator validator, Set allowedUsers) { + super(handler); + + this.validator = validator; + this.allowedUsers = allowedUsers; + + // Only the requests arriving through the configured connector are authenticated. The other requests are passed + // to the next handler by the onConditionsNotMet method of the ElseNext superclass. This separates the internal + // and external connectors and makes sure the internal connector is skipped. + include(new ConnectorPredicate(connectorName)); + } + + @Override + protected boolean onConditionsMet(Request request, Response response, Callback callback) throws Exception { + String authorization = request.getHeaders().get(HttpHeader.AUTHORIZATION); + + if (authorization == null || !authorization.startsWith(BEARER_PREFIX)) { + LOGGER.warn("Request to {} is missing the Bearer token", request.getHttpURI()); + return unauthorized(response, callback, "Missing Bearer token"); + } + + String principal; + + try { + TokenInfo token = validator.validate(authorization.substring(BEARER_PREFIX.length()).trim()); + principal = token.principal(); + } catch (RuntimeException e) { + LOGGER.warn("Failed to validate the token of the request to {}", request.getHttpURI(), e); + return unauthorized(response, callback, "Invalid Bearer token"); + } + + if (!allowedUsers.contains(principal)) { + LOGGER.warn("User {} is not allowed to access {}", principal, request.getHttpURI()); + response.getHeaders().put(HttpHeader.CONTENT_TYPE, "text/plain; charset=UTF-8"); + response.setStatus(HttpServletResponse.SC_FORBIDDEN); + response.write(true, StandardCharsets.UTF_8.encode("User is not allowed to access this endpoint"), callback); + return true; + } + + LOGGER.trace("Request to {} was authenticated as user {}", request.getHttpURI(), principal); + + return nextHandler(request, response, callback); + } + + @Override + protected void doStop() throws Exception { + super.doStop(); + validator.close(); + } + + private static boolean unauthorized(Response response, Callback callback, String message) { + response.getHeaders().put(HttpHeader.CONTENT_TYPE, "text/plain; charset=UTF-8"); + response.getHeaders().put(HttpHeader.WWW_AUTHENTICATE, "Bearer"); + response.setStatus(HttpServletResponse.SC_UNAUTHORIZED); + response.write(true, StandardCharsets.UTF_8.encode(message), callback); + return true; + } + + /** + * Creates the validator which validates the signature and the claims of the Service Account tokens. The keys used + * to validate the signatures are downloaded from the JWKS endpoint of the Kubernetes API server and periodically + * refreshed in the background. + * + * @param jwksUri URI of the JWKS endpoint of the Kubernetes API server + * @param jwksCaPath Path to the PEM file with the certificates trusted when connecting to the JWKS + * endpoint. When null, the default trust of the JVM is used. + * @param issuer Expected issuer of the tokens + * @param audience Expected audience of the tokens + * @param kubernetesTokenPath Path to the Service Account token of this Pod which is used to authenticate against + * the JWKS endpoint + * + * @return Token validator + */ + /* test */ static TokenValidator createValidator(String jwksUri, String jwksCaPath, String issuer, String audience, String kubernetesTokenPath) { + return new JWTSignatureValidator( + "kafka-agent", // ID of this validator + null, // Client ID is not used, we authenticate with the token of this Pod + null, // Client secret is not used, we authenticate with the token of this Pod + new FileBasedTokenProvider(kubernetesTokenPath), // The JWKS endpoint of the Kubernetes API server requires authentication + jwksUri, // The JWKS endpoint of the Kubernetes API server + SSLUtil.createSSLFactory(jwksCaPath, null, null, "PEM", null), // TLS trust used when connecting to the JWKS endpoint + null, // Default hostname verifier is used + new PrincipalExtractor("sub"), // The username is taken from the sub claim + null, // Groups are not extracted from the token + null, // Groups are not extracted from the token + issuer, // Expected issuer of the token + JWKS_REFRESH_SECONDS, + JWKS_REFRESH_MIN_PAUSE_SECONDS, + JWKS_EXPIRY_SECONDS, + false, // Only the keys marked for signing are used + false, // Service Account tokens do not have the token type claim + audience, // Expected audience of the token + null, // No custom claim check is needed + CONNECT_TIMEOUT_SECONDS, + READ_TIMEOUT_SECONDS, + false, // Metrics are not collected by the Kafka Agent + false, // The Kafka Agent should not prevent the broker from starting when the keys cannot be downloaded + false); // The Kubernetes API server does not handle the Accept header well + } + + /** + * Parses the comma-separated list of the users which are allowed to access the Kafka Agent endpoints. + * + * @param allowedUsers Comma-separated list of the allowed users + * + * @return Set with the allowed users + */ + private static Set parseAllowedUsers(String allowedUsers) { + if (allowedUsers == null) { + return Set.of(); + } + + return Arrays.stream(allowedUsers.split(",")) + .map(String::trim) + .filter(user -> !user.isEmpty()) + .collect(Collectors.toSet()); + } +} diff --git a/kafka-agent/src/test/java/io/strimzi/kafka/agent/KafkaAgentTest.java b/kafka-agent/src/test/java/io/strimzi/kafka/agent/KafkaAgentTest.java index 747edd1d32a..3bec53e8b90 100644 --- a/kafka-agent/src/test/java/io/strimzi/kafka/agent/KafkaAgentTest.java +++ b/kafka-agent/src/test/java/io/strimzi/kafka/agent/KafkaAgentTest.java @@ -17,20 +17,25 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; import javax.net.ssl.KeyManagerFactory; import javax.net.ssl.SSLContext; import javax.net.ssl.TrustManagerFactory; import java.io.IOException; +import java.net.ServerSocket; import java.net.URI; import java.net.URISyntaxException; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; +import java.nio.file.Files; +import java.nio.file.Path; import java.security.GeneralSecurityException; import java.security.SecureRandom; import java.time.Duration; +import java.util.HashMap; import java.util.Map; import static org.hamcrest.CoreMatchers.is; @@ -50,6 +55,9 @@ public class KafkaAgentTest { "namespace", NAMESPACE, "sslKeyStoreSecretName", NODE_CERT_SECRET_NAME); + @TempDir + Path tempDir; + private Server server; private HttpClient httpsClient; private HttpRequest httpsReq; @@ -240,6 +248,86 @@ public void testReadinessFail() throws Exception { assertThat(response.statusCode(), is(HttpServletResponse.SC_SERVICE_UNAVAILABLE)); } + /** + * Creates the Kafka Agent configuration with TLS encryption and with Service Account authentication enabled. The + * JWKS endpoint points to an unused port, because these tests check only the requests which are rejected before + * the token is validated. The Kafka Agent is expected to start anyway and keep retrying the download of the keys + * in the background. + * + * @return Map with the Kafka Agent configuration + */ + private Map serviceAccountAuthenticationConfig() throws IOException { + int unusedPort; + + try (ServerSocket socket = new ServerSocket(0)) { + unusedPort = socket.getLocalPort(); + } + + Path tokenPath = tempDir.resolve("token"); + Files.writeString(tokenPath, "my-kubernetes-token"); + + Map config = new HashMap<>(TLS_CONFIG); + config.put("tokenIssuer", "https://kubernetes.default.svc.cluster.local"); + config.put("tokenJwksUri", "http://localhost:" + unusedPort + "/openid/v1/jwks"); + config.put("tokenAudience", "strimzi.io/kafka/my-namespace/my-cluster"); + config.put("tokenAllowedUsers", "system:serviceaccount:my-namespace:my-cluster-cluster-operator"); + config.put("tokenPath", tokenPath.toString()); + + return config; + } + + @Test + public void testExternalConnectorRequiresTokenWithServiceAccountAuthentication() throws Exception { + @SuppressWarnings({ "rawtypes" }) + final Gauge brokerState = mock(Gauge.class); + when(brokerState.value()).thenReturn((byte) 3); + + KafkaAgent agent = new KafkaAgent(client, serviceAccountAuthenticationConfig(), brokerState, null, null); + server = agent.startHttpServer(); + + HttpClient tlsClient = HttpClient.newBuilder() + .connectTimeout(Duration.ofSeconds(10)) + .sslContext(getClientSSLContext(caCertSecret, null)) + .build(); + HttpResponse response = tlsClient.send(httpsReq, HttpResponse.BodyHandlers.ofString()); + + assertThat(response.statusCode(), is(HttpServletResponse.SC_UNAUTHORIZED)); + } + + @Test + public void testInternalConnectorDoesNotRequireTokenWithServiceAccountAuthentication() throws Exception { + @SuppressWarnings({ "rawtypes" }) + final Gauge brokerState = mock(Gauge.class); + when(brokerState.value()).thenReturn((byte) 3); + + KafkaAgent agent = new KafkaAgent(client, serviceAccountAuthenticationConfig(), brokerState, null, null); + server = agent.startHttpServer(); + + HttpResponse response = HttpClient.newBuilder() + .build() + .send(httpReq, HttpResponse.BodyHandlers.ofString()); + + assertThat(response.statusCode(), is(HttpServletResponse.SC_NO_CONTENT)); + } + + @Test + public void testExternalConnectorDoesNotRequireTokenWithoutServiceAccountAuthentication() throws Exception { + @SuppressWarnings({ "rawtypes" }) + final Gauge brokerState = mock(Gauge.class); + when(brokerState.value()).thenReturn((byte) 3); + + KafkaAgent agent = new KafkaAgent(client, TLS_CONFIG, brokerState, null, null); + server = agent.startHttpServer(); + + HttpClient tlsClient = HttpClient.newBuilder() + .connectTimeout(Duration.ofSeconds(10)) + .sslContext(getClientSSLContext(caCertSecret, null)) + .build(); + HttpResponse response = tlsClient.send(httpsReq, HttpResponse.BodyHandlers.ofString()); + + assertThat(response.statusCode(), is(HttpServletResponse.SC_OK)); + } + @Test public void testReadinessFailWithBrokerUnknownState() throws Exception { @SuppressWarnings({ "rawtypes" }) diff --git a/kafka-agent/src/test/java/io/strimzi/kafka/agent/ServiceAccountAuthenticationHandlerTest.java b/kafka-agent/src/test/java/io/strimzi/kafka/agent/ServiceAccountAuthenticationHandlerTest.java new file mode 100644 index 00000000000..97fd74acfdc --- /dev/null +++ b/kafka-agent/src/test/java/io/strimzi/kafka/agent/ServiceAccountAuthenticationHandlerTest.java @@ -0,0 +1,363 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.kafka.agent; + +import com.sun.net.httpserver.HttpServer; +import io.strimzi.kafka.oauth.common.TokenInfo; +import io.strimzi.kafka.oauth.validator.TokenValidationException; +import io.strimzi.kafka.oauth.validator.TokenValidator; +import jakarta.servlet.http.HttpServletResponse; +import org.eclipse.jetty.http.HttpHeader; +import org.eclipse.jetty.server.Connector; +import org.eclipse.jetty.server.Handler; +import org.eclipse.jetty.server.Request; +import org.eclipse.jetty.server.Response; +import org.eclipse.jetty.server.Server; +import org.eclipse.jetty.server.ServerConnector; +import org.eclipse.jetty.util.Callback; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.math.BigInteger; +import java.net.InetSocketAddress; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.security.KeyPair; +import java.security.KeyPairGenerator; +import java.security.PrivateKey; +import java.security.Signature; +import java.security.interfaces.RSAPublicKey; +import java.util.Base64; +import java.util.Set; + +import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class ServiceAccountAuthenticationHandlerTest { + private static final String EXTERNAL_CONNECTOR_NAME = "external"; + private static final String INTERNAL_CONNECTOR_NAME = "internal"; + private static final String ALLOWED_USER = "system:serviceaccount:my-namespace:my-cluster-cluster-operator"; + private static final Set ALLOWED_USERS = Set.of(ALLOWED_USER); + private static final String ISSUER = "https://kubernetes.default.svc.cluster.local"; + private static final String AUDIENCE = "strimzi.io/kafka/my-namespace/my-cluster"; + private static final String KEY_ID = "my-signing-key"; + private static final String KUBERNETES_TOKEN = "my-kubernetes-token"; + + @TempDir + static Path tempDir; + + private static KeyPair signingKey; + private static HttpServer jwksServer; + private static String jwksAuthorization; + private static TokenValidator validator; + + private Server server; + private int externalPort; + private int internalPort; + + /** + * Starts a JWKS endpoint with a single signing key and creates the validator which uses it. This is the same + * validator as the one used by the Kafka Agent at runtime. + */ + @BeforeAll + public static void setUpValidator() throws Exception { + KeyPairGenerator generator = KeyPairGenerator.getInstance("RSA"); + generator.initialize(2048); + signingKey = generator.generateKeyPair(); + + jwksServer = HttpServer.create(new InetSocketAddress("localhost", 0), 0); + jwksServer.createContext("/openid/v1/jwks", exchange -> { + jwksAuthorization = exchange.getRequestHeaders().getFirst("Authorization"); + + byte[] body = jwks((RSAPublicKey) signingKey.getPublic()).getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "application/json"); + exchange.sendResponseHeaders(HttpServletResponse.SC_OK, body.length); + exchange.getResponseBody().write(body); + exchange.close(); + }); + jwksServer.start(); + + Path tokenPath = tempDir.resolve("token"); + Files.writeString(tokenPath, KUBERNETES_TOKEN); + + validator = ServiceAccountAuthenticationHandler.createValidator( + "http://localhost:" + jwksServer.getAddress().getPort() + "/openid/v1/jwks", + null, + ISSUER, + AUDIENCE, + tokenPath.toString()); + } + + @AfterAll + public static void tearDownValidator() { + if (validator != null) { + validator.close(); + } + + if (jwksServer != null) { + jwksServer.stop(0); + } + } + + @AfterEach + public void tearDown() throws Exception { + if (server != null) { + server.stop(); + } + } + + /** + * Starts a server with two connectors and with the authentication handler in front of a handler which always + * returns HTTP 200. Both connectors are bound to a random port on localhost. + * + * @param validator Validator which should be used to validate the tokens + */ + private void startServer(TokenValidator validator) throws Exception { + server = new Server(); + + ServerConnector externalConnector = new ServerConnector(server); + externalConnector.setName(EXTERNAL_CONNECTOR_NAME); + externalConnector.setHost("localhost"); + + ServerConnector internalConnector = new ServerConnector(server); + internalConnector.setName(INTERNAL_CONNECTOR_NAME); + internalConnector.setHost("localhost"); + + server.setConnectors(new Connector[] {externalConnector, internalConnector}); + server.setHandler(new ServiceAccountAuthenticationHandler(new OkHandler(), EXTERNAL_CONNECTOR_NAME, validator, ALLOWED_USERS)); + server.start(); + + externalPort = externalConnector.getLocalPort(); + internalPort = internalConnector.getLocalPort(); + } + + private HttpResponse get(int port, String authorization) throws IOException, InterruptedException { + HttpRequest.Builder builder = HttpRequest.newBuilder() + .uri(URI.create("http://localhost:" + port + "/v1/broker-state")) + .GET(); + + if (authorization != null) { + builder.header(HttpHeader.AUTHORIZATION.asString(), authorization); + } + + return HttpClient.newHttpClient().send(builder.build(), HttpResponse.BodyHandlers.ofString()); + } + + private TokenValidator mockValidator(String principal) { + TokenValidator validator = mock(TokenValidator.class); + when(validator.validate("my-token")).thenReturn(new TokenInfo("my-token", (String) null, principal, null, 0L, Long.MAX_VALUE)); + return validator; + } + + @Test + public void testAllowedUserIsAuthenticated() throws Exception { + TokenValidator validator = mockValidator(ALLOWED_USER); + startServer(validator); + + HttpResponse response = get(externalPort, "Bearer my-token"); + + assertThat(response.statusCode(), is(HttpServletResponse.SC_OK)); + assertThat(response.body(), is("OK")); + verify(validator).validate(eq("my-token")); + } + + @Test + public void testUserWhichIsNotAllowedIsRejected() throws Exception { + startServer(mockValidator("system:serviceaccount:my-namespace:my-user")); + + HttpResponse response = get(externalPort, "Bearer my-token"); + + assertThat(response.statusCode(), is(HttpServletResponse.SC_FORBIDDEN)); + assertThat(response.body(), is("User is not allowed to access this endpoint")); + } + + @Test + public void testMissingAuthorizationHeaderIsRejected() throws Exception { + startServer(mockValidator(ALLOWED_USER)); + + HttpResponse response = get(externalPort, null); + + assertThat(response.statusCode(), is(HttpServletResponse.SC_UNAUTHORIZED)); + assertThat(response.body(), is("Missing Bearer token")); + assertThat(response.headers().firstValue(HttpHeader.WWW_AUTHENTICATE.asString()).orElse(null), is("Bearer")); + } + + @Test + public void testAuthorizationHeaderWithoutBearerTokenIsRejected() throws Exception { + startServer(mockValidator(ALLOWED_USER)); + + HttpResponse response = get(externalPort, "Basic dXNlcjpwYXNzd29yZA=="); + + assertThat(response.statusCode(), is(HttpServletResponse.SC_UNAUTHORIZED)); + assertThat(response.body(), is("Missing Bearer token")); + } + + @Test + public void testInvalidTokenIsRejected() throws Exception { + TokenValidator validator = mock(TokenValidator.class); + when(validator.validate("my-token")).thenThrow(new TokenValidationException("Token validation failed")); + startServer(validator); + + HttpResponse response = get(externalPort, "Bearer my-token"); + + assertThat(response.statusCode(), is(HttpServletResponse.SC_UNAUTHORIZED)); + assertThat(response.body(), is("Invalid Bearer token")); + assertThat(response.headers().firstValue(HttpHeader.WWW_AUTHENTICATE.asString()).orElse(null), is("Bearer")); + } + + @Test + public void testRequestsOnInternalConnectorAreNotAuthenticated() throws Exception { + TokenValidator validator = mockValidator(ALLOWED_USER); + startServer(validator); + + HttpResponse response = get(internalPort, null); + + assertThat(response.statusCode(), is(HttpServletResponse.SC_OK)); + assertThat(response.body(), is("OK")); + verify(validator, never()).validate(anyString()); + } + + @Test + public void testValidTokenIsAccepted() { + TokenInfo token = validator.validate(token(signingKey.getPrivate(), KEY_ID, ISSUER, ALLOWED_USER, AUDIENCE, 3600)); + + assertThat(token.principal(), is(ALLOWED_USER)); + } + + @Test + public void testJwksEndpointIsCalledWithTheKubernetesToken() { + assertThat(jwksAuthorization, is("Bearer " + KUBERNETES_TOKEN)); + } + + @Test + public void testTokenWithWrongIssuerIsRejected() { + String token = token(signingKey.getPrivate(), KEY_ID, "https://my-other-issuer.io", ALLOWED_USER, AUDIENCE, 3600); + + assertThrows(TokenValidationException.class, () -> validator.validate(token)); + } + + @Test + public void testTokenWithWrongAudienceIsRejected() { + String token = token(signingKey.getPrivate(), KEY_ID, ISSUER, ALLOWED_USER, "strimzi.io/kafka/my-namespace/my-other-cluster", 3600); + + assertThrows(TokenValidationException.class, () -> validator.validate(token)); + } + + @Test + public void testExpiredTokenIsRejected() { + String token = token(signingKey.getPrivate(), KEY_ID, ISSUER, ALLOWED_USER, AUDIENCE, -3600); + + assertThrows(TokenValidationException.class, () -> validator.validate(token)); + } + + @Test + public void testTokenSignedWithAnotherKeyIsRejected() throws Exception { + KeyPairGenerator generator = KeyPairGenerator.getInstance("RSA"); + generator.initialize(2048); + String token = token(generator.generateKeyPair().getPrivate(), KEY_ID, ISSUER, ALLOWED_USER, AUDIENCE, 3600); + + assertThrows(TokenValidationException.class, () -> validator.validate(token)); + } + + @Test + public void testTokenSignedWithUnknownKeyIdIsRejected() { + String token = token(signingKey.getPrivate(), "my-unknown-key", ISSUER, ALLOWED_USER, AUDIENCE, 3600); + + assertThrows(TokenValidationException.class, () -> validator.validate(token)); + } + + /** + * Creates the JWKS response with the public part of the signing key + * + * @param key Public key which should be advertised in the JWKS response + * + * @return The JWKS response as a String + */ + private static String jwks(RSAPublicKey key) { + return "{\"keys\":[{\"use\":\"sig\",\"kty\":\"RSA\",\"alg\":\"RS256\",\"kid\":\"" + KEY_ID + "\"," + + "\"n\":\"" + base64Url(unsigned(key.getModulus())) + "\"," + + "\"e\":\"" + base64Url(unsigned(key.getPublicExponent())) + "\"}]}"; + } + + /** + * Creates a signed JWT token which looks like a Kubernetes Service Account token + * + * @param key Private key used to sign the token + * @param keyId ID of the key which is set in the token header + * @param issuer Issuer of the token + * @param subject Subject of the token + * @param audience Audience of the token + * @param expirySeconds For how many seconds from now is the token valid. Negative value creates an expired token. + * + * @return The signed token + */ + private static String token(PrivateKey key, String keyId, String issuer, String subject, String audience, long expirySeconds) { + String header = "{\"alg\":\"RS256\",\"kid\":\"" + keyId + "\"}"; + String payload = "{\"iss\":\"" + issuer + "\",\"sub\":\"" + subject + "\",\"aud\":[\"" + audience + "\"]," + + "\"exp\":" + ((System.currentTimeMillis() / 1000) + expirySeconds) + "}"; + String signingInput = base64Url(header.getBytes(StandardCharsets.UTF_8)) + "." + base64Url(payload.getBytes(StandardCharsets.UTF_8)); + + try { + Signature signature = Signature.getInstance("SHA256withRSA"); + signature.initSign(key); + signature.update(signingInput.getBytes(StandardCharsets.US_ASCII)); + return signingInput + "." + base64Url(signature.sign()); + } catch (Exception e) { + throw new RuntimeException("Failed to sign the token", e); + } + } + + private static String base64Url(byte[] bytes) { + return Base64.getUrlEncoder().withoutPadding().encodeToString(bytes); + } + + /** + * Converts the number into its unsigned big-endian representation as required by the JWKS format. The two's + * complement representation used by BigInteger might have an extra leading zero byte. + * + * @param value Number which should be converted + * + * @return The unsigned big-endian bytes of the number + */ + private static byte[] unsigned(BigInteger value) { + byte[] bytes = value.toByteArray(); + + if (bytes.length > 1 && bytes[0] == 0) { + byte[] stripped = new byte[bytes.length - 1]; + System.arraycopy(bytes, 1, stripped, 0, stripped.length); + return stripped; + } + + return bytes; + } + + /** + * Handler which always returns HTTP 200 and is used to check whether the request passed the authentication + */ + private static class OkHandler extends Handler.Abstract { + @Override + public boolean handle(Request request, Response response, Callback callback) { + response.setStatus(HttpServletResponse.SC_OK); + response.write(true, StandardCharsets.UTF_8.encode("OK"), callback); + return true; + } + } +} diff --git a/pom.xml b/pom.xml index 14e0a087e43..45fbeb69f47 100644 --- a/pom.xml +++ b/pom.xml @@ -590,6 +590,11 @@ kafka-oauth-client ${strimzi-oauth.version} + + io.strimzi + kafka-oauth-common + ${strimzi-oauth.version} + io.skodjob.kubetest4j