fix: fix fetch_id_token credential lookup order to match adc (#748)

* fix: fix fetch_id_token credential lookup order to match adc

* fix tests

* fix linter

* update

* update

* add comments

4 files changed, 227 insertions, 155 deletions

diff --git a/google/oauth2/_id_token_async.py b/google/oauth2/_id_token_async.py
index f5ef8ba..ab681a9 100644
--- a/google/oauth2/_id_token_async.py
+++ b/google/oauth2/_id_token_async.py
@@ -180,13 +180,14 @@ async def verify_firebase_token(id_token, request, audience=None):
 async def fetch_id_token(request, audience):
     """Fetch the ID Token from the current environment.
 
-    This function acquires ID token from the environment in the following order:
+    This function acquires ID token from the environment in the following order.
+    See https://google.aip.dev/auth/4110.
 
-    1. If the application is running in Compute Engine, App Engine or Cloud Run,
-       then the ID token are obtained from the metadata server.
-    2. If the environment variable ``GOOGLE_APPLICATION_CREDENTIALS`` is set
+    1. If the environment variable ``GOOGLE_APPLICATION_CREDENTIALS`` is set
        to the path of a valid service account JSON file, then ID token is
        acquired using this service account credentials.
+    2. If the application is running in Compute Engine, App Engine or Cloud Run,
+       then the ID token are obtained from the metadata server.
     3. If metadata server doesn't exist and no valid service account credentials
        are found, :class:`~google.auth.exceptions.DefaultCredentialsError` will
        be raised.
@@ -214,54 +215,52 @@ async def fetch_id_token(request, audience):
             If metadata server doesn't exist and no valid service account
             credentials are found.
     """
-    # 1. First try to fetch ID token from metadata server if it exists. The code
-    # works for GAE and Cloud Run metadata server as well.
-    try:
-        from google.auth import compute_engine
-
-        request_new = requests.Request()
-        credentials = compute_engine.IDTokenCredentials(
-            request_new, audience, use_metadata_identity_endpoint=True
-        )
-        credentials.refresh(request_new)
-
-        return credentials.token
-
-    except (ImportError, exceptions.TransportError, exceptions.RefreshError):
-        pass
-
-    # 2. Try to use service account credentials to get ID token.
-
-    # Try to get credentials from the GOOGLE_APPLICATION_CREDENTIALS environment
+    # 1. Try to get credentials from the GOOGLE_APPLICATION_CREDENTIALS environment
     # variable.
     credentials_filename = os.environ.get(environment_vars.CREDENTIALS)
-    if not (
-        credentials_filename
-        and os.path.exists(credentials_filename)
-        and os.path.isfile(credentials_filename)
-    ):
-        raise exceptions.DefaultCredentialsError(
-            "Neither metadata server or valid service account credentials are found."
-        )
+    if credentials_filename:
+        if not (
+            os.path.exists(credentials_filename)
+            and os.path.isfile(credentials_filename)
+        ):
+            raise exceptions.DefaultCredentialsError(
+                "GOOGLE_APPLICATION_CREDENTIALS path is either not found or invalid."
+            )
 
-    try:
-        with open(credentials_filename, "r") as f:
-            info = json.load(f)
-            credentials_content = (
-                (info.get("type") == "service_account") and info or None
+        try:
+            with open(credentials_filename, "r") as f:
+                from google.oauth2 import _service_account_async as service_account
+
+                info = json.load(f)
+                if info.get("type") == "service_account":
+                    credentials = service_account.IDTokenCredentials.from_service_account_info(
+                        info, target_audience=audience
+                    )
+                    await credentials.refresh(request)
+                    return credentials.token
+        except ValueError as caught_exc:
+            new_exc = exceptions.DefaultCredentialsError(
+                "GOOGLE_APPLICATION_CREDENTIALS is not valid service account credentials.",
+                caught_exc,
             )
+            six.raise_from(new_exc, caught_exc)
 
-            from google.oauth2 import _service_account_async as service_account
+    # 2. Try to fetch ID token from metada server if it exists. The code works for GAE and
+    # Cloud Run metadata server as well.
+    try:
+        from google.auth import compute_engine
+        from google.auth.compute_engine import _metadata
 
-            credentials = service_account.IDTokenCredentials.from_service_account_info(
-                credentials_content, target_audience=audience
+        request_new = requests.Request()
+        if _metadata.ping(request_new):
+            credentials = compute_engine.IDTokenCredentials(
+                request_new, audience, use_metadata_identity_endpoint=True
             )
-    except ValueError as caught_exc:
-        new_exc = exceptions.DefaultCredentialsError(
-            "Neither metadata server or valid service account credentials are found.",
-            caught_exc,
-        )
-        six.raise_from(new_exc, caught_exc)
+            credentials.refresh(request_new)
+            return credentials.token
+    except (ImportError, exceptions.TransportError):
+        pass
 
-    await credentials.refresh(request)
-    return credentials.token
+    raise exceptions.DefaultCredentialsError(
+        "Neither metadata server or valid service account credentials are found."
+    )
diff --git a/google/oauth2/id_token.py b/google/oauth2/id_token.py
index 5fbb6a1..540ccd1 100644
--- a/google/oauth2/id_token.py
+++ b/google/oauth2/id_token.py
@@ -179,13 +179,14 @@ def verify_firebase_token(id_token, request, audience=None):
 def fetch_id_token(request, audience):
     """Fetch the ID Token from the current environment.
 
-    This function acquires ID token from the environment in the following order:
+    This function acquires ID token from the environment in the following order.
+    See https://google.aip.dev/auth/4110.
 
-    1. If the application is running in Compute Engine, App Engine or Cloud Run,
-       then the ID token are obtained from the metadata server.
-    2. If the environment variable ``GOOGLE_APPLICATION_CREDENTIALS`` is set
+    1. If the environment variable ``GOOGLE_APPLICATION_CREDENTIALS`` is set
        to the path of a valid service account JSON file, then ID token is
        acquired using this service account credentials.
+    2. If the application is running in Compute Engine, App Engine or Cloud Run,
+       then the ID token are obtained from the metadata server.
     3. If metadata server doesn't exist and no valid service account credentials
        are found, :class:`~google.auth.exceptions.DefaultCredentialsError` will
        be raised.
@@ -213,51 +214,51 @@ def fetch_id_token(request, audience):
             If metadata server doesn't exist and no valid service account
             credentials are found.
     """
-    # 1. First try to fetch ID token from metada server if it exists. The code
-    # works for GAE and Cloud Run metadata server as well.
-    try:
-        from google.auth import compute_engine
-
-        credentials = compute_engine.IDTokenCredentials(
-            request, audience, use_metadata_identity_endpoint=True
-        )
-        credentials.refresh(request)
-        return credentials.token
-    except (ImportError, exceptions.TransportError, exceptions.RefreshError):
-        pass
-
-    # 2. Try to use service account credentials to get ID token.
-
-    # Try to get credentials from the GOOGLE_APPLICATION_CREDENTIALS environment
+    # 1. Try to get credentials from the GOOGLE_APPLICATION_CREDENTIALS environment
     # variable.
     credentials_filename = os.environ.get(environment_vars.CREDENTIALS)
-    if not (
-        credentials_filename
-        and os.path.exists(credentials_filename)
-        and os.path.isfile(credentials_filename)
-    ):
-        raise exceptions.DefaultCredentialsError(
-            "Neither metadata server or valid service account credentials are found."
-        )
+    if credentials_filename:
+        if not (
+            os.path.exists(credentials_filename)
+            and os.path.isfile(credentials_filename)
+        ):
+            raise exceptions.DefaultCredentialsError(
+                "GOOGLE_APPLICATION_CREDENTIALS path is either not found or invalid."
+            )
 
-    try:
-        with open(credentials_filename, "r") as f:
-            info = json.load(f)
-            credentials_content = (
-                (info.get("type") == "service_account") and info or None
+        try:
+            with open(credentials_filename, "r") as f:
+                from google.oauth2 import service_account
+
+                info = json.load(f)
+                if info.get("type") == "service_account":
+                    credentials = service_account.IDTokenCredentials.from_service_account_info(
+                        info, target_audience=audience
+                    )
+                    credentials.refresh(request)
+                    return credentials.token
+        except ValueError as caught_exc:
+            new_exc = exceptions.DefaultCredentialsError(
+                "GOOGLE_APPLICATION_CREDENTIALS is not valid service account credentials.",
+                caught_exc,
             )
+            six.raise_from(new_exc, caught_exc)
 
-            from google.oauth2 import service_account
+    # 2. Try to fetch ID token from metada server if it exists. The code works for GAE and
+    # Cloud Run metadata server as well.
+    try:
+        from google.auth import compute_engine
+        from google.auth.compute_engine import _metadata
 
-            credentials = service_account.IDTokenCredentials.from_service_account_info(
-                credentials_content, target_audience=audience
+        if _metadata.ping(request):
+            credentials = compute_engine.IDTokenCredentials(
+                request, audience, use_metadata_identity_endpoint=True
             )
-    except ValueError as caught_exc:
-        new_exc = exceptions.DefaultCredentialsError(
-            "Neither metadata server or valid service account credentials are found.",
-            caught_exc,
-        )
-        six.raise_from(new_exc, caught_exc)
+            credentials.refresh(request)
+            return credentials.token
+    except (ImportError, exceptions.TransportError):
+        pass
 
-    credentials.refresh(request)
-    return credentials.token
+    raise exceptions.DefaultCredentialsError(
+        "Neither metadata server or valid service account credentials are found."
+    )
diff --git a/tests/oauth2/test_id_token.py b/tests/oauth2/test_id_token.py
index 0c70d68..ab67743 100644
--- a/tests/oauth2/test_id_token.py
+++ b/tests/oauth2/test_id_token.py
@@ -23,6 +23,7 @@ from google.auth import exceptions
 from google.auth import transport
 import google.auth.compute_engine._metadata
 from google.oauth2 import id_token
+from google.oauth2 import service_account
 
 SERVICE_ACCOUNT_FILE = os.path.join(
     os.path.dirname(__file__), "../data/service_account.json"
@@ -134,62 +135,93 @@ def test_verify_firebase_token(verify_token):
     )
 
 
-def test_fetch_id_token_from_metadata_server():
+def test_fetch_id_token_from_metadata_server(monkeypatch):
+    monkeypatch.delenv(environment_vars.CREDENTIALS, raising=False)
+
     def mock_init(self, request, audience, use_metadata_identity_endpoint):
         assert use_metadata_identity_endpoint
         self.token = "id_token"
 
-    with mock.patch.multiple(
-        google.auth.compute_engine.IDTokenCredentials,
-        __init__=mock_init,
-        refresh=mock.Mock(),
-    ):
-        request = mock.Mock()
-        token = id_token.fetch_id_token(request, "https://pubsub.googleapis.com")
-        assert token == "id_token"
+    with mock.patch("google.auth.compute_engine._metadata.ping", return_value=True):
+        with mock.patch.multiple(
+            google.auth.compute_engine.IDTokenCredentials,
+            __init__=mock_init,
+            refresh=mock.Mock(),
+        ):
+            request = mock.Mock()
+            token = id_token.fetch_id_token(request, "https://pubsub.googleapis.com")
+            assert token == "id_token"
 
 
-@mock.patch.object(
-    google.auth.compute_engine.IDTokenCredentials,
-    "__init__",
-    side_effect=exceptions.TransportError(),
-)
-def test_fetch_id_token_from_explicit_cred_json_file(mock_init, monkeypatch):
+def test_fetch_id_token_from_explicit_cred_json_file(monkeypatch):
     monkeypatch.setenv(environment_vars.CREDENTIALS, SERVICE_ACCOUNT_FILE)
 
     def mock_refresh(self, request):
         self.token = "id_token"
 
-    with mock.patch.object(
-        google.oauth2.service_account.IDTokenCredentials, "refresh", mock_refresh
-    ):
+    with mock.patch.object(service_account.IDTokenCredentials, "refresh", mock_refresh):
         request = mock.Mock()
         token = id_token.fetch_id_token(request, "https://pubsub.googleapis.com")
         assert token == "id_token"
 
 
-@mock.patch.object(
-    google.auth.compute_engine.IDTokenCredentials,
-    "__init__",
-    side_effect=exceptions.TransportError(),
-)
-def test_fetch_id_token_no_cred_json_file(mock_init, monkeypatch):
+def test_fetch_id_token_no_cred_exists(monkeypatch):
     monkeypatch.delenv(environment_vars.CREDENTIALS, raising=False)
 
-    with pytest.raises(exceptions.DefaultCredentialsError):
+    with mock.patch(
+        "google.auth.compute_engine._metadata.ping",
+        side_effect=exceptions.TransportError(),
+    ):
+        with pytest.raises(exceptions.DefaultCredentialsError) as excinfo:
+            request = mock.Mock()
+            id_token.fetch_id_token(request, "https://pubsub.googleapis.com")
+        assert excinfo.match(
+            r"Neither metadata server or valid service account credentials are found."
+        )
+
+    with mock.patch("google.auth.compute_engine._metadata.ping", return_value=False):
+        with pytest.raises(exceptions.DefaultCredentialsError) as excinfo:
+            request = mock.Mock()
+            id_token.fetch_id_token(request, "https://pubsub.googleapis.com")
+        assert excinfo.match(
+            r"Neither metadata server or valid service account credentials are found."
+        )
+
+
+def test_fetch_id_token_invalid_cred_file_type(monkeypatch):
+    user_credentials_file = os.path.join(
+        os.path.dirname(__file__), "../data/authorized_user.json"
+    )
+    monkeypatch.setenv(environment_vars.CREDENTIALS, user_credentials_file)
+
+    with mock.patch("google.auth.compute_engine._metadata.ping", return_value=False):
+        with pytest.raises(exceptions.DefaultCredentialsError) as excinfo:
+            request = mock.Mock()
+            id_token.fetch_id_token(request, "https://pubsub.googleapis.com")
+        assert excinfo.match(
+            r"Neither metadata server or valid service account credentials are found."
+        )
+
+
+def test_fetch_id_token_invalid_json(monkeypatch):
+    not_json_file = os.path.join(os.path.dirname(__file__), "../data/public_cert.pem")
+    monkeypatch.setenv(environment_vars.CREDENTIALS, not_json_file)
+
+    with pytest.raises(exceptions.DefaultCredentialsError) as excinfo:
         request = mock.Mock()
         id_token.fetch_id_token(request, "https://pubsub.googleapis.com")
+    assert excinfo.match(
+        r"GOOGLE_APPLICATION_CREDENTIALS is not valid service account credentials."
+    )
 
 
-@mock.patch.object(
-    google.auth.compute_engine.IDTokenCredentials,
-    "__init__",
-    side_effect=exceptions.TransportError(),
-)
-def test_fetch_id_token_invalid_cred_file(mock_init, monkeypatch):
-    not_json_file = os.path.join(os.path.dirname(__file__), "../data/public_cert.pem")
+def test_fetch_id_token_invalid_cred_path(monkeypatch):
+    not_json_file = os.path.join(os.path.dirname(__file__), "../data/not_exists.json")
     monkeypatch.setenv(environment_vars.CREDENTIALS, not_json_file)
 
-    with pytest.raises(exceptions.DefaultCredentialsError):
+    with pytest.raises(exceptions.DefaultCredentialsError) as excinfo:
         request = mock.Mock()
         id_token.fetch_id_token(request, "https://pubsub.googleapis.com")
+    assert excinfo.match(
+        r"GOOGLE_APPLICATION_CREDENTIALS path is either not found or invalid."
+    )
diff --git a/tests_async/oauth2/test_id_token.py b/tests_async/oauth2/test_id_token.py
index a46bd61..1deb9ef 100644
--- a/tests_async/oauth2/test_id_token.py
+++ b/tests_async/oauth2/test_id_token.py
@@ -21,6 +21,7 @@ from google.auth import environment_vars
 from google.auth import exceptions
 import google.auth.compute_engine._metadata
 from google.oauth2 import _id_token_async as id_token
+from google.oauth2 import _service_account_async
 from google.oauth2 import id_token as sync_id_token
 from tests.oauth2 import test_id_token
 
@@ -139,67 +140,106 @@ async def test_verify_firebase_token(verify_token):
 
 
 @pytest.mark.asyncio
-async def test_fetch_id_token_from_metadata_server():
+async def test_fetch_id_token_from_metadata_server(monkeypatch):
+    monkeypatch.delenv(environment_vars.CREDENTIALS, raising=False)
+
     def mock_init(self, request, audience, use_metadata_identity_endpoint):
         assert use_metadata_identity_endpoint
         self.token = "id_token"
 
-    with mock.patch.multiple(
-        google.auth.compute_engine.IDTokenCredentials,
-        __init__=mock_init,
-        refresh=mock.Mock(),
-    ):
-        request = mock.AsyncMock()
-        token = await id_token.fetch_id_token(request, "https://pubsub.googleapis.com")
-        assert token == "id_token"
+    with mock.patch("google.auth.compute_engine._metadata.ping", return_value=True):
+        with mock.patch.multiple(
+            google.auth.compute_engine.IDTokenCredentials,
+            __init__=mock_init,
+            refresh=mock.Mock(),
+        ):
+            request = mock.AsyncMock()
+            token = await id_token.fetch_id_token(
+                request, "https://pubsub.googleapis.com"
+            )
+            assert token == "id_token"
 
 
-@mock.patch.object(
-    google.auth.compute_engine.IDTokenCredentials,
-    "__init__",
-    side_effect=exceptions.TransportError(),
-)
 @pytest.mark.asyncio
-async def test_fetch_id_token_from_explicit_cred_json_file(mock_init, monkeypatch):
+async def test_fetch_id_token_from_explicit_cred_json_file(monkeypatch):
     monkeypatch.setenv(environment_vars.CREDENTIALS, test_id_token.SERVICE_ACCOUNT_FILE)
 
     async def mock_refresh(self, request):
         self.token = "id_token"
 
     with mock.patch.object(
-        google.oauth2._service_account_async.IDTokenCredentials, "refresh", mock_refresh
+        _service_account_async.IDTokenCredentials, "refresh", mock_refresh
     ):
         request = mock.AsyncMock()
         token = await id_token.fetch_id_token(request, "https://pubsub.googleapis.com")
         assert token == "id_token"
 
 
-@mock.patch.object(
-    google.auth.compute_engine.IDTokenCredentials,
-    "__init__",
-    side_effect=exceptions.TransportError(),
-)
 @pytest.mark.asyncio
-async def test_fetch_id_token_no_cred_json_file(mock_init, monkeypatch):
+async def test_fetch_id_token_no_cred_exists(monkeypatch):
     monkeypatch.delenv(environment_vars.CREDENTIALS, raising=False)
 
-    with pytest.raises(exceptions.DefaultCredentialsError):
+    with mock.patch(
+        "google.auth.compute_engine._metadata.ping",
+        side_effect=exceptions.TransportError(),
+    ):
+        with pytest.raises(exceptions.DefaultCredentialsError) as excinfo:
+            request = mock.AsyncMock()
+            await id_token.fetch_id_token(request, "https://pubsub.googleapis.com")
+        assert excinfo.match(
+            r"Neither metadata server or valid service account credentials are found."
+        )
+
+    with mock.patch("google.auth.compute_engine._metadata.ping", return_value=False):
+        with pytest.raises(exceptions.DefaultCredentialsError) as excinfo:
+            request = mock.AsyncMock()
+            await id_token.fetch_id_token(request, "https://pubsub.googleapis.com")
+        assert excinfo.match(
+            r"Neither metadata server or valid service account credentials are found."
+        )
+
+
+@pytest.mark.asyncio
+async def test_fetch_id_token_invalid_cred_file(monkeypatch):
+    not_json_file = os.path.join(
+        os.path.dirname(__file__), "../../tests/data/public_cert.pem"
+    )
+    monkeypatch.setenv(environment_vars.CREDENTIALS, not_json_file)
+
+    with pytest.raises(exceptions.DefaultCredentialsError) as excinfo:
         request = mock.AsyncMock()
         await id_token.fetch_id_token(request, "https://pubsub.googleapis.com")
+    assert excinfo.match(
+        r"GOOGLE_APPLICATION_CREDENTIALS is not valid service account credentials."
+    )
 
 
-@mock.patch.object(
-    google.auth.compute_engine.IDTokenCredentials,
-    "__init__",
-    side_effect=exceptions.TransportError(),
-)
 @pytest.mark.asyncio
-async def test_fetch_id_token_invalid_cred_file(mock_init, monkeypatch):
+async def test_fetch_id_token_invalid_cred_type(monkeypatch):
+    user_credentials_file = os.path.join(
+        os.path.dirname(__file__), "../../tests/data/authorized_user.json"
+    )
+    monkeypatch.setenv(environment_vars.CREDENTIALS, user_credentials_file)
+
+    with mock.patch("google.auth.compute_engine._metadata.ping", return_value=False):
+        with pytest.raises(exceptions.DefaultCredentialsError) as excinfo:
+            request = mock.AsyncMock()
+            await id_token.fetch_id_token(request, "https://pubsub.googleapis.com")
+        assert excinfo.match(
+            r"Neither metadata server or valid service account credentials are found."
+        )
+
+
+@pytest.mark.asyncio
+async def test_fetch_id_token_invalid_cred_path(monkeypatch):
     not_json_file = os.path.join(
-        os.path.dirname(__file__), "../../tests/data/public_cert.pem"
+        os.path.dirname(__file__), "../../tests/data/not_exists.json"
     )
     monkeypatch.setenv(environment_vars.CREDENTIALS, not_json_file)
 
-    with pytest.raises(exceptions.DefaultCredentialsError):
+    with pytest.raises(exceptions.DefaultCredentialsError) as excinfo:
         request = mock.AsyncMock()
         await id_token.fetch_id_token(request, "https://pubsub.googleapis.com")
+    assert excinfo.match(
+        r"GOOGLE_APPLICATION_CREDENTIALS path is either not found or invalid."
+    )