First pass add Basic-Auth capabilities to client

Author: Stephen Powis

src/main/java/org/sourcelab/kafka/connect/apiclient/Configuration.java

@@ -29,9 +29,13 @@ public final class Configuration {
     // Defines the URL/Hostname of Kafka-Connect
     private final String apiHost;
 
-    // Optional Connection settings
+    // Optional Connection options
     private int requestTimeoutInSeconds = 300;
 
+    // Optional BasicAuth options
+    private String basicAuthUsername = null;
+    private String basicAuthPassword = null;
+
     // Optional SSL options
     private boolean ignoreInvalidSslCertificates = false;
     private File trustStoreFile = null;
@@ -64,6 +68,19 @@ public Configuration(final String kafkaConnectHost) {
         }
     }
 
+    /**
+     * Allow setting http Basic-Authentication username and password to authenticate requests.
+     *
+     * @param username username to authenticate requests to Kafka-Connect with.
+     * @param password password to authenticate requests to Kafka-Connect with.
+     * @return Configuration instance.
+     */
+    public Configuration useBasicAuth(final String username, final String password) {
+        this.basicAuthUsername = username;
+        this.basicAuthPassword = password;
+        return this;
+    }
+
     /**
      * Allow setting optional proxy configuration.
      *
@@ -168,6 +185,14 @@ public int getRequestTimeoutInSeconds() {
         return requestTimeoutInSeconds;
     }
 
+    public String getBasicAuthUsername() {
+        return basicAuthUsername;
+    }
+
+    public String getBasicAuthPassword() {
+        return basicAuthPassword;
+    }
+
     @Override
     public String toString() {
         final StringBuilder stringBuilder = new StringBuilder("Configuration{")
@@ -191,6 +216,11 @@ public String toString() {
                 stringBuilder.append(", sslTrustStorePassword='******'");
             }
         }
+        if (basicAuthUsername != null) {
+            stringBuilder
+                .append(", basicAuthUsername='").append(basicAuthUsername).append('\'')
+                .append(", basicAuthPassword='******'");
+        }
         stringBuilder.append('}');
 
         return stringBuilder.toString();

src/main/java/org/sourcelab/kafka/connect/apiclient/KafkaConnectClient.java

@@ -17,6 +17,7 @@
 
 package org.sourcelab.kafka.connect.apiclient;
 
+import org.apache.http.HttpStatus;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.sourcelab.kafka.connect.apiclient.request.JacksonFactory;
@@ -49,6 +50,7 @@
 import org.sourcelab.kafka.connect.apiclient.rest.RestClient;
 import org.sourcelab.kafka.connect.apiclient.rest.RestResponse;
 import org.sourcelab.kafka.connect.apiclient.rest.exceptions.InvalidRequestException;
+import org.sourcelab.kafka.connect.apiclient.rest.exceptions.UnauthorizedRequestException;
 
 import java.io.IOException;
 import java.util.Collection;
@@ -289,6 +291,19 @@ private <T> T submitRequest(final Request<T> request) {
             }
         }
 
+        // Server reject's client's authentication.
+        if (responseCode == HttpStatus.SC_UNAUTHORIZED) {
+            // Throw contextual error msg based on if credentials are configured or not.
+            String errorMsg;
+            if (configuration.getBasicAuthUsername() == null) {
+                errorMsg = "Server required authentication credentials but none were provided in client configuration.";
+            } else {
+                errorMsg = "Client authentication credentials (username=" + configuration.getBasicAuthUsername() + ") was rejected by server.";
+            }
+            errorMsg = errorMsg + " Server responded with: \"" + responseStr + "\"";
+            throw new UnauthorizedRequestException(errorMsg, responseCode);
+        }
+
         // Attempt to parse error response
         try {
             final RequestErrorResponse errorResponse = JacksonFactory.newInstance().readValue(responseStr, RequestErrorResponse.class);

src/main/java/org/sourcelab/kafka/connect/apiclient/rest/HttpClientRestClient.java

@@ -21,6 +21,7 @@
 import org.apache.http.HttpHost;
 import org.apache.http.auth.AuthScope;
 import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.AuthCache;
 import org.apache.http.client.ClientProtocolException;
 import org.apache.http.client.CredentialsProvider;
 import org.apache.http.client.ResponseHandler;
@@ -29,8 +30,11 @@
 import org.apache.http.client.methods.HttpGet;
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.protocol.HttpClientContext;
 import org.apache.http.client.utils.URIBuilder;
 import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.auth.BasicScheme;
+import org.apache.http.impl.client.BasicAuthCache;
 import org.apache.http.impl.client.BasicCredentialsProvider;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.impl.client.HttpClientBuilder;
@@ -45,8 +49,11 @@
 import org.sourcelab.kafka.connect.apiclient.rest.handlers.RestResponseHandler;
 
 import java.io.IOException;
+import java.net.MalformedURLException;
 import java.net.SocketException;
+import java.net.URI;
 import java.net.URISyntaxException;
+import java.net.URL;
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.Collection;
@@ -78,6 +85,8 @@ public class HttpClientRestClient implements RestClient {
      */
     private CloseableHttpClient httpClient;
 
+    private HttpClientContext httpClientContext;
+
     /**
      * Constructor.
      */
@@ -109,6 +118,15 @@ public void init(final Configuration configuration) {
         // Define our RequestConfigBuilder
         final RequestConfig.Builder requestConfigBuilder = RequestConfig.custom();
 
+        // Define our Credentials Provider
+        final CredentialsProvider credsProvider = new BasicCredentialsProvider();
+
+        // Define our context
+        httpClientContext = HttpClientContext.create();
+
+        // Define our auth cache
+        final AuthCache authCache = new BasicAuthCache();
+
         // If we have a configured proxy host
         if (configuration.getProxyHost() != null) {
             // Define proxy host
@@ -120,21 +138,53 @@ public void init(final Configuration configuration) {
 
             // If we have proxy auth enabled
             if (configuration.getProxyUsername() != null) {
-                // Create credential provider
-                final CredentialsProvider credsProvider = new BasicCredentialsProvider();
+                // Add proxy credentials
                 credsProvider.setCredentials(
                     new AuthScope(configuration.getProxyHost(), configuration.getProxyPort()),
                     new UsernamePasswordCredentials(configuration.getProxyUsername(), configuration.getProxyPassword())
                 );
 
-                // Attach Credentials provider to client builder.
-                clientBuilder.setDefaultCredentialsProvider(credsProvider);
+                // Preemptive load context with authentication.
+                authCache.put(
+                    new HttpHost(configuration.getProxyHost(), configuration.getProxyPort(), configuration.getProxyScheme()), new BasicScheme()
+                );
             }
 
             // Attach Proxy to request config builder
             requestConfigBuilder.setProxy(proxyHost);
         }
 
+        // If BasicAuth credentials are configured.
+        if (configuration.getBasicAuthUsername() != null) {
+            try {
+                // parse ApiHost for Hostname and port.
+                final URL apiUrl = new URL(configuration.getApiHost());
+
+                // Add Kafka-Connect credentials
+                credsProvider.setCredentials(
+                    new AuthScope(apiUrl.getHost(), apiUrl.getPort()),
+                    new UsernamePasswordCredentials(
+                        configuration.getBasicAuthUsername(),
+                        configuration.getBasicAuthPassword()
+                    )
+                );
+
+                // Preemptive load context with authentication.
+                authCache.put(
+                    new HttpHost(apiUrl.getHost(), apiUrl.getPort(), apiUrl.getProtocol()), new BasicScheme()
+                );
+            } catch (final MalformedURLException exception) {
+                throw new RuntimeException(exception.getMessage(), exception);
+            }
+        }
+
+        // Configure context.
+        httpClientContext.setAuthCache(authCache);
+        httpClientContext.setCredentialsProvider(credsProvider);
+
+        // Attach Credentials provider to client builder.
+        clientBuilder.setDefaultCredentialsProvider(credsProvider);
+
         // Attach default request config
         clientBuilder.setDefaultRequestConfig(requestConfigBuilder.build());
 
@@ -211,7 +261,7 @@ private <T> T submitGetRequest(final String url, final Map<String, String> getPa
             logger.debug("Executing request {}", get.getRequestLine());
 
             // Execute and return
-            return httpClient.execute(get, responseHandler);
+            return httpClient.execute(get, responseHandler, httpClientContext);
         } catch (final ClientProtocolException | SocketException | URISyntaxException connectionException) {
             // Typically this is a connection issue.
             throw new ConnectionException(connectionException.getMessage(), connectionException);
@@ -244,7 +294,7 @@ private <T> T submitPostRequest(final String url, final Object requestBody, fina
             logger.debug("Executing request {} with {}", post.getRequestLine(), jsonPayloadStr);
 
             // Execute and return
-            return httpClient.execute(post, responseHandler);
+            return httpClient.execute(post, responseHandler, httpClientContext);
         } catch (final ClientProtocolException | SocketException connectionException) {
             // Typically this is a connection issue.
             throw new ConnectionException(connectionException.getMessage(), connectionException);
@@ -276,7 +326,7 @@ private <T> T submitPutRequest(final String url, final Object requestBody, final
             logger.debug("Executing request {} with {}", put.getRequestLine(), jsonPayloadStr);
 
             // Execute and return
-            return httpClient.execute(put, responseHandler);
+            return httpClient.execute(put, responseHandler, httpClientContext);
         } catch (final ClientProtocolException | SocketException connectionException) {
             // Typically this is a connection issue.
             throw new ConnectionException(connectionException.getMessage(), connectionException);
@@ -307,7 +357,7 @@ private <T> T submitDeleteRequest(final String url, final Object requestBody, fi
             logger.debug("Executing request {} with {}", delete.getRequestLine(), jsonPayloadStr);
 
             // Execute and return
-            return httpClient.execute(delete, responseHandler);
+            return httpClient.execute(delete, responseHandler, httpClientContext);
         } catch (final ClientProtocolException | SocketException connectionException) {
             // Typically this is a connection issue.
             throw new ConnectionException(connectionException.getMessage(), connectionException);

src/main/java/org/sourcelab/kafka/connect/apiclient/rest/exceptions/InvalidRequestException.java

@@ -64,6 +64,8 @@ public static InvalidRequestException factory(final RequestErrorResponse errorRe
         Objects.requireNonNull(errorResponse, "Invalid RequestErrorResponse parameter, must not be null");
 
         switch (errorResponse.getErrorCode()) {
+            case 401:
+                return new UnauthorizedRequestException(errorResponse.getMessage(), errorResponse.getErrorCode());
             case 404:
                 return new ResourceNotFoundException(errorResponse.getMessage());
             case 409:

src/main/java/org/sourcelab/kafka/connect/apiclient/rest/exceptions/UnauthorizedRequestException.java

@@ -0,0 +1,17 @@
+package org.sourcelab.kafka.connect.apiclient.rest.exceptions;
+
+/**
+ * Thrown if the server required Authentication, but the client was either not configured to provide credentials,
+ * or those credentials were rejected/invalid.
+ */
+public class UnauthorizedRequestException extends InvalidRequestException {
+
+    /**
+     * Constructor.
+     * @param message Error message.
+     * @param errorCode Error code.
+     */
+    public UnauthorizedRequestException(final String message, final int errorCode) {
+        super(message, errorCode);
+    }
+}

src/test/java/org/sourcelab/kafka/connect/apiclient/KafkaConnectClientTest.java

@@ -61,6 +61,11 @@ public void setup() {
             );
         }
 
+        final String basicAuthUsername = System.getenv("KAFKA_CONNECT_BASICAUTH_USERNAME");
+        if (basicAuthUsername != null && !basicAuthUsername.isEmpty()) {
+            configuration.useBasicAuth(basicAuthUsername, System.getenv("KAFKA_CONNECT_BASICAUTH_PASSWORD"));
+        }
+
         // Build api client
         this.kafkaConnectClient = new KafkaConnectClient(configuration);
     }