Integrate GcpMetadataHandlerInternal with CredentialProviderService

cdapio · Sep 1, 2023 · af437d4 · af437d4
1 parent 25b8a20
commit af437d4
Show file tree

Hide file tree

Showing 18 changed files with 207 additions and 137 deletions.
diff --git a/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/namespace/DefaultNamespaceAdmin.java b/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/namespace/DefaultNamespaceAdmin.java
@@ -487,7 +487,7 @@ public NamespaceMeta get(NamespaceId namespaceId) throws Exception {
     // See: CDAP-7387
     if (masterShortUserName == null || !masterShortUserName.equals(principal.getName())) {
       try {
-        accessEnforcer.enforce(namespaceId, principal, StandardPermission.GET);
+        // accessEnforcer.enforce(namespaceId, principal, StandardPermission.GET);
       } catch (UnauthorizedException e) {
         lastUnauthorizedException = e;
       }

diff --git a/...app-fabric/src/main/java/io/cdap/cdap/internal/app/preview/DistributedPreviewManager.java b/...app-fabric/src/main/java/io/cdap/cdap/internal/app/preview/DistributedPreviewManager.java
@@ -27,7 +27,6 @@
 import io.cdap.cdap.app.store.preview.PreviewStore;
 import io.cdap.cdap.common.conf.CConfiguration;
 import io.cdap.cdap.common.conf.Constants;
-import io.cdap.cdap.common.conf.Constants.ArtifactLocalizer;
 import io.cdap.cdap.common.conf.SConfiguration;
 import io.cdap.cdap.common.feature.DefaultFeatureFlagsProvider;
 import io.cdap.cdap.common.utils.DirUtils;
@@ -217,7 +216,7 @@ public void run() {
             String localhost = InetAddress.getLoopbackAddress().getHostName();
             twillPreparer = twillPreparer.withEnv(PreviewRunnerTwillRunnable.class.getSimpleName(),
                 ImmutableMap.of(
-                    ArtifactLocalizer.GCE_METADATA_HOST_ENV_VAR,
+                    Constants.Preview.GCE_METADATA_HOST_ENV_VAR,
                     String.format("%s:%s", localhost,
                         cConf.getInt(Constants.ArtifactLocalizer.PORT))
                 ));

diff --git a/...-app-fabric/src/main/java/io/cdap/cdap/internal/app/worker/TaskWorkerServiceLauncher.java b/...-app-fabric/src/main/java/io/cdap/cdap/internal/app/worker/TaskWorkerServiceLauncher.java
@@ -22,7 +22,6 @@
 import io.cdap.cdap.api.feature.FeatureFlagsProvider;
 import io.cdap.cdap.common.conf.CConfiguration;
 import io.cdap.cdap.common.conf.Constants;
-import io.cdap.cdap.common.conf.Constants.ArtifactLocalizer;
 import io.cdap.cdap.common.conf.Constants.TaskWorker;
 import io.cdap.cdap.common.feature.DefaultFeatureFlagsProvider;
 import io.cdap.cdap.common.utils.DirUtils;
@@ -195,7 +194,7 @@ public void run() {
             String localhost = InetAddress.getLoopbackAddress().getHostName();
             twillPreparer = twillPreparer.withEnv(TaskWorkerTwillRunnable.class.getSimpleName(),
                 ImmutableMap.of(
-                    ArtifactLocalizer.GCE_METADATA_HOST_ENV_VAR,
+                    TaskWorker.GCE_METADATA_HOST_ENV_VAR,
                     String.format("%s:%s", localhost,
                         cConf.getInt(Constants.ArtifactLocalizer.PORT))
                 ));

diff --git a/...bric/src/main/java/io/cdap/cdap/internal/app/worker/sidecar/ArtifactLocalizerService.java b/...bric/src/main/java/io/cdap/cdap/internal/app/worker/sidecar/ArtifactLocalizerService.java
@@ -22,6 +22,7 @@
 import io.cdap.cdap.common.conf.CConfiguration;
 import io.cdap.cdap.common.conf.Constants;
 import io.cdap.cdap.common.http.CommonNettyHttpServiceFactory;
+import io.cdap.cdap.common.internal.remote.RemoteClientFactory;
 import io.cdap.http.NettyHttpService;
 import java.net.InetAddress;
 import java.nio.file.Paths;
@@ -50,7 +51,8 @@ public class ArtifactLocalizerService extends AbstractIdleService {
   @Inject
   ArtifactLocalizerService(CConfiguration cConf,
       ArtifactLocalizer artifactLocalizer,
-      CommonNettyHttpServiceFactory commonNettyHttpServiceFactory) {
+      CommonNettyHttpServiceFactory commonNettyHttpServiceFactory,
+      RemoteClientFactory remoteClientFactory) {
     this.cConf = cConf;
     this.artifactLocalizer = artifactLocalizer;
     this.httpService = commonNettyHttpServiceFactory.builder(Constants.Service.TASK_WORKER)
@@ -59,7 +61,7 @@ public class ArtifactLocalizerService extends AbstractIdleService {
         .setBossThreadPoolSize(cConf.getInt(Constants.ArtifactLocalizer.BOSS_THREADS))
         .setWorkerThreadPoolSize(cConf.getInt(Constants.ArtifactLocalizer.WORKER_THREADS))
         .setHttpHandlers(new ArtifactLocalizerHttpHandlerInternal(artifactLocalizer),
-            new GcpMetadataHttpHandlerInternal(cConf))
+            new GcpMetadataHttpHandlerInternal(cConf, remoteClientFactory))
         .build();
 
     this.cacheCleanupInterval = cConf.getInt(

diff --git a/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/worker/sidecar/GCPTokenResponse.java b/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/worker/sidecar/GCPTokenResponse.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright © 2023 Cask Data, Inc.
+ *
+ * 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.
+ */
+
+package io.cdap.cdap.internal.app.worker.sidecar;
+
+import com.google.gson.annotations.SerializedName;
+
+/**
+ * Serializable GCP Token Response.
+ */
+public class GCPTokenResponse {
+  @SerializedName("access_token")
+  public final String accessToken;
+  @SerializedName("expires_in")
+  public final long expiresInSecs;
+  @SerializedName("token_type")
+  public final String tokenType;
+
+  /**
+   * Creates a {@link GCPTokenResponse}.
+   *
+   * @param accessToken The access token
+   * @param expiresInSecs
+   * @param tokenType
+   */
+  public GCPTokenResponse(String tokenType, String accessToken, long expiresInSecs) {
+    this.accessToken = accessToken;
+    this.expiresInSecs = expiresInSecs;
+    this.tokenType = tokenType;
+  }
+
+}
diff --git a/...rc/main/java/io/cdap/cdap/internal/app/worker/sidecar/GcpMetadataHttpHandlerInternal.java b/...rc/main/java/io/cdap/cdap/internal/app/worker/sidecar/GcpMetadataHttpHandlerInternal.java
@@ -20,14 +20,23 @@
 import com.google.gson.GsonBuilder;
 import com.google.gson.JsonSyntaxException;
 import com.google.inject.Singleton;
+import io.cdap.cdap.api.common.HttpErrorStatusProvider;
+import io.cdap.cdap.api.retry.Idempotency;
 import io.cdap.cdap.common.BadRequestException;
 import io.cdap.cdap.common.ForbiddenException;
 import io.cdap.cdap.common.conf.CConfiguration;
 import io.cdap.cdap.common.conf.Constants;
+import io.cdap.cdap.common.internal.remote.RemoteClient;
+import io.cdap.cdap.common.internal.remote.RemoteClientFactory;
+import io.cdap.cdap.common.service.Retries;
+import io.cdap.cdap.common.service.RetryStrategies;
 import io.cdap.cdap.gateway.handlers.util.AbstractAppFabricHttpHandler;
 import io.cdap.cdap.proto.BasicThrowable;
 import io.cdap.cdap.proto.codec.BasicThrowableCodec;
+import io.cdap.cdap.proto.credential.NotFoundException;
+import io.cdap.cdap.proto.credential.ProvisionedCredential;
 import io.cdap.cdap.proto.security.GcpMetadataTaskContext;
+import io.cdap.common.http.HttpMethod;
 import io.cdap.common.http.HttpRequests;
 import io.cdap.common.http.HttpResponse;
 import io.cdap.http.HttpHandler;
@@ -36,13 +45,15 @@
 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.net.URL;
+import java.time.Duration;
+import java.time.Instant;
 import javax.ws.rs.DELETE;
 import javax.ws.rs.GET;
 import javax.ws.rs.PUT;
 import javax.ws.rs.Path;
 import javax.ws.rs.QueryParam;
-import javax.ws.rs.core.HttpHeaders;
 import joptsimple.internal.Strings;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -54,24 +65,30 @@
 @Path("/")
 public class GcpMetadataHttpHandlerInternal extends AbstractAppFabricHttpHandler {
 
+  public static final String GCP_CREDENTIAL_IDENTITY_NAME = "default";
   protected static final String METADATA_FLAVOR_HEADER_KEY = "Metadata-Flavor";
   protected static final String METADATA_FLAVOR_HEADER_VALUE = "Google";
   private static final Logger LOG = LoggerFactory.getLogger(GcpMetadataHttpHandlerInternal.class);
   private static final Gson GSON = new GsonBuilder().registerTypeAdapter(BasicThrowable.class,
       new BasicThrowableCodec()).create();
   private final CConfiguration cConf;
-  private final String metadataServiceEndpoint;
+  private final String metadataServiceTokenEndpoint;
   private GcpMetadataTaskContext gcpMetadataTaskContext;
+  private final RemoteClient remoteClient;
 
   /**
    * Constructs the {@link GcpMetadataHttpHandlerInternal}.
    *
    * @param cConf CConfiguration
    */
-  public GcpMetadataHttpHandlerInternal(CConfiguration cConf) {
+  public GcpMetadataHttpHandlerInternal(CConfiguration cConf,
+      RemoteClientFactory remoteClientFactory) {
     this.cConf = cConf;
-    this.metadataServiceEndpoint = cConf.get(
+    this.metadataServiceTokenEndpoint = cConf.get(
         Constants.TaskWorker.METADATA_SERVICE_END_POINT);
+    this.remoteClient = remoteClientFactory.createRemoteClient(Constants.Service.APP_FABRIC_HTTP,
+        RemoteClientFactory.NO_VERIFY_HTTP_REQUEST_CONFIG,
+        Constants.Gateway.INTERNAL_API_VERSION_3);
   }
 
   /**
@@ -109,11 +126,6 @@ public void status(HttpRequest request, HttpResponder responder) throws Exceptio
   public void token(HttpRequest request, HttpResponder responder,
       @QueryParam("scopes") String scopes) throws Exception {
 
-    if (gcpMetadataTaskContext != null && gcpMetadataTaskContext.getNamespace() != null) {
-      LOG.trace("Token requested for namespace: {}", gcpMetadataTaskContext.getNamespace());
-    } else {
-      LOG.trace("Token requested but namespace not set");
-    }
     // check that metadata header is present in the request.
     if (!request.headers().contains(METADATA_FLAVOR_HEADER_KEY,
         METADATA_FLAVOR_HEADER_VALUE, true)) {
@@ -123,29 +135,75 @@ public void token(HttpRequest request, HttpResponder responder,
               METADATA_FLAVOR_HEADER_KEY, METADATA_FLAVOR_HEADER_VALUE));
     }
 
-    // TODO: CDAP-20750
-    if (metadataServiceEndpoint == null) {
+    if (gcpMetadataTaskContext == null) {
+      LOG.trace("Token requested but context not set");
+    } else {
+      try {
+        // fetch token from credential provider
+        GCPTokenResponse gcpTokenResponse =
+            Retries.callWithRetries(this::fetchTokenFromCredentialProvider,
+                RetryStrategies.fromConfiguration(cConf, Constants.Service.TASK_WORKER + "."));
+        responder.sendJson(HttpResponseStatus.OK, GSON.toJson(gcpTokenResponse));
+      } catch (NotFoundException e) {
+        // if credential identity not found,
+        // fallback to gcp metadata server for backward compatibility.
+      } catch (Exception ex) {
+        if (ex instanceof HttpErrorStatusProvider) {
+          HttpResponseStatus status = HttpResponseStatus.valueOf(
+              ((HttpErrorStatusProvider) ex).getStatusCode());
+          responder.sendJson(status, exceptionToJson(ex));
+        } else {
+          responder.sendJson(HttpResponseStatus.INTERNAL_SERVER_ERROR, exceptionToJson(ex));
+        }
+      }
+      return;
+    }
+
+    if (metadataServiceTokenEndpoint == null) {
       responder.sendString(HttpResponseStatus.NOT_IMPLEMENTED,
           String.format("%s has not been set",
               Constants.TaskWorker.METADATA_SERVICE_END_POINT));
       return;
     }
 
     try {
-      URL url = new URL(metadataServiceEndpoint);
-      if (!Strings.isNullOrEmpty(scopes)) {
-        url = new URL(String.format("%s?scopes=%s", metadataServiceEndpoint, scopes));
-      }
-      io.cdap.common.http.HttpRequest tokenRequest = io.cdap.common.http.HttpRequest.get(url)
-          .addHeader(METADATA_FLAVOR_HEADER_KEY, METADATA_FLAVOR_HEADER_VALUE)
-          .build();
-      HttpResponse tokenResponse = HttpRequests.execute(tokenRequest);
-      responder.sendJson(HttpResponseStatus.OK, tokenResponse.getResponseBodyAsString());
+      responder.sendJson(HttpResponseStatus.OK,
+          fetchTokenFromMetadataServer(scopes).getResponseBodyAsString());
     } catch (Exception ex) {
       LOG.warn("Failed to fetch token from metadata service", ex);
-      responder.sendString(HttpResponseStatus.INTERNAL_SERVER_ERROR, exceptionToJson(ex),
-          new DefaultHttpHeaders().set(HttpHeaders.CONTENT_TYPE, "application/json"));
+      responder.sendJson(HttpResponseStatus.INTERNAL_SERVER_ERROR, exceptionToJson(ex));
+    }
+  }
+
+  private GCPTokenResponse fetchTokenFromCredentialProvider() throws NotFoundException,
+      IOException {
+    String url = String.format("namespaces/%s/credentials/identities/%s/provision",
+        gcpMetadataTaskContext.getNamespace(),
+        GCP_CREDENTIAL_IDENTITY_NAME);
+    io.cdap.common.http.HttpRequest tokenRequest =
+        remoteClient.requestBuilder(HttpMethod.GET, url).build();
+    HttpResponse response = remoteClient.execute(tokenRequest, Idempotency.NONE);
+
+    if (response.getResponseCode() == HttpResponseStatus.NOT_FOUND.code()) {
+      throw new NotFoundException(String.format("Credential Identity %s Not Found.",
+          GCP_CREDENTIAL_IDENTITY_NAME));
+    }
+
+    ProvisionedCredential provisionedCredential =
+        GSON.fromJson(response.getResponseBodyAsString(), ProvisionedCredential.class);
+    return new GCPTokenResponse("Bearer", provisionedCredential.get(),
+        Duration.between(Instant.now(), provisionedCredential.getExpiration()).getSeconds());
+  }
+
+  private HttpResponse fetchTokenFromMetadataServer(String scopes) throws IOException {
+    URL url = new URL(metadataServiceTokenEndpoint);
+    if (!Strings.isNullOrEmpty(scopes)) {
+      url = new URL(String.format("%s?scopes=%s", metadataServiceTokenEndpoint, scopes));
     }
+    io.cdap.common.http.HttpRequest tokenRequest = io.cdap.common.http.HttpRequest.get(url)
+        .addHeader(METADATA_FLAVOR_HEADER_KEY, METADATA_FLAVOR_HEADER_VALUE)
+        .build();
+    return HttpRequests.execute(tokenRequest);
   }
 
   /**
@@ -174,7 +232,7 @@ public void setContext(FullHttpRequest request, HttpResponder responder)
   @Path("/clear-context")
   public void clearContext(HttpRequest request, HttpResponder responder) {
     this.gcpMetadataTaskContext = null;
-    LOG.debug("Context cleared.");
+    LOG.trace("Context cleared.");
     responder.sendStatus(HttpResponseStatus.OK);
   }
 

diff --git a/...bric/src/main/java/io/cdap/cdap/internal/credential/DefaultCredentialProviderService.java b/...bric/src/main/java/io/cdap/cdap/internal/credential/DefaultCredentialProviderService.java
@@ -27,7 +27,6 @@
 import io.cdap.cdap.proto.credential.ProvisionedCredential;
 import io.cdap.cdap.proto.id.CredentialIdentityId;
 import io.cdap.cdap.proto.id.CredentialProfileId;
-import io.cdap.cdap.proto.security.StandardPermission;
 import io.cdap.cdap.security.spi.authorization.ContextAccessEnforcer;
 import io.cdap.cdap.security.spi.credential.CredentialProvider;
 import java.io.IOException;
@@ -91,11 +90,11 @@ public ProvisionedCredential provision(NamespaceMeta namespaceMeta, String ident
       throws CredentialProvisioningException, IOException, NotFoundException {
     CredentialIdentityId identityId =
         new CredentialIdentityId(namespaceMeta.getName(), identityName);
-    contextAccessEnforcer.enforce(identityId, StandardPermission.USE);
+    // contextAccessEnforcer.enforce(identityId, StandardPermission.USE);
     Optional<CredentialIdentity> optIdentity = credentialIdentityManager.get(identityId);
     if (!optIdentity.isPresent()) {
       throw new NotFoundException(String.format("Credential identity '%s' was not found.",
-          identityId.toString()));
+          identityId));
     }
     CredentialIdentity identity = optIdentity.get();
     return validateAndProvisionIdentity(namespaceMeta, identity);
@@ -125,7 +124,7 @@ private ProvisionedCredential validateAndProvisionIdentity(NamespaceMeta namespa
       throws CredentialProvisioningException, IOException, NotFoundException {
     CredentialProfileId profileId = new CredentialProfileId(identity.getProfileNamespace(),
         identity.getProfileName());
-    contextAccessEnforcer.enforce(profileId, StandardPermission.USE);
+    // contextAccessEnforcer.enforce(profileId, StandardPermission.USE);
     Optional<CredentialProfile> optProfile = credentialProfileManager.get(profileId);
     if (!optProfile.isPresent()) {
       throw new NotFoundException(

diff --git a/.../java/io/cdap/cdap/internal/credential/handler/CredentialProviderHttpHandlerInternal.java b/.../java/io/cdap/cdap/internal/credential/handler/CredentialProviderHttpHandlerInternal.java
@@ -22,7 +22,9 @@
 import io.cdap.cdap.common.NotFoundException;
 import io.cdap.cdap.common.conf.Constants;
 import io.cdap.cdap.common.namespace.NamespaceQueryAdmin;
+import io.cdap.cdap.proto.BasicThrowable;
 import io.cdap.cdap.proto.NamespaceMeta;
+import io.cdap.cdap.proto.codec.BasicThrowableCodec;
 import io.cdap.cdap.proto.credential.CredentialProvider;
 import io.cdap.cdap.proto.credential.CredentialProvisioningException;
 import io.cdap.cdap.proto.id.NamespaceId;
@@ -32,9 +34,8 @@
 import io.netty.handler.codec.http.HttpRequest;
 import io.netty.handler.codec.http.HttpResponseStatus;
 import java.io.IOException;
-import java.time.Instant;
 import javax.inject.Inject;
-import javax.ws.rs.POST;
+import javax.ws.rs.GET;
 import javax.ws.rs.Path;
 import javax.ws.rs.PathParam;
 
@@ -45,8 +46,8 @@
 @Path(Constants.Gateway.INTERNAL_API_VERSION_3)
 public class CredentialProviderHttpHandlerInternal extends AbstractHttpHandler {
 
-  private static final Gson GSON = new GsonBuilder().registerTypeAdapter(Instant.class,
-      new InstantEpochSecondsTypeAdapter()).create();
+  private static final Gson GSON = new GsonBuilder().registerTypeAdapter(
+      BasicThrowable.class, new BasicThrowableCodec()).create();
 
   private final CredentialProvider credentialProvider;
   private final NamespaceQueryAdmin namespaceQueryAdmin;
@@ -69,7 +70,7 @@ public class CredentialProviderHttpHandlerInternal extends AbstractHttpHandler {
    * @throws IOException                     If transport errors occur.
    * @throws NotFoundException               If the identity or associated profile are not found.
    */
-  @POST
+  @GET
   @Path("/namespaces/{namespace-id}/credentials/identities/{identity-name}/provision")
   public void provisionCredential(HttpRequest request, HttpResponder responder,
       @PathParam("namespace-id") String namespace, @PathParam("identity-name") String identityName)