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

sdk/src/main/java/com/google/cloud/dataflow/sdk/util/Credentials.java

@@ -36,12 +36,14 @@
 import org.slf4j.LoggerFactory;
 
 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;
 
 /**
  * Provides support for loading credentials.
@@ -105,7 +107,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 +137,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;

sdk/src/test/java/com/google/cloud/dataflow/sdk/util/GcpCredentialFactoryTest.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright (C) 2015 Google 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 com.google.cloud.dataflow.sdk.util;
+
+import com.google.cloud.dataflow.sdk.options.GcpOptions;
+import com.google.cloud.dataflow.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;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+
+/** 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();
+  }
+}