diff --git a/packages/google-auth/google/auth/jwt.py b/packages/google-auth/google/auth/jwt.py index 9eff792e7bc2..4f256b79cddc 100644 --- a/packages/google-auth/google/auth/jwt.py +++ b/packages/google-auth/google/auth/jwt.py @@ -41,10 +41,10 @@ """ try: - from collections.abc import Mapping + from collections.abc import Iterable, Mapping # Python 2.7 compatibility except ImportError: # pragma: NO COVER - from collections import Mapping # type: ignore + from collections import Iterable, Mapping # type: ignore import copy import datetime import json @@ -269,8 +269,8 @@ def decode(token, certs=None, verify=True, audience=None, clock_skew_in_seconds= try: verifier_cls = _ALGORITHM_TO_VERIFIER_CLASS[key_alg] - except KeyError as exc: - if key_alg in _CRYPTOGRAPHY_BASED_ALGORITHMS: + except (KeyError, TypeError) as exc: + if isinstance(key_alg, str) and key_alg in _CRYPTOGRAPHY_BASED_ALGORITHMS: raise exceptions.InvalidValue( "The key algorithm {} requires the cryptography package to be installed.".format( key_alg @@ -284,11 +284,12 @@ def decode(token, certs=None, verify=True, audience=None, clock_skew_in_seconds= # use the certificate identified by the key ID in the token header. if isinstance(certs, Mapping): if key_id: - if key_id not in certs: + try: + certs_to_check = [certs[key_id]] + except (KeyError, TypeError) as exc: raise exceptions.MalformedError( "Certificate for key id {} not found.".format(key_id) - ) - certs_to_check = [certs[key_id]] + ) from exc # If there's no key id in the header, check against all of the certs. else: certs_to_check = certs.values() @@ -296,9 +297,19 @@ def decode(token, certs=None, verify=True, audience=None, clock_skew_in_seconds= certs_to_check = certs # Verify that the signature matches the message. - if not crypt.verify_signature( - signed_section, signature, certs_to_check, verifier_cls - ): + try: + valid_signature = crypt.verify_signature( + signed_section, signature, certs_to_check, verifier_cls + ) + except TypeError as exc: + if not isinstance(certs_to_check, Iterable): + raise + raise exceptions.InvalidValue( + "Certificate key type does not match signature algorithm {}.".format( + key_alg + ) + ) from exc + if not valid_signature: raise exceptions.MalformedError("Could not verify token signature.") # Verify the issued at and created times in the payload. diff --git a/packages/google-auth/google/oauth2/_id_token_async.py b/packages/google-auth/google/oauth2/_id_token_async.py index 024a305c0808..2a1fe7466623 100644 --- a/packages/google-auth/google/oauth2/_id_token_async.py +++ b/packages/google-auth/google/oauth2/_id_token_async.py @@ -61,6 +61,7 @@ import http.client as http_client import json import os +import warnings from google.auth import environment_vars, exceptions, jwt from google.auth.transport import requests @@ -176,16 +177,30 @@ async def verify_firebase_token( id_token (Union[str, bytes]): The encoded token. request (google.auth.transport.Request): The object used to make HTTP requests. This must be an aiohttp request. - audience (str): The audience that this token is intended for. This is - typically your Firebase application ID. If None then the audience - is not verified. + audience (Optional[Union[str, Iterable[str]]]): The audience that this + token is intended for. This is your Firebase project ID. If + ``None`` or empty, a ``UserWarning`` is issued. Passing a + non-empty ``audience`` will be required in a future release. clock_skew_in_seconds (int): The clock skew used for `iat` and `exp` validation. Returns: Mapping[str, Any]: The decoded token. + + Raises: + exceptions.InvalidValue: If the issuer, audience, or subject is invalid. + ValueError: If token verification fails. """ - return await verify_token( + if not audience: + warnings.warn( + "Calling verify_firebase_token without an 'audience' leaves the " + "token's audience claim unverified. Pass your Firebase project ID " + "as 'audience', which will be required in a future release.", + UserWarning, + stacklevel=2, + ) + + idinfo = await verify_token( id_token, request, audience=audience, @@ -193,6 +208,22 @@ async def verify_firebase_token( clock_skew_in_seconds=clock_skew_in_seconds, ) + token_aud = idinfo.get("aud") + if not isinstance(token_aud, str) or not token_aud: + raise exceptions.InvalidValue("Token does not contain a valid 'aud' claim.") + + expected_issuer = "{}{}".format(sync_id_token._FIREBASE_ISSUER_PREFIX, token_aud) + if idinfo.get("iss") != expected_issuer: + raise exceptions.InvalidValue( + "Wrong issuer. 'iss' should be '{}'.".format(expected_issuer) + ) + + token_sub = idinfo.get("sub") + if not isinstance(token_sub, str) or not token_sub: + raise exceptions.InvalidValue("Token does not contain a valid 'sub' claim.") + + return idinfo + async def fetch_id_token(request, audience, bind_id_token=None): """Fetch the ID Token from the current environment. diff --git a/packages/google-auth/google/oauth2/id_token.py b/packages/google-auth/google/oauth2/id_token.py index c73d11b50f94..cac9fe2f18f6 100644 --- a/packages/google-auth/google/oauth2/id_token.py +++ b/packages/google-auth/google/oauth2/id_token.py @@ -60,6 +60,7 @@ import http.client as http_client import json import os +import warnings from typing import Any, Mapping, Union from google.auth import environment_vars, exceptions, jwt, transport @@ -76,6 +77,7 @@ ) _GOOGLE_ISSUERS = ["accounts.google.com", "https://accounts.google.com"] +_FIREBASE_ISSUER_PREFIX = "https://securetoken.google.com/" def _fetch_certs(request, certs_url): @@ -218,16 +220,30 @@ def verify_firebase_token(id_token, request, audience=None, clock_skew_in_second id_token (Union[str, bytes]): The encoded token. request (google.auth.transport.Request): The object used to make HTTP requests. - audience (str): The audience that this token is intended for. This is - typically your Firebase application ID. If None then the audience - is not verified. + audience (Optional[Union[str, Iterable[str]]]): The audience that this + token is intended for. This is your Firebase project ID. If + ``None`` or empty, a ``UserWarning`` is issued. Passing a + non-empty ``audience`` will be required in a future release. clock_skew_in_seconds (int): The clock skew used for `iat` and `exp` validation. Returns: Mapping[str, Any]: The decoded token. + + Raises: + exceptions.InvalidValue: If the issuer, audience, or subject is invalid. + ValueError: If token verification fails. """ - return verify_token( + if not audience: + warnings.warn( + "Calling verify_firebase_token without an 'audience' leaves the " + "token's audience claim unverified. Pass your Firebase project ID " + "as 'audience', which will be required in a future release.", + UserWarning, + stacklevel=2, + ) + + idinfo = verify_token( id_token, request, audience=audience, @@ -235,6 +251,22 @@ def verify_firebase_token(id_token, request, audience=None, clock_skew_in_second clock_skew_in_seconds=clock_skew_in_seconds, ) + token_aud = idinfo.get("aud") + if not isinstance(token_aud, str) or not token_aud: + raise exceptions.InvalidValue("Token does not contain a valid 'aud' claim.") + + expected_issuer = "{}{}".format(_FIREBASE_ISSUER_PREFIX, token_aud) + if idinfo.get("iss") != expected_issuer: + raise exceptions.InvalidValue( + "Wrong issuer. 'iss' should be '{}'.".format(expected_issuer) + ) + + token_sub = idinfo.get("sub") + if not isinstance(token_sub, str) or not token_sub: + raise exceptions.InvalidValue("Token does not contain a valid 'sub' claim.") + + return idinfo + def fetch_id_token_credentials(audience, request=None, bind_id_token=None): """Create the ID Token credentials from the current environment. diff --git a/packages/google-auth/tests/oauth2/test_id_token.py b/packages/google-auth/tests/oauth2/test_id_token.py index 07fd2c3f1935..b00844f79981 100644 --- a/packages/google-auth/tests/oauth2/test_id_token.py +++ b/packages/google-auth/tests/oauth2/test_id_token.py @@ -12,20 +12,29 @@ # See the License for the specific language governing permissions and # limitations under the License. +import base64 import json import os +import warnings from unittest import mock import pytest # type: ignore from google.auth import ( + _helpers, + crypt, environment_vars, exceptions, impersonated_credentials, + jwt, transport, ) from google.oauth2 import id_token, service_account +PRIVATE_KEY_FILE = os.path.join(os.path.dirname(__file__), "../data/privatekey.pem") + +PUBLIC_CERT_FILE = os.path.join(os.path.dirname(__file__), "../data/public_cert.pem") + SERVICE_ACCOUNT_FILE = os.path.join( os.path.dirname(__file__), "../data/service_account.json" ) @@ -237,9 +246,16 @@ def test_verify_oauth2_token_invalid_iss(verify_token): @mock.patch("google.oauth2.id_token.verify_token", autospec=True) def test_verify_firebase_token(verify_token): - result = id_token.verify_firebase_token( - mock.sentinel.token, mock.sentinel.request, audience=mock.sentinel.audience - ) + verify_token.return_value = { + "aud": "my-project-id", + "iss": "https://securetoken.google.com/my-project-id", + "sub": "my-user-id", + } + with warnings.catch_warnings(): + warnings.simplefilter("error", UserWarning) + result = id_token.verify_firebase_token( + mock.sentinel.token, mock.sentinel.request, audience=mock.sentinel.audience + ) assert result == verify_token.return_value verify_token.assert_called_once_with( @@ -253,6 +269,11 @@ def test_verify_firebase_token(verify_token): @mock.patch("google.oauth2.id_token.verify_token", autospec=True) def test_verify_firebase_token_clock_skew(verify_token): + verify_token.return_value = { + "aud": "my-project-id", + "iss": "https://securetoken.google.com/my-project-id", + "sub": "my-user-id", + } result = id_token.verify_firebase_token( mock.sentinel.token, mock.sentinel.request, @@ -270,6 +291,186 @@ def test_verify_firebase_token_clock_skew(verify_token): ) +@pytest.mark.parametrize( + "invalid_payload", + [ + {"aud": "my-project-id"}, + { + "aud": "my-project-id", + "iss": "https://securetoken.google.com/other-project-id", + }, + ], +) +@mock.patch("google.oauth2.id_token.verify_token", autospec=True) +def test_verify_firebase_token_invalid_iss(verify_token, invalid_payload): + verify_token.return_value = invalid_payload + + with pytest.raises(exceptions.InvalidValue, match="Wrong issuer"): + id_token.verify_firebase_token( + mock.sentinel.token, + mock.sentinel.request, + audience=mock.sentinel.audience, + ) + + +@pytest.mark.parametrize( + "invalid_payload", + [ + {"iss": "https://securetoken.google.com/"}, + {"aud": "", "iss": "https://securetoken.google.com/"}, + { + "aud": ["my-project-id"], + "iss": "https://securetoken.google.com/my-project-id", + }, + ], +) +@mock.patch("google.oauth2.id_token.verify_token", autospec=True) +def test_verify_firebase_token_missing_aud(verify_token, invalid_payload): + verify_token.return_value = invalid_payload + + with pytest.raises( + exceptions.InvalidValue, match="Token does not contain a valid 'aud' claim" + ): + id_token.verify_firebase_token( + mock.sentinel.token, + mock.sentinel.request, + audience=mock.sentinel.audience, + ) + + +@pytest.mark.parametrize( + "invalid_payload", + [ + { + "aud": "my-project-id", + "iss": "https://securetoken.google.com/my-project-id", + }, + { + "aud": "my-project-id", + "iss": "https://securetoken.google.com/my-project-id", + "sub": "", + }, + { + "aud": "my-project-id", + "iss": "https://securetoken.google.com/my-project-id", + "sub": 123, + }, + ], +) +@mock.patch("google.oauth2.id_token.verify_token", autospec=True) +def test_verify_firebase_token_invalid_sub(verify_token, invalid_payload): + verify_token.return_value = invalid_payload + + with pytest.raises( + exceptions.InvalidValue, match="Token does not contain a valid 'sub' claim" + ): + id_token.verify_firebase_token( + mock.sentinel.token, + mock.sentinel.request, + audience=mock.sentinel.audience, + ) + + +@pytest.mark.parametrize("audience", [None, "", []]) +@mock.patch("google.oauth2.id_token.verify_token", autospec=True) +def test_verify_firebase_token_warns_without_audience(verify_token, audience): + verify_token.return_value = { + "aud": "my-project-id", + "iss": "https://securetoken.google.com/my-project-id", + "sub": "my-user-id", + } + + with pytest.warns( + UserWarning, + match=( + "Calling verify_firebase_token without an 'audience' leaves the " + "token's audience claim unverified" + ), + ) as record: + result = id_token.verify_firebase_token( + mock.sentinel.token, mock.sentinel.request, audience=audience + ) + + assert len(record) == 1 + assert record[0].filename == __file__ + assert result == verify_token.return_value + verify_token.assert_called_once_with( + mock.sentinel.token, + mock.sentinel.request, + audience=audience, + certs_url=id_token._GOOGLE_APIS_CERTS_URL, + clock_skew_in_seconds=0, + ) + + +@pytest.mark.parametrize("alg", ["ES256", "ES384"]) +def test_verify_firebase_token_mismatched_alg_for_rsa_certs(alg): + with open(PUBLIC_CERT_FILE, "r") as fh: + public_cert = fh.read() + request = make_request(200, {"1": public_cert}) + header = base64.urlsafe_b64encode( + json.dumps({"alg": alg, "kid": "1", "typ": "JWT"}).encode("utf-8") + ) + payload = base64.urlsafe_b64encode( + json.dumps( + { + "aud": "my-project-id", + "iss": "https://securetoken.google.com/my-project-id", + "sub": "my-user-id", + } + ).encode("utf-8") + ) + sig = base64.urlsafe_b64encode(b"sig") + token = b".".join([header, payload, sig]) + + with pytest.raises( + exceptions.InvalidValue, + match="Certificate key type does not match signature algorithm {}".format(alg), + ): + id_token.verify_firebase_token(token, request, audience="my-project-id") + + +def test_verify_firebase_token_end_to_end(): + with open(PRIVATE_KEY_FILE, "rb") as fh: + private_key = fh.read() + with open(PUBLIC_CERT_FILE, "r") as fh: + public_cert = fh.read() + + signer = crypt.RSASigner.from_string(private_key, "1") + now = _helpers.datetime_to_secs(_helpers.utcnow()) + + valid_payload = { + "aud": "my-project-id", + "iss": "https://securetoken.google.com/my-project-id", + "sub": "my-user-id", + "iat": now, + "exp": now + 300, + } + valid_token = jwt.encode(signer, valid_payload) + request = make_request(200, {"1": public_cert}) + + result = id_token.verify_firebase_token( + valid_token, request, audience="my-project-id" + ) + assert result == valid_payload + + cross_tenant_token = jwt.encode( + signer, + { + "aud": "other-project-id", + "iss": "https://securetoken.google.com/other-project-id", + "sub": "my-user-id", + "iat": now, + "exp": now + 300, + }, + ) + request = make_request(200, {"1": public_cert}) + with pytest.raises(exceptions.InvalidValue, match="Token has wrong audience"): + id_token.verify_firebase_token( + cross_tenant_token, request, audience="my-project-id" + ) + + def test_fetch_id_token_credentials_optional_request(monkeypatch): monkeypatch.delenv(environment_vars.CREDENTIALS, raising=False) diff --git a/packages/google-auth/tests/test_jwt.py b/packages/google-auth/tests/test_jwt.py index 207648aa06fa..ffc3fe5a0112 100644 --- a/packages/google-auth/tests/test_jwt.py +++ b/packages/google-auth/tests/test_jwt.py @@ -361,6 +361,64 @@ def test_decode_missing_crytography_alg(monkeypatch): assert excinfo.match(r"cryptography") +@pytest.mark.parametrize( + "token_kwargs,certs,expected_alg", + [ + ({"use_es256_signer": True}, PUBLIC_CERT_BYTES, "ES256"), + ({"use_es256_signer": True}, {"1": PUBLIC_CERT_BYTES}, "ES256"), + ({"use_es384_signer": True}, PUBLIC_CERT_BYTES, "ES384"), + ({"use_es384_signer": True}, {"1": PUBLIC_CERT_BYTES}, "ES384"), + ({}, EC_PUBLIC_CERT_BYTES, "RS256"), + ({}, {"1": EC_PUBLIC_CERT_BYTES}, "RS256"), + ], +) +def test_decode_mismatched_cert_key_type( + token_factory, token_kwargs, certs, expected_alg +): + token = token_factory(**token_kwargs) + + with pytest.raises(exceptions.InvalidValue) as excinfo: + jwt.decode(token, certs=certs) + assert excinfo.match( + r"Certificate key type does not match signature algorithm {}".format( + expected_alg + ) + ) + + +def test_decode_missing_certs(token_factory): + with pytest.raises(TypeError): + jwt.decode(token_factory()) + + +@pytest.mark.parametrize("certs", [None, object()]) +def test_decode_non_iterable_certs(token_factory, certs): + with pytest.raises(TypeError): + jwt.decode(token_factory(), certs=certs) + + +def test_decode_unhashable_alg(): + headers = json.dumps({"kid": "1", "alg": ["RS256"]}) + token = b".".join( + map(lambda seg: base64.b64encode(seg.encode("utf-8")), [headers, "{}", "sig"]) + ) + + with pytest.raises(exceptions.InvalidValue) as excinfo: + jwt.decode(token, PUBLIC_CERT_BYTES) + assert excinfo.match(r"Unsupported signature algorithm") + + +def test_decode_unhashable_kid(): + headers = json.dumps({"kid": ["1"], "alg": "RS256"}) + token = b".".join( + map(lambda seg: base64.b64encode(seg.encode("utf-8")), [headers, "{}", "sig"]) + ) + + with pytest.raises(exceptions.MalformedError) as excinfo: + jwt.decode(token, {"1": PUBLIC_CERT_BYTES}) + assert excinfo.match(r"Certificate for key id .* not found") + + def test_roundtrip_explicit_key_id(token_factory): token = token_factory(key_id="3") certs = {"2": OTHER_CERT_BYTES, "3": PUBLIC_CERT_BYTES} diff --git a/packages/google-auth/tests_async/oauth2/test_id_token.py b/packages/google-auth/tests_async/oauth2/test_id_token.py index 883a9ee9c5ec..c5f09cbf7c17 100644 --- a/packages/google-auth/tests_async/oauth2/test_id_token.py +++ b/packages/google-auth/tests_async/oauth2/test_id_token.py @@ -12,14 +12,16 @@ # See the License for the specific language governing permissions and # limitations under the License. +import base64 import json import os +import warnings from unittest import mock import pytest # type: ignore import google.auth.compute_engine._metadata -from google.auth import environment_vars, exceptions +from google.auth import _helpers, crypt, environment_vars, exceptions, jwt 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 @@ -174,9 +176,16 @@ async def test_verify_oauth2_token_invalid_iss(verify_token): @mock.patch("google.oauth2._id_token_async.verify_token", autospec=True) @pytest.mark.asyncio async def test_verify_firebase_token(verify_token): - result = await id_token.verify_firebase_token( - mock.sentinel.token, mock.sentinel.request, audience=mock.sentinel.audience - ) + verify_token.return_value = { + "aud": "my-project-id", + "iss": "https://securetoken.google.com/my-project-id", + "sub": "my-user-id", + } + with warnings.catch_warnings(): + warnings.simplefilter("error", UserWarning) + result = await id_token.verify_firebase_token( + mock.sentinel.token, mock.sentinel.request, audience=mock.sentinel.audience + ) assert result == verify_token.return_value verify_token.assert_called_once_with( @@ -191,6 +200,11 @@ async def test_verify_firebase_token(verify_token): @mock.patch("google.oauth2._id_token_async.verify_token", autospec=True) @pytest.mark.asyncio async def test_verify_firebase_token_clock_skew(verify_token): + verify_token.return_value = { + "aud": "my-project-id", + "iss": "https://securetoken.google.com/my-project-id", + "sub": "my-user-id", + } result = await id_token.verify_firebase_token( mock.sentinel.token, mock.sentinel.request, @@ -208,6 +222,192 @@ async def test_verify_firebase_token_clock_skew(verify_token): ) +@pytest.mark.parametrize( + "invalid_payload", + [ + {"aud": "my-project-id"}, + { + "aud": "my-project-id", + "iss": "https://securetoken.google.com/other-project-id", + }, + ], +) +@mock.patch("google.oauth2._id_token_async.verify_token", autospec=True) +@pytest.mark.asyncio +async def test_verify_firebase_token_invalid_iss(verify_token, invalid_payload): + verify_token.return_value = invalid_payload + + with pytest.raises(exceptions.InvalidValue, match="Wrong issuer"): + await id_token.verify_firebase_token( + mock.sentinel.token, + mock.sentinel.request, + audience=mock.sentinel.audience, + ) + + +@pytest.mark.parametrize( + "invalid_payload", + [ + {"iss": "https://securetoken.google.com/"}, + {"aud": "", "iss": "https://securetoken.google.com/"}, + { + "aud": ["my-project-id"], + "iss": "https://securetoken.google.com/my-project-id", + }, + ], +) +@mock.patch("google.oauth2._id_token_async.verify_token", autospec=True) +@pytest.mark.asyncio +async def test_verify_firebase_token_missing_aud(verify_token, invalid_payload): + verify_token.return_value = invalid_payload + + with pytest.raises( + exceptions.InvalidValue, match="Token does not contain a valid 'aud' claim" + ): + await id_token.verify_firebase_token( + mock.sentinel.token, + mock.sentinel.request, + audience=mock.sentinel.audience, + ) + + +@pytest.mark.parametrize( + "invalid_payload", + [ + { + "aud": "my-project-id", + "iss": "https://securetoken.google.com/my-project-id", + }, + { + "aud": "my-project-id", + "iss": "https://securetoken.google.com/my-project-id", + "sub": "", + }, + { + "aud": "my-project-id", + "iss": "https://securetoken.google.com/my-project-id", + "sub": 123, + }, + ], +) +@mock.patch("google.oauth2._id_token_async.verify_token", autospec=True) +@pytest.mark.asyncio +async def test_verify_firebase_token_invalid_sub(verify_token, invalid_payload): + verify_token.return_value = invalid_payload + + with pytest.raises( + exceptions.InvalidValue, match="Token does not contain a valid 'sub' claim" + ): + await id_token.verify_firebase_token( + mock.sentinel.token, + mock.sentinel.request, + audience=mock.sentinel.audience, + ) + + +@pytest.mark.parametrize("audience", [None, "", []]) +@mock.patch("google.oauth2._id_token_async.verify_token", autospec=True) +@pytest.mark.asyncio +async def test_verify_firebase_token_warns_without_audience(verify_token, audience): + verify_token.return_value = { + "aud": "my-project-id", + "iss": "https://securetoken.google.com/my-project-id", + "sub": "my-user-id", + } + + with pytest.warns( + UserWarning, + match=( + "Calling verify_firebase_token without an 'audience' leaves the " + "token's audience claim unverified" + ), + ) as record: + result = await id_token.verify_firebase_token( + mock.sentinel.token, mock.sentinel.request, audience=audience + ) + + assert len(record) == 1 + assert record[0].filename == __file__ + assert result == verify_token.return_value + verify_token.assert_called_once_with( + mock.sentinel.token, + mock.sentinel.request, + audience=audience, + certs_url=sync_id_token._GOOGLE_APIS_CERTS_URL, + clock_skew_in_seconds=0, + ) + + +@pytest.mark.parametrize("alg", ["ES256", "ES384"]) +@pytest.mark.asyncio +async def test_verify_firebase_token_mismatched_alg_for_rsa_certs(alg): + with open(test_id_token.PUBLIC_CERT_FILE, "r") as fh: + public_cert = fh.read() + request = make_request(200, {"1": public_cert}) + header = base64.urlsafe_b64encode( + json.dumps({"alg": alg, "kid": "1", "typ": "JWT"}).encode("utf-8") + ) + payload = base64.urlsafe_b64encode( + json.dumps( + { + "aud": "my-project-id", + "iss": "https://securetoken.google.com/my-project-id", + "sub": "my-user-id", + } + ).encode("utf-8") + ) + sig = base64.urlsafe_b64encode(b"sig") + token = b".".join([header, payload, sig]) + + with pytest.raises( + exceptions.InvalidValue, + match="Certificate key type does not match signature algorithm {}".format(alg), + ): + await id_token.verify_firebase_token(token, request, audience="my-project-id") + + +@pytest.mark.asyncio +async def test_verify_firebase_token_end_to_end(): + with open(test_id_token.PRIVATE_KEY_FILE, "rb") as fh: + private_key = fh.read() + with open(test_id_token.PUBLIC_CERT_FILE, "r") as fh: + public_cert = fh.read() + + signer = crypt.RSASigner.from_string(private_key, "1") + now = _helpers.datetime_to_secs(_helpers.utcnow()) + + valid_payload = { + "aud": "my-project-id", + "iss": "https://securetoken.google.com/my-project-id", + "sub": "my-user-id", + "iat": now, + "exp": now + 300, + } + valid_token = jwt.encode(signer, valid_payload) + request = make_request(200, {"1": public_cert}) + + result = await id_token.verify_firebase_token( + valid_token, request, audience="my-project-id" + ) + assert result == valid_payload + + cross_tenant_token = jwt.encode( + signer, + { + "aud": "other-project-id", + "iss": "https://securetoken.google.com/other-project-id", + "sub": "my-user-id", + "iat": now, + "exp": now + 300, + }, + ) + request = make_request(200, {"1": public_cert}) + with pytest.raises(exceptions.InvalidValue, match="Token has wrong audience"): + await id_token.verify_firebase_token( + cross_tenant_token, request, audience="my-project-id" + ) + + @pytest.mark.asyncio @pytest.mark.parametrize( "call_kwargs,expected_bind_id_token",