1296: Added reactive binding

pom.xml

@@ -51,6 +51,7 @@
     <module>credentials</module>
     <module>oauth2_http</module>
     <module>appengine</module>
+    <module>reactor</module>
     <module>bom</module>
   </modules>
 

reactor/java/com/google/auth/oauth2/CacheableTokenProvider.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2017-2023 the original author or authors.
+ *
+ * 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
+ *
+ *      https://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.
+ */
+
+package com.google.auth.oauth2;
+
+import java.util.concurrent.atomic.AtomicReference;
+import reactor.core.publisher.Mono;
+
+public class CacheableTokenProvider implements ReactiveTokenProvider {
+
+  private ReactiveTokenProvider reactiveTokenProvider;
+
+  private AtomicReference<Mono<AccessToken>> atomicReference;
+
+  private final RefreshThreshold refreshThreshold;
+
+  public CacheableTokenProvider(ReactiveTokenProvider reactiveTokenProvider) {
+    this(reactiveTokenProvider, new RefreshThreshold());
+  }
+
+  public CacheableTokenProvider(
+      ReactiveTokenProvider reactiveTokenProvider, RefreshThreshold refreshThreshold) {
+    this.reactiveTokenProvider = reactiveTokenProvider;
+    this.atomicReference = new AtomicReference<>(cachedRetrieval());
+    this.refreshThreshold = refreshThreshold;
+  }
+
+  @Override
+  public Mono<AccessToken> retrieve() {
+    Mono<AccessToken> currentInstance = atomicReference.get();
+    return currentInstance.flatMap(t -> createNewIfCloseToExpiration(currentInstance, t));
+  }
+
+  private Mono<AccessToken> createNewIfCloseToExpiration(
+      Mono<AccessToken> currentInstance, AccessToken t) {
+    boolean expiresSoon = refreshThreshold.over(t);
+    if (expiresSoon) {
+      return refreshOnExpiration(currentInstance);
+    } else {
+      return Mono.just(t);
+    }
+  }
+
+  private Mono<AccessToken> refreshOnExpiration(Mono<AccessToken> expected) {
+    Mono<AccessToken> toExecute = cachedRetrieval();
+    atomicReference.compareAndSet(expected, toExecute);
+    return atomicReference.get();
+  }
+
+  Mono<AccessToken> cachedRetrieval() {
+    return reactiveTokenProvider.retrieve().cache();
+  }
+}

reactor/java/com/google/auth/oauth2/ComputeEngineTokenProvider.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2017-2023 the original author or authors.
+ *
+ * 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
+ *
+ *      https://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.
+ */
+
+package com.google.auth.oauth2;
+
+import static com.google.auth.oauth2.Constants.ACCESS_TOKEN;
+import static com.google.auth.oauth2.Constants.ERROR_PARSING_TOKEN_REFRESH_RESPONSE;
+import static com.google.auth.oauth2.Constants.EXPIRES_IN;
+
+import com.google.api.client.util.GenericData;
+import java.io.IOException;
+import java.util.Date;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Mono;
+
+public class ComputeEngineTokenProvider implements ReactiveTokenProvider {
+
+  private final WebClient webClient;
+  private final ComputeEngineCredentials computeEngineCredentials;
+
+  private final String tokenUrl;
+
+  public ComputeEngineTokenProvider(
+      WebClient webClient, ComputeEngineCredentials computeEngineCredentials) {
+    this.webClient = webClient;
+    this.computeEngineCredentials = computeEngineCredentials;
+    tokenUrl = computeEngineCredentials.createTokenUrlWithScopes();
+  }
+
+  public ComputeEngineTokenProvider(
+      WebClient webClient, ComputeEngineCredentials computeEngineCredentials, String tokenUrl) {
+    this.webClient = webClient;
+    this.computeEngineCredentials = computeEngineCredentials;
+    this.tokenUrl = tokenUrl;
+  }
+
+  @Override
+  public Mono<AccessToken> retrieve() {
+    return webClient
+        .get()
+        .uri(tokenUrl)
+        .header("Metadata-Flavor", "Google")
+        .retrieve()
+        .bodyToMono(GenericData.class)
+        .flatMap(
+            gd -> {
+              try {
+                String tokenValue =
+                    OAuth2Utils.validateString(
+                        gd, ACCESS_TOKEN, ERROR_PARSING_TOKEN_REFRESH_RESPONSE);
+                int expiresInSeconds =
+                    OAuth2Utils.validateInt32(gd, EXPIRES_IN, ERROR_PARSING_TOKEN_REFRESH_RESPONSE);
+                long expiresAtMilliseconds =
+                    computeEngineCredentials.clock.currentTimeMillis() + (expiresInSeconds * 1000L);
+                AccessToken accessToken =
+                    new AccessToken(tokenValue, new Date(expiresAtMilliseconds));
+                return Mono.just(accessToken);
+              } catch (IOException e) {
+                return Mono.error(e);
+              }
+            });
+  }
+}

reactor/java/com/google/auth/oauth2/Constants.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2017-2023 the original author or authors.
+ *
+ * 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
+ *
+ *      https://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.
+ */
+
+package com.google.auth.oauth2;
+
+public class Constants {
+
+  static final String ACCESS_TOKEN = "access_token";
+  static final String EXPIRES_IN = "expires_in";
+  static final String ERROR_PARSING_TOKEN_REFRESH_RESPONSE =
+      "Error parsing token refresh response. ";
+
+  private Constants() {}
+}

reactor/java/com/google/auth/oauth2/ReactiveTokenProvider.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2017-2023 the original author or authors.
+ *
+ * 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
+ *
+ *      https://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.
+ */
+
+package com.google.auth.oauth2;
+
+import com.google.auth.Credentials;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Mono;
+
+public interface ReactiveTokenProvider {
+
+  Mono<AccessToken> retrieve();
+
+  static ReactiveTokenProvider createCacheable(Credentials credentials) {
+    ReactiveTokenProvider reactiveTokenProvider = create(credentials);
+    return new CacheableTokenProvider(reactiveTokenProvider);
+  }
+
+  static ReactiveTokenProvider create(Credentials credentials) {
+    WebClient webClient = WebClient.builder().build();
+    return create(credentials, webClient);
+  }
+
+  static ReactiveTokenProvider create(Credentials credentials, WebClient webClient) {
+    if (credentials instanceof UserCredentials) {
+      return new UserCredentialsTokenProvider(webClient, (UserCredentials) credentials);
+    } else if (credentials instanceof ServiceAccountCredentials) {
+      return new ServiceAccountTokenProvider(webClient, (ServiceAccountCredentials) credentials);
+    } else if (credentials instanceof ComputeEngineCredentials) {
+      return new ComputeEngineTokenProvider(webClient, (ComputeEngineCredentials) credentials);
+    } else {
+      throw new UnsupportedOperationException(
+          "Unsupported credentials type. UserCredentials,ServiceAccountCredentials,ComputeEngineCredentials are supported");
+    }
+  }
+}

reactor/java/com/google/auth/oauth2/RefreshThreshold.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2017-2023 the original author or authors.
+ *
+ * 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
+ *
+ *      https://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.
+ */
+
+package com.google.auth.oauth2;
+
+class RefreshThreshold {
+
+  static final int DEFAULT_MILLIS_BEFORE_EXPIRATION = 60 * 1000;
+  private final int millisBeforeExpiration;
+
+  public RefreshThreshold() {
+    this(DEFAULT_MILLIS_BEFORE_EXPIRATION);
+  }
+
+  public RefreshThreshold(int millisBeforeExpiration) {
+    this.millisBeforeExpiration = millisBeforeExpiration;
+  }
+
+  boolean over(AccessToken accessToken) {
+    long currentMillis = System.currentTimeMillis();
+    long expirationMillis = accessToken.getExpirationTime().getTime();
+    return currentMillis > expirationMillis - millisBeforeExpiration;
+  }
+}

reactor/java/com/google/auth/oauth2/ServiceAccountTokenProvider.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2017-2023 the original author or authors.
+ *
+ * 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
+ *
+ *      https://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.
+ */
+
+package com.google.auth.oauth2;
+
+import com.google.api.client.json.JsonFactory;
+import com.google.api.client.util.GenericData;
+import java.io.IOException;
+import java.util.Date;
+import org.springframework.http.MediaType;
+import org.springframework.web.reactive.function.BodyInserters;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Mono;
+
+public class ServiceAccountTokenProvider implements ReactiveTokenProvider {
+
+  private static final String GRANT_TYPE = "grant_type";
+  private static final String ASSERTION = "assertion";
+  private final WebClient webClient;
+  private final ServiceAccountCredentials serviceAccountCredentials;
+
+  private final String tokenUrl;
+
+  public ServiceAccountTokenProvider(
+      WebClient webClient, ServiceAccountCredentials serviceAccountCredentials) {
+    this.webClient = webClient;
+    this.serviceAccountCredentials = serviceAccountCredentials;
+    tokenUrl = OAuth2Utils.TOKEN_SERVER_URI.toString();
+  }
+
+  public ServiceAccountTokenProvider(
+      WebClient webClient, ServiceAccountCredentials serviceAccountCredentials, String tokenUrl) {
+    this.webClient = webClient;
+    this.serviceAccountCredentials = serviceAccountCredentials;
+    this.tokenUrl = tokenUrl;
+  }
+
+  @Override
+  public Mono<AccessToken> retrieve() {
+    JsonFactory jsonFactory = OAuth2Utils.JSON_FACTORY;
+    long currentTime = serviceAccountCredentials.clock.currentTimeMillis();
+
+    try {
+      String assertion = serviceAccountCredentials.createAssertion(jsonFactory, currentTime);
+
+      return webClient
+          .post()
+          .uri(tokenUrl)
+          .contentType(MediaType.APPLICATION_FORM_URLENCODED)
+          .body(refreshForm(assertion))
+          .retrieve()
+          .bodyToMono(GenericData.class)
+          .flatMap(
+              gd -> {
+                try {
+                  AccessToken accessToken = getAccessToken(gd);
+                  return Mono.just(accessToken);
+                } catch (IOException e) {
+                  return Mono.error(e);
+                }
+              });
+
+    } catch (IOException e) {
+      return Mono.error(e);
+    }
+  }
+
+  private AccessToken getAccessToken(GenericData gd) throws IOException {
+    String tokenValue =
+        OAuth2Utils.validateString(gd, "access_token", "Error parsing token refresh response. ");
+    int expiresInSeconds =
+        OAuth2Utils.validateInt32(gd, "expires_in", "Error parsing token refresh response. ");
+    long expiresAtMilliseconds =
+        serviceAccountCredentials.clock.currentTimeMillis() + (long) expiresInSeconds * 1000L;
+    AccessToken accessToken = new AccessToken(tokenValue, new Date(expiresAtMilliseconds));
+    return accessToken;
+  }
+
+  private static BodyInserters.FormInserter<String> refreshForm(String assertion) {
+    return BodyInserters.fromFormData(GRANT_TYPE, OAuth2Utils.GRANT_TYPE_JWT_BEARER)
+        .with(ASSERTION, assertion);
+  }
+}