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

@@ -20,6 +20,7 @@
 """
 
 import datetime
+import json
 
 from google.auth import _helpers
 from google.auth import credentials
@@ -157,6 +158,15 @@ def universe_domain(self):
         self._universe_domain_cached = True
         return self._universe_domain
 
+    @_helpers.copy_docstring(credentials.Credentials)
+    def get_cred_info(self):
+        cred_info_json = {
+            "credential_source": "metadata_server",
+            "credential_type": "VM credentials",
+            "principal": self.service_account_email,
+        }
+        return json.dumps(cred_info_json)
+
     @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
     def with_quota_project(self, quota_project_id):
         creds = self.__class__(

google/auth/credentials.py

@@ -128,6 +128,10 @@ def universe_domain(self):
         """The universe domain value."""
         return self._universe_domain
 
+    def get_cred_info(self):
+        """The credential information string."""
+        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 json.dumps(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
 
@@ -448,6 +462,7 @@ def with_quota_project(self, quota_project_id):
         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
 
@@ -456,6 +471,7 @@ def with_token_uri(self, token_uri):
         kwargs = self._constructor_args()
         kwargs.update(token_url=token_uri)
         new_cred = self.__class__(**kwargs)
+        new_cred._cred_file_path = self._cred_file_path
         new_cred._metrics_options = self._metrics_options
         return new_cred
 
@@ -464,6 +480,7 @@ def with_universe_domain(self, universe_domain):
         kwargs = self._constructor_args()
         kwargs.update(universe_domain=universe_domain)
         new_cred = self.__class__(**kwargs)
+        new_cred._cred_file_path = self._cred_file_path
         new_cred._metrics_options = self._metrics_options
         return new_cred
 

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,39 @@ 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:
+            cred_info_json = {
+                "credential_source": self._cred_file_path,
+                "credential_type": "external account authorized user credentials",
+            }
+            return json.dumps(cred_info_json)
+        return None
+
     @_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.__class__(**kwargs)
+        cred._cred_file_path = self._cred_file_path
+        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.__class__(**kwargs)
+        cred._cred_file_path = self._cred_file_path
+        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.__class__(**kwargs)
+        cred._cred_file_path = self._cred_file_path
+        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,9 +317,20 @@ def signer(self):
     def requires_scopes(self):
         return not self._target_scopes
 
+    @_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": "impersonated credentials",
+                "principal": self._target_principal,
+            }
+            return json.dumps(cred_info_json)
+        return None
+
     @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
     def with_quota_project(self, quota_project_id):
-        return self.__class__(
+        cred = self.__class__(
             self._source_credentials,
             target_principal=self._target_principal,
             target_scopes=self._target_scopes,
@@ -327,10 +339,12 @@ def with_quota_project(self, quota_project_id):
             quota_project_id=quota_project_id,
             iam_endpoint_override=self._iam_endpoint_override,
         )
+        cred._cred_file_path = self._cred_file_path
+        return cred
 
     @_helpers.copy_docstring(credentials.Scoped)
     def with_scopes(self, scopes, default_scopes=None):
-        return self.__class__(
+        cred = self.__class__(
             self._source_credentials,
             target_principal=self._target_principal,
             target_scopes=scopes or default_scopes,
@@ -339,6 +353,8 @@ def with_scopes(self, scopes, default_scopes=None):
             quota_project_id=self._quota_project_id,
             iam_endpoint_override=self._iam_endpoint_override,
         )
+        cred._cred_file_path = self._cred_file_path
+        return cred
 
 
 class IDTokenCredentials(credentials.CredentialsWithQuotaProject):

google/oauth2/credentials.py

@@ -151,6 +151,7 @@ def __init__(
         self._trust_boundary = trust_boundary
         self._universe_domain = universe_domain or credentials.DEFAULT_UNIVERSE_DOMAIN
         self._account = account or ""
+        self._cred_file_path = None
 
     def __getstate__(self):
         """A __getstate__ method must exist for the __setstate__ to be called
@@ -189,6 +190,7 @@ def __setstate__(self, d):
         self._universe_domain = (
             d.get("_universe_domain") or credentials.DEFAULT_UNIVERSE_DOMAIN
         )
+        self._cred_file_path = d.get("_cred_file_path")
         # The refresh_handler setter should be used to repopulate this.
         self._refresh_handler = None
         self._refresh_worker = None
@@ -278,10 +280,8 @@ def account(self):
         """str: The user account associated with the credential. If the account is unknown an empty string is returned."""
         return self._account
 
-    @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
-    def with_quota_project(self, quota_project_id):
-
-        return self.__class__(
+    def _make_copy(self):
+        cred = self.__class__(
             self.token,
             refresh_token=self.refresh_token,
             id_token=self.id_token,
@@ -291,34 +291,37 @@ def with_quota_project(self, quota_project_id):
             scopes=self.scopes,
             default_scopes=self.default_scopes,
             granted_scopes=self.granted_scopes,
-            quota_project_id=quota_project_id,
+            quota_project_id=self.quota_project_id,
             rapt_token=self.rapt_token,
             enable_reauth_refresh=self._enable_reauth_refresh,
             trust_boundary=self._trust_boundary,
             universe_domain=self._universe_domain,
             account=self._account,
         )
+        cred._cred_file_path = self._cred_file_path
+        return cred
+
+    @_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": "user credentials",
+            }
+            return json.dumps(cred_info_json)
+        return None
+
+    @_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.CredentialsWithTokenUri)
     def with_token_uri(self, token_uri):
-
-        return self.__class__(
-            self.token,
-            refresh_token=self.refresh_token,
-            id_token=self.id_token,
-            token_uri=token_uri,
-            client_id=self.client_id,
-            client_secret=self.client_secret,
-            scopes=self.scopes,
-            default_scopes=self.default_scopes,
-            granted_scopes=self.granted_scopes,
-            quota_project_id=self.quota_project_id,
-            rapt_token=self.rapt_token,
-            enable_reauth_refresh=self._enable_reauth_refresh,
-            trust_boundary=self._trust_boundary,
-            universe_domain=self._universe_domain,
-            account=self._account,
-        )
+        cred = self._make_copy()
+        cred._token_uri = token_uri
+        return cred
 
     def with_account(self, account):
         """Returns a copy of these credentials with a modified account.
@@ -329,45 +332,15 @@ def with_account(self, account):
         Returns:
             google.oauth2.credentials.Credentials: A new credentials instance.
         """
-
-        return self.__class__(
-            self.token,
-            refresh_token=self.refresh_token,
-            id_token=self.id_token,
-            token_uri=self._token_uri,
-            client_id=self.client_id,
-            client_secret=self.client_secret,
-            scopes=self.scopes,
-            default_scopes=self.default_scopes,
-            granted_scopes=self.granted_scopes,
-            quota_project_id=self.quota_project_id,
-            rapt_token=self.rapt_token,
-            enable_reauth_refresh=self._enable_reauth_refresh,
-            trust_boundary=self._trust_boundary,
-            universe_domain=self._universe_domain,
-            account=account,
-        )
+        cred = self._make_copy()
+        cred._account = account
+        return cred
 
     @_helpers.copy_docstring(credentials.CredentialsWithUniverseDomain)
     def with_universe_domain(self, universe_domain):
-
-        return self.__class__(
-            self.token,
-            refresh_token=self.refresh_token,
-            id_token=self.id_token,
-            token_uri=self._token_uri,
-            client_id=self.client_id,
-            client_secret=self.client_secret,
-            scopes=self.scopes,
-            default_scopes=self.default_scopes,
-            granted_scopes=self.granted_scopes,
-            quota_project_id=self.quota_project_id,
-            rapt_token=self.rapt_token,
-            enable_reauth_refresh=self._enable_reauth_refresh,
-            trust_boundary=self._trust_boundary,
-            universe_domain=universe_domain,
-            account=self._account,
-        )
+        cred = self._make_copy()
+        cred._universe_domain = universe_domain
+        return cred
 
     def _metric_header_for_usage(self):
         return metrics.CRED_TYPE_USER