Add support for three-legged OAuth for Google connections. Useful for…
diff --git a/airflow/contrib/hooks/bigquery_hook.py b/airflow/contrib/hooks/bigquery_hook.py
index c8ae39e9aee11..6d410e6415e2c 100644
--- a/airflow/contrib/hooks/bigquery_hook.py
+++ b/airflow/contrib/hooks/bigquery_hook.py
@@ -3,56 +3,43 @@
 import pandas
 import time
 
-from airflow.hooks.base_hook import BaseHook
+from airflow.contrib.hooks.gc_base_hook import GoogleCloudBaseHook
 from apiclient.discovery import build
-from oauth2client.client import SignedJwtAssertionCredentials
 from pandas.io.gbq import GbqConnector, _parse_data as gbq_parse_data
 from pandas.tools.merge import concat
 
 logging.getLogger("bigquery").setLevel(logging.INFO)
 
-BQ_SCOPE = 'https://www.googleapis.com/auth/bigquery'
-
-class BigQueryHook(BaseHook):
+class BigQueryHook(GoogleCloudBaseHook):
     """
-    Interact with BigQuery. Connections must be defined with an extras JSON field containing:
+    Interact with BigQuery. Connections must be defined with an extras JSON 
+    field containing:
 
     {
         "project": "<google project ID>",
         "service_account": "<google service account email>",
         "key_path": "<p12 key path>"
     }
+
+    If you have used ``gcloud auth`` to authenticate on the machine that's 
+    running Airflow, you can exclude the service_account and key_path 
+    parameters.
     """
     conn_name_attr = 'bigquery_conn_id'
 
-    def __init__(self, bigquery_conn_id='bigquery_default'):
-        self.bigquery_conn_id = bigquery_conn_id
+    def __init__(self, scope='https://www.googleapis.com/auth/bigquery', bigquery_conn_id='bigquery_default'):
+        """
+        :param scope: The scope of the hook.
+        :type scope: string
+        """
+        super(BigQueryHook, self).__init__(scope, bigquery_conn_id)
 
     def get_conn(self):
         """
         Returns a BigQuery service object.
         """
-        connection_info = self.get_connection(self.bigquery_conn_id)
-        connection_extras = connection_info.extra_dejson
-        service_account = connection_extras['service_account']
-        key_path = connection_extras['key_path']
-
-        with file(key_path, 'rb') as key_file:
-            key = key_file.read()
-
-        credentials = SignedJwtAssertionCredentials(
-            service_account,
-            key,
-            scope=BQ_SCOPE)
-            # TODO Support domain delegation, which will allow us to set a sub-account to execute as. We can then
-            # pass DAG owner emails into the connection_info, and use it here.
-            # sub='some@email.com')
-
-        http = httplib2.Http()
-        http_authorized = credentials.authorize(http)
-        service = build('bigquery', 'v2', http=http_authorized)
-
-        return service
+        http_authorized = self._authorize()
+        return build('bigquery', 'v2', http=http_authorized)
 
     def get_pandas_df(self, bql, parameters=None):
         """
@@ -63,8 +50,7 @@ def get_pandas_df(self, bql, parameters=None):
         :type bql: string
         """
         service = self.get_conn()
-        connection_info = self.get_connection(self.bigquery_conn_id)
-        connection_extras = connection_info.extra_dejson
+        connection_extras = self._extras_dejson()
         project = connection_extras['project']
         connector = BigQueryPandasConnector(project, service)
         schema, pages = connector.run_query(bql, verbose=False)
@@ -95,8 +81,7 @@ def run(self, bql, destination_dataset_table = False, write_disposition = 'WRITE
         :param write_disposition: What to do if the table already exists in 
             BigQuery.
         """
-        connection_info = self.get_connection(self.bigquery_conn_id)
-        connection_extras = connection_info.extra_dejson
+        connection_extras = self._extras_dejson()
         project = connection_extras['project']
         configuration = {
             'query': {
@@ -145,8 +130,7 @@ def run_extract(self, source_dataset_table, destination_cloud_storage_uris, comp
         assert '.' in source_dataset_table, \
             'Expected source_dataset_table in the format of <dataset>.<table>. Got: {}'.format(source_dataset_table)
 
-        connection_info = self.get_connection(self.bigquery_conn_id)
-        connection_extras = connection_info.extra_dejson
+        connection_extras = self._extras_dejson()
         project = connection_extras['project']
         source_dataset, source_table = source_dataset_table.split('.', 1)
         configuration = {
@@ -179,8 +163,7 @@ def run_with_configuration(self, configuration):
         https://cloud.google.com/bigquery/docs/reference/v2/jobs for details.
         """
         service = self.get_conn()
-        connection_info = self.get_connection(self.bigquery_conn_id)
-        connection_extras = connection_info.extra_dejson
+        connection_extras = self._extras_dejson()
         project = connection_extras['project']
         jobs = service.jobs()
         job_data = {
diff --git a/airflow/contrib/hooks/gc_base_hook.py b/airflow/contrib/hooks/gc_base_hook.py
new file mode 100644
index 0000000000000..75f7f3454bfaf
--- /dev/null
+++ b/airflow/contrib/hooks/gc_base_hook.py
@@ -0,0 +1,63 @@
+import httplib2
+import logging
+
+from airflow.hooks.base_hook import BaseHook
+from airflow.utils import AirflowException
+from oauth2client.client import SignedJwtAssertionCredentials, GoogleCredentials
+
+class GoogleCloudBaseHook(BaseHook):
+    """
+    A base hook for Google cloud-related hooks. Google cloud has a shared REST
+    API client that is built in the same way no matter which service you use.
+    This class helps construct and authorize the credentials needed to then
+    call apiclient.discovery.build() to actually discover and build a client
+    for a Google cloud service.
+
+    The class also contains some miscellaneous helper functions.
+    """
+
+    def __init__(self, scope, conn_id):
+        """
+        :param scope: The scope of the hook.
+        :type scope: string
+        :param conn_id: The connection ID to use when fetching connection info.
+        :type conn_id: string
+        """
+        self.scope = scope
+        self.conn_id = conn_id
+
+    def _authorize(self):
+        """
+        Returns an authorized HTTP object to be used to build a Google cloud 
+        service hook connection.
+        """
+        connection_info = self.get_connection(self.conn_id)
+        connection_extras = connection_info.extra_dejson
+        service_account = connection_extras.get('service_account', False)
+        key_path = connection_extras.get('key_path', False)
+
+        if not key_path or not service_account:
+            logging.info('Getting connection using `gcloud auth` user, since no service_account/key_path are defined for hook.')
+            credentials = GoogleCredentials.get_application_default()
+        elif self.scope:
+            with file(key_path, 'rb') as key_file:
+                key = key_file.read()
+                credentials = SignedJwtAssertionCredentials(
+                    service_account,
+                    key,
+                    scope=self.scope)
+                    # TODO Support domain delegation, which will allow us to set a sub-account to execute as. We can then
+                    # pass DAG owner emails into the connection_info, and use it here.
+                    # sub='some@email.com')
+        else:
+            raise AirflowException('Scope undefined, or either key_path/service_account config was missing.')
+
+        http = httplib2.Http()
+        return credentials.authorize(http)
+
+    def _extras_dejson(self):
+        """
+        A little helper method that returns the JSON-deserialized extras in a 
+        single call.
+        """
+        return self.get_connection(self.conn_id).extra_dejson
diff --git a/airflow/contrib/hooks/gcs_hook.py b/airflow/contrib/hooks/gcs_hook.py
index cb8e682035359..8e5556ad2d845 100644
--- a/airflow/contrib/hooks/gcs_hook.py
+++ b/airflow/contrib/hooks/gcs_hook.py
@@ -1,21 +1,27 @@
 import httplib2
 import logging
 
+from airflow.contrib.hooks.gc_base_hook import GoogleCloudBaseHook
 from airflow.hooks.base_hook import BaseHook
 from apiclient.discovery import build
 from oauth2client.client import SignedJwtAssertionCredentials
 
 logging.getLogger("google_cloud_storage").setLevel(logging.INFO)
 
-class GoogleCloudStorageHook(BaseHook):
+class GoogleCloudStorageHook(GoogleCloudBaseHook):
     """
-    Interact with Google Cloud Storage. Connections must be defined with an extras JSON field containing:
+    Interact with Google Cloud Storage. Connections must be defined with an 
+    extras JSON field containing:
 
     {
         "project": "<google project ID>",
         "service_account": "<google service account email>",
         "key_path": "<p12 key path>"
     }
+
+    If you have used ``gcloud auth`` to authenticate on the machine that's 
+    running Airflow, you can exclude the service_account and key_path 
+    parameters.
     """
     conn_name_attr = 'google_cloud_storage_conn_id'
 
@@ -25,34 +31,14 @@ def __init__(self, scope='https://www.googleapis.com/auth/devstorage.read_only',
             https://cloud.google.com/storage/docs/authentication?hl=en#oauth-scopes
         :type scope: string
         """
-        self.scope = scope
-        self.google_cloud_storage_conn_id = google_cloud_storage_conn_id
+        super(GoogleCloudStorageHook, self).__init__(scope, google_cloud_storage_conn_id)
 
     def get_conn(self):
         """
         Returns a Google Cloud Storage service object.
         """
-        connection_info = self.get_connection(self.google_cloud_storage_conn_id)
-        connection_extras = connection_info.extra_dejson
-        service_account = connection_extras['service_account']
-        key_path = connection_extras['key_path']
-
-        with file(key_path, 'rb') as key_file:
-            key = key_file.read()
-
-        credentials = SignedJwtAssertionCredentials(
-            service_account,
-            key,
-            scope=self.scope)
-            # TODO Support domain delegation, which will allow us to set a sub-account to execute as. We can then
-            # pass DAG owner emails into the connection_info, and use it here.
-            # sub='some@email.com')
-
-        http = httplib2.Http()
-        http_authorized = credentials.authorize(http)
-        service = build('storage', 'v1', http=http_authorized)
-
-        return service
+        http_authorized = self._authorize()
+        return build('storage', 'v1', http=http_authorized)
 
     def download(self, bucket, object, filename=False):
         """

… developer-mode.
