fix(oauth2): validate Firebase issuer/subject and warn on omitted audience in verify_firebase_token

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,21 +284,32 @@ 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()
     else:
         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.

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,23 +177,53 @@ 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,
         certs_url=sync_id_token._GOOGLE_APIS_CERTS_URL,
         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.

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,23 +220,53 @@ 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,
         certs_url=_GOOGLE_APIS_CERTS_URL,
         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.