fix: compute engine id token credentials "with_target_audience" method (#438)

Co-authored-by: Lorenzo Migliorino <[email protected]>
Co-authored-by: Bu Sun Kim <[email protected]>

Author: 3 people

.gitignore

@@ -31,8 +31,11 @@ tests/data/user-key.json
 
 # PyCharm configuration:
 .idea
+venv/
 
 # Generated files
 pylintrc
 pylintrc.test
 pytype_output/
+
+.python-version

google/auth/compute_engine/credentials.py

@@ -136,6 +136,7 @@ def __init__(
         token_uri=_DEFAULT_TOKEN_URI,
         additional_claims=None,
         service_account_email=None,
+        signer=None,
     ):
         """
         Args:
@@ -150,6 +151,9 @@ def __init__(
             service_account_email (str): Optional explicit service account to
                 use to sign JWT tokens.
                 By default, this is the default GCE service account.
+            signer (google.auth.crypt.Signer): The signer used to sign JWTs.
+                In case the signer is specified, the request argument will be
+                ignored.
         """
         super(IDTokenCredentials, self).__init__()
 
@@ -158,11 +162,13 @@ def __init__(
             service_account_email = sa_info["email"]
         self._service_account_email = service_account_email
 
-        self._signer = iam.Signer(
-            request=request,
-            credentials=Credentials(),
-            service_account_email=service_account_email,
-        )
+        if signer is None:
+            signer = iam.Signer(
+                request=request,
+                credentials=Credentials(),
+                service_account_email=service_account_email,
+            )
+        self._signer = signer
 
         self._token_uri = token_uri
         self._target_audience = target_audience
@@ -182,12 +188,15 @@ def with_target_audience(self, target_audience):
             google.auth.service_account.IDTokenCredentials: A new credentials
                 instance.
         """
+        # since the signer is already instantiated,
+        # the request is not needed
         return self.__class__(
-            self._signer,
+            None,
             service_account_email=self._service_account_email,
             token_uri=self._token_uri,
             target_audience=target_audience,
             additional_claims=self._additional_claims.copy(),
+            signer=self.signer,
         )
 
     def _make_authorization_grant_assertion(self):

noxfile.py

@@ -25,6 +25,7 @@
     "requests",
     "urllib3",
     "cryptography",
+    "responses",
     "grpcio",
 ]
 BLACK_VERSION = "black==19.3b0"

tests/compute_engine/test_credentials.py

@@ -11,17 +11,19 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-
+import base64
 import datetime
 
 import mock
 import pytest
+import responses
 
 from google.auth import _helpers
 from google.auth import exceptions
 from google.auth import jwt
 from google.auth import transport
 from google.auth.compute_engine import credentials
+from google.auth.transport import requests
 
 
 class TestCredentials(object):
@@ -270,6 +272,84 @@ def test_with_target_audience(self, sign, get, utcnow):
             "target_audience": "https://actually.not",
         }
 
+        # Check that the signer have been initialized with a Request object
+        assert isinstance(self.credentials._signer._request, transport.Request)
+
+    @responses.activate
+    def test_with_target_audience_integration(self):
+        """ Test that it is possible to refresh credentials
+        generated from `with_target_audience`.
+
+        Instead of mocking the methods, the HTTP responses
+        have been mocked.
+        """
+
+        # mock information about credentials
+        responses.add(
+            responses.GET,
+            "http://metadata.google.internal/computeMetadata/v1/instance/"
+            "service-accounts/default/?recursive=true",
+            status=200,
+            content_type="application/json",
+            json={
+                "scopes": "email",
+                "email": "[email protected]",
+                "aliases": ["default"],
+            },
+        )
+
+        # mock token for credentials
+        responses.add(
+            responses.GET,
+            "http://metadata.google.internal/computeMetadata/v1/instance/"
+            "service-accounts/[email protected]/token",
+            status=200,
+            content_type="application/json",
+            json={
+                "access_token": "some-token",
+                "expires_in": 3210,
+                "token_type": "Bearer",
+            },
+        )
+
+        # mock sign blob endpoint
+        signature = base64.b64encode(b"some-signature").decode("utf-8")
+        responses.add(
+            responses.POST,
+            "https://iam.googleapis.com/v1/projects/-/serviceAccounts/"
+            "[email protected]:signBlob?alt=json",
+            status=200,
+            content_type="application/json",
+            json={"keyId": "some-key-id", "signature": signature},
+        )
+
+        id_token = "{}.{}.{}".format(
+            base64.b64encode(b'{"some":"some"}').decode("utf-8"),
+            base64.b64encode(b'{"exp": 3210}').decode("utf-8"),
+            base64.b64encode(b"token").decode("utf-8"),
+        )
+
+        # mock id token endpoint
+        responses.add(
+            responses.POST,
+            "https://www.googleapis.com/oauth2/v4/token",
+            status=200,
+            content_type="application/json",
+            json={"id_token": id_token, "expiry": 3210},
+        )
+
+        self.credentials = credentials.IDTokenCredentials(
+            request=requests.Request(),
+            service_account_email="[email protected]",
+            target_audience="https://audience.com",
+        )
+
+        self.credentials = self.credentials.with_target_audience("https://actually.not")
+
+        self.credentials.refresh(requests.Request())
+
+        assert self.credentials.token is not None
+
     @mock.patch(
         "google.auth._helpers.utcnow",
         return_value=datetime.datetime.utcfromtimestamp(0),