ESP-221_BasicAuth - adding HeaderPreprocessor and support for Basic Authentication mechanism. (#26)

* ESP-221_BasicAuth - adding HeaderPreprocessor.java

Signed-off-by: Krzysztof Chmielewski <krzysztof.chmielewski@getindata.com>

* ESP-221_BasicAuth - Changes after code review.

Signed-off-by: Krzysztof Chmielewski <krzysztof.chmielewski@getindata.com>

Signed-off-by: Krzysztof Chmielewski <krzysztof.chmielewski@getindata.com>

Author: kristoffSC

CHANGELOG.md

@@ -4,12 +4,15 @@
 
 ### Added
 
+- Add Http Header value preprocessor mechanism, that can preprocess defined header value before setting it on the request.
+- Allow user to specify `Authorization` header for Basic Authentication. The value will be converted to Base64,
+  or if it starts from prefix `Basic `, it will be used as is (without any extra modification).
 - Add TLS and mTLS support for Http Sink and Lookup Source connectors.  
 New properties are:
   - `gid.connector.http.security.cert.server` - path to server's certificate.
   - `gid.connector.http.security.cert.client` - path to connector's certificate.
   - `gid.connector.http.security.key.client` - path to connector's private key.
-  - `gid.connector.http.security.cert.server.allowSelfSigned` - allowing for self signed certificates without adding them to KeyStore (not recommended for a production).
+  - `gid.connector.http.security.cert.server.allowSelfSigned` - allowing for self-signed certificates without adding them to KeyStore (not recommended for a production).
 
 ## [0.4.0] - 2022-08-31
 

README.md

@@ -231,6 +231,11 @@ For non production environments it is sometimes necessary to use Https connectio
 In this special case, you can configure connector to trust all certificates without adding them to keystore.
 To enable this option use `gid.connector.http.security.cert.server.allowSelfSigned` property setting its value to `true`.
 
+## Basic Authentication
+Connector supports Basic Authentication mechanism using HTTP `Authorization` header.
+The header value can set via properties same as other headers. Connector will convert passed value to Base64 and use it for request.
+If the used value starts from prefix `Basic `, it will be used as header value as is, without any extra modification.
+
 ## Table API Connector Options
 ### HTTP TableLookup Source
 | Option                                                  | Required | Description/Value                                                                                                                                    |

src/main/java/com/getindata/connectors/http/HttpSink.java

@@ -5,6 +5,7 @@
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.connector.base.sink.writer.ElementConverter;
 
+import com.getindata.connectors.http.internal.HeaderPreprocessor;
 import com.getindata.connectors.http.internal.SinkHttpClientBuilder;
 import com.getindata.connectors.http.internal.sink.HttpSinkInternal;
 import com.getindata.connectors.http.internal.sink.HttpSinkRequestEntry;
@@ -41,8 +42,10 @@ public class HttpSink<InputT> extends HttpSinkInternal<InputT> {
             long maxRecordSizeInBytes,
             String endpointUrl,
             HttpPostRequestCallback<HttpSinkRequestEntry> httpPostRequestCallback,
+            HeaderPreprocessor headerPreprocessor,
             SinkHttpClientBuilder sinkHttpClientBuilder,
             Properties properties) {
+
         super(elementConverter,
             maxBatchSize,
             maxInFlightRequests,
@@ -52,6 +55,7 @@ public class HttpSink<InputT> extends HttpSinkInternal<InputT> {
             maxRecordSizeInBytes,
             endpointUrl,
             httpPostRequestCallback,
+            headerPreprocessor,
             sinkHttpClientBuilder,
             properties
         );

src/main/java/com/getindata/connectors/http/HttpSinkBuilder.java

@@ -7,11 +7,13 @@
 import org.apache.flink.connector.base.sink.AsyncSinkBaseBuilder;
 import org.apache.flink.connector.base.sink.writer.ElementConverter;
 
+import com.getindata.connectors.http.internal.HeaderPreprocessor;
 import com.getindata.connectors.http.internal.SinkHttpClient;
 import com.getindata.connectors.http.internal.SinkHttpClientBuilder;
 import com.getindata.connectors.http.internal.sink.HttpSinkRequestEntry;
 import com.getindata.connectors.http.internal.sink.httpclient.JavaNetSinkHttpClient;
 import com.getindata.connectors.http.internal.table.sink.Slf4jHttpPostRequestCallback;
+import com.getindata.connectors.http.internal.utils.HttpHeaderUtils;
 
 /**
  * Builder to construct {@link HttpSink}.
@@ -63,6 +65,9 @@ public class HttpSinkBuilder<InputT> extends
     private static final HttpPostRequestCallback<HttpSinkRequestEntry>
         DEFAULT_POST_REQUEST_CALLBACK = new Slf4jHttpPostRequestCallback();
 
+    private static final HeaderPreprocessor DEFAULT_HEADER_PREPROCESSOR =
+        HttpHeaderUtils.createDefaultHeaderPreprocessor();
+
     private final Properties properties = new Properties();
 
     // Mandatory field
@@ -77,9 +82,13 @@ public class HttpSinkBuilder<InputT> extends
     // If not defined, should be set to DEFAULT_POST_REQUEST_CALLBACK
     private HttpPostRequestCallback<HttpSinkRequestEntry> httpPostRequestCallback;
 
+    // If not defined, should be set to DEFAULT_HEADER_PREPROCESSOR
+    private HeaderPreprocessor headerPreprocessor;
+
     HttpSinkBuilder() {
         this.sinkHttpClientBuilder = DEFAULT_CLIENT_BUILDER;
         this.httpPostRequestCallback = DEFAULT_POST_REQUEST_CALLBACK;
+        this.headerPreprocessor = DEFAULT_HEADER_PREPROCESSOR;
     }
 
     /**
@@ -119,6 +128,12 @@ public HttpSinkBuilder<InputT> setHttpPostRequestCallback(
         return this;
     }
 
+    public HttpSinkBuilder<InputT> setHttpHeaderPreprocessor(
+            HeaderPreprocessor headerPreprocessor) {
+        this.headerPreprocessor = headerPreprocessor;
+        return this;
+    }
+
     /**
      * Set property for Http Sink.
      * @param propertyName property name
@@ -152,6 +167,7 @@ public HttpSink<InputT> build() {
             Optional.ofNullable(getMaxRecordSizeInBytes()).orElse(DEFAULT_MAX_RECORD_SIZE_IN_B),
             endpointUrl,
             httpPostRequestCallback,
+            headerPreprocessor,
             sinkHttpClientBuilder,
             properties
         );

src/main/java/com/getindata/connectors/http/internal/BasicAuthHeaderValuePreprocessor.java

@@ -0,0 +1,32 @@
+package com.getindata.connectors.http.internal;
+
+import java.util.Base64;
+
+/**
+ * Header processor for HTTP Basic Authentication mechanism.
+ * Only "Basic" authentication is supported currently.
+ */
+public class BasicAuthHeaderValuePreprocessor implements HeaderValuePreprocessor {
+
+    public static final String BASIC = "Basic ";
+
+    /**
+     * Calculates {@link Base64} value of provided header value. For Basic authentication mechanism,
+     * the raw value is expected to match user:password pattern.
+     * <p>
+     * If rawValue starts with "Basic " prefix it is assumed that this value is already converted to
+     * expected "Authorization" header value.
+     *
+     * @param rawValue header original value to modify.
+     * @return value of "Authorization" header with format "Basic " + Base64 from rawValue or
+     * rawValue without any changes if it starts with "Basic " prefix.
+     */
+    @Override
+    public String preprocessHeaderValue(String rawValue) {
+        if (rawValue.startsWith(BASIC)) {
+            return rawValue;
+        } else {
+            return BASIC + Base64.getEncoder().encodeToString(rawValue.getBytes());
+        }
+    }
+}

src/main/java/com/getindata/connectors/http/internal/ComposeHeaderPreprocessor.java

@@ -0,0 +1,44 @@
+package com.getindata.connectors.http.internal;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * This implementation of {@link HeaderPreprocessor} acts as a registry for all {@link
+ * HeaderValuePreprocessor} that should be applied on HTTP request.
+ */
+public class ComposeHeaderPreprocessor implements HeaderPreprocessor {
+
+    /**
+     * Default, pass through header value preprocessor used whenever dedicated preprocessor for a
+     * given header does not exist.
+     */
+    private static final HeaderValuePreprocessor DEFAULT_VALUE_PREPROCESSOR = rawValue -> rawValue;
+
+    /**
+     * Map with {@link HeaderValuePreprocessor} to apply.
+     */
+    private final Map<String, HeaderValuePreprocessor> valuePreprocessors;
+
+    /**
+     * Creates a new instance of ComposeHeaderPreprocessor for provided {@link
+     * HeaderValuePreprocessor} map.
+     *
+     * @param valuePreprocessors map of {@link HeaderValuePreprocessor} that should be used for this
+     *                           processor. If null, then default, pass through header value
+     *                           processor will be used for every header.
+     */
+    public ComposeHeaderPreprocessor(Map<String, HeaderValuePreprocessor> valuePreprocessors) {
+        this.valuePreprocessors = (valuePreprocessors == null)
+            ? Collections.emptyMap()
+            : new HashMap<>(valuePreprocessors);
+    }
+
+    @Override
+    public String preprocessValueForHeader(String headerName, String headerRawValue) {
+        return valuePreprocessors
+            .getOrDefault(headerName, DEFAULT_VALUE_PREPROCESSOR)
+            .preprocessHeaderValue(headerRawValue);
+    }
+}

src/main/java/com/getindata/connectors/http/internal/HeaderPreprocessor.java

@@ -0,0 +1,17 @@
+package com.getindata.connectors.http.internal;
+
+import java.io.Serializable;
+
+/**
+ * Interface for header preprocessing
+ */
+public interface HeaderPreprocessor extends Serializable {
+
+    /**
+     * Preprocess value of a header.Preprocessing can change or validate header value.
+     * @param headerName header name which value should be preprocessed.
+     * @param headerRawValue header value to process.
+     * @return preprocessed header value.
+     */
+    String preprocessValueForHeader(String headerName, String headerRawValue);
+}

src/main/java/com/getindata/connectors/http/internal/HeaderValuePreprocessor.java

@@ -0,0 +1,18 @@
+package com.getindata.connectors.http.internal;
+
+import java.io.Serializable;
+
+/**
+ * Processor interface which modifies header value based on implemented logic.
+ * An example would be calculation of Value of Authorization header.
+ */
+public interface HeaderValuePreprocessor extends Serializable {
+
+    /**
+     * Modifies header rawValue according to the implemented logic.
+     * @param rawValue header original value to modify
+     * @return modified header value.
+     */
+    String preprocessHeaderValue(String rawValue);
+
+}

src/main/java/com/getindata/connectors/http/internal/SinkHttpClientBuilder.java

@@ -13,8 +13,12 @@
  */
 @PublicEvolving
 public interface SinkHttpClientBuilder extends Serializable {
+
+    // TODO Consider moving HttpPostRequestCallback and HeaderPreprocessor to be a
+    //  SinkHttpClientBuilder fields. This method is getting more and more arguments.
     SinkHttpClient build(
         Properties properties,
-        HttpPostRequestCallback<HttpSinkRequestEntry> httpPostRequestCallback
+        HttpPostRequestCallback<HttpSinkRequestEntry> httpPostRequestCallback,
+        HeaderPreprocessor headerPreprocessor
     );
 }

src/main/java/com/getindata/connectors/http/internal/sink/HttpSinkInternal.java

@@ -13,9 +13,9 @@
 import org.apache.flink.util.StringUtils;
 
 import com.getindata.connectors.http.HttpPostRequestCallback;
+import com.getindata.connectors.http.internal.HeaderPreprocessor;
 import com.getindata.connectors.http.internal.SinkHttpClientBuilder;
 
-
 /**
  * An internal implementation of HTTP Sink that performs async requests against a specified HTTP
  * endpoint using the buffering protocol specified in {@link AsyncSinkBase}.
@@ -60,6 +60,8 @@ public class HttpSinkInternal<InputT> extends AsyncSinkBase<InputT, HttpSinkRequ
 
     private final HttpPostRequestCallback<HttpSinkRequestEntry> httpPostRequestCallback;
 
+    private final HeaderPreprocessor headerPreprocessor;
+
     private final Properties properties;
 
     protected HttpSinkInternal(
@@ -72,13 +74,20 @@ protected HttpSinkInternal(
         long maxRecordSizeInBytes,
         String endpointUrl,
         HttpPostRequestCallback<HttpSinkRequestEntry> httpPostRequestCallback,
+        HeaderPreprocessor headerPreprocessor,
         SinkHttpClientBuilder sinkHttpClientBuilder,
-        Properties properties
-    ) {
-        super(elementConverter, maxBatchSize, maxInFlightRequests, maxBufferedRequests,
+        Properties properties) {
+
+        super(
+            elementConverter,
+            maxBatchSize,
+            maxInFlightRequests,
+            maxBufferedRequests,
             maxBatchSizeInBytes,
-            maxTimeInBufferMS, maxRecordSizeInBytes
+            maxTimeInBufferMS,
+            maxRecordSizeInBytes
         );
+
         Preconditions.checkArgument(!StringUtils.isNullOrWhitespaceOnly(endpointUrl),
             "The endpoint URL must be set when initializing HTTP Sink.");
         this.endpointUrl = endpointUrl;
@@ -87,6 +96,10 @@ protected HttpSinkInternal(
                 httpPostRequestCallback,
                 "Post request callback must be set when initializing HTTP Sink."
             );
+        this.headerPreprocessor = Preconditions.checkNotNull(
+            headerPreprocessor,
+            "Header Preprocessor must be set when initializing HTTP Sink."
+        );
         this.sinkHttpClientBuilder =
             Preconditions.checkNotNull(sinkHttpClientBuilder,
                 "The HTTP client builder must not be null when initializing HTTP Sink.");
@@ -107,7 +120,7 @@ public StatefulSinkWriter<InputT, BufferedRequestState<HttpSinkRequestEntry>> cr
             getMaxTimeInBufferMS(),
             getMaxRecordSizeInBytes(),
             endpointUrl,
-            sinkHttpClientBuilder.build(properties, httpPostRequestCallback),
+            sinkHttpClientBuilder.build(properties, httpPostRequestCallback, headerPreprocessor),
             Collections.emptyList()
         );
     }
@@ -126,7 +139,7 @@ public StatefulSinkWriter<InputT, BufferedRequestState<HttpSinkRequestEntry>> re
             getMaxTimeInBufferMS(),
             getMaxRecordSizeInBytes(),
             endpointUrl,
-            sinkHttpClientBuilder.build(properties, httpPostRequestCallback),
+            sinkHttpClientBuilder.build(properties, httpPostRequestCallback, headerPreprocessor),
             recoveredState
         );
     }

src/main/java/com/getindata/connectors/http/internal/sink/httpclient/JavaNetSinkHttpClient.java

@@ -15,6 +15,7 @@
 import org.apache.flink.annotation.VisibleForTesting;
 
 import com.getindata.connectors.http.HttpPostRequestCallback;
+import com.getindata.connectors.http.internal.HeaderPreprocessor;
 import com.getindata.connectors.http.internal.SinkHttpClient;
 import com.getindata.connectors.http.internal.SinkHttpClientResponse;
 import com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants;
@@ -23,9 +24,8 @@
 import com.getindata.connectors.http.internal.status.ComposeHttpStatusCodeChecker.ComposeHttpStatusCodeCheckerConfig;
 import com.getindata.connectors.http.internal.status.HttpStatusCodeChecker;
 import com.getindata.connectors.http.internal.table.sink.Slf4jHttpPostRequestCallback;
-import com.getindata.connectors.http.internal.utils.ConfigUtils;
+import com.getindata.connectors.http.internal.utils.HttpHeaderUtils;
 import com.getindata.connectors.http.internal.utils.JavaNetHttpClientFactory;
-import static com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants.SINK_HEADER_PREFIX;
 
 /**
  * An implementation of {@link SinkHttpClient} that uses Java 11's {@link HttpClient}.
@@ -36,32 +36,31 @@ public class JavaNetSinkHttpClient implements SinkHttpClient {
 
     private final HttpClient httpClient;
 
-    private final Map<String, String> headerMap;
-
     private final String[] headersAndValues;
 
+    private final Map<String, String> headerMap;
+
     private final HttpStatusCodeChecker statusCodeChecker;
 
     private final HttpPostRequestCallback<HttpSinkRequestEntry> httpPostRequestCallback;
 
-    public JavaNetSinkHttpClient(Properties properties) {
-        this(properties, new Slf4jHttpPostRequestCallback());
+    public JavaNetSinkHttpClient(Properties properties, HeaderPreprocessor headerPreprocessor) {
+        this(properties, new Slf4jHttpPostRequestCallback(), headerPreprocessor);
     }
 
     public JavaNetSinkHttpClient(
             Properties properties,
-            HttpPostRequestCallback<HttpSinkRequestEntry> httpPostRequestCallback) {
+            HttpPostRequestCallback<HttpSinkRequestEntry> httpPostRequestCallback,
+            HeaderPreprocessor headerPreprocessor) {
 
         this.httpClient = JavaNetHttpClientFactory.createClient(properties);
         this.httpPostRequestCallback = httpPostRequestCallback;
-
-        var propertiesHeaderMap =
-            ConfigUtils.propertiesToMap(properties, SINK_HEADER_PREFIX, String.class);
-        this.headersAndValues = ConfigUtils.toHeaderAndValueArray(propertiesHeaderMap);
-        this.headerMap = new HashMap<>();
-        propertiesHeaderMap
-            .forEach((key, value) ->
-                         this.headerMap.put(ConfigUtils.extractPropertyLastElement(key), value));
+        this.headerMap = HttpHeaderUtils.prepareHeaderMap(
+            HttpConnectorConfigConstants.SINK_HEADER_PREFIX,
+            properties,
+            headerPreprocessor
+        );
+        this.headersAndValues = HttpHeaderUtils.toHeaderAndValueArray(this.headerMap);
 
         // TODO Inject this via constructor when implementing a response processor.
         //  Processor will be injected and it will wrap statusChecker implementation.