fix: Do not create new JWT credentials if they make the same claims as the existing.

google/auth/jwt.py

@@ -590,6 +590,11 @@ def signer_email(self):
     def signer(self):
         return self._signer
 
+    @property  # type: ignore
+    def additonal_claims(self):
+        """Additional claims the JWT object was created with."""
+        return self._additional_claims
+
 
 class OnDemandCredentials(
     google.auth.credentials.Signing, google.auth.credentials.CredentialsWithQuotaProject

google/oauth2/service_account.py

@@ -441,19 +441,32 @@ def _create_self_signed_jwt(self, audience):
         # https://google.aip.dev/auth/4111
         if self._always_use_jwt_access:
             if self._scopes:
-                self._jwt_credentials = jwt.Credentials.from_signing_credentials(
-                    self, None, additional_claims={"scope": " ".join(self._scopes)}
-                )
+                additional_claims = {"scope": " ".join(self._scopes)}
+                if (
+                    self._jwt_credentials is None
+                    or self._jwt_credentials.additional_claims != additional_claims
+                ):
+                    self._jwt_credentials = jwt.Credentials.from_signing_credentials(
+                        self, None, additional_claims=additional_claims
+                    )
             elif audience:
-                self._jwt_credentials = jwt.Credentials.from_signing_credentials(
-                    self, audience
-                )
+                if (
+                    self._jwt_credentials is None
+                    or self._jwt_credentials._audience != audience
+                ):
+
+                    self._jwt_credentials = jwt.Credentials.from_signing_credentials(
+                        self, audience
+                    )
             elif self._default_scopes:
-                self._jwt_credentials = jwt.Credentials.from_signing_credentials(
-                    self,
-                    None,
-                    additional_claims={"scope": " ".join(self._default_scopes)},
-                )
+                additional_claims = {"scope": " ".join(self._default_scopes)}
+                if (
+                    self._jwt_credentials is None
+                    or additional_claims != self._jwt_credentials.additional_claims
+                ):
+                    self._jwt_credentials = jwt.Credentials.from_signing_credentials(
+                        self, None, additional_claims=additional_claims
+                    )
         elif not self._scopes and audience:
             self._jwt_credentials = jwt.Credentials.from_signing_credentials(
                 self, audience

tests/oauth2/test_service_account.py

@@ -253,6 +253,24 @@ def test__create_self_signed_jwt_always_use_jwt_access_with_audience(self, jwt):
         credentials._create_self_signed_jwt(audience)
         jwt.from_signing_credentials.assert_called_once_with(credentials, audience)
 
+    @mock.patch("google.auth.jwt.Credentials", instance=True, autospec=True)
+    def test__create_self_signed_jwt_always_use_jwt_access_with_audience_similar_jwt_is_reused(
+        self, jwt
+    ):
+        credentials = service_account.Credentials(
+            SIGNER,
+            self.SERVICE_ACCOUNT_EMAIL,
+            self.TOKEN_URI,
+            default_scopes=["bar", "foo"],
+            always_use_jwt_access=True,
+        )
+
+        audience = "https://pubsub.googleapis.com"
+        credentials._create_self_signed_jwt(audience)
+        credentials._jwt_credentials._audience = audience
+        credentials._create_self_signed_jwt(audience)
+        jwt.from_signing_credentials.assert_called_once_with(credentials, audience)
+
     @mock.patch("google.auth.jwt.Credentials", instance=True, autospec=True)
     def test__create_self_signed_jwt_always_use_jwt_access_with_scopes(self, jwt):
         credentials = service_account.Credentials(
@@ -269,6 +287,26 @@ def test__create_self_signed_jwt_always_use_jwt_access_with_scopes(self, jwt):
             credentials, None, additional_claims={"scope": "bar foo"}
         )
 
+    @mock.patch("google.auth.jwt.Credentials", instance=True, autospec=True)
+    def test__create_self_signed_jwt_always_use_jwt_access_with_scopes_similar_jwt_is_reused(
+        self, jwt
+    ):
+        credentials = service_account.Credentials(
+            SIGNER,
+            self.SERVICE_ACCOUNT_EMAIL,
+            self.TOKEN_URI,
+            scopes=["bar", "foo"],
+            always_use_jwt_access=True,
+        )
+
+        audience = "https://pubsub.googleapis.com"
+        credentials._create_self_signed_jwt(audience)
+        credentials._jwt_credentials.additional_claims = {"scope": "bar foo"}
+        credentials._create_self_signed_jwt(audience)
+        jwt.from_signing_credentials.assert_called_once_with(
+            credentials, None, additional_claims={"scope": "bar foo"}
+        )
+
     @mock.patch("google.auth.jwt.Credentials", instance=True, autospec=True)
     def test__create_self_signed_jwt_always_use_jwt_access_with_default_scopes(
         self, jwt
@@ -286,6 +324,25 @@ def test__create_self_signed_jwt_always_use_jwt_access_with_default_scopes(
             credentials, None, additional_claims={"scope": "bar foo"}
         )
 
+    @mock.patch("google.auth.jwt.Credentials", instance=True, autospec=True)
+    def test__create_self_signed_jwt_always_use_jwt_access_with_default_scopes_similar_jwt_is_reused(
+        self, jwt
+    ):
+        credentials = service_account.Credentials(
+            SIGNER,
+            self.SERVICE_ACCOUNT_EMAIL,
+            self.TOKEN_URI,
+            default_scopes=["bar", "foo"],
+            always_use_jwt_access=True,
+        )
+
+        credentials._create_self_signed_jwt(None)
+        credentials._jwt_credentials.additional_claims = {"scope": "bar foo"}
+        credentials._create_self_signed_jwt(None)
+        jwt.from_signing_credentials.assert_called_once_with(
+            credentials, None, additional_claims={"scope": "bar foo"}
+        )
+
     @mock.patch("google.auth.jwt.Credentials", instance=True, autospec=True)
     def test__create_self_signed_jwt_always_use_jwt_access(self, jwt):
         credentials = service_account.Credentials(