[BEAM-582] Allow usage of the new GCP service account JSON key

sdks/java/core/src/main/java/org/apache/beam/sdk/util/Credentials.java

@@ -33,16 +33,19 @@
 import com.google.api.client.json.jackson2.JacksonFactory;
 import com.google.api.client.util.store.FileDataStoreFactory;
 import java.io.File;
+import java.io.FileInputStream;
 import java.io.FileReader;
 import java.io.IOException;
 import java.security.GeneralSecurityException;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
+import javax.annotation.Nullable;
 import org.apache.beam.sdk.options.GcpOptions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+
 /**
  * Provides support for loading credentials.
  */
@@ -105,7 +108,10 @@ public static Credential getCredential(GcpOptions options)
     String keyFile = options.getServiceAccountKeyfile();
     String accountName = options.getServiceAccountName();
 
-    if (keyFile != null && accountName != null) {
+    if (keyFile != null || accountName != null) {
+      if (keyFile == null) {
+        throw new IOException("If accountName is given, also supply a keyFile");
+      }
       try {
         return getCredentialFromFile(keyFile, accountName, SCOPES);
       } catch (GeneralSecurityException e) {
@@ -132,15 +138,29 @@ public static Credential getCredential(GcpOptions options)
    * Loads OAuth2 credential from a local file.
    */
   private static Credential getCredentialFromFile(
-      String keyFile, String accountId, Collection<String> scopes)
+          String keyFile, @Nullable String accountName, Collection<String> scopes)
       throws IOException, GeneralSecurityException {
-    GoogleCredential credential = new GoogleCredential.Builder()
+
+    GoogleCredential credential;
+    if (keyFile.toLowerCase().endsWith("json")) {
+      if (accountName != null) {
+        throw new IOException("Only use an accountName with legacy P12 key files");
+      }
+      credential = GoogleCredential.fromStream(new FileInputStream(keyFile))
+        .createScoped(SCOPES);
+    } else {
+      if (accountName == null) {
+        throw new IOException("You need an accountName with P12 key files "
+          + "or use preferred JSON key files");
+      }
+      credential = new GoogleCredential.Builder()
         .setTransport(Transport.getTransport())
         .setJsonFactory(Transport.getJsonFactory())
-        .setServiceAccountId(accountId)
+        .setServiceAccountId(accountName)
         .setServiceAccountScopes(scopes)
         .setServiceAccountPrivateKeyFromP12File(new File(keyFile))
         .build();
+    }
 
     LOG.info("Created credential from file {}", keyFile);
     return credential;

sdks/java/core/src/test/java/org/apache/beam/sdk/util/GcpCredentialFactoryTest.java

@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.beam.sdk.util;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+
+import org.apache.beam.sdk.options.GcpOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.hamcrest.core.StringContains;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link GcpCredentialFactory}. */
+@RunWith(JUnit4.class)
+public class GcpCredentialFactoryTest {
+
+  @Rule
+  public ExpectedException thrown = ExpectedException.none();
+
+  @Test
+  public void testP12KeyFileNeedsAccountName() throws Exception {
+    GcpOptions options = PipelineOptionsFactory.as(GcpOptions.class);
+    options.setServiceAccountKeyfile("key.p12");
+
+    thrown.expect(IOException.class);
+    thrown.expectMessage(new StringContains("You need an accountName"));
+    GcpCredentialFactory.fromOptions(options).getCredential();
+  }
+
+  @Test
+  public void testJSONKeyFileDoesntAllowAccountName() throws Exception {
+    GcpOptions options = PipelineOptionsFactory.as(GcpOptions.class);
+    options.setServiceAccountKeyfile("key.json");
+    options.setServiceAccountName("test@gcloud");
+
+    thrown.expect(IOException.class);
+    thrown.expectMessage(new StringContains("Only use an accountName"));
+    GcpCredentialFactory.fromOptions(options).getCredential();
+  }
+
+  @Test
+  public void testAccountNameWithoutKeyFile() throws Exception {
+    GcpOptions options = PipelineOptionsFactory.as(GcpOptions.class);
+    options.setServiceAccountName("test@gcloud");
+
+    thrown.expect(IOException.class);
+    thrown.expectMessage(new StringContains("also supply a keyFile"));
+    GcpCredentialFactory.fromOptions(options).getCredential();
+  }
+
+  @Test
+  public void testCorrectP12() throws Exception {
+    GcpOptions options = PipelineOptionsFactory.as(GcpOptions.class);
+    options.setServiceAccountKeyfile("key.p12");
+    options.setServiceAccountName("test@gcloud");
+
+    thrown.expect(FileNotFoundException.class);
+    thrown.expectMessage(new StringContains("No such file or directory"));
+    GcpCredentialFactory.fromOptions(options).getCredential();
+  }
+
+  @Test
+  public void testCorrectJSON() throws Exception {
+    GcpOptions options = PipelineOptionsFactory.as(GcpOptions.class);
+    options.setServiceAccountKeyfile("key.json");
+
+    thrown.expect(FileNotFoundException.class);
+    thrown.expectMessage(new StringContains("No such file or directory"));
+    GcpCredentialFactory.fromOptions(options).getCredential();
+  }
+}