Skip to content

Commit

Permalink
Improve container setup for kafka-oauth test
Browse files Browse the repository at this point in the history
  • Loading branch information
@jamesnetherton
jamesnetherton committed Nov 16, 2023
1 parent ef1028e commit aec25ff
Show file tree
Hide file tree
Showing 6 changed files with 84 additions and 95 deletions.
11 changes: 7 additions & 4 deletions integration-tests/kafka-oauth/pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,11 @@
<artifactId>strimzi-test-container</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.quarkus</groupId>
<artifactId>quarkus-test-keycloak-server</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
Expand Down Expand Up @@ -126,8 +131,7 @@
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<systemPropertyVariables>
<!-- Configure the app to test to resolve "keycloak" hostname to the Docker hostname -->
<jdk.net.hosts.file>target/hosts</jdk.net.hosts.file>
<keycloak.docker.image>${keycloak.container.image}</keycloak.docker.image>
</systemPropertyVariables>
</configuration>
</plugin>
Expand Down Expand Up @@ -156,9 +160,8 @@
<goal>verify</goal>
</goals>
<configuration>
<rerunFailingTestsCount>${rerun.failing.test.count}</rerunFailingTestsCount>
<systemPropertyVariables>
<quarkus.test.arg-line>-Djdk.net.hosts.file=target/hosts</quarkus.test.arg-line>
<keycloak.docker.image>${keycloak.container.image}</keycloak.docker.image>
</systemPropertyVariables>
</configuration>
</execution>
Expand Down
11 changes: 1 addition & 10 deletions integration-tests/kafka-oauth/src/main/resources/application.properties
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,26 +23,17 @@ kafka.topic.name=test
timer.period = 100
timer.delay = 100

camel.component.kafka.security-protocol = SASL_PLAINTEXT
camel.component.kafka.sasl-mechanism = OAUTHBEARER
camel.component.kafka.sasl-jaas-config = org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
oauth.client.id="kafka-client" \
oauth.client.secret="kafka-client-secret" \
oauth.token.endpoint.uri="http://keycloak:8080/auth/realms/kafka-authz/protocol/openid-connect/token" ;
camel.component.kafka.additional-properties[sasl.login.callback.handler.class] = io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

# enable health check
quarkus.kafka.health.enabled=true

# using QuarkusTestResourceLifecycleManager in this test: Kafka configuration needs to be tuned to work with Keycloak
quarkus.kafka.devservices.enabled=false


#
# Quarkus - Log
#
quarkus.log.category."org.apache.camel.quarkus.core.deployment".level = INFO
quarkus.log.category."org.apache.camel.quarkus.component.kafka".level = DEBUG
quarkus.log.category."org.apache.camel.quarkus.component.kafka".level = INFO
quarkus.log.category."org.apache.zookeeper".level = WARNING
quarkus.log.category."org.apache.kafka".level = WARNING

Expand Down
80 changes: 73 additions & 7 deletions ...auth/src/test/java/org/apache/camel/quarkus/kafka/oauth/it/KafkaKeycloakTestResource.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,13 +16,19 @@
*/
package org.apache.camel.quarkus.kafka.oauth.it;

import java.io.IOException;
import java.net.URL;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

import io.quarkus.test.common.QuarkusTestResourceLifecycleManager;
import io.quarkus.test.keycloak.client.KeycloakTestClient;
import io.quarkus.test.keycloak.server.KeycloakContainer;
import io.strimzi.test.container.StrimziKafkaContainer;
import org.apache.camel.quarkus.kafka.oauth.it.container.KeycloakContainer;
import org.apache.commons.io.IOUtils;
import org.eclipse.microprofile.config.ConfigProvider;
import org.jboss.logging.Logger;
import org.testcontainers.utility.MountableFile;
Expand All @@ -34,7 +40,8 @@
*/
public class KafkaKeycloakTestResource implements QuarkusTestResourceLifecycleManager {

private static final Logger log = Logger.getLogger(KafkaKeycloakTestResource.class);
private static final Logger LOG = Logger.getLogger(KafkaKeycloakTestResource.class);
private static final String REALM_JSON = "keycloak/realms/kafka-authz-realm.json";
private StrimziKafkaContainer kafka;
private KeycloakContainer keycloak;

Expand All @@ -46,23 +53,56 @@ public Map<String, String> start() {
keycloak = new KeycloakContainer();
keycloak.withStartupTimeout(Duration.ofMinutes(5));
keycloak.start();
log.info(keycloak.getLogs());
keycloak.createHostsFile();
LOG.info(keycloak.getLogs());

Path realmJson = null;
try {
URL resource = Thread.currentThread().getContextClassLoader().getResource(REALM_JSON);
if (resource == null) {
throw new RuntimeException("Unable to load " + REALM_JSON);
}

realmJson = Files.createTempFile("keycloak-auth", ".json");
IOUtils.copy(resource, realmJson.toFile());

KeycloakTestClient client = new KeycloakTestClient(keycloak.getServerUrl());
client.createRealmFromPath(realmJson.toAbsolutePath().toString());
} catch (IOException e) {
throw new RuntimeException(e);
} finally {
if (realmJson != null) {
try {
Files.deleteIfExists(realmJson);
} catch (IOException e) {
// Ignored
}
}
}

//Start kafka container
String imageName = ConfigProvider.getConfig().getValue("strimzi-kafka.container.image", String.class);
this.kafka = new StrimziKafkaContainer(imageName)
.withBrokerId(1)
.withKafkaConfigurationMap(Map.of("listener.security.protocol.map", "JWT:SASL_PLAINTEXT,BROKER1:PLAINTEXT"))
.withKafkaConfigurationMap(Map.of("listener.security.protocol.map",
"JWT:SASL_PLAINTEXT,BROKER1:PLAINTEXT",
"listener.name.jwt.oauthbearer.sasl.jaas.config",
getOauthSaslJaasConfig(keycloak.getInternalUrl(), keycloak.getServerUrl()),
"listener.name.jwt.plain.sasl.jaas.config",
getPlainSaslJaasConfig(keycloak.getInternalUrl(), keycloak.getServerUrl())))
.withNetworkAliases("kafka")
.withServerProperties(MountableFile.forClasspathResource("kafkaServer.properties"))
.withBootstrapServers(
c -> String.format("JWT://%s:%s", c.getHost(), c.getMappedPort(KAFKA_PORT)));
this.kafka.start();
log.info(this.kafka.getLogs());
LOG.info(this.kafka.getLogs());

properties.put("kafka.bootstrap.servers", this.kafka.getBootstrapServers());
properties.put("camel.component.kafka.brokers", kafka.getBootstrapServers());

properties.put("camel.component.kafka.security-protocol", "SASL_PLAINTEXT");
properties.put("camel.component.kafka.sasl-mechanism", "OAUTHBEARER");
properties.put("camel.component.kafka.additional-properties[sasl.login.callback.handler.class]",
"io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler");
properties.put("camel.component.kafka.sasl-jaas-config", getClientSaslJaasConfig(keycloak.getServerUrl()));
return properties;
}

Expand All @@ -75,4 +115,30 @@ public void stop() {
keycloak.stop();
}
}

private String getClientSaslJaasConfig(String keycloakServerUrl) {
return "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required" +
" oauth.client.id=\"kafka-client\"" +
" oauth.client.secret=\"kafka-client-secret\"" +
" oauth.token.endpoint.uri=\"" + keycloakServerUrl + "/realms/kafka-authz/protocol/openid-connect/token\";";
}

private String getPlainSaslJaasConfig(String keycloakInternalUrl, String keycloakServerUrl) {
return "'org.apache.kafka.common.security.plain.PlainLoginModule required " +
"oauth.jwks.endpoint.uri=\"" + keycloakInternalUrl + "/realms/kafka-authz/protocol/openid-connect/certs\" " +
"oauth.valid.issuer.uri=\"" + keycloakServerUrl + "/realms/kafka-authz\" " +
"oauth.token.endpoint.uri=\"" + keycloakInternalUrl + "/realms/kafka-authz/protocol/openid-connect/token\" " +
"oauth.client.id=\"kafka\" " +
"oauth.client.secret=\"kafka-secret\" " +
"unsecuredLoginStringClaim_sub=\"admin\";'";
}

private String getOauthSaslJaasConfig(String keycloakInternalUrl, String keycloakServerUrl) {
return "'org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required " +
"oauth.jwks.endpoint.uri=\"" + keycloakInternalUrl + "/realms/kafka-authz/protocol/openid-connect/certs\" " +
"oauth.valid.issuer.uri=\"" + keycloakServerUrl + "/realms/kafka-authz\" " +
"oauth.token.endpoint.uri=\"" + keycloakInternalUrl + "/realms/kafka-authz/protocol/openid-connect/token\" " +
"oauth.client.id=\"kafka\" " +
"oauth.client.secret=\"kafka-secret\";'";
}
}
61 changes: 0 additions & 61 deletions ...th/src/test/java/org/apache/camel/quarkus/kafka/oauth/it/container/KeycloakContainer.java
View file

This file was deleted.

15 changes: 2 additions & 13 deletions integration-tests/kafka-oauth/src/test/resources/kafkaServer.properties
View file
Original file line number Diff line number Diff line change
Expand Up @@ -76,22 +76,11 @@ oauth.username.claim=preferred_username
principal.builder.class=io.strimzi.kafka.oauth.server.OAuthKafkaPrincipalBuilder

listener.name.jwt.sasl.enabled.mechanisms=OAUTHBEARER,PLAIN
listener.name.jwt.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
oauth.jwks.endpoint.uri="http://keycloak:8080/auth/realms/kafka-authz/protocol/openid-connect/certs" \
oauth.valid.issuer.uri="http://keycloak:8080/auth/realms/kafka-authz" \
oauth.token.endpoint.uri="http://keycloak:8080/auth/realms/kafka-authz/protocol/openid-connect/token" \
oauth.client.id="kafka" \
oauth.client.secret="kafka-secret";
listener.name.jwt.oauthbearer.sasl.jaas.config=set_by_test

listener.name.jwt.oauthbearer.sasl.server.callback.handler.class=io.strimzi.kafka.oauth.server.JaasServerOauthValidatorCallbackHandler
listener.name.jwt.oauthbearer.sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
listener.name.jwt.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
oauth.jwks.endpoint.uri="http://keycloak:8080/auth/realms/kafka-authz/protocol/openid-connect/certs" \
oauth.valid.issuer.uri="http://keycloak:8080/auth/realms/kafka-authz" \
oauth.token.endpoint.uri="http://keycloak:8080/auth/realms/kafka-authz/protocol/openid-connect/token" \
oauth.client.id="kafka" \
oauth.client.secret="kafka-secret" \
unsecuredLoginStringClaim_sub="admin";
#listener.name.jwt.plain.sasl.jaas.config=set_by_test

listener.name.jwt.plain.sasl.server.callback.handler.class=io.strimzi.kafka.oauth.server.plain.JaasServerOauthOverPlainValidatorCallbackHandler

Expand Down
1 change: 1 addition & 0 deletions pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -230,6 +230,7 @@
<infinispan.container.image>docker.io/infinispan/server:14.0</infinispan.container.image>
<influxdb.container.image>docker.io/influxdb:1.8.10</influxdb.container.image>
<kafka.container.image>confluentinc/cp-kafka:7.4.0</kafka.container.image>
<keycloak.container.image>quay.io/keycloak/keycloak:22.0.5</keycloak.container.image>
<kudu.container.image>docker.io/apache/kudu:1.15.0</kudu.container.image>
<localstack.container.image>docker.io/localstack/localstack:2.2.0</localstack.container.image>
<lra-coordinator.container.image>quay.io/jbosstm/lra-coordinator:7.0.0.Final-3.2.2.Final</lra-coordinator.container.image>
Expand Down

0 comments on commit aec25ff

Please sign in to comment.