Add audience claim to requests to UserRepository

Author: mpgxvii

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

@@ -90,6 +90,8 @@ class ServiceUserRepository : UserRepository {
             tokenUrl = URLBuilder(config.fitbitUserRepositoryTokenUrl.toString()).build(),
             clientId = config.fitbitUserRepositoryClientId,
             clientSecret = config.fitbitUserRepositoryClientSecret,
+            scope = "SUBJECT.READ MEASUREMENT.CREATE",
+            audience = "res_restAuthorizer"
         )
 
         val refreshDuration = config.userCacheRefreshInterval.toKotlinDuration()
@@ -113,6 +115,8 @@ class ServiceUserRepository : UserRepository {
         tokenUrl: Url?,
         clientId: String?,
         clientSecret: String?,
+        scope: String?,
+        audience: String?,
     ): HttpClient = HttpClient(CIO) {
         if (tokenUrl != null) {
             install(Auth) {
@@ -121,6 +125,8 @@ class ServiceUserRepository : UserRepository {
                         tokenUrl.toString(),
                         clientId,
                         clientSecret,
+                        scope,
+                        audience
                     ).copyWithEnv("MANAGEMENT_PORTAL"),
                     baseUrl.host,
                 )

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

@@ -64,6 +64,8 @@ public class ServiceUserRepositoryLegacy implements UserRepository {
    private static final Duration CONNECTION_TIMEOUT = Duration.ofSeconds(60);
    private static final Duration CONNECTION_READ_TIMEOUT = Duration.ofSeconds(90);
 
+   private static final String CLIENT_AUDIENCE = "res_restAuthorizer"; 
+
    private final OkHttpClient client;
    private final Map<String, OAuth2UserCredentials> cachedCredentials;
    private final AtomicReference<Instant> nextFetch = new AtomicReference<>(MIN_INSTANT);
@@ -76,6 +78,17 @@ public class ServiceUserRepositoryLegacy implements UserRepository {
 
    public ServiceUserRepositoryLegacy() {
      this.client = new OkHttpClient.Builder()
+          .addInterceptor(chain -> {
+            Request original = chain.request();
+            HttpUrl originalUrl = original.url();
+            HttpUrl newUrl = originalUrl.newBuilder()
+                .addQueryParameter("audience", CLIENT_AUDIENCE)  
+                .build();
+            Request newRequest = original.newBuilder()
+                .url(newUrl)
+                .build();
+            return chain.proceed(newRequest);
+        })
          .connectTimeout(CONNECTION_TIMEOUT)
          .readTimeout(CONNECTION_READ_TIMEOUT)
          .build();

kafka-connect-oura-source/src/main/java/org/radarbase/connect/rest/oura/user/OuraServiceUserRepository.kt

@@ -90,6 +90,8 @@ class OuraServiceUserRepository : OuraUserRepository() {
                 tokenUrl = URLBuilder(config.ouraUserRepositoryTokenUrl.toString()).build(),
                 clientId = config.ouraUserRepositoryClientId,
                 clientSecret = config.ouraUserRepositoryClientSecret,
+                scope = "SUBJECT.READ MEASUREMENT.CREATE",
+                audience = "res_restAuthorizer"
             )
 
         userCache =
@@ -111,6 +113,8 @@ class OuraServiceUserRepository : OuraUserRepository() {
         tokenUrl: Url?,
         clientId: String?,
         clientSecret: String?,
+        scope: String?,
+        audience: String?,
     ): HttpClient =
         HttpClient(CIO) {
             if (tokenUrl != null) {
@@ -120,6 +124,8 @@ class OuraServiceUserRepository : OuraUserRepository() {
                             tokenUrl.toString(),
                             clientId,
                             clientSecret,
+                            scope,
+                            audience
                         ).copyWithEnv("MANAGEMENT_PORTAL"),
                         baseUrl.host,
                     )

kafka-connect-oura-source/src/main/java/org/radarbase/connect/rest/oura/user/OuraServiceUserRepositoryLegacy.java

@@ -78,6 +78,8 @@ public class OuraServiceUserRepositoryLegacy extends OuraUserRepository {
    private static final Duration CONNECTION_TIMEOUT = Duration.ofSeconds(60);
    private static final Duration CONNECTION_READ_TIMEOUT = Duration.ofSeconds(90);
 
+   private static final String CLIENT_AUDIENCE = "res_restAuthorizer"; 
+
    private final OkHttpClient client;
    private final Map<String, OAuth2UserCredentials> cachedCredentials;
    private final AtomicReference<Instant> nextFetch = new AtomicReference<>(MIN_INSTANT);
@@ -90,6 +92,17 @@ public class OuraServiceUserRepositoryLegacy extends OuraUserRepository {
 
    public OuraServiceUserRepositoryLegacy() {
      this.client = new OkHttpClient.Builder()
+          .addInterceptor(chain -> {
+            Request original = chain.request();
+            HttpUrl originalUrl = original.url();
+            HttpUrl newUrl = originalUrl.newBuilder()
+                .addQueryParameter("audience", CLIENT_AUDIENCE)  
+                .build();
+            Request newRequest = original.newBuilder()
+                .url(newUrl)
+                .build();
+            return chain.proceed(newRequest);
+        })
          .connectTimeout(CONNECTION_TIMEOUT)
          .readTimeout(CONNECTION_READ_TIMEOUT)
          .build();