diff --git a/pyproject.toml b/pyproject.toml index ded479a..3fed6aa 100644 --- a/pyproject.toml +++ b/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", diff --git a/src/firebase_functions/private/token_verifier.py b/src/firebase_functions/private/token_verifier.py index 096b113..3b21f06 100644 --- a/src/firebase_functions/private/token_verifier.py +++ b/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): diff --git a/tests/test_identity_fn.py b/tests/test_identity_fn.py index b3d43fb..19421a3 100644 --- a/tests/test_identity_fn.py +++ b/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()) diff --git a/tests/test_token_verifier.py b/tests/test_token_verifier.py new file mode 100644 index 0000000..f3f9d15 --- /dev/null +++ b/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)) diff --git a/uv.lock b/uv.lock index 7948369..e537a66 100644 --- a/uv.lock +++ b/uv.lock @@ -472,6 +472,7 @@ dependencies = [ [package.dev-dependencies] dev = [ { name = "build" }, + { name = "cryptography" }, { name = "google-cloud-tasks" }, { name = "mypy" }, { name = "pytest" }, @@ -502,6 +503,7 @@ requires-dist = [ [package.metadata.requires-dev] dev = [ { name = "build", specifier = ">=1.0.0" }, + { name = "cryptography", specifier = ">=3.4.0" }, { name = "google-cloud-tasks", specifier = ">=2.13.1" }, { name = "mypy", specifier = ">=1.0.0" }, { name = "pytest", specifier = ">=7.1.2,<10" },