Support JSON based service auth (#81)

* Support authentication from json service account file

* Add tests

* mark p12 tests with skip

Author: isidentical

pydrive2/auth.py

@@ -161,11 +161,7 @@ class GoogleAuth(ApiAttributeMixin, object):
         "revoke_uri",
         "redirect_uri",
     ]
-    SERVICE_CONFIGS_LIST = [
-        "client_service_email",
-        "client_user_email",
-        "client_pkcs12_file_path",
-    ]
+    SERVICE_CONFIGS_LIST = ["client_user_email"]
     settings = ApiAttribute("settings")
     client_config = ApiAttribute("client_config")
     flow = ApiAttribute("flow")
@@ -296,14 +292,21 @@ def ServiceAuth(self):
         if set(self.SERVICE_CONFIGS_LIST) - set(self.client_config):
             self.LoadServiceConfigSettings()
         scopes = scopes_to_string(self.settings["oauth_scope"])
+        client_service_json = self.client_config.get("client_json_file_path")
+        if client_service_json:
+            self.credentials = ServiceAccountCredentials.from_json_keyfile_name(
+                filename=client_service_json, scopes=scopes
+            )
+        else:
+            service_email = self.client_config["client_service_email"]
+            file_path = self.client_config["client_pkcs12_file_path"]
+            self.credentials = ServiceAccountCredentials.from_p12_keyfile(
+                service_account_email=service_email,
+                filename=file_path,
+                scopes=scopes,
+            )
+
         user_email = self.client_config.get("client_user_email")
-        service_email = self.client_config["client_service_email"]
-        file_path = self.client_config["client_pkcs12_file_path"]
-        self.credentials = ServiceAccountCredentials.from_p12_keyfile(
-            service_account_email=service_email,
-            filename=file_path,
-            scopes=scopes,
-        )
         if user_email:
             self.credentials = self.credentials.create_delegated(
                 sub=user_email
@@ -322,6 +325,8 @@ def LoadCredentials(self, backend=None):
                 raise InvalidConfigError("Please specify credential backend")
         if backend == "file":
             self.LoadCredentialsFile()
+        elif backend == "json":
+            self.LoadCredentialsFileJson()
         else:
             raise InvalidConfigError("Unknown save_credentials_backend")
 
@@ -348,6 +353,19 @@ def LoadCredentialsFile(self, credentials_file=None):
                 "Credentials file cannot be symbolic link"
             )
 
+    def LoadCredentialsFileJson(self, credentials_file=None):
+        if credentials_file is None:
+            credentials_file = self.settings.get("save_credentials_file")
+            if credentials_file is None:
+                raise InvalidConfigError(
+                    "Please specify credentials file to read"
+                )
+
+        scopes = scopes_to_string(self.settings["oauth_scope"])
+        self.credentials = ServiceAccountCredentials.from_json_keyfile_name(
+            filename=credentials_file, scopes=scopes
+        )
+
     def SaveCredentials(self, backend=None):
         """Saves credentials according to specified backend.
 
@@ -471,6 +489,21 @@ def LoadServiceConfigSettings(self):
         """Loads client configuration from settings file.
     :raises: InvalidConfigError
     """
+        for file_format in ["json", "pkcs12"]:
+            config = f"client_{file_format}_file_path"
+            value = self.settings["service_config"].get(config)
+            if value:
+                self.client_config[config] = value
+                break
+        else:
+            raise InvalidConfigError(
+                "Either json or pkcs12 file path required "
+                "for service authentication"
+            )
+
+        if file_format == "pkcs12":
+            self.SERVICE_CONFIGS_LIST.append("client_service_email")
+
         for config in self.SERVICE_CONFIGS_LIST:
             try:
                 self.client_config[config] = self.settings["service_config"][
@@ -516,7 +549,7 @@ def GetFlow(self):
             self.client_config["client_id"],
             self.client_config["client_secret"],
             scopes_to_string(self.settings["oauth_scope"]),
-            **constructor_kwargs
+            **constructor_kwargs,
         )
         if self.settings.get("get_refresh_token"):
             self.flow.params.update(

pydrive2/settings.py

@@ -72,8 +72,9 @@
                 "required": True,
                 "default": None,
             },
-            "client_service_email": {"type": str, "required": True},
-            "client_pkcs12_file_path": {"type": str, "required": True},
+            "client_service_email": {"type": str, "required": False},
+            "client_pkcs12_file_path": {"type": str, "required": False},
+            "client_json_file_path": {"type": str, "required": False},
         },
     },
     "oauth_scope": {

pydrive2/test/settings/default.yaml

@@ -1,10 +1,9 @@
 client_config_backend: service
 service_config:
-  client_service_email: your-service-account-email
-  client_pkcs12_file_path: your-file-path.p12
+  client_json_file_path: your-file-path.json
 
 save_credentials: True
-save_credentials_backend: file
+save_credentials_backend: json
 save_credentials_file: credentials/default.dat
 
 oauth_scope:

pydrive2/test/settings/test_oauth_test_07.yaml

@@ -0,0 +1,10 @@
+client_config_backend: service
+service_config:
+  client_json_file_path: your-file-path.json
+
+save_credentials: True
+save_credentials_backend: json
+save_credentials_file: credentials/7.dat
+
+oauth_scope:
+  - https://www.googleapis.com/auth/drive

pydrive2/test/test_oauth.py

@@ -73,13 +73,21 @@ def test_05_ConfigFromSettingsWithoutOauthScope(self):
         self.assertEqual(ga.access_token_expired, False)
         time.sleep(1)
 
-    def test_06_ServiceAuthFromSavedCredentialsFile(self):
+    @pytest.mark.skip(reason="P12 authentication is deprecated")
+    def test_06_ServiceAuthFromSavedCredentialsP12File(self):
         setup_credentials("credentials/6.dat")
         ga = GoogleAuth(settings_file_path("test_oauth_test_06.yaml"))
         ga.ServiceAuth()
         self.assertEqual(ga.access_token_expired, False)
         time.sleep(1)
 
+    def test_07_ServiceAuthFromSavedCredentialsJsonFile(self):
+        setup_credentials("credentials/7.dat")
+        ga = GoogleAuth(settings_file_path("test_oauth_test_07.yaml"))
+        ga.ServiceAuth()
+        self.assertEqual(ga.access_token_expired, False)
+        time.sleep(1)
+
     def CheckCredentialsFile(self, credentials, no_file=False):
         ga = GoogleAuth(settings_file_path("test_oauth_default.yaml"))
         ga.LoadCredentialsFile(credentials)