Add Kafka REST API authentication Add basic username/password authentication to all calls to the Kafka REST API. Change-Id: I9b91b7750d49fe5664ed4e837ed16cdca633dff1
diff --git a/BUILD b/BUILD index d9f938b..8889c80 100644 --- a/BUILD +++ b/BUILD
@@ -28,6 +28,7 @@ junit_tests( name = "events_kafka_tests", srcs = glob(["src/test/java/**/*.java"]), + resources = glob(["src/test/resources/**/*"]), tags = ["events-kafka"], timeout = "long", deps = [
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaProperties.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaProperties.java index 62330e3..1b29748 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaProperties.java +++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaProperties.java
@@ -40,6 +40,8 @@ private static final String PROPERTY_HTTP_WIRE_LOG = "httpWireLog"; private static final boolean DEFAULT_HTTP_WIRE_LOG = false; private static final String PROPERTY_REST_API_URI = "restApiUri"; + private static final String PROPERTY_REST_API_USERNAME = "restApiUsername"; + private static final String PROPERTY_REST_API_PASSWORD = "restApiPassword"; private static final String PROPERTY_REST_API_TIMEOUT = "restApiTimeout"; private static final Duration DEFAULT_REST_API_TIMEOUT = Duration.ofSeconds(60); private static final String PROPERTY_REST_API_THREADS = "restApiThreads"; @@ -64,6 +66,8 @@ private final boolean sendAsync; private final ClientType clientType; private final String restApiUriString; + private final String restApiUsername; + private final String restApiPassword; private final boolean httpWireLog; private final Duration restApiTimeout; private final int restApiThreads; @@ -86,6 +90,12 @@ throw new IllegalArgumentException("Missing REST API URI in Kafka properties"); } + restApiUsername = fromGerritConfig.getString(PROPERTY_REST_API_USERNAME); + restApiPassword = fromGerritConfig.getString(PROPERTY_REST_API_PASSWORD); + if (!Strings.isNullOrEmpty(restApiUsername) && Strings.isNullOrEmpty(restApiPassword)) { + throw new IllegalArgumentException("Missing REST API password in kafka properties"); + } + httpWireLog = fromGerritConfig.getBoolean(PROPERTY_HTTP_WIRE_LOG, DEFAULT_HTTP_WIRE_LOG); restApiTimeout = Duration.ofMillis( @@ -99,6 +109,8 @@ case NATIVE: default: restApiUriString = null; + restApiUsername = null; + restApiPassword = null; httpWireLog = false; restApiTimeout = null; restApiThreads = 0; @@ -111,7 +123,11 @@ @VisibleForTesting public KafkaProperties( - boolean sendAsync, ClientType clientType, @Nullable String restApiUriString) { + boolean sendAsync, + ClientType clientType, + @Nullable String restApiUriString, + @Nullable String restApiUsername, + @Nullable String restApiPassword) { super(); setDefaults(); topic = DEFAULT_STREAM_EVENTS_TOPIC_NAME; @@ -122,6 +138,8 @@ this.httpWireLog = false; restApiTimeout = DEFAULT_REST_API_TIMEOUT; restApiThreads = DEFAULT_REST_API_THREADS; + this.restApiUsername = restApiUsername; + this.restApiPassword = restApiPassword; } private void setDefaults() { @@ -176,6 +194,14 @@ return getRestApiUri(""); } + public String getRestApiUsername() { + return restApiUsername; + } + + public String getRestApiPassword() { + return restApiPassword; + } + public URI getRestApiUri(String kafkaRestId) throws URISyntaxException { return new URI(restApiUriString.replace(REST_API_URI_ID_PLACEHOLDER, kafkaRestId)); }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaSubscriberProperties.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaSubscriberProperties.java index 9b4ef8f..7e6a49c 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaSubscriberProperties.java +++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaSubscriberProperties.java
@@ -45,7 +45,7 @@ @VisibleForTesting public KafkaSubscriberProperties( int pollingInterval, String groupId, int numberOfSubscribers, ClientType clientType) { - this(pollingInterval, groupId, numberOfSubscribers, clientType, null); + this(pollingInterval, groupId, numberOfSubscribers, clientType, null, null, null); } @VisibleForTesting @@ -54,8 +54,10 @@ String groupId, int numberOfSubscribers, ClientType clientType, - String restApiUriString) { - super(true, clientType, restApiUriString); + String restApiUriString, + String restApiUsername, + String restApiPassword) { + super(true, clientType, restApiUriString, restApiUsername, restApiPassword); this.pollingInterval = pollingInterval; this.groupId = groupId; this.numberOfSubscribers = numberOfSubscribers;
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/HttpAsyncClientBuilderFactory.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/HttpAsyncClientBuilderFactory.java new file mode 100644 index 0000000..ec50bf6 --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/HttpAsyncClientBuilderFactory.java
@@ -0,0 +1,50 @@ +// Copyright (C) 2011 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.kafka.rest; + +import com.google.common.base.Strings; +import com.google.inject.Inject; +import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties; +import java.net.URISyntaxException; +import org.apache.http.auth.AuthScope; +import org.apache.http.auth.UsernamePasswordCredentials; +import org.apache.http.client.CredentialsProvider; +import org.apache.http.impl.client.BasicCredentialsProvider; +import org.apache.http.impl.nio.client.HttpAsyncClientBuilder; +import org.apache.http.impl.nio.client.HttpAsyncClients; + +/** Looks up a remote's password in secure.config. */ +public class HttpAsyncClientBuilderFactory { + private final KafkaProperties config; + + @Inject + public HttpAsyncClientBuilderFactory(KafkaProperties config) { + this.config = config; + } + + public HttpAsyncClientBuilder create() throws URISyntaxException { + String user = config.getRestApiUsername(); + String pass = config.getRestApiPassword(); + HttpAsyncClientBuilder httpAsyncClientBuilder = HttpAsyncClients.custom(); + if (!Strings.isNullOrEmpty(user) && !Strings.isNullOrEmpty(pass)) { + CredentialsProvider credsProvider = new BasicCredentialsProvider(); + credsProvider.setCredentials( + new AuthScope(config.getRestApiUri().getHost(), config.getRestApiUri().getPort()), + new UsernamePasswordCredentials(user, pass)); + httpAsyncClientBuilder.setDefaultCredentialsProvider(credsProvider); + } + return httpAsyncClientBuilder; + } +}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/KafkaRestClient.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/KafkaRestClient.java index d06b427..047a96e 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/KafkaRestClient.java +++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/rest/KafkaRestClient.java
@@ -50,7 +50,6 @@ import org.apache.http.entity.ContentType; import org.apache.http.entity.StringEntity; import org.apache.http.impl.nio.client.CloseableHttpAsyncClient; -import org.apache.http.impl.nio.client.HttpAsyncClients; import org.apache.log4j.Appender; import org.apache.log4j.AppenderSkeleton; import org.apache.log4j.Level; @@ -78,9 +77,11 @@ public KafkaRestClient( HttpHostProxy httpHostProxy, @FutureExecutor ExecutorService executor, - @Assisted KafkaProperties configuration) { + HttpAsyncClientBuilderFactory credentialsFactory, + @Assisted KafkaProperties configuration) + throws URISyntaxException { proxy = httpHostProxy; - httpclient = proxy.apply(HttpAsyncClients.custom()).build(); + httpclient = proxy.apply(credentialsFactory.create()).build(); httpclient.start(); this.configuration = configuration; kafkaRestApiTimeoutMsec = (int) configuration.getRestApiTimeout().toMillis();
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md index 24bdc8c..bfac4c5 100644 --- a/src/main/resources/Documentation/config.md +++ b/src/main/resources/Documentation/config.md
@@ -93,3 +93,19 @@ : Send messages to Kafka asynchronously, detaching the calling process from the acknowledge of the message being sent. Default: true + +secure.config +-------------------- + +`plugin.@PLUGIN@.restApiUsername` +: Username used for the authentication to the + [Confluent REST-API Proxy](https://docs.confluent.io/platform/current/kafka-rest/index.html) + for sending/receiving messages through REST-API instead of using the native Kafka client. + **NOTE**: when `plugin.@PLUGIN@.restApiUri` is unset or set to `NATIVE`, this setting is ignored. + Default: unset + +`plugin.@PLUGIN@.restApiPassword` +: Password used for the authentication to the + [Confluent REST-API Proxy](https://docs.confluent.io/platform/current/kafka-rest/index.html) + **NOTE**: when `plugin.@PLUGIN@.restApiUri` is unset or set to `NATIVE`, this setting is ignored. + Default: unset
diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/KafkaRestContainer.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/KafkaRestContainer.java index 06c6801..5cbf585 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/kafka/KafkaRestContainer.java +++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/KafkaRestContainer.java
@@ -20,6 +20,7 @@ import java.net.URISyntaxException; import java.util.Map; import org.junit.Ignore; +import org.testcontainers.containers.BindMode; import org.testcontainers.containers.GenericContainer; import org.testcontainers.containers.KafkaContainer; import org.testcontainers.utility.DockerImageName; @@ -31,11 +32,12 @@ public static final int KAFKA_REST_PORT = 8082; private final String kafkaRestHostname; - public KafkaRestContainer(KafkaContainer kafkaContainer) { - this(kafkaContainer, null); + public KafkaRestContainer(KafkaContainer kafkaContainer, Boolean enableAuthentication) { + this(kafkaContainer, null, enableAuthentication); } - public KafkaRestContainer(KafkaContainer kafkaContainer, String kafkaRestId) { + public KafkaRestContainer( + KafkaContainer kafkaContainer, String kafkaRestId, Boolean enableAuthentication) { super(restProxyImageFor(kafkaContainer)); kafkaRestHostname = KAFKA_REST_PROXY_HOSTNAME + Strings.nullToEmpty(kafkaRestId); @@ -54,6 +56,18 @@ if (kafkaRestId != null) { withEnv("KAFKA_REST_ID", kafkaRestId); } + if (enableAuthentication) { + withEnv("KAFKA_REST_AUTHENTICATION_METHOD", "BASIC"); + withEnv("KAFKA_REST_AUTHENTICATION_REALM", "KafkaRest"); + withEnv("KAFKA_REST_AUTHENTICATION_ROLES", "GerritRole"); + withEnv( + "KAFKAREST_OPTS", + "-Djava.security.auth.login.config=/etc/kafka-rest/rest-jaas.properties"); + withClasspathResourceMapping( + "rest-jaas.properties", "/etc/kafka-rest/rest-jaas.properties", BindMode.READ_ONLY); + withClasspathResourceMapping( + "password.properties", "/etc/kafka-rest/password.properties", BindMode.READ_ONLY); + } withCreateContainerCmdModifier(cmd -> cmd.withHostName(kafkaRestHostname)); }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApiTest.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApiTest.java index 37f15da..1991467 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApiTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApiTest.java
@@ -17,6 +17,7 @@ import static com.google.common.truth.Truth.assertThat; import static org.mockito.Mockito.mock; +import com.google.common.base.Strings; import com.google.gerrit.metrics.MetricMaker; import com.google.gerrit.server.events.Event; import com.google.gerrit.server.events.EventGson; @@ -65,6 +66,8 @@ static KafkaRestContainer kafkaRest; static KafkaRestContainer kafkaRestWithId; static GenericContainer<?> nginx; + static String restApiUsername; + static String restApiPassword; static final int TEST_NUM_SUBSCRIBERS = 1; static final String TEST_GROUP_ID = KafkaBrokerApiTest.class.getName(); @@ -167,9 +170,9 @@ public static void beforeClass() throws Exception { kafka = KafkaContainerProvider.get(); kafka.start(); - kafkaRestWithId = new KafkaRestContainer(kafka, KAFKA_REST_ID); + kafkaRestWithId = new KafkaRestContainer(kafka, KAFKA_REST_ID, isAuthenticationProvided()); kafkaRestWithId.start(); - kafkaRest = new KafkaRestContainer(kafka); + kafkaRest = new KafkaRestContainer(kafka, isAuthenticationProvided()); kafkaRest.start(); System.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers()); @@ -194,6 +197,10 @@ } } + private static boolean isAuthenticationProvided() { + return !Strings.isNullOrEmpty(restApiUsername) && !Strings.isNullOrEmpty(restApiPassword); + } + protected TestModule newTestModule(KafkaProperties kafkaProperties) { return new TestModule(kafkaProperties); } @@ -221,7 +228,9 @@ @Test public void shouldSendSyncAndReceiveToTopic() { - connectToKafka(new KafkaProperties(false, clientType, getKafkaRestApiUriString())); + connectToKafka( + new KafkaProperties( + false, clientType, getKafkaRestApiUriString(), restApiUsername, restApiPassword)); KafkaBrokerApi kafkaBrokerApi = injector.getInstance(KafkaBrokerApi.class); String testTopic = "test_topic_sync"; TestConsumer testConsumer = new TestConsumer(1); @@ -240,7 +249,9 @@ @Test public void shouldSendAsyncAndReceiveToTopic() { - connectToKafka(new KafkaProperties(true, clientType, getKafkaRestApiUriString())); + connectToKafka( + new KafkaProperties( + true, clientType, getKafkaRestApiUriString(), restApiUsername, restApiPassword)); KafkaBrokerApi kafkaBrokerApi = injector.getInstance(KafkaBrokerApi.class); String testTopic = "test_topic_async"; TestConsumer testConsumer = new TestConsumer(1); @@ -259,7 +270,9 @@ @Test public void shouldSendToTopicAndResetOffset() { - connectToKafka(new KafkaProperties(false, clientType, getKafkaRestApiUriString())); + connectToKafka( + new KafkaProperties( + false, clientType, getKafkaRestApiUriString(), restApiUsername, restApiPassword)); KafkaBrokerApi kafkaBrokerApi = injector.getInstance(KafkaBrokerApi.class); String testTopic = "test_topic_reset"; Event testEventMessage = new ProjectCreatedEvent();
diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTest.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTest.java index c0c2bc7..5be61ab 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTest.java
@@ -56,7 +56,9 @@ TEST_GROUP_ID, TEST_NUM_SUBSCRIBERS, ClientType.REST, - getKafkaRestApiUriString()); + getKafkaRestApiUriString(), + restApiUsername, + restApiPassword); bind(KafkaSubscriberProperties.class).toInstance(kafkaSubscriberProperties); bind(HttpHostProxy.class).toInstance(new HttpHostProxy(null, null, null));
diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTestBase.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTestBase.java index 91b6286..cee073f 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTestBase.java +++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiTestBase.java
@@ -35,9 +35,11 @@ + " listen [::]:80 default_server;\\n" + " location /%s/ {\\n" + " proxy_pass http://%s:%d/; \\n" + + " proxy_set_header Authorization \"Basic Z2Vycml0OnNlY3JldA==\";\\n" + " }\\n" + " location / {\\n" + " proxy_pass http://%s:%d; \\n" + + " proxy_set_header Authorization \"Basic Z2Vycml0OnNlY3JldA==\";\\n" + " }\\n" + "}", KAFKA_REST_ID,
diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiWithAuthenticationTest.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiWithAuthenticationTest.java new file mode 100644 index 0000000..abd054f --- /dev/null +++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiWithAuthenticationTest.java
@@ -0,0 +1,30 @@ +// Copyright (C) 2022 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.kafka.api; + +import org.junit.BeforeClass; +import org.junit.runner.RunWith; +import org.mockito.junit.MockitoJUnitRunner; + +@RunWith(MockitoJUnitRunner.class) +public class KafkaBrokerRestApiWithAuthenticationTest extends KafkaBrokerRestApiTest { + + @BeforeClass + public static void beforeClass() throws Exception { + restApiUsername = "gerrit"; + restApiPassword = "secret"; + KafkaBrokerRestApiTest.beforeClass(); + } +}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiWithIdPrefixTest.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiWithIdPrefixTest.java index 9125989..6ee06e7 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiWithIdPrefixTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerRestApiWithIdPrefixTest.java
@@ -56,7 +56,9 @@ TEST_GROUP_ID, TEST_NUM_SUBSCRIBERS, ClientType.REST, - getApiUriString()); + getApiUriString(), + null, + null); bind(KafkaSubscriberProperties.class).toInstance(kafkaSubscriberProperties); bind(HttpHostProxy.class).toInstance(new HttpHostProxy(null, null, null));
diff --git a/src/test/resources/password.properties b/src/test/resources/password.properties new file mode 100644 index 0000000..bcff92d --- /dev/null +++ b/src/test/resources/password.properties
@@ -0,0 +1 @@ +gerrit: secret,GerritRole \ No newline at end of file
diff --git a/src/test/resources/rest-jaas.properties b/src/test/resources/rest-jaas.properties new file mode 100644 index 0000000..719a030 --- /dev/null +++ b/src/test/resources/rest-jaas.properties
@@ -0,0 +1,5 @@ +KafkaRest { + org.eclipse.jetty.jaas.spi.PropertyFileLoginModule required + debug="true" + file="/etc/kafka-rest/password.properties"; +}; \ No newline at end of file