fix(identity): accept cloudfunctions.net auth blocking token audiences

pyproject.toml

@@ -45,6 +45,7 @@ dev = [
   "pytest>=7.1.2,<10",
   "setuptools>=63.4.2",
   "pytest-cov>=3.0.0",
+  "cryptography>=3.4.0",
   "mypy>=1.0.0",
   "sphinx>=6.1.3",
   "sphinxcontrib-napoleon>=0.7",

src/firebase_functions/private/token_verifier.py

@@ -15,6 +15,8 @@
 Module for internal token verification.
 """
 
+from urllib.parse import urlsplit
+
 import google.auth.exceptions
 import google.oauth2.id_token
 import google.oauth2.service_account
@@ -29,12 +31,32 @@
 )
 from google.auth import jwt
 
+_BLOCKING_HOST_SUFFIXES = (".run.app", ".cloudfunctions.net")
+
+
+def _blocking_audience_matcher(audience):
+    """The `audience_matcher` for auth blocking tokens.
+
+    A blocking token's `aud` is the function's own URL. firebase-tools registers the
+    cloudfunctions.net URL when it creates a blocking function and the run.app URL
+    when it updates one, so either host has to be accepted.
+
+    Neither host is project-scoped: `iss` ties the token to this project. The `aud`
+    check only has to separate blocking tokens from regular ID tokens, whose `aud` is
+    the bare project id.
+    """
+    if not isinstance(audience, str):
+        return False
+    parts = urlsplit(audience)
+    host = parts.hostname or ""
+    return parts.scheme == "https" and host.endswith(_BLOCKING_HOST_SUFFIXES)
+
 
 # pylint: disable=consider-using-f-string
 # mypy: ignore-errors
 # TODO remove once firebase-admin supports this directly.
 # Modified from src/firebase_admin/_token_gen.py to add
-# support for app_check tokens (expected_audience kwarg and
+# support for app_check tokens (audience_matcher kwarg and
 # usage are new, plus None for audience on google.oauth2.id_token.verify_token call)
 class _JWTVerifier:
     """Verifies Firebase JWTs (ID tokens or session cookies)."""
@@ -46,7 +68,8 @@ def __init__(self, **kwargs):
         self.url = kwargs.pop("doc_url")
         self.cert_url = kwargs.pop("cert_url")
         self.issuer = kwargs.pop("issuer")
-        self.expected_audience = kwargs.pop("expected_audience")
+        self.audience_matcher = kwargs.pop("audience_matcher", None)
+        self.expected_audience_msg = kwargs.pop("expected_audience_msg", None)
         if self.short_name[0].lower() in "aeiou":
             self.articled_short_name = f"an {self.short_name}"
         else:
@@ -102,12 +125,12 @@ def verify(self, token, request):
                     self.short_name, header.get("alg"), verify_id_token_msg
                 )
             )
-        elif not emulated and self.expected_audience and self.expected_audience not in audience:
+        elif not emulated and self.audience_matcher and not self.audience_matcher(audience):
             error_message = (
-                f'Firebase {self.short_name} has incorrect "aud" (audience) claim. Expected "{self.expected_audience}" but '
+                f'Firebase {self.short_name} has incorrect "aud" (audience) claim. Expected {self.expected_audience_msg} but '
                 f'got "{audience}". {project_id_match_msg} {verify_id_token_msg}'
             )
-        elif not emulated and not self.expected_audience and audience != self.project_id:
+        elif not emulated and not self.audience_matcher and audience != self.project_id:
             error_message = (
                 f'Firebase {self.short_name} has incorrect "aud" (audience) claim. Expected "{self.project_id}" but '
                 f'got "{audience}". {project_id_match_msg} {verify_id_token_msg}'
@@ -136,9 +159,9 @@ def verify(self, token, request):
                 verified_claims = google.oauth2.id_token.verify_token(
                     token,
                     request=request,
-                    # If expected_audience is set then we have already verified
+                    # If audience_matcher is set then we have already verified
                     # the audience above.
-                    audience=(None if self.expected_audience else self.project_id),
+                    audience=(None if self.audience_matcher else self.project_id),
                     certs_url=self.cert_url,
                 )
             verified_claims["uid"] = verified_claims["sub"]
@@ -189,7 +212,8 @@ def __init__(self, app):
             issuer=_token_gen.ID_TOKEN_ISSUER_PREFIX,
             invalid_token_error=InvalidAuthBlockingTokenError,
             expired_token_error=ExpiredAuthBlockingTokenError,
-            expected_audience="run.app",  # v2 only
+            audience_matcher=_blocking_audience_matcher,
+            expected_audience_msg="a https://*.run.app or https://*.cloudfunctions.net function URL",
         )
 
     def verify_auth_blocking_token(self, auth_blocking_token):

tests/test_identity_fn.py

@@ -9,6 +9,7 @@
 from werkzeug.test import EnvironBuilder
 
 from firebase_functions import core, identity_fn
+from firebase_functions.private import _identity_fn
 
 token_verifier_mock = MagicMock()
 token_verifier_mock.verify_auth_blocking_token = Mock(
@@ -20,9 +21,6 @@
         "iat": 0,
     }
 )
-mocked_modules = {
-    "firebase_functions.private.token_verifier": token_verifier_mock,
-}
 
 
 class TestIdentity(unittest.TestCase):
@@ -38,7 +36,7 @@ def init():
             nonlocal hello
             hello = "world"
 
-        with patch.dict("sys.modules", mocked_modules):
+        with patch.object(_identity_fn, "_token_verifier", token_verifier_mock):
             app = Flask(__name__)
 
             func = Mock(__name__="example_func", return_value=identity_fn.BeforeSignInResponse())

tests/test_token_verifier.py

@@ -0,0 +1,148 @@
+# Copyright 2026 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# 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.
+"""
+Tests for auth blocking token verification.
+"""
+
+import base64
+import datetime
+import json
+import time
+
+import firebase_admin
+import google.oauth2.id_token
+import pytest
+from cryptography import x509
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import padding, rsa
+from cryptography.x509.oid import NameOID
+
+from firebase_functions.private import token_verifier
+
+PROJECT_ID = "test-project"
+KEY_ID = "test-key-id"
+RUN_APP_AUDIENCE = "https://before-create-7kndfybk7q-ue.a.run.app"
+CLOUDFUNCTIONS_AUDIENCE = f"https://us-east1-{PROJECT_ID}.cloudfunctions.net/before_create"
+
+
+@pytest.fixture(scope="module")
+def signing_key():
+    return rsa.generate_private_key(public_exponent=65537, key_size=2048)
+
+
+@pytest.fixture(autouse=True)
+def _signing_certs(monkeypatch, signing_key):
+    """Serve the test signing key in place of Google's public certs."""
+    subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "test")])
+    now = datetime.datetime.now(datetime.timezone.utc)
+    certificate = (
+        x509.CertificateBuilder()
+        .subject_name(subject)
+        .issuer_name(subject)
+        .public_key(signing_key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(now - datetime.timedelta(days=1))
+        .not_valid_after(now + datetime.timedelta(days=1))
+        .sign(signing_key, hashes.SHA256())
+    )
+    certs = {KEY_ID: certificate.public_bytes(serialization.Encoding.PEM).decode()}
+    monkeypatch.setattr(google.oauth2.id_token, "_fetch_certs", lambda request, certs_url: certs)
+
+
+@pytest.fixture(autouse=True)
+def app():
+    app = firebase_admin.initialize_app(
+        options={"projectId": PROJECT_ID}, name="token-verifier-test"
+    )
+    yield app
+    firebase_admin.delete_app(app)
+
+
+def _segment(value: dict) -> str:
+    return base64.urlsafe_b64encode(json.dumps(value).encode()).decode().rstrip("=")
+
+
+def _token(signing_key, audience: str, issuer: str | None = None) -> str:
+    header = {"alg": "RS256", "kid": KEY_ID, "typ": "JWT"}
+    payload = {
+        "aud": audience,
+        "iss": issuer or f"https://securetoken.google.com/{PROJECT_ID}",
+        "sub": "uid123",
+        "iat": int(time.time()) - 10,
+        "exp": int(time.time()) + 3600,
+        "event_type": "beforeCreate",
+    }
+    signing_input = f"{_segment(header)}.{_segment(payload)}"
+    signature = signing_key.sign(signing_input.encode(), padding.PKCS1v15(), hashes.SHA256())
+    return f"{signing_input}.{base64.urlsafe_b64encode(signature).decode().rstrip('=')}"
+
+
+def _verify(app, token: str):
+    return token_verifier.AuthBlockingTokenVerifier(app).verify_auth_blocking_token(token)
+
+
+@pytest.mark.parametrize(
+    "audience",
+    [
+        RUN_APP_AUDIENCE,
+        CLOUDFUNCTIONS_AUDIENCE,
+        # Run hosts: the current form carries the project number, the legacy form a
+        # hash, and a revision tag prefixes the service with `tag---`.
+        "https://beforecreate-123456789.us-central1.run.app",
+        "https://tag---beforecreate-123456789.us-central1.run.app",
+        f"https://northamerica-northeast1-{PROJECT_ID}.cloudfunctions.net/before_create",
+    ],
+)
+def test_accepts_function_url_audiences(app, signing_key, audience):
+    assert _verify(app, _token(signing_key, audience))["uid"] == "uid123"
+
+
+@pytest.mark.parametrize(
+    "audience",
+    [
+        # A regular Firebase ID token for the same project.
+        PROJECT_ID,
+        "https://example.com/before_create",
+        # The expected host must be the host, not text anywhere in the URL.
+        f"https://us-east1-{PROJECT_ID}.cloudfunctions.net.example.com/before_create",
+        f"https://example.com/{PROJECT_ID}.cloudfunctions.net/before_create",
+        f"https://example.com#us-east1-{PROJECT_ID}.cloudfunctions.net/",
+        "https://example.com/?x=run.app",
+        "https://run.app.example.com/before_create",
+        f"http://us-east1-{PROJECT_ID}.cloudfunctions.net/before_create",
+    ],
+)
+def test_rejects_foreign_audience(app, signing_key, audience):
+    with pytest.raises(token_verifier.InvalidAuthBlockingTokenError, match='"aud"'):
+        _verify(app, _token(signing_key, audience))
+
+
+@pytest.mark.parametrize(
+    "other_project",
+    ["other-project", f"other-{PROJECT_ID}", f"x1-{PROJECT_ID}", f"a-{PROJECT_ID}"],
+)
+def test_rejects_function_url_from_another_project(app, signing_key, other_project):
+    token = _token(
+        signing_key,
+        f"https://us-east1-{other_project}.cloudfunctions.net/before_create",
+        issuer=f"https://securetoken.google.com/{other_project}",
+    )
+    with pytest.raises(token_verifier.InvalidAuthBlockingTokenError, match='"iss"'):
+        _verify(app, token)
+
+
+def test_rejects_token_signed_by_another_key(app):
+    other_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    with pytest.raises(token_verifier.InvalidAuthBlockingTokenError, match="signature"):
+        _verify(app, _token(other_key, CLOUDFUNCTIONS_AUDIENCE))