Add a UserSecrets.get_gcloud_credentials() method.

https://b.corp.google.com/issues/158133824

Author: mcollins42

patches/kaggle_secrets.py

@@ -30,6 +30,9 @@ class BackendError(Exception):
 class ValidationError(Exception):
     pass
 
+class NotFoundError(Exception):
+    pass
+
 @unique
 class GcpTarget(Enum):
     """Enum class to store GCP targets."""
@@ -116,6 +119,22 @@ def get_secret(self, label) -> str:
                 f'Unexpected response from the service. Response: {response_json}')
         return response_json['secret']
 
+    def get_gcloud_credential(self) -> str:
+        """Retrieves the Google Cloud SDK credential attached to the current
+        kernel.
+        Example usage:
+            client = UserSecretsClient()
+            credential_json = client.get_gcloud_credential()
+        """
+        try:
+            return self.get_secret("__gcloud_sdk_auth__")
+        except BackendError as backend_error:
+            message = str(backend_error.args)
+            if message.find('No user secrets exist') != -1:
+              raise NotFoundError('Google Cloud SDK credential not found.')
+            else:
+              raise
+
     def get_bigquery_access_token(self) -> Tuple[str, Optional[datetime]]:
         """Retrieves BigQuery access token information from the UserSecrets service.
 

tests/test_user_secrets.py

@@ -13,7 +13,7 @@
 from kaggle_secrets import (_KAGGLE_URL_BASE_ENV_VAR_NAME,
                             _KAGGLE_USER_SECRETS_TOKEN_ENV_VAR_NAME,
                             CredentialError, GcpTarget, UserSecretsClient,
-                            BackendError, ValidationError)
+                            BackendError, NotFoundError, ValidationError)
 
 _TEST_JWT = 'test-secrets-key'
 
@@ -55,7 +55,7 @@ def get_response(self):
                 if success:
                     return {'result': {'secret': secret, 'secretType': 'refreshToken', 'secretProvider': 'google', 'expiresInSeconds': 3600}, 'wasSuccessful': "true"}
                 else:
-                    return {'wasSuccessful': "false"}
+                    return {'wasSuccessful': "false", 'errors': ['No user secrets exist for kernel']}
 
         env = EnvironmentVarGuard()
         env.set(_KAGGLE_USER_SECRETS_TOKEN_ENV_VAR_NAME, _TEST_JWT)
@@ -95,7 +95,7 @@ def call_get_secret():
         self._test_client(call_get_secret,
                           '/requests/GetUserSecretByLabelRequest', {'Label': "secret_label", 'JWE': _TEST_JWT},
                           secret=secret)
-    
+
     def test_get_secret_handles_unsuccessful(self):
         def call_get_secret():
             client = UserSecretsClient()
@@ -112,7 +112,28 @@ def test_get_secret_validates_label(self):
             client = UserSecretsClient()
             with self.assertRaises(ValidationError):
                 secret_response = client.get_secret("")
-                          
+
+    def test_get_gcloud_secret_succeeds(self):
+        secret = '{"client_id":"gcloud","type":"authorized_user"}'
+
+        def call_get_secret():
+            client = UserSecretsClient()
+            secret_response = client.get_gcloud_credential()
+            self.assertEqual(secret_response, secret)
+        self._test_client(call_get_secret,
+                          '/requests/GetUserSecretByLabelRequest', {'Label': "__gcloud_sdk_auth__", 'JWE': _TEST_JWT},
+                          secret=secret)
+
+    def test_get_gcloud_secret_handles_unsuccessful(self):
+        def call_get_secret():
+            client = UserSecretsClient()
+            with self.assertRaises(NotFoundError):
+              secret_response = client.get_gcloud_credential()
+        self._test_client(call_get_secret,
+                          '/requests/GetUserSecretByLabelRequest', {'Label': "__gcloud_sdk_auth__", 'JWE': _TEST_JWT},
+                          success=False)
+
+
     @mock.patch('kaggle_secrets.datetime')
     def test_get_access_token_succeeds(self, mock_dt):
         secret = '12345'