Merge pull request #59 from RADAR-base/clientCredentialsSupport

Add OAuth 2.0 client credentials support for user repository

Author: blootsvoets

README.md

@@ -63,6 +63,12 @@ your Fitbit App client ID and client secret. The following tables shows the poss
 <tr>
 <td>fitbit.user.repository.url</td></td><td>URL for webservice containing user credentials. Only used if a webservice-based user repository is configured.</td></td><td>string</td></td><td>""</td></td><td></td></td><td>low</td></td></tr>
 <tr>
+<td>fitbit.user.repository.client.id</td></td><td>Client ID for connecting to the service repository.</td></td><td>string</td></td><td>""</td></td><td></td></td><td>medium</td></td></tr>
+<tr>
+<td>fitbit.user.repository.client.secret</td></td><td>Client secret for connecting to the service repository.</td></td><td>string</td></td><td>""</td></td><td></td></td><td>medium</td></td></tr>
+<tr>
+<td>fitbit.user.repository.oauth2.token.url</td></td><td>OAuth 2.0 token url for retrieving client credentials.</td></td><td>string</td></td><td>""</td></td><td></td></td><td>medium</td></td></tr>
+<tr>
 <td>fitbit.intraday.steps.topic</td></td><td>Topic for Fitbit intraday steps</td></td><td>string</td></td><td>connect_fitbit_intraday_steps</td></td><td>non-empty string without control characters</td></td><td>low</td></td></tr>
 <tr>
 <td>fitbit.intraday.heart.rate.topic</td></td><td>Topic for Fitbit intraday heart_rate</td></td><td>string</td></td><td>connect_fitbit_intraday_heart_rate</td></td><td>non-empty string without control characters</td></td><td>low</td></td></tr>
@@ -76,8 +82,26 @@ your Fitbit App client ID and client secret. The following tables shows the poss
 <td>fitbit.activity.log.topic</td></td><td>Topic for Fitbit activity log.</td></td><td>string</td></td><td>connect_fitbit_activity_log</td></td><td>non-empty string without control characters</td></td><td>low</td></td></tr>
 <tr>
 <td>fitbit.intraday.calories.topic</td></td><td>Topic for Fitbit intraday calories</td></td><td>string</td></td><td>connect_fitbit_intraday_calories</td></td><td>non-empty string without control characters</td></td><td>low</td></td></tr>
+<tr>
+<td>fitbit.user.firebase.collection.fitbit.name</td></td><td>Firestore Collection for retrieving Fitbit Auth details. Only used when a Firebase based user repository is used.</td></td><td>string</td></td><td>fitbit</td></td><td></td></td><td>low</td></td></tr>
+<tr>
+<td>fitbit.user.firebase.collection.user.name</td></td><td>Firestore Collection for retrieving User details. Only used when a Firebase based user repository is used.</td></td><td>string</td></td><td>users</td></td><td></td></td><td>low</td></td></tr>
 </tbody></table>
 
+If the ManagementPortal is used to authenticate against the user repository, please add an OAuth client to ManagementPortal with the following properties:
+
+```
+Client ID: fitbit.user.repository.client.id
+Client Secret: fitbit.user.repository.client.secret
+Scope: SUBJECT.READ
+Resources: res_restAuthorizer
+Grant types: client_credentials
+Access Token validity: 600
+Refresh Token validity: 0
+```
+
+Finally set the `fitbit.user.repository.oauth.token.url` to `http://managementportal-app:8080/managementportal/oauth/token`.
+
 Now you can run a full Kafka stack using
 
 ```shell

build.gradle

@@ -21,6 +21,7 @@ subprojects {
         maven { url "https://packages.confluent.io/maven/" }
         maven { url "https://repo.maven.apache.org/maven2" }
         jcenter()
+        maven { url  "https://dl.bintray.com/radar-cns/org.radarcns" }
         maven { url  'https://oss.jfrog.org/artifactory/oss-snapshot-local/' }
     }
 }

kafka-connect-fitbit-source/build.gradle

@@ -3,6 +3,7 @@ dependencies {
     api group: 'io.confluent', name: 'kafka-connect-avro-converter', version: confluentVersion
     api group: 'org.radarcns', name: 'radar-schemas-commons', version: '0.5.3'
 
+    implementation group: 'org.radarcns', name: 'oauth-client-util', version: '0.5.8'
 
     implementation group: 'com.fasterxml.jackson.dataformat', name: 'jackson-dataformat-yaml', version: jacksonVersion
     implementation group: 'com.fasterxml.jackson.datatype', name: 'jackson-datatype-jsr310', version: jacksonVersion

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/FitbitRestSourceConnectorConfig.java

@@ -20,6 +20,8 @@
 import static org.apache.kafka.common.config.ConfigDef.NO_DEFAULT_VALUE;
 
 import java.lang.reflect.InvocationTargetException;
+import java.net.MalformedURLException;
+import java.net.URL;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -122,6 +124,19 @@ public class FitbitRestSourceConnectorConfig extends RestSourceConnectorConfig {
   private static final String FITBIT_INTRADAY_CALORIES_TOPIC_DISPLAY = "Intraday calories topic";
   private static final String FITBIT_INTRADAY_CALORIES_TOPIC_DEFAULT = "connect_fitbit_intraday_calories";
 
+
+  public static final String FITBIT_USER_REPOSITORY_CLIENT_ID_CONFIG = "fitbit.user.repository.client.id";
+  private static final String FITBIT_USER_REPOSITORY_CLIENT_ID_DOC = "Client ID for connecting to the service repository.";
+  private static final String FITBIT_USER_REPOSITORY_CLIENT_ID_DISPLAY = "Client ID for user repository.";
+
+  public static final String FITBIT_USER_REPOSITORY_CLIENT_SECRET_CONFIG = "fitbit.user.repository.client.secret";
+  private static final String FITBIT_USER_REPOSITORY_CLIENT_SECRET_DOC = "Client secret for connecting to the service repository.";
+  private static final String FITBIT_USER_REPOSITORY_CLIENT_SECRET_DISPLAY = "Client Secret for user repository.";
+
+  public static final String FITBIT_USER_REPOSITORY_TOKEN_URL_CONFIG = "fitbit.user.repository.oauth2.token.url";
+  private static final String FITBIT_USER_REPOSITORY_TOKEN_URL_DOC = "OAuth 2.0 token url for retrieving client credentials.";
+  private static final String FITBIT_USER_REPOSITORY_TOKEN_URL_DISPLAY = "OAuth 2.0 token URL.";
+
   public static final String FITBIT_USER_REPOSITORY_FIRESTORE_FITBIT_COLLECTION_CONFIG = "fitbit.user.firebase.collection.fitbit.name";
   private static final String FITBIT_USER_REPOSITORY_FIRESTORE_FITBIT_COLLECTION_DOC = "Firestore Collection for retrieving Fitbit Auth details. Only used when a Firebase based user repository is used.";
   private static final String FITBIT_USER_REPOSITORY_FIRESTORE_FITBIT_COLLECTION_DISPLAY = "Firebase Fitbit collection name.";
@@ -242,6 +257,37 @@ public String toString() {
             Width.SHORT,
             FITBIT_USER_REPOSITORY_URL_DISPLAY)
 
+
+        .define(FITBIT_USER_REPOSITORY_CLIENT_ID_CONFIG,
+            Type.STRING,
+            "",
+            Importance.MEDIUM,
+            FITBIT_USER_REPOSITORY_CLIENT_ID_DOC,
+            group,
+            ++orderInGroup,
+            Width.SHORT,
+            FITBIT_USER_REPOSITORY_CLIENT_ID_DISPLAY)
+
+        .define(FITBIT_USER_REPOSITORY_CLIENT_SECRET_CONFIG,
+            Type.STRING,
+            "",
+            Importance.MEDIUM,
+            FITBIT_USER_REPOSITORY_CLIENT_SECRET_DOC,
+            group,
+            ++orderInGroup,
+            Width.SHORT,
+            FITBIT_USER_REPOSITORY_CLIENT_SECRET_DISPLAY)
+
+        .define(FITBIT_USER_REPOSITORY_TOKEN_URL_CONFIG,
+            Type.STRING,
+            "",
+            Importance.MEDIUM,
+            FITBIT_USER_REPOSITORY_TOKEN_URL_DOC,
+            group,
+            ++orderInGroup,
+            Width.SHORT,
+            FITBIT_USER_REPOSITORY_TOKEN_URL_DISPLAY)
+
         .define(FITBIT_INTRADAY_STEPS_TOPIC_CONFIG,
             Type.STRING,
             FITBIT_INTRADAY_STEPS_TOPIC_DEFAULT,
@@ -447,4 +493,25 @@ public String getFitbitUserRepositoryFirestoreFitbitCollection() {
   public String getFitbitUserRepositoryFirestoreUserCollection() {
     return getString(FITBIT_USER_REPOSITORY_FIRESTORE_USER_COLLECTION_CONFIG);
   }
+
+  public String getFitbitUserRepositoryClientId() {
+    return getString(FITBIT_USER_REPOSITORY_CLIENT_ID_CONFIG);
+  }
+
+  public String getFitbitUserRepositoryClientSecret() {
+    return getString(FITBIT_USER_REPOSITORY_CLIENT_SECRET_CONFIG);
+  }
+
+  public URL getFitbitUserRepositoryTokenUrl() {
+    String value = getString(FITBIT_USER_REPOSITORY_TOKEN_URL_CONFIG);
+    if (value == null || value.isEmpty()) {
+      return null;
+    } else {
+      try {
+        return new URL(getString(FITBIT_USER_REPOSITORY_TOKEN_URL_CONFIG));
+      } catch (MalformedURLException e) {
+        throw new ConfigException("Fitbit user repository token URL is invalid.");
+      }
+    }
+  }
 }

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/user/ServiceUserRepository.java

@@ -22,7 +22,10 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectReader;
+import io.confluent.common.config.ConfigException;
 import java.io.IOException;
+import java.net.MalformedURLException;
+import java.net.URL;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.HashMap;
@@ -34,15 +37,19 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import javax.ws.rs.NotAuthorizedException;
+import okhttp3.Credentials;
 import okhttp3.HttpUrl;
 import okhttp3.MediaType;
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
 import okhttp3.RequestBody;
 import okhttp3.Response;
 import okhttp3.ResponseBody;
+import org.apache.kafka.connect.errors.ConnectException;
 import org.radarbase.connect.rest.RestSourceConnectorConfig;
 import org.radarbase.connect.rest.fitbit.FitbitRestSourceConnectorConfig;
+import org.radarcns.exception.TokenException;
+import org.radarcns.oauth.OAuth2Client;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -62,8 +69,10 @@ public class ServiceUserRepository implements UserRepository {
   private final AtomicReference<Instant> nextFetch = new AtomicReference<>(MIN_INSTANT);
 
   private HttpUrl baseUrl;
-  private HashSet<String> containedUsers;
+  private final HashSet<String> containedUsers;
   private Set<? extends User> timedCachedUsers = new HashSet<>();
+  private OAuth2Client repositoryClient;
+  private String basicCredentials;
 
   public ServiceUserRepository() {
     this.client = new OkHttpClient();
@@ -82,6 +91,24 @@ public void initialize(RestSourceConnectorConfig config) {
     FitbitRestSourceConnectorConfig fitbitConfig = (FitbitRestSourceConnectorConfig) config;
     this.baseUrl = fitbitConfig.getFitbitUserRepositoryUrl();
     this.containedUsers.addAll(fitbitConfig.getFitbitUsers());
+
+    URL tokenUrl = fitbitConfig.getFitbitUserRepositoryTokenUrl();
+    String clientId = fitbitConfig.getFitbitUserRepositoryClientId();
+    String clientSecret = fitbitConfig.getFitbitUserRepositoryClientSecret();
+
+    if (tokenUrl != null) {
+      if (clientId.isEmpty()) {
+        throw new ConfigException("Client ID for user repository is not set.");
+      }
+      this.repositoryClient = new OAuth2Client.Builder()
+          .credentials(clientId, clientSecret)
+          .endpoint(tokenUrl)
+          .scopes("SUBJECT.READ")
+          .httpClient(client)
+          .build();
+    } else if (clientId != null) {
+      basicCredentials = Credentials.basic(clientId, clientSecret);
+    }
   }
 
   @Override
@@ -121,21 +148,39 @@ public void applyPendingUpdates() throws IOException {
     Request request = requestFor("users?source-type=FitBit").build();
     this.timedCachedUsers =
         this.<Users>makeRequest(request, USER_LIST_READER).getUsers().stream()
-            .filter(
-                u ->
-                    u.isComplete()
-                        && (containedUsers.isEmpty()
-                            || containedUsers.contains(u.getVersionedId())))
+            .filter(u -> u.isComplete()
+                && (containedUsers.isEmpty()
+                || containedUsers.contains(u.getVersionedId())))
             .collect(Collectors.toSet());
     nextFetch.set(Instant.now().plus(FETCH_THRESHOLD));
   }
 
-  private Request.Builder requestFor(String relativeUrl) {
+  private Request.Builder requestFor(String relativeUrl) throws IOException {
     HttpUrl url = baseUrl.resolve(relativeUrl);
     if (url == null) {
       throw new IllegalArgumentException("Relative URL is invalid");
     }
-    return new Request.Builder().url(url);
+    Request.Builder builder = new Request.Builder().url(url);
+    String authorization = requestAuthorization();
+    if (authorization != null) {
+      builder.addHeader("Authorization", authorization);
+    }
+
+    return builder;
+  }
+
+  private String requestAuthorization() throws IOException {
+    if (repositoryClient != null) {
+      try {
+        return "Bearer " + repositoryClient.getValidToken().getAccessToken();
+      } catch (TokenException ex) {
+        throw new IOException(ex);
+      }
+    } else if (basicCredentials != null) {
+      return basicCredentials;
+    } else {
+      return null;
+    }
   }
 
   private <T> T makeRequest(Request request, ObjectReader reader) throws IOException {