[CDAP-20957] Add scopes support to Remote Authenticator

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/worker/sidecar/GcpMetadataHttpHandlerInternal.java

@@ -47,10 +47,12 @@
 import io.netty.handler.codec.http.FullHttpRequest;
 import io.netty.handler.codec.http.HttpRequest;
 import io.netty.handler.codec.http.HttpResponseStatus;
+import java.io.IOException;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
 import javax.ws.rs.DELETE;
 import javax.ws.rs.GET;
 import javax.ws.rs.PUT;
@@ -76,7 +78,8 @@ public class GcpMetadataHttpHandlerInternal extends AbstractAppFabricHttpHandler
   private final GcpWorkloadIdentityInternalAuthenticator gcpWorkloadIdentityInternalAuthenticator;
   private GcpMetadataTaskContext gcpMetadataTaskContext;
   private final LoadingCache<ProvisionedCredentialCacheKey,
-        ProvisionedCredential> credentialLoadingCache;
+      GcpTokenResponse> credentialLoadingCache;
+  private boolean credentialIdentityPresent;
 
   /**
    * Constructs the {@link GcpMetadataHttpHandlerInternal}.
@@ -91,12 +94,13 @@ public GcpMetadataHttpHandlerInternal(CConfiguration cConf,
         new GcpWorkloadIdentityInternalAuthenticator(gcpMetadataTaskContext);
     this.credentialProvider = new RemoteNamespaceCredentialProvider(remoteClientFactory,
         this.gcpWorkloadIdentityInternalAuthenticator);
+    this.credentialIdentityPresent = true;
     this.credentialLoadingCache = CacheBuilder.newBuilder()
         // Provisioned credential expire after 60mins, assuming 20% buffer in cache exp (0.8*60).
         .expireAfterWrite(48, TimeUnit.MINUTES)
-        .build(new CacheLoader<ProvisionedCredentialCacheKey, ProvisionedCredential>() {
+        .build(new CacheLoader<ProvisionedCredentialCacheKey, GcpTokenResponse>() {
           @Override
-          public ProvisionedCredential load(ProvisionedCredentialCacheKey
+          public GcpTokenResponse load(ProvisionedCredentialCacheKey
               provisionedCredentialCacheKey) throws Exception {
             return fetchTokenFromCredentialProvider(
                 provisionedCredentialCacheKey.getGcpMetadataTaskContext(),
@@ -159,47 +163,62 @@ public void token(HttpRequest request, HttpResponder responder,
       return;
     }
 
-    try {
-      // fetch token from credential provider
-      ProvisionedCredential provisionedCredential =
-          credentialLoadingCache.get(
-              new ProvisionedCredentialCacheKey(this.gcpMetadataTaskContext, scopes));
-      GcpTokenResponse gcpTokenResponse = new GcpTokenResponse("Bearer",
-          provisionedCredential.get(),
-          Duration.between(Instant.now(), provisionedCredential.getExpiration()).getSeconds());
-      responder.sendJson(HttpResponseStatus.OK, GSON.toJson(gcpTokenResponse));
-      return;
-    } catch (ExecutionException e) {
-      if (!(e.getCause() instanceof NotFoundException)) {
-        LOG.error("Failed to fetch token from credential provider", e.getCause());
-        throw e;
+    if (credentialIdentityPresent) {
+      try {
+        GcpTokenResponse gcpTokenResponse =
+            credentialLoadingCache.get(
+                new ProvisionedCredentialCacheKey(gcpMetadataTaskContext, scopes));
+        responder.sendJson(HttpResponseStatus.OK, GSON.toJson(gcpTokenResponse));
+        return;
+      } catch (ExecutionException e) {
+        if (!(e.getCause() instanceof NotFoundException)) {
+          LOG.error("Failed to fetch token from credential provider", e.getCause());
+          throw e;
+        }
+        // if credential identity not found,
+        // fallback to gcp metadata server for backward compatibility.
+        credentialIdentityPresent = false;
       }
-      // if credential identity not found,
-      // fallback to gcp metadata server for backward compatibility.
     }
 
     try {
-      Credential credential = remoteAuthenticator.getCredentials();
-      if (credential == null || Strings.isNullOrEmpty(credential.getValue())) {
-        responder.sendJson(HttpResponseStatus.INTERNAL_SERVER_ERROR,
-            "Failed to fetch token from metadata server");
-        return;
+      GcpTokenResponse gcpTokenResponse;
+      if (Strings.isNullOrEmpty(scopes)) {
+        gcpTokenResponse = convert(remoteAuthenticator.getCredentials());
+      } else {
+        gcpTokenResponse = credentialLoadingCache.get(
+            new ProvisionedCredentialCacheKey(null, scopes));
       }
-      GcpTokenResponse gcpTokenResponse =
-          new GcpTokenResponse(credential.getType().getQualifiedName(), credential.getValue(),
-              credential.getExpirationTimeSecs());
       responder.sendJson(HttpResponseStatus.OK, GSON.toJson(gcpTokenResponse));
-    } catch (Exception ex) {
-      LOG.error("Failed to fetch token from metadata server", ex);
+    } catch (ExecutionException ex) {
+      LOG.error("Failed to fetch token from metadata server", ex.getCause());
       responder.sendJson(HttpResponseStatus.INTERNAL_SERVER_ERROR, exceptionToJson(ex));
     }
   }
 
-  private ProvisionedCredential fetchTokenFromCredentialProvider(
-      GcpMetadataTaskContext gcpMetadataTaskContext, String scopes) throws Exception {
-    return Retries.callWithRetries(() ->
+  private GcpTokenResponse fetchTokenFromCredentialProvider(
+      @Nullable GcpMetadataTaskContext gcpMetadataTaskContext, String scopes) throws Exception {
+    if (gcpMetadataTaskContext == null) {
+      return convert(remoteAuthenticator.getCredentials(scopes));
+    }
+
+    ProvisionedCredential provisionedCredential = Retries.callWithRetries(() ->
             this.credentialProvider.provision(gcpMetadataTaskContext.getNamespace(), scopes),
         RetryStrategies.fromConfiguration(cConf, Constants.Service.TASK_WORKER + "."));
+    return convert(provisionedCredential);
+  }
+
+  private GcpTokenResponse convert(ProvisionedCredential provisionedCredential) {
+    return new GcpTokenResponse("Bearer", provisionedCredential.get(),
+        Duration.between(Instant.now(), provisionedCredential.getExpiration()).getSeconds());
+  }
+
+  private GcpTokenResponse convert(@Nullable Credential credential) throws IOException {
+    if (credential == null || Strings.isNullOrEmpty(credential.getValue())) {
+      throw new IOException("Unable to fetch credential");
+    }
+    return new GcpTokenResponse(credential.getType().getQualifiedName(), credential.getValue(),
+        credential.getExpirationTimeSecs());
   }
 
   /**
@@ -231,6 +250,7 @@ public void clearContext(HttpRequest request, HttpResponder responder) {
     this.gcpMetadataTaskContext = null;
     this.gcpWorkloadIdentityInternalAuthenticator.setGcpMetadataTaskContext(gcpMetadataTaskContext);
     this.credentialLoadingCache.invalidateAll();
+    this.credentialIdentityPresent = true;
     LOG.trace("Context cleared.");
     responder.sendStatus(HttpResponseStatus.OK);
   }

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/worker/sidecar/ProvisionedCredentialCacheKey.java

@@ -48,6 +48,15 @@ public boolean equals(Object o) {
       return  false;
     }
     ProvisionedCredentialCacheKey that = (ProvisionedCredentialCacheKey) o;
+
+    if (gcpMetadataTaskContext == null && that.gcpMetadataTaskContext == null) {
+      return Objects.equals(scopes, that.scopes);
+    }
+
+    if (gcpMetadataTaskContext == null || that.gcpMetadataTaskContext == null) {
+      return false;
+    }
+
     return Objects.equals(gcpMetadataTaskContext.getNamespace(),
         that.gcpMetadataTaskContext.getNamespace())
         && Objects.equals(gcpMetadataTaskContext.getUserCredential().toString(),
@@ -63,9 +72,14 @@ public boolean equals(Object o) {
   public int hashCode() {
     Integer hashCode = this.hashCode;
     if (hashCode == null) {
-      this.hashCode = hashCode = Objects.hash(gcpMetadataTaskContext.getNamespace(),
-          gcpMetadataTaskContext.getUserCredential().toString(),
-          gcpMetadataTaskContext.getUserId(), gcpMetadataTaskContext.getUserIp(), scopes);
+
+      if (gcpMetadataTaskContext == null) {
+        this.hashCode = hashCode = Objects.hash(scopes);
+      } else {
+        this.hashCode = hashCode = Objects.hash(gcpMetadataTaskContext.getNamespace(),
+            gcpMetadataTaskContext.getUserCredential().toString(),
+            gcpMetadataTaskContext.getUserId(), gcpMetadataTaskContext.getUserIp(), scopes);
+      }
     }
     return hashCode;
   }

cdap-app-fabric/src/test/java/io/cdap/cdap/internal/app/runtime/monitor/RuntimeServiceRoutingTest.java

@@ -41,7 +41,7 @@
 import io.cdap.cdap.gateway.handlers.PingHandler;
 import io.cdap.cdap.messaging.DefaultTopicMetadata;
 import io.cdap.cdap.messaging.spi.MessagingService;
 import io.cdap.cdap.messaging.guice.MessagingServerRuntimeModule;
 import io.cdap.cdap.proto.ProgramRunStatus;
 import io.cdap.cdap.proto.id.NamespaceId;
 import io.cdap.cdap.proto.id.ProgramRunId;
@@ -62,6 +62,7 @@
 import java.util.Arrays;
 import java.util.Base64;
 import java.util.EnumSet;
+import javax.annotation.Nullable;
 import org.apache.twill.common.Cancellable;
 import org.apache.twill.discovery.DiscoveryService;
 import org.junit.After;
@@ -268,6 +269,15 @@
                                                                     .asBytes());
       return new Credential(credentialValue, Credential.CredentialType.EXTERNAL_BEARER);
     }
+
+    /**
+     * Returns the credentials for the authentication with scopes.
+     */
+    @Nullable
+    @Override
+    public Credential getCredentials(String scopes) throws IOException {
+      return getCredentials();
+    }
   }
 
   /**

cdap-authenticator-ext-gcp/src/main/java/io/cdap/cdap/authenticator/gcp/GCPRemoteAuthenticator.java

@@ -19,6 +19,7 @@
 import com.google.auth.oauth2.AccessToken;
 import com.google.auth.oauth2.GoogleCredentials;
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
 import io.cdap.cdap.proto.security.Credential;
 import io.cdap.cdap.security.spi.authenticator.RemoteAuthenticator;
 import java.io.IOException;
@@ -30,7 +31,7 @@
  * {@link RemoteAuthenticator} Authenticator which returns Google application default credentials.
  * For additional information, see https://google.aip.dev/auth/4110.
  */
 public class GCPRemoteAuthenticator implements RemoteAuthenticator {
 
   public static final String GCP_REMOTE_AUTHENTICATOR_NAME = "gcp-remote-authenticator";
 
@@ -67,4 +68,19 @@
     return new Credential(accessToken.getTokenValue(), Credential.CredentialType.EXTERNAL_BEARER,
         accessToken.getExpirationTime().getTime() / 1000L);
   }
+
+  /**
+   * Returns the credentials for the authentication with scopes.
+   */
+  @Nullable
+  @Override
+  public Credential getCredentials(@Nullable String scopes) throws IOException {
+    if (Strings.isNullOrEmpty(scopes)) {
+      return getCredentials();
+    }
+    AccessToken accessToken =
+        GoogleCredentials.getApplicationDefault().createScoped(scopes).refreshAccessToken();
+    return new Credential(accessToken.getTokenValue(), Credential.CredentialType.EXTERNAL_BEARER,
+        accessToken.getExpirationTime().getTime() / 1000L);
+  }
 }

cdap-common/src/main/java/io/cdap/cdap/common/internal/remote/GceRemoteAuthenticator.java

@@ -16,16 +16,20 @@
 
 package io.cdap.cdap.common.internal.remote;
 
+import com.google.common.base.Strings;
 import com.google.gson.Gson;
 import com.google.gson.JsonObject;
+import io.cdap.cdap.common.discovery.URIScheme;
 import io.cdap.cdap.proto.security.Credential;
 import io.cdap.cdap.security.spi.authenticator.RemoteAuthenticator;
 import io.cdap.common.http.HttpRequest;
 import io.cdap.common.http.HttpRequests;
 import io.cdap.common.http.HttpResponse;
 import java.io.IOException;
-import java.net.URL;
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
 
 /**
  * A {@link RemoteAuthenticator} that authenticate remote calls using Google Cloud token acquired
@@ -46,22 +50,45 @@ public String getName() {
 
   @Override
   public Credential getCredentials() throws IOException {
-    return new Credential(getAccessToken().getToken(), Credential.CredentialType.EXTERNAL_BEARER);
+    return new Credential(getAccessToken(null).getToken(),
+        Credential.CredentialType.EXTERNAL_BEARER);
+  }
+
+  /**
+   * Returns the credentials for the authentication with scopes.
+   */
+  @Nullable
+  @Override
+  public Credential getCredentials(String scopes) throws IOException {
+    return new Credential(getAccessToken(scopes).getToken(),
+        Credential.CredentialType.EXTERNAL_BEARER);
   }
 
   /**
    * Returns an unexpired access token for authentication.
    */
-  private AccessToken getAccessToken() throws IOException {
+  private AccessToken getAccessToken(@Nullable String scopes) throws IOException {
     AccessToken accessToken = this.accessToken;
     if (accessToken != null && !accessToken.isExpired()) {
       return accessToken;
     }
 
-    URL url = new URL(
-        "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token");
+    URI uri;
+    try {
+      uri = new URI(URIScheme.HTTP.getScheme(), "metadata.google.internal",
+          "/computeMetadata/v1/instance/service-accounts/default/token",
+          null, null);
+      if (!Strings.isNullOrEmpty(scopes)) {
+        uri = new URI(URIScheme.HTTP.getScheme(), "metadata.google.internal",
+            "/computeMetadata/v1/instance/service-accounts/default/token",
+            String.format("scopes=%s", scopes), null);
+      }
+    } catch (URISyntaxException e) {
+      throw new IOException(e);
+    }
+
     HttpResponse response = HttpRequests.execute(
-        HttpRequest.get(url).addHeader("Metadata-Flavor", "Google").build());
+        HttpRequest.get(uri.toURL()).addHeader("Metadata-Flavor", "Google").build());
     if (response.getResponseCode() != 200) {
       throw new IOException("Failed to default service account token");
     }

cdap-common/src/main/java/io/cdap/cdap/common/internal/remote/NoOpRemoteAuthenticator.java

@@ -38,4 +38,13 @@ public String getName() {
   public Credential getCredentials() throws IOException {
     return null;
   }
+
+  /**
+   * Returns the credentials for the authentication with scopes.
+   */
+  @Nullable
+  @Override
+  public Credential getCredentials(String scopes) throws IOException {
+    return null;
+  }
 }

cdap-security-spi/src/main/java/io/cdap/cdap/security/spi/authenticator/RemoteAuthenticator.java

@@ -29,7 +29,7 @@
  */
 public interface RemoteAuthenticator {
 
   /**
    * @return the name of the remote authenticator.
    */
   String getName();
@@ -39,4 +39,10 @@
    */
   @Nullable
   Credential getCredentials() throws IOException;
+
+  /**
+   * Returns the credentials for the authentication with scopes.
+   */
+  @Nullable
+  Credential getCredentials(String scopes) throws IOException;
 }