feat: add cred info to ADC creds

google/auth/_default.py

@@ -237,6 +237,7 @@ def _get_gcloud_sdk_credentials(quota_project_id=None):
     credentials, project_id = load_credentials_from_file(
         credentials_filename, quota_project_id=quota_project_id
     )
+    credentials._cred_file_path = credentials_filename
 
     if not project_id:
         project_id = _cloud_sdk.get_project_id()
@@ -270,6 +271,7 @@ def _get_explicit_environ_credentials(quota_project_id=None):
         credentials, project_id = load_credentials_from_file(
             os.environ[environment_vars.CREDENTIALS], quota_project_id=quota_project_id
         )
+        credentials._cred_file_path = f"{explicit_file} file via the GOOGLE_APPLICATION_CREDENTIALS environment variable"
 
         return credentials, project_id
 

google/auth/compute_engine/credentials.py

@@ -157,6 +157,14 @@ def universe_domain(self):
         self._universe_domain_cached = True
         return self._universe_domain
 
+    @_helpers.copy_docstring(credentials.Credentials)
+    def get_cred_info(self):
+        return {
+            "credential_source": "metadata server",
+            "credential_type": "VM credentials",
+            "principal": self.service_account_email,
+        }
+
     @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
     def with_quota_project(self, quota_project_id):
         creds = self.__class__(

google/auth/credentials.py

@@ -128,6 +128,17 @@ def universe_domain(self):
         """The universe domain value."""
         return self._universe_domain
 
+    def get_cred_info(self):
+        """The credential information JSON.
+
+        The credential information will be added to auth related error messages
+        by client library.
+
+        Returns:
+            Mapping[str, str]: The credential information JSON.
+        """
+        return None
+
     @abc.abstractmethod
     def refresh(self, request):
         """Refreshes the access token.

google/auth/external_account.py

@@ -186,6 +186,7 @@ def __init__(
         self._supplier_context = SupplierContext(
             self._subject_token_type, self._audience
         )
+        self._cred_file_path = None
 
         if not self.is_workforce_pool and self._workforce_pool_user_project:
             # Workload identity pools do not support workforce pool user projects.
@@ -321,11 +322,24 @@ def token_info_url(self):
 
         return self._token_info_url
 
+    @_helpers.copy_docstring(credentials.Credentials)
+    def get_cred_info(self):
+        if self._cred_file_path:
+            cred_info_json = {
+                "credential_source": self._cred_file_path,
+                "credential_type": "external account credentials",
+            }
+            if self.service_account_email:
+                cred_info_json["principal"] = self.service_account_email
+            return cred_info_json
+        return None
+
     @_helpers.copy_docstring(credentials.Scoped)
     def with_scopes(self, scopes, default_scopes=None):
         kwargs = self._constructor_args()
         kwargs.update(scopes=scopes, default_scopes=default_scopes)
         scoped = self.__class__(**kwargs)
+        scoped._cred_file_path = self._cred_file_path
         scoped._metrics_options = self._metrics_options
         return scoped
 
@@ -442,30 +456,31 @@ def refresh(self, request):
 
             self.expiry = now + lifetime
 
-    @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
-    def with_quota_project(self, quota_project_id):
-        # Return copy of instance with the provided quota project ID.
+    def _make_copy(self):
         kwargs = self._constructor_args()
-        kwargs.update(quota_project_id=quota_project_id)
         new_cred = self.__class__(**kwargs)
+        new_cred._cred_file_path = self._cred_file_path
         new_cred._metrics_options = self._metrics_options
         return new_cred
 
+    @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
+    def with_quota_project(self, quota_project_id):
+        # Return copy of instance with the provided quota project ID.
+        cred = self._make_copy()
+        cred._quota_project_id = quota_project_id
+        return cred
+
     @_helpers.copy_docstring(credentials.CredentialsWithTokenUri)
     def with_token_uri(self, token_uri):
-        kwargs = self._constructor_args()
-        kwargs.update(token_url=token_uri)
-        new_cred = self.__class__(**kwargs)
-        new_cred._metrics_options = self._metrics_options
-        return new_cred
+        cred = self._make_copy()
+        cred._token_url = token_uri
+        return cred
 
     @_helpers.copy_docstring(credentials.CredentialsWithUniverseDomain)
     def with_universe_domain(self, universe_domain):
-        kwargs = self._constructor_args()
-        kwargs.update(universe_domain=universe_domain)
-        new_cred = self.__class__(**kwargs)
-        new_cred._metrics_options = self._metrics_options
-        return new_cred
+        cred = self._make_copy()
+        cred._universe_domain = universe_domain
+        return cred
 
     def _should_initialize_impersonated_credentials(self):
         return (

google/auth/external_account_authorized_user.py

@@ -120,6 +120,7 @@ def __init__(
         self._quota_project_id = quota_project_id
         self._scopes = scopes
         self._universe_domain = universe_domain or credentials.DEFAULT_UNIVERSE_DOMAIN
+        self._cred_file_path = None
 
         if not self.valid and not self.can_refresh:
             raise exceptions.InvalidOperation(
@@ -290,23 +291,38 @@ def refresh(self, request):
     def _make_sts_request(self, request):
         return self._sts_client.refresh_token(request, self._refresh_token)
 
+    @_helpers.copy_docstring(credentials.Credentials)
+    def get_cred_info(self):
+        if self._cred_file_path:
+            return {
+                "credential_source": self._cred_file_path,
+                "credential_type": "external account authorized user credentials",
+            }
+        return None
+
+    def _make_copy(self):
+        kwargs = self.constructor_args()
+        cred = self.__class__(**kwargs)
+        cred._cred_file_path = self._cred_file_path
+        return cred
+
     @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
     def with_quota_project(self, quota_project_id):
-        kwargs = self.constructor_args()
-        kwargs.update(quota_project_id=quota_project_id)
-        return self.__class__(**kwargs)
+        cred = self._make_copy()
+        cred._quota_project_id = quota_project_id
+        return cred
 
     @_helpers.copy_docstring(credentials.CredentialsWithTokenUri)
     def with_token_uri(self, token_uri):
-        kwargs = self.constructor_args()
-        kwargs.update(token_url=token_uri)
-        return self.__class__(**kwargs)
+        cred = self._make_copy()
+        cred._token_url = token_uri
+        return cred
 
     @_helpers.copy_docstring(credentials.CredentialsWithUniverseDomain)
     def with_universe_domain(self, universe_domain):
-        kwargs = self.constructor_args()
-        kwargs.update(universe_domain=universe_domain)
-        return self.__class__(**kwargs)
+        cred = self._make_copy()
+        cred._universe_domain = universe_domain
+        return cred
 
     @classmethod
     def from_info(cls, info, **kwargs):

google/auth/impersonated_credentials.py

@@ -226,6 +226,7 @@ def __init__(
         self.expiry = _helpers.utcnow()
         self._quota_project_id = quota_project_id
         self._iam_endpoint_override = iam_endpoint_override
+        self._cred_file_path = None
 
     def _metric_header_for_usage(self):
         return metrics.CRED_TYPE_SA_IMPERSONATE
@@ -316,29 +317,40 @@ def signer(self):
     def requires_scopes(self):
         return not self._target_scopes
 
-    @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
-    def with_quota_project(self, quota_project_id):
-        return self.__class__(
+    @_helpers.copy_docstring(credentials.Credentials)
+    def get_cred_info(self):
+        if self._cred_file_path:
+            return {
+                "credential_source": self._cred_file_path,
+                "credential_type": "impersonated credentials",
+                "principal": self._target_principal,
+            }
+        return None
+
+    def _make_copy(self):
+        cred = self.__class__(
             self._source_credentials,
             target_principal=self._target_principal,
             target_scopes=self._target_scopes,
             delegates=self._delegates,
             lifetime=self._lifetime,
-            quota_project_id=quota_project_id,
+            quota_project_id=self._quota_project_id,
             iam_endpoint_override=self._iam_endpoint_override,
         )
+        cred._cred_file_path = self._cred_file_path
+        return cred
+
+    @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
+    def with_quota_project(self, quota_project_id):
+        cred = self._make_copy()
+        cred._quota_project_id = quota_project_id
+        return cred
 
     @_helpers.copy_docstring(credentials.Scoped)
     def with_scopes(self, scopes, default_scopes=None):
-        return self.__class__(
-            self._source_credentials,
-            target_principal=self._target_principal,
-            target_scopes=scopes or default_scopes,
-            delegates=self._delegates,
-            lifetime=self._lifetime,
-            quota_project_id=self._quota_project_id,
-            iam_endpoint_override=self._iam_endpoint_override,
-        )
+        cred = self._make_copy()
+        cred._target_scopes = scopes or default_scopes
+        return cred
 
 
 class IDTokenCredentials(credentials.CredentialsWithQuotaProject):