Http authentication support including None and Basic authentication

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 ### v0.8.0 (TBD)
 -   Provided different log levels for `OkHttpClient`.`TRACE`: `BODY`, `DEBUG`: `BASIC`, `*`: `NONE`
 -   Refactored throttler adding the notion of timer
+-   Support for authentication extension. Initial implementations for None and Basic authentication types
 
 ### v0.7.6 (05/28/2020)
 -   Fix typo in `http.throttler.catchup.interval.millis` configuration property

README.md

@@ -265,6 +265,49 @@ Uses a [OkHttp](https://square.github.io/okhttp/) client.
 > Maximum number of idle connections in the connection pool
 > *   Type: `Integer`
 > *   Default: `1`
+---
+<a name="auth"/>
+
+### HttpAuthenticator: Authenticating a HttpRequest
+When executing the request, authentication might be required. The HttpAuthenticator is responsible for resolving the authentication header
+to be included in the request 
+`HttpAuthenticator`
+
+> #### `http.auth`
+> ```java
+> public interface HttpAuthenticator extends Configurable {
+> 
+>     Optional<String> getAuthorizationHeader();
+> }
+> ```
+> *   Type: `Class`
+> *   Default: `com.github.castorm.kafka.connect.http.auth.ConfigurableHttpAuthenticator`
+> *   Available implementations:
+>     *   `com.github.castorm.kafka.connect.http.auth.ConfigurableHttpAuthenticator`
+>     *   `com.github.castorm.kafka.connect.http.auth.NoneHttpAuthenticator`
+>     *   `com.github.castorm.kafka.connect.http.auth.BasicHttpAuthenticator`
+
+#### Authenticating a HttpRequest with ConfigurableHttpAuthenticator
+Allows selecting the athentication type via configuration property
+
+> ##### `http.auth.type`
+> Type of authentication
+> *   Type: `String`
+> *   Default: `None`
+> *   Available options:
+>     *   `None`
+>     *   `Basic`
+
+#### Authenticating a HttpRequest with BasicHttpAuthenticator
+Allows selecting the athentication type via configuration property
+
+> ##### `http.auth.user`
+> *   Type: `String`
+> *   Default: ``
+>
+> ##### `http.auth.password`
+> *   Type: `String`
+> *   Default: ``
 
 ---
 <a name="response"/>

examples/jira-issues-search.md

@@ -48,8 +48,11 @@ And based on the results we would be updating the `updated` filter for subsequen
         "tasks.max": "1",
         "http.offset.initial": "timestamp=2020-05-08T07:55:44Z",
         "http.request.url": "https://your-host-here/rest/api/2/search",
-        "http.request.headers": "Authorization: Basic TBD, Accept: application/json",
+        "http.request.headers": "Accept: application/json",
         "http.request.params": "jql=updated>=\"${offset.timestamp?datetime.iso?string['yyyy/MM/dd HH:mm']}\" ORDER BY updated ASC&maxResults=100",
+        "http.auth.type": "Basic",
+        "http.auth.user": "username",
+        "http.auth.password": "password",
         "http.response.list.pointer": "/issues",
         "http.response.record.key.pointer": "/id",
         "http.response.record.offset.pointer": "timestamp=/fields/updated",

src/main/java/com/github/castorm/kafka/connect/http/auth/BasicHttpAuthenticator.java

@@ -0,0 +1,62 @@
+package com.github.castorm.kafka.connect.http.auth;
+
+/*-
+ * #%L
+ * Kafka Connect HTTP
+ * %%
+ * Copyright (C) 2020 CastorM
+ * %%
+ * 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.
+ * #L%
+ */
+
+import com.github.castorm.kafka.connect.http.auth.spi.HttpAuthenticator;
+import okhttp3.Credentials;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+
+import static org.apache.commons.lang.StringUtils.isEmpty;
+
+public class BasicHttpAuthenticator implements HttpAuthenticator {
+
+    private final Function<Map<String, ?>, BasicHttpAuthenticatorConfig> configFactory;
+
+    Optional<String> header;
+
+    public BasicHttpAuthenticator() {
+        this(BasicHttpAuthenticatorConfig::new);
+    }
+
+    public BasicHttpAuthenticator(Function<Map<String, ?>, BasicHttpAuthenticatorConfig> configFactory) {
+        this.configFactory = configFactory;
+    }
+
+    @Override
+    public void configure(Map<String, ?> configs) {
+
+        BasicHttpAuthenticatorConfig config = configFactory.apply(configs);
+
+        if (!isEmpty(config.getUser()) || !isEmpty(config.getPassword().value())) {
+            header = Optional.of(Credentials.basic(config.getUser(), config.getPassword().value()));
+        } else {
+            header = Optional.empty();
+        }
+    }
+
+    @Override
+    public Optional<String> getAuthorizationHeader() {
+        return header;
+    }
+}

src/main/java/com/github/castorm/kafka/connect/http/auth/BasicHttpAuthenticatorConfig.java

@@ -0,0 +1,53 @@
+package com.github.castorm.kafka.connect.http.auth;
+
+/*-
+ * #%L
+ * kafka-connect-http
+ * %%
+ * Copyright (C) 2020 CastorM
+ * %%
+ * 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.
+ * #L%
+ */
+
+import lombok.Getter;
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.types.Password;
+
+import java.util.Map;
+
+import static org.apache.kafka.common.config.ConfigDef.Importance.HIGH;
+import static org.apache.kafka.common.config.ConfigDef.Type.STRING;
+
+@Getter
+public class BasicHttpAuthenticatorConfig extends AbstractConfig {
+
+    private static final String USER = "http.auth.user";
+    private static final String PASSWORD = "http.auth.password";
+
+    private final String user;
+    private final Password password;
+
+    BasicHttpAuthenticatorConfig(Map<String, ?> originals) {
+        super(config(), originals);
+        user = getString(USER);
+        password = getPassword(PASSWORD);
+    }
+
+    public static ConfigDef config() {
+        return new ConfigDef()
+                .define(USER, STRING, "", HIGH, "Username")
+                .define(PASSWORD, ConfigDef.Type.PASSWORD, "", HIGH, "Password");
+    }
+}

src/main/java/com/github/castorm/kafka/connect/http/auth/ConfigurableHttpAuthenticator.java

@@ -0,0 +1,55 @@
+package com.github.castorm.kafka.connect.http.auth;
+
+/*-
+ * #%L
+ * Kafka Connect HTTP
+ * %%
+ * Copyright (C) 2020 CastorM
+ * %%
+ * 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.
+ * #L%
+ */
+
+import com.github.castorm.kafka.connect.http.auth.spi.HttpAuthenticator;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+
+public class ConfigurableHttpAuthenticator implements HttpAuthenticator {
+
+    private final Function<Map<String, ?>, ConfigurableHttpAuthenticatorConfig> configFactory;
+
+    private HttpAuthenticator delegate;
+
+    public ConfigurableHttpAuthenticator() {
+        this(ConfigurableHttpAuthenticatorConfig::new);
+    }
+
+    public ConfigurableHttpAuthenticator(Function<Map<String, ?>, ConfigurableHttpAuthenticatorConfig> configFactory) {
+        this.configFactory = configFactory;
+    }
+
+    @Override
+    public void configure(Map<String, ?> configs) {
+
+        ConfigurableHttpAuthenticatorConfig config = configFactory.apply(configs);
+
+        delegate = config.getAuthenticator();
+    }
+
+    @Override
+    public Optional<String> getAuthorizationHeader() {
+        return delegate.getAuthorizationHeader();
+    }
+}

src/main/java/com/github/castorm/kafka/connect/http/auth/ConfigurableHttpAuthenticatorConfig.java

@@ -0,0 +1,61 @@
+package com.github.castorm.kafka.connect.http.auth;
+
+/*-
+ * #%L
+ * kafka-connect-http
+ * %%
+ * Copyright (C) 2020 CastorM
+ * %%
+ * 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.
+ * #L%
+ */
+
+import com.github.castorm.kafka.connect.http.auth.spi.HttpAuthenticationType;
+import com.github.castorm.kafka.connect.http.auth.spi.HttpAuthenticator;
+import lombok.Getter;
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+
+import java.util.Map;
+
+import static org.apache.kafka.common.config.ConfigDef.Importance.MEDIUM;
+import static org.apache.kafka.common.config.ConfigDef.Type.STRING;
+
+@Getter
+public class ConfigurableHttpAuthenticatorConfig extends AbstractConfig {
+
+    private static final String AUTH_TYPE = "http.auth.type";
+
+    private final HttpAuthenticator authenticator;
+
+    ConfigurableHttpAuthenticatorConfig(Map<String, ?> originals) {
+        super(config(), originals);
+        authenticator = getAuthenticator(originals);
+    }
+
+    private HttpAuthenticator getAuthenticator(Map<String, ?> originals) {
+        switch (HttpAuthenticationType.valueOf(getString(AUTH_TYPE).toUpperCase())) {
+            case BASIC:
+                BasicHttpAuthenticator auth = new BasicHttpAuthenticator();
+                auth.configure(originals);
+                return auth;
+            default:
+                return new NoneHttpAuthenticator();
+        }
+    }
+
+    public static ConfigDef config() {
+        return new ConfigDef()
+                .define(AUTH_TYPE, STRING, HttpAuthenticationType.NONE.name(), MEDIUM, "Authentication Type");
+    }
+}

src/main/java/com/github/castorm/kafka/connect/http/auth/NoneHttpAuthenticator.java

@@ -0,0 +1,33 @@
+package com.github.castorm.kafka.connect.http.auth;
+
+/*-
+ * #%L
+ * Kafka Connect HTTP
+ * %%
+ * Copyright (C) 2020 CastorM
+ * %%
+ * 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.
+ * #L%
+ */
+
+import com.github.castorm.kafka.connect.http.auth.spi.HttpAuthenticator;
+
+import java.util.Optional;
+
+public class NoneHttpAuthenticator implements HttpAuthenticator {
+
+    @Override
+    public Optional<String> getAuthorizationHeader() {
+        return Optional.empty();
+    }
+}

src/main/java/com/github/castorm/kafka/connect/http/auth/spi/HttpAuthenticationType.java

@@ -0,0 +1,25 @@
+package com.github.castorm.kafka.connect.http.auth.spi;
+
+/*-
+ * #%L
+ * Kafka Connect HTTP
+ * %%
+ * Copyright (C) 2020 CastorM
+ * %%
+ * 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.
+ * #L%
+ */
+
+public enum HttpAuthenticationType {
+    NONE, BASIC
+}