gax: Using DirectPath fails with NPE when using NoCredentialsProvider to support Downscoped tokens

[frankyn]

@mohanli-ml can you help with this issue?
cc: @danielduhh , @sydney-munro , and @arunkumarchacko

Environment details

2. OS type and version:
3. Java version:
4. artifact version(s): com.google.api:gax-grpc:jar:2.39.0, com.google.cloud:google-cloud-storage:2.31.0

Steps to reproduce

Run the following example in a GCE instance:

package org.example;

import com.google.cloud.NoCredentials;
import com.google.cloud.storage.Storage;
import com.google.cloud.storage.StorageOptions;

public class FailureExample {
    public static void main(String[] args) throws Exception {
        // GCS client converts NoCredentials type to NoCredentialsProvider 
        // and hands that over to underlying Storage v2 GAPIC
        // https://github.com/googleapis/java-storage/blob/main/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java#L175-L177
        Storage storage = StorageOptions.grpc()
                .setAttemptDirectPath(true)
                .setCredentials(NoCredentials.getInstance())
                .build().getService();
    }
}

Stack trace

Exception in thread "main" java.lang.NullPointerException: creds
        at com.google.common.base.Preconditions.checkNotNull(Preconditions.java:921)
        at io.grpc.auth.GoogleAuthLibraryCallCredentials.<init>(GoogleAuthLibraryCallCredentials.java:74)
        at io.grpc.auth.GoogleAuthLibraryCallCredentials.<init>(GoogleAuthLibraryCallCredentials.java:69)
        at io.grpc.auth.MoreCallCredentials.from(MoreCallCredentials.java:35)
        at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.createSingleChannel(InstantiatingGrpcChannelProvider.java:371)
        at com.google.api.gax.grpc.ChannelPool.<init>(ChannelPool.java:107)
        at com.google.api.gax.grpc.ChannelPool.create(ChannelPool.java:85)
        at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.createChannel(InstantiatingGrpcChannelProvider.java:243)
        at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.getTransportChannel(InstantiatingGrpcChannelProvider.java:237)
        at com.google.api.gax.rpc.ClientContext.create(ClientContext.java:226)
        at com.google.storage.v2.stub.GrpcStorageStub.create(GrpcStorageStub.java:535)
        at com.google.storage.v2.stub.StorageStubSettings.createStub(StorageStubSettings.java:622)
        at com.google.storage.v2.StorageClient.<init>(StorageClient.java:166)
        at com.google.storage.v2.StorageClient.create(StorageClient.java:149)
        at com.google.cloud.storage.GrpcStorageOptions$GrpcStorageFactory.create(GrpcStorageOptions.java:663)
        at com.google.cloud.storage.GrpcStorageOptions$GrpcStorageFactory.create(GrpcStorageOptions.java:627)
        at com.google.cloud.ServiceOptions.getService(ServiceOptions.java:563)
        at org.example.ArunFailureSimple.main(ArunFailureSimple.java:28)

Potential issue

Digging into the stacktrace, I found that in InstantiatingGrpcChannelProvider.java, there's logic to handle null creds when not using DirectPath. However, when using DirectPath, no handling for null creds exists and is passed along as-is which I think is causing the NPE.

E2E Code example

Following is Downscoped Token example that reads data from GCS using DirectPath:

package org.example;
import com.google.auth.Credentials;
import static com.google.auth.http.AuthHttpConstants.AUTHORIZATION;
import com.google.auth.oauth2.AccessToken;
import com.google.auth.oauth2.CredentialAccessBoundary;
import com.google.auth.oauth2.DownscopedCredentials;
import com.google.auth.oauth2.GoogleCredentials;
import com.google.cloud.NoCredentials;
import com.google.cloud.storage.*;
import com.google.common.collect.ImmutableList;
import com.google.storage.v2.BucketName;
import com.google.storage.v2.GetObjectRequest;
import io.grpc.CallOptions;
import io.grpc.Channel;
import io.grpc.ClientCall;
import io.grpc.ClientInterceptor;
import io.grpc.ForwardingClientCall;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;

import java.io.IOException;

public class FailureWithCABToken {
    private static AccessToken getTokenFromBroker(String bucketName, String objectPrefix)
            throws IOException {
        // Retrieve the source credentials from ADC.
        GoogleCredentials sourceCredentials =
                GoogleCredentials.getApplicationDefault()
                        .createScoped("https://www.googleapis.com/auth/cloud-platform");
        // Initialize the Credential Access Boundary rules.
        String availableResource = "//storage.googleapis.com/projects/_/buckets/" + bucketName;
        // Downscoped credentials will have readonly access to the resource.
        String availablePermission = "inRole:roles/storage.objectViewer";
        // Only objects starting with the specified prefix string in the object name will be allowed
        // read access.
        String expression =
                "resource.name.startsWith('projects/_/buckets/"
                        + bucketName
                        + "/objects/"
                        + objectPrefix
                        + "')";
        // Build the AvailabilityCondition.
        CredentialAccessBoundary.AccessBoundaryRule.AvailabilityCondition availabilityCondition =
                CredentialAccessBoundary.AccessBoundaryRule.AvailabilityCondition.newBuilder()
                        .setExpression(expression)
                        .build();
        // Define the single access boundary rule using the above properties.
        CredentialAccessBoundary.AccessBoundaryRule rule =
                CredentialAccessBoundary.AccessBoundaryRule.newBuilder()
                        .setAvailableResource(availableResource)
                        .addAvailablePermission(availablePermission)
                        .setAvailabilityCondition(availabilityCondition)
                        .build();
        // Define the Credential Access Boundary with all the relevant rules.
        CredentialAccessBoundary credentialAccessBoundary =
                CredentialAccessBoundary.newBuilder().addRule(rule).build();
        // Create the downscoped credentials.
        DownscopedCredentials downscopedCredentials =
                DownscopedCredentials.newBuilder()
                        .setSourceCredential(sourceCredentials)
                        .setCredentialAccessBoundary(credentialAccessBoundary)
                        .build();
        // Retrieve the token.
        // This will need to be passed to the Token Consumer.
        AccessToken accessToken = downscopedCredentials.refreshAccessToken();
        return accessToken;
    }
    private static class DownscopedTokenByRequestInterceptor implements ClientInterceptor {
        public final Metadata.Key<String> AUTH_KEY =
                Metadata.Key.of(AUTHORIZATION, Metadata.ASCII_STRING_MARSHALLER);
        @Override
        public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
            if (!method.getFullMethodName().equals("google.storage.v2.Storage/GetObject")) {
                // Only support PCU based operations
                return next.newCall(method, callOptions);
            }
            return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
                Listener responseListener;
                Metadata headers;
                int flowControlRequests;
                String methodName;
                @Override
                public void start(Listener<RespT> responseListener, Metadata headers) {
                    this.responseListener = responseListener;
                    this.headers = headers;
                    this.methodName = method.getFullMethodName();
                }
                @Override
                public void sendMessage(ReqT message) {
                    if (headers != null) { // start() is required before sendMessage()
                        try {
                            GetObjectRequest req = (GetObjectRequest)message;
                            String bucketName = BucketName.parse(req.getBucket()).getBucket();
                            String token = getTokenFromBroker(bucketName, req.getObject()).getTokenValue();
                            headers.put(AUTH_KEY, "Bearer " + token);
                        } catch (Exception e) {
                            halfClose();
                        }
                        delegate().start(responseListener, headers);
                        if (flowControlRequests != 0) {
                            super.request(flowControlRequests);
                        }
                        headers = null;
                    }
                    super.sendMessage(message);
                }
                @Override
                public void request(int numMessages) {
                    if (headers != null) {
                        this.flowControlRequests += numMessages;
                    } else {
                        super.request(numMessages);
                    }
                }
            };
        }
    }

    private static Credentials getNoCredentialsWorkaround() {
        GoogleCredentials theCred = GoogleCredentials.create(new AccessToken("", null));
        return theCred;
    }

    public static void main(String[] args) throws Exception {
        Storage storage = StorageOptions.grpc()
                .setAttemptDirectPath(true)
                .setGrpcInterceptorProvider(() -> ImmutableList.of(new DownscopedTokenByRequestInterceptor()))
                .setCredentials(NoCredentials.getInstance())
                .build().getService();
        Blob blob = storage.get("bucket-name", "object-name");
        System.out.println("Downloaded blob?: " + (blob != null));
    }
}

Workaround

package org.example;

import com.google.cloud.storage.Storage;
import com.google.cloud.storage.StorageOptions;

public class ArunFailureSimple {
    public static void main(String[] args) throws Exception {
        Storage storage = StorageOptions.grpc()
                .setAttemptDirectPath(true)
                .setCredentials(GoogleCredentials.create(new AccessToken("", null)))
                .build().getService();
    }
}

[mohanli-ml]

Hi Frank, we recently fixed this bug that a call credential will be required to build a DirectPath channel, and the code is just merged this Monday, #2281. With this fix I think when you use NoCredentials, the code will use CloudPath instead of reporting exception.

[frankyn]

Thanks @mohanli-ml, IIUC the expectation that NoCredentials should default to CloudPath.
However, that means that using Downscoped Tokens over DirectPath would not work with NoCredentails.

Is there an option to use NoCredentials and still support DirectPath or would it need to be a new Credential type alltogether?

[mohanli-ml]

Hi Frank, can you give more context about why you want to use NoCredentials in this use case? And if NoCredentials is used, it means RPCs do not have call credentials. Is this understanding correct?

cc @blakeli0

[frankyn]

The credentials are currently coming in through an interceptor that's passed into client at init, example below.

Use Case

Customer wants to use a CAB token per RPC type when using Dataproc hadoop connector, connector calls storage client that has configured an interceptor to provide a token per RPC. To prevent accidental usage of ADC token for uninstrumented RPCs, NoCredentials is passed in at client init as well. Subsequent requests out of java-storage use CAB token when instrumented otherwise if not instrumented use no token.

E2E Sample Code

package org.example;
import com.google.auth.Credentials;
import static com.google.auth.http.AuthHttpConstants.AUTHORIZATION;
import com.google.auth.oauth2.AccessToken;
import com.google.auth.oauth2.CredentialAccessBoundary;
import com.google.auth.oauth2.DownscopedCredentials;
import com.google.auth.oauth2.GoogleCredentials;
import com.google.cloud.NoCredentials;
import com.google.cloud.storage.*;
import com.google.common.collect.ImmutableList;
import com.google.storage.v2.BucketName;
import com.google.storage.v2.GetObjectRequest;
import io.grpc.CallOptions;
import io.grpc.Channel;
import io.grpc.ClientCall;
import io.grpc.ClientInterceptor;
import io.grpc.ForwardingClientCall;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;

import java.io.IOException;

public class FailureWithCABToken {
    private static AccessToken getTokenFromBroker(String bucketName, String objectPrefix)
            throws IOException {
        // Retrieve the source credentials from ADC.
        GoogleCredentials sourceCredentials =
                GoogleCredentials.getApplicationDefault()
                        .createScoped("https://www.googleapis.com/auth/cloud-platform");
        // Initialize the Credential Access Boundary rules.
        String availableResource = "//storage.googleapis.com/projects/_/buckets/" + bucketName;
        // Downscoped credentials will have readonly access to the resource.
        String availablePermission = "inRole:roles/storage.objectViewer";
        // Only objects starting with the specified prefix string in the object name will be allowed
        // read access.
        String expression =
                "resource.name.startsWith('projects/_/buckets/"
                        + bucketName
                        + "/objects/"
                        + objectPrefix
                        + "')";
        // Build the AvailabilityCondition.
        CredentialAccessBoundary.AccessBoundaryRule.AvailabilityCondition availabilityCondition =
                CredentialAccessBoundary.AccessBoundaryRule.AvailabilityCondition.newBuilder()
                        .setExpression(expression)
                        .build();
        // Define the single access boundary rule using the above properties.
        CredentialAccessBoundary.AccessBoundaryRule rule =
                CredentialAccessBoundary.AccessBoundaryRule.newBuilder()
                        .setAvailableResource(availableResource)
                        .addAvailablePermission(availablePermission)
                        .setAvailabilityCondition(availabilityCondition)
                        .build();
        // Define the Credential Access Boundary with all the relevant rules.
        CredentialAccessBoundary credentialAccessBoundary =
                CredentialAccessBoundary.newBuilder().addRule(rule).build();
        // Create the downscoped credentials.
        DownscopedCredentials downscopedCredentials =
                DownscopedCredentials.newBuilder()
                        .setSourceCredential(sourceCredentials)
                        .setCredentialAccessBoundary(credentialAccessBoundary)
                        .build();
        // Retrieve the token.
        // This will need to be passed to the Token Consumer.
        AccessToken accessToken = downscopedCredentials.refreshAccessToken();
        return accessToken;
    }
    private static class DownscopedTokenByRequestInterceptor implements ClientInterceptor {
        public final Metadata.Key<String> AUTH_KEY =
                Metadata.Key.of(AUTHORIZATION, Metadata.ASCII_STRING_MARSHALLER);
        @Override
        public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
            if (!method.getFullMethodName().equals("google.storage.v2.Storage/GetObject")) {
                // Only support PCU based operations
                return next.newCall(method, callOptions);
            }
            return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
                Listener responseListener;
                Metadata headers;
                int flowControlRequests;
                String methodName;
                @Override
                public void start(Listener<RespT> responseListener, Metadata headers) {
                    this.responseListener = responseListener;
                    this.headers = headers;
                    this.methodName = method.getFullMethodName();
                }
                @Override
                public void sendMessage(ReqT message) {
                    if (headers != null) { // start() is required before sendMessage()
                        try {
                            GetObjectRequest req = (GetObjectRequest)message;
                            String bucketName = BucketName.parse(req.getBucket()).getBucket();
                            String token = getTokenFromBroker(bucketName, req.getObject()).getTokenValue();
                            headers.put(AUTH_KEY, "Bearer " + token);
                        } catch (Exception e) {
                            halfClose();
                        }
                        delegate().start(responseListener, headers);
                        if (flowControlRequests != 0) {
                            super.request(flowControlRequests);
                        }
                        headers = null;
                    }
                    super.sendMessage(message);
                }
                @Override
                public void request(int numMessages) {
                    if (headers != null) {
                        this.flowControlRequests += numMessages;
                    } else {
                        super.request(numMessages);
                    }
                }
            };
        }
    }

    private static Credentials getNoCredentialsWorkaround() {
        GoogleCredentials theCred = GoogleCredentials.create(new AccessToken("", null));
        return theCred;
    }

    public static void main(String[] args) throws Exception {
        Storage storage = StorageOptions.grpc()
                .setAttemptDirectPath(true)
                .setGrpcInterceptorProvider(() -> ImmutableList.of(new DownscopedTokenByRequestInterceptor()))
                .setCredentials(NoCredentials.getInstance())
                .build().getService();
        Blob blob = storage.get("bucket-name", "object-name");
        System.out.println("Downloaded blob?: " + (blob != null));
    }
}

[mohanli-ml]

Hi Frank, is it possible to use custom call credential instead of using interceptors?

cc @apolcyn

[blakeli0]

Can they pass FixedCredentialsProvider instead of NoCredentialProvider to StorageOptions? See an example here

[frankyn]

Thanks for the follow-up team!

is it possible to use custom call credential instead of using interceptors?

Assuming you're talking about call context credentials that can be passed in to GAPIC per RPC made. No not right now, the veneer doesn't expose this option to users. We are considering how to support CAB tokens in veneer but it's an unscheduled feature request right now.

Can they pass FixedCredentialsProvider instead of NoCredentialProvider to StorageOptions? See an example here

This is technically the workaround we are doing right now. The following sample code sets credentials to a GoogleCredentials instance which inside java-storage GrpcStorageOptions is wrapped with FixedCredentialsProvider. Are there any cons with using this workaround? We consider a workaround because it create a GoogleCredentials instance that returns no token and has no expiration which could be a feature added to auth iff NoCredentials is expected to use Cloud Path by default.

package org.example;

import com.google.cloud.storage.Storage;
import com.google.cloud.storage.StorageOptions;

public class ArunFailureSimple {
    public static void main(String[] args) throws Exception {
        Storage storage = StorageOptions.grpc()
                .setAttemptDirectPath(true)
                .setCredentials(GoogleCredentials.create(new AccessToken("", null)))
                .build().getService();
    }
}

[blakeli0]

Are there any cons with using this workaround?

This is not ideal, but I think it's fine for now. Currently, if we should enable direct path is decided on channel creation, which is on client level, not on RPC level. And looks like the problem comes down to a normal channel can be created without a credentials object, but a direct path channel has to be created with a credentials object. Not sure this is something we can refactor though @mohanli-ml

[blakeli0]

The fix requires non-trivial work in both gax and grpc-java. Considering that there is an existing workaround, closing it as not planned.