From 3e328cef4049b5ebb29f1caa4fbff60a2706d212 Mon Sep 17 00:00:00 2001 From: Yvonne Pan <103622026+yvonnep165@users.noreply.github.com> Date: Mon, 17 Aug 2026 14:10:37 -0400 Subject: [PATCH 01/14] add one time token verification support --- firebase_admin/app_check.py | 35 +++++++++++++++++++++++++++++------ 1 file changed, 29 insertions(+), 6 deletions(-) diff --git a/firebase_admin/app_check.py b/firebase_admin/app_check.py index 40d857f4..2c712493 100644 --- a/firebase_admin/app_check.py +++ b/firebase_admin/app_check.py @@ -15,38 +15,47 @@ """Firebase App Check module.""" from typing import Any, Dict +import requests import jwt from jwt import PyJWKClient, ExpiredSignatureError, InvalidTokenError, DecodeError from jwt import InvalidAudienceError, InvalidIssuerError, InvalidSignatureError -from firebase_admin import _utils +from firebase_admin import _http_client, _utils _APP_CHECK_ATTRIBUTE = '_app_check' def _get_app_check_service(app) -> Any: return _utils.get_app_service(app, _APP_CHECK_ATTRIBUTE, _AppCheckService) -def verify_token(token: str, app=None) -> Dict[str, Any]: +def verify_token(token: str, app=None, consume: bool = False) -> Dict[str, Any]: """Verifies a Firebase App Check token. Args: token: A token from App Check. app: An App instance (optional). + consume: If set to ``True``, performs stateful verification with the App Check + backend to mark the token as consumed for replay protection. Defaults to ``False``. Returns: - Dict[str, Any]: The token's decoded claims. + Dict[str, Any]: The token's decoded claims. If ``consume`` is ``True``, the dictionary + also includes an ``already_consumed`` boolean key indicating whether the token was + previously consumed. Raises: ValueError: If the app's ``project_id`` is invalid or unspecified, - or if the token's headers or payload are invalid. + or if the token's headers or payload are invalid. + FirebaseError: If an error occurs while communicating with the App Check service. PyJWKClientError: If PyJWKClient fails to fetch a valid signing key. """ - return _get_app_check_service(app).verify_token(token) + return _get_app_check_service(app).verify_token(token, consume=consume) class _AppCheckService: """Service class that implements Firebase App Check functionality.""" _APP_CHECK_ISSUER = 'https://firebaseappcheck.googleapis.com/' _JWKS_URL = 'https://firebaseappcheck.googleapis.com/v1/jwks' + _VERIFY_URL_FORMAT = ( + 'https://firebaseappcheck.googleapis.com/v1/projects/{project_id}:verifyAppCheckToken' + ) _project_id = None _scoped_project_id = None _jwks_client = None @@ -68,9 +77,12 @@ def __init__(self, app): # Default lifespan is 300 seconds (5 minutes) so we change it to 21600 seconds (6 hours). self._jwks_client = PyJWKClient( self._JWKS_URL, lifespan=21600, headers=self._APP_CHECK_HEADERS) + timeout = app.options.get('httpTimeout', _http_client.DEFAULT_TIMEOUT_SECONDS) + self._http_client = _http_client.JsonHttpClient( + credential=app.credential.get_credential(), timeout=timeout) - def verify_token(self, token: str) -> Dict[str, Any]: + def verify_token(self, token: str, consume: bool = False) -> Dict[str, Any]: """Verifies a Firebase App Check token.""" _Validators.check_string("app check token", token) @@ -87,6 +99,17 @@ def verify_token(self, token: str) -> Dict[str, Any]: ) from exception verified_claims['app_id'] = verified_claims.get('sub') + + if consume: + url = self._VERIFY_URL_FORMAT.format(project_id=self._project_id) + try: + body = self._http_client.body('post', url, json={'app_check_token': token}) + except requests.exceptions.RequestException as error: + raise _utils.handle_requests_error(error) + + already_consumed = body.get('alreadyConsumed', False) if isinstance(body, dict) else False + verified_claims['already_consumed'] = bool(already_consumed) + return verified_claims def _has_valid_token_headers(self, headers: Any) -> None: From 02f5f74d2b30598e4f465e1003a04279bf93072e Mon Sep 17 00:00:00 2001 From: Yvonne Pan <103622026+yvonnep165@users.noreply.github.com> Date: Mon, 17 Aug 2026 14:47:53 -0400 Subject: [PATCH 02/14] add unit tests --- tests/test_app_check.py | 50 ++++++++++++++++++++++++++++++++++++++++- 1 file changed, 49 insertions(+), 1 deletion(-) diff --git a/tests/test_app_check.py b/tests/test_app_check.py index e55ae39d..152420f3 100644 --- a/tests/test_app_check.py +++ b/tests/test_app_check.py @@ -19,7 +19,7 @@ from jwt import PyJWK, InvalidAudienceError, InvalidIssuerError from jwt import ExpiredSignatureError, InvalidSignatureError import firebase_admin -from firebase_admin import app_check +from firebase_admin import app_check, exceptions from tests import testutils NON_STRING_ARGS = [[], tuple(), {}, True, False, 1, 0] @@ -273,3 +273,51 @@ def test_verify_token_with_incorrect_issuer_raises_error(self, mocker): expected = 'Token does not contain the correct "iss" (issuer).' assert str(excinfo.value) == expected + + def test_verify_token_with_consume_true_not_consumed(self, mocker): + mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE) + mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key)) + mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers")) + app = firebase_admin.get_app() + app_check_service = app_check._get_app_check_service(app) + mock_body = mocker.patch.object(app_check_service._http_client, "body", return_value={"alreadyConsumed": False}) + + payload = app_check.verify_token("encoded", app=app, consume=True) + expected = JWT_PAYLOAD_SAMPLE.copy() + expected["app_id"] = APP_ID + expected["already_consumed"] = False + assert payload == expected + + expected_url = f"https://firebaseappcheck.googleapis.com/v1/projects/{PROJECT_ID}:verifyAppCheckToken" + mock_body.assert_called_once_with("post", expected_url, json={"app_check_token": "encoded"}) + + def test_verify_token_with_consume_true_already_consumed(self, mocker): + mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE) + mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key)) + mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers")) + app = firebase_admin.get_app() + app_check_service = app_check._get_app_check_service(app) + mock_body = mocker.patch.object(app_check_service._http_client, "body", return_value={"alreadyConsumed": True}) + + payload = app_check.verify_token("encoded", app=app, consume=True) + expected = JWT_PAYLOAD_SAMPLE.copy() + expected["app_id"] = APP_ID + expected["already_consumed"] = True + assert payload == expected + + expected_url = f"https://firebaseappcheck.googleapis.com/v1/projects/{PROJECT_ID}:verifyAppCheckToken" + mock_body.assert_called_once_with("post", expected_url, json={"app_check_token": "encoded"}) + + def test_verify_token_with_consume_true_backend_error(self, mocker): + import requests + mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE) + mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key)) + mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers")) + app = firebase_admin.get_app() + app_check_service = app_check._get_app_check_service(app) + + req_exc = requests.exceptions.RequestException("Backend error") + mocker.patch.object(app_check_service._http_client, "body", side_effect=req_exc) + + with pytest.raises(exceptions.FirebaseError): + app_check.verify_token("encoded", app=app, consume=True) From bd14d8192d84273f960585db22bea6d9a6eb370f Mon Sep 17 00:00:00 2001 From: Yvonne Pan <103622026+yvonnep165@users.noreply.github.com> Date: Mon, 17 Aug 2026 15:03:11 -0400 Subject: [PATCH 03/14] fix lint error --- firebase_admin/app_check.py | 4 +++- tests/test_app_check.py | 26 +++++++++++++++++++------- 2 files changed, 22 insertions(+), 8 deletions(-) diff --git a/firebase_admin/app_check.py b/firebase_admin/app_check.py index 2c712493..abdd1d8c 100644 --- a/firebase_admin/app_check.py +++ b/firebase_admin/app_check.py @@ -107,7 +107,9 @@ def verify_token(self, token: str, consume: bool = False) -> Dict[str, Any]: except requests.exceptions.RequestException as error: raise _utils.handle_requests_error(error) - already_consumed = body.get('alreadyConsumed', False) if isinstance(body, dict) else False + already_consumed = False + if isinstance(body, dict): + already_consumed = body.get('alreadyConsumed', False) verified_claims['already_consumed'] = bool(already_consumed) return verified_claims diff --git a/tests/test_app_check.py b/tests/test_app_check.py index 152420f3..0ae15c6b 100644 --- a/tests/test_app_check.py +++ b/tests/test_app_check.py @@ -15,6 +15,7 @@ """Test cases for the firebase_admin.app_check module.""" import base64 import pytest +import requests from jwt import PyJWK, InvalidAudienceError, InvalidIssuerError from jwt import ExpiredSignatureError, InvalidSignatureError @@ -280,7 +281,9 @@ def test_verify_token_with_consume_true_not_consumed(self, mocker): mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers")) app = firebase_admin.get_app() app_check_service = app_check._get_app_check_service(app) - mock_body = mocker.patch.object(app_check_service._http_client, "body", return_value={"alreadyConsumed": False}) + mock_body = mocker.patch.object( + app_check_service._http_client, "body", return_value={"alreadyConsumed": False} + ) payload = app_check.verify_token("encoded", app=app, consume=True) expected = JWT_PAYLOAD_SAMPLE.copy() @@ -288,8 +291,12 @@ def test_verify_token_with_consume_true_not_consumed(self, mocker): expected["already_consumed"] = False assert payload == expected - expected_url = f"https://firebaseappcheck.googleapis.com/v1/projects/{PROJECT_ID}:verifyAppCheckToken" - mock_body.assert_called_once_with("post", expected_url, json={"app_check_token": "encoded"}) + expected_url = ( + f"https://firebaseappcheck.googleapis.com/v1/projects/{PROJECT_ID}:verifyAppCheckToken" + ) + mock_body.assert_called_once_with( + "post", expected_url, json={"app_check_token": "encoded"} + ) def test_verify_token_with_consume_true_already_consumed(self, mocker): mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE) @@ -297,7 +304,9 @@ def test_verify_token_with_consume_true_already_consumed(self, mocker): mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers")) app = firebase_admin.get_app() app_check_service = app_check._get_app_check_service(app) - mock_body = mocker.patch.object(app_check_service._http_client, "body", return_value={"alreadyConsumed": True}) + mock_body = mocker.patch.object( + app_check_service._http_client, "body", return_value={"alreadyConsumed": True} + ) payload = app_check.verify_token("encoded", app=app, consume=True) expected = JWT_PAYLOAD_SAMPLE.copy() @@ -305,11 +314,14 @@ def test_verify_token_with_consume_true_already_consumed(self, mocker): expected["already_consumed"] = True assert payload == expected - expected_url = f"https://firebaseappcheck.googleapis.com/v1/projects/{PROJECT_ID}:verifyAppCheckToken" - mock_body.assert_called_once_with("post", expected_url, json={"app_check_token": "encoded"}) + expected_url = ( + f"https://firebaseappcheck.googleapis.com/v1/projects/{PROJECT_ID}:verifyAppCheckToken" + ) + mock_body.assert_called_once_with( + "post", expected_url, json={"app_check_token": "encoded"} + ) def test_verify_token_with_consume_true_backend_error(self, mocker): - import requests mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE) mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key)) mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers")) From df4fcf2b1d904191c01a526e6db8421722254d3d Mon Sep 17 00:00:00 2001 From: Yvonne Pan <103622026+yvonnep165@users.noreply.github.com> Date: Mon, 17 Aug 2026 15:18:17 -0400 Subject: [PATCH 04/14] change to use _utils.handle_platform_error_from_requests --- firebase_admin/app_check.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/firebase_admin/app_check.py b/firebase_admin/app_check.py index abdd1d8c..19c208cb 100644 --- a/firebase_admin/app_check.py +++ b/firebase_admin/app_check.py @@ -105,7 +105,7 @@ def verify_token(self, token: str, consume: bool = False) -> Dict[str, Any]: try: body = self._http_client.body('post', url, json={'app_check_token': token}) except requests.exceptions.RequestException as error: - raise _utils.handle_requests_error(error) + raise _utils.handle_platform_error_from_requests(error) already_consumed = False if isinstance(body, dict): From 96c7083ffb4b53af5ad9d9acebdf95d1e4effbea Mon Sep 17 00:00:00 2001 From: Yvonne Pan <103622026+yvonnep165@users.noreply.github.com> Date: Mon, 17 Aug 2026 15:24:24 -0400 Subject: [PATCH 05/14] change the function description --- firebase_admin/app_check.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/firebase_admin/app_check.py b/firebase_admin/app_check.py index 19c208cb..e865f524 100644 --- a/firebase_admin/app_check.py +++ b/firebase_admin/app_check.py @@ -32,8 +32,8 @@ def verify_token(token: str, app=None, consume: bool = False) -> Dict[str, Any]: Args: token: A token from App Check. app: An App instance (optional). - consume: If set to ``True``, performs stateful verification with the App Check - backend to mark the token as consumed for replay protection. Defaults to ``False``. + consume: Set to ``True`` only if the token is a limited-use (one-time) token + that should be consumed upon verification (optional, defaults to ``False``). Returns: Dict[str, Any]: The token's decoded claims. If ``consume`` is ``True``, the dictionary From ccca46964b30520bcc82fd68e5dafe3960a85f43 Mon Sep 17 00:00:00 2001 From: Yvonne Pan <103622026+yvonnep165@users.noreply.github.com> Date: Mon, 17 Aug 2026 15:32:10 -0400 Subject: [PATCH 06/14] change docstring to state limited-use tokens --- firebase_admin/app_check.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/firebase_admin/app_check.py b/firebase_admin/app_check.py index e865f524..f92277d3 100644 --- a/firebase_admin/app_check.py +++ b/firebase_admin/app_check.py @@ -27,7 +27,7 @@ def _get_app_check_service(app) -> Any: return _utils.get_app_service(app, _APP_CHECK_ATTRIBUTE, _AppCheckService) def verify_token(token: str, app=None, consume: bool = False) -> Dict[str, Any]: - """Verifies a Firebase App Check token. + """Verifies a Firebase App Check token, optionally consuming limited-use tokens. Args: token: A token from App Check. @@ -83,7 +83,7 @@ def __init__(self, app): def verify_token(self, token: str, consume: bool = False) -> Dict[str, Any]: - """Verifies a Firebase App Check token.""" + """Verifies a Firebase App Check token, optionally consuming limited-use tokens.""" _Validators.check_string("app check token", token) # Obtain the Firebase App Check Public Keys From 86d356d4e0686b8497fb059adca0a6637b740714 Mon Sep 17 00:00:00 2001 From: Yvonne Pan <103622026+yvonnep165@users.noreply.github.com> Date: Mon, 24 Aug 2026 18:19:11 -0400 Subject: [PATCH 07/14] add validation for consume variable --- firebase_admin/app_check.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/firebase_admin/app_check.py b/firebase_admin/app_check.py index f92277d3..76eb214b 100644 --- a/firebase_admin/app_check.py +++ b/firebase_admin/app_check.py @@ -100,6 +100,8 @@ def verify_token(self, token: str, consume: bool = False) -> Dict[str, Any]: verified_claims['app_id'] = verified_claims.get('sub') + if not isinstance(consume, bool): + raise ValueError('consume must be a boolean.') if consume: url = self._VERIFY_URL_FORMAT.format(project_id=self._project_id) try: From 4e2f1087f6399e08e0f4396d77fd460b8e7034d2 Mon Sep 17 00:00:00 2001 From: Yvonne Pan <103622026+yvonnep165@users.noreply.github.com> Date: Wed, 16 Sep 2026 15:43:42 -0400 Subject: [PATCH 08/14] add check_boolean to _Validators and add unit tests to check non-boolean consume --- firebase_admin/app_check.py | 10 ++++++++-- tests/test_app_check.py | 6 ++++++ 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/firebase_admin/app_check.py b/firebase_admin/app_check.py index 76eb214b..6b6dda75 100644 --- a/firebase_admin/app_check.py +++ b/firebase_admin/app_check.py @@ -85,6 +85,7 @@ def __init__(self, app): def verify_token(self, token: str, consume: bool = False) -> Dict[str, Any]: """Verifies a Firebase App Check token, optionally consuming limited-use tokens.""" _Validators.check_string("app check token", token) + _Validators.check_boolean("consume", consume) # Obtain the Firebase App Check Public Keys # Note: It is not recommended to hard code these keys as they rotate, @@ -100,8 +101,6 @@ def verify_token(self, token: str, consume: bool = False) -> Dict[str, Any]: verified_claims['app_id'] = verified_claims.get('sub') - if not isinstance(consume, bool): - raise ValueError('consume must be a boolean.') if consume: url = self._VERIFY_URL_FORMAT.format(project_id=self._project_id) try: @@ -186,3 +185,10 @@ def check_string(cls, label: str, value: Any): raise ValueError(f'{label} "{value}" must be a non-empty string.') if not isinstance(value, str): raise ValueError(f'{label} "{value}" must be a string.') + + @classmethod + def check_boolean(cls, label: str, value: Any): + """Checks if the given value is a boolean.""" + if not isinstance(value, bool): + raise ValueError(f'{label} must be a boolean.') + return value diff --git a/tests/test_app_check.py b/tests/test_app_check.py index 0ae15c6b..2ce34a00 100644 --- a/tests/test_app_check.py +++ b/tests/test_app_check.py @@ -75,6 +75,12 @@ def test_verify_token_with_non_string_raises_error(self, token): expected = f'app check token "{token}" must be a string.' assert str(excinfo.value) == expected + @pytest.mark.parametrize('consume', [[], tuple(), {}, 1, 0, 'true', 'false', None]) + def test_verify_token_with_non_boolean_consume_raises_error(self, consume): + with pytest.raises(ValueError) as excinfo: + app_check.verify_token('app_check_token', consume=consume) + assert str(excinfo.value) == 'consume must be a boolean.' + def test_has_valid_token_headers(self): app = firebase_admin.get_app() app_check_service = app_check._get_app_check_service(app) From 40e96d2d73d872a6f13bb8d7d5fcd35e4e044963 Mon Sep 17 00:00:00 2001 From: Yvonne Pan <103622026+yvonnep165@users.noreply.github.com> Date: Wed, 16 Sep 2026 16:00:02 -0400 Subject: [PATCH 09/14] raise `exceptions.UnknownError` if body is not a dict --- firebase_admin/app_check.py | 12 ++++++++---- tests/test_app_check.py | 15 +++++++++++++++ 2 files changed, 23 insertions(+), 4 deletions(-) diff --git a/firebase_admin/app_check.py b/firebase_admin/app_check.py index 6b6dda75..bbf22795 100644 --- a/firebase_admin/app_check.py +++ b/firebase_admin/app_check.py @@ -19,7 +19,7 @@ import jwt from jwt import PyJWKClient, ExpiredSignatureError, InvalidTokenError, DecodeError from jwt import InvalidAudienceError, InvalidIssuerError, InvalidSignatureError -from firebase_admin import _http_client, _utils +from firebase_admin import _http_client, _utils, exceptions _APP_CHECK_ATTRIBUTE = '_app_check' @@ -108,9 +108,13 @@ def verify_token(self, token: str, consume: bool = False) -> Dict[str, Any]: except requests.exceptions.RequestException as error: raise _utils.handle_platform_error_from_requests(error) - already_consumed = False - if isinstance(body, dict): - already_consumed = body.get('alreadyConsumed', False) + if not isinstance(body, dict): + raise exceptions.UnknownError( + 'Unexpected response from App Check service. ' + f'Expected a JSON object, but got {type(body).__name__}.' + ) + + already_consumed = body.get('alreadyConsumed', False) verified_claims['already_consumed'] = bool(already_consumed) return verified_claims diff --git a/tests/test_app_check.py b/tests/test_app_check.py index 2ce34a00..b487b4ce 100644 --- a/tests/test_app_check.py +++ b/tests/test_app_check.py @@ -339,3 +339,18 @@ def test_verify_token_with_consume_true_backend_error(self, mocker): with pytest.raises(exceptions.FirebaseError): app_check.verify_token("encoded", app=app, consume=True) + + @pytest.mark.parametrize('malformed_body', ['string_response', [1, 2], 123, None]) + def test_verify_token_with_consume_true_malformed_response_raises_error( + self, mocker, malformed_body + ): + mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE) + mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key)) + mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers")) + app = firebase_admin.get_app() + app_check_service = app_check._get_app_check_service(app) + mocker.patch.object(app_check_service._http_client, "body", return_value=malformed_body) + + with pytest.raises(exceptions.UnknownError) as excinfo: + app_check.verify_token("encoded", app=app, consume=True) + assert 'Unexpected response from App Check service' in str(excinfo.value) From 7a95f911a17ca8e6f913dfbd53779f103f953e13 Mon Sep 17 00:00:00 2001 From: Yvonne Pan <103622026+yvonnep165@users.noreply.github.com> Date: Wed, 16 Sep 2026 16:11:39 -0400 Subject: [PATCH 10/14] move the replay verification to a separate helper method and add unit tests --- firebase_admin/app_check.py | 35 +++++++++++++++++++++-------------- tests/test_app_check.py | 18 ++++++++++++++++++ 2 files changed, 39 insertions(+), 14 deletions(-) diff --git a/firebase_admin/app_check.py b/firebase_admin/app_check.py index bbf22795..efefea33 100644 --- a/firebase_admin/app_check.py +++ b/firebase_admin/app_check.py @@ -102,23 +102,30 @@ def verify_token(self, token: str, consume: bool = False) -> Dict[str, Any]: verified_claims['app_id'] = verified_claims.get('sub') if consume: - url = self._VERIFY_URL_FORMAT.format(project_id=self._project_id) - try: - body = self._http_client.body('post', url, json={'app_check_token': token}) - except requests.exceptions.RequestException as error: - raise _utils.handle_platform_error_from_requests(error) - - if not isinstance(body, dict): - raise exceptions.UnknownError( - 'Unexpected response from App Check service. ' - f'Expected a JSON object, but got {type(body).__name__}.' - ) - - already_consumed = body.get('alreadyConsumed', False) - verified_claims['already_consumed'] = bool(already_consumed) + verified_claims['already_consumed'] = self._verify_replay_protection(token) return verified_claims + def _verify_replay_protection(self, token: str) -> bool: + """Verifies replay protection with the backend and returns the alreadyConsumed status.""" + url = self._VERIFY_URL_FORMAT.format(project_id=self._project_id) + try: + body = self._http_client.body('post', url, json={'app_check_token': token}) + except requests.exceptions.RequestException as error: + raise _utils.handle_platform_error_from_requests(error) + except ValueError as error: + raise exceptions.UnknownError( + f'Unexpected response from App Check service: {error}' + ) from error + + if not isinstance(body, dict): + raise exceptions.UnknownError( + 'Unexpected response from App Check service. ' + f'Expected a JSON object, but got {type(body).__name__}.' + ) + + return bool(body.get('alreadyConsumed', False)) + def _has_valid_token_headers(self, headers: Any) -> None: """Checks whether the token has valid headers for App Check.""" # Ensure the token's header has type JWT diff --git a/tests/test_app_check.py b/tests/test_app_check.py index b487b4ce..e4a46251 100644 --- a/tests/test_app_check.py +++ b/tests/test_app_check.py @@ -354,3 +354,21 @@ def test_verify_token_with_consume_true_malformed_response_raises_error( with pytest.raises(exceptions.UnknownError) as excinfo: app_check.verify_token("encoded", app=app, consume=True) assert 'Unexpected response from App Check service' in str(excinfo.value) + + def test_verify_token_with_consume_true_json_decode_error(self, mocker): + mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE) + mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key)) + mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers")) + app = firebase_admin.get_app() + app_check_service = app_check._get_app_check_service(app) + mocker.patch.object( + app_check_service._http_client, + "body", + side_effect=ValueError("Expecting value: line 1 column 1 (char 0)"), + ) + + with pytest.raises(exceptions.UnknownError) as excinfo: + app_check.verify_token("encoded", app=app, consume=True) + assert "Unexpected response from App Check service: Expecting value: line 1 column 1 (char 0)" in str( + excinfo.value + ) From 5ce0edb26382011f7849509ed4d6168bacbcd238 Mon Sep 17 00:00:00 2001 From: Yvonne Pan <103622026+yvonnep165@users.noreply.github.com> Date: Wed, 16 Sep 2026 16:16:38 -0400 Subject: [PATCH 11/14] add test_verify_token_with_consume_false_makes_no_http_call unit test --- tests/test_app_check.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/tests/test_app_check.py b/tests/test_app_check.py index e4a46251..dfd8374f 100644 --- a/tests/test_app_check.py +++ b/tests/test_app_check.py @@ -281,6 +281,22 @@ def test_verify_token_with_incorrect_issuer_raises_error(self, mocker): expected = 'Token does not contain the correct "iss" (issuer).' assert str(excinfo.value) == expected + @pytest.mark.parametrize('consume_kwargs', [{}, {'consume': False}]) + def test_verify_token_with_consume_false_makes_no_http_call(self, mocker, consume_kwargs): + mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE) + mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key)) + mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers")) + app = firebase_admin.get_app() + app_check_service = app_check._get_app_check_service(app) + mock_body = mocker.patch.object(app_check_service._http_client, "body") + + payload = app_check.verify_token("encoded", app=app, **consume_kwargs) + expected = JWT_PAYLOAD_SAMPLE.copy() + expected["app_id"] = APP_ID + assert payload == expected + assert 'already_consumed' not in payload + mock_body.assert_not_called() + def test_verify_token_with_consume_true_not_consumed(self, mocker): mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE) mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key)) From 0957eed5d59e95db47413693c49804e653a83452 Mon Sep 17 00:00:00 2001 From: Yvonne Pan <103622026+yvonnep165@users.noreply.github.com> Date: Wed, 16 Sep 2026 16:22:24 -0400 Subject: [PATCH 12/14] replace the broad exceptions.FirebaseError assertion with more specific error types --- tests/test_app_check.py | 20 +++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/tests/test_app_check.py b/tests/test_app_check.py index dfd8374f..92ccfce0 100644 --- a/tests/test_app_check.py +++ b/tests/test_app_check.py @@ -353,8 +353,26 @@ def test_verify_token_with_consume_true_backend_error(self, mocker): req_exc = requests.exceptions.RequestException("Backend error") mocker.patch.object(app_check_service._http_client, "body", side_effect=req_exc) - with pytest.raises(exceptions.FirebaseError): + with pytest.raises(exceptions.UnknownError) as excinfo: + app_check.verify_token("encoded", app=app, consume=True) + assert "Unknown error while making a remote service call" in str(excinfo.value) + + def test_verify_token_with_consume_true_http_error(self, mocker): + mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE) + mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key)) + mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers")) + app = firebase_admin.get_app() + app_check_service = app_check._get_app_check_service(app) + + response = requests.Response() + response.status_code = 403 + response._content = b'{"error": {"status": "PERMISSION_DENIED", "message": "Permission denied."}}' + http_exc = requests.exceptions.HTTPError(response=response) + mocker.patch.object(app_check_service._http_client, "body", side_effect=http_exc) + + with pytest.raises(exceptions.PermissionDeniedError) as excinfo: app_check.verify_token("encoded", app=app, consume=True) + assert "Permission denied." in str(excinfo.value) @pytest.mark.parametrize('malformed_body', ['string_response', [1, 2], 123, None]) def test_verify_token_with_consume_true_malformed_response_raises_error( From d43471f22097c2d542d5af9cd663f4cb75559765 Mon Sep 17 00:00:00 2001 From: Yvonne Pan <103622026+yvonnep165@users.noreply.github.com> Date: Wed, 16 Sep 2026 16:31:46 -0400 Subject: [PATCH 13/14] Refactor token verification tests to use mock_jwt fixture --- tests/test_app_check.py | 56 ++++++++++++++++++++--------------------- 1 file changed, 27 insertions(+), 29 deletions(-) diff --git a/tests/test_app_check.py b/tests/test_app_check.py index 92ccfce0..b7a7bcd3 100644 --- a/tests/test_app_check.py +++ b/tests/test_app_check.py @@ -59,6 +59,15 @@ def setup_class(cls): def teardown_class(cls): testutils.cleanup_apps() + +@pytest.fixture +def mock_jwt(mocker): + """Mocks JWT decoding and JWKS key retrieval for a valid App Check token.""" + mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE) + mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key)) + mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers")) + + class TestVerifyToken(TestBatch): def test_no_project_id(self): @@ -228,10 +237,8 @@ def test_decode_and_verify_with_non_string_sub_raises_error(self, mocker): f'"{sub_number}" must be a string.') assert str(excinfo.value) == expected - def test_verify_token(self, mocker): - mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE) - mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key)) - mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers")) + @pytest.mark.usefixtures("mock_jwt") + def test_verify_token(self): app = firebase_admin.get_app() payload = app_check.verify_token("encoded", app) @@ -281,11 +288,11 @@ def test_verify_token_with_incorrect_issuer_raises_error(self, mocker): expected = 'Token does not contain the correct "iss" (issuer).' assert str(excinfo.value) == expected + @pytest.mark.usefixtures("mock_jwt") @pytest.mark.parametrize('consume_kwargs', [{}, {'consume': False}]) - def test_verify_token_with_consume_false_makes_no_http_call(self, mocker, consume_kwargs): - mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE) - mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key)) - mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers")) + def test_verify_token_with_consume_false_makes_no_http_call( + self, mocker, consume_kwargs + ): app = firebase_admin.get_app() app_check_service = app_check._get_app_check_service(app) mock_body = mocker.patch.object(app_check_service._http_client, "body") @@ -297,10 +304,8 @@ def test_verify_token_with_consume_false_makes_no_http_call(self, mocker, consum assert 'already_consumed' not in payload mock_body.assert_not_called() + @pytest.mark.usefixtures("mock_jwt") def test_verify_token_with_consume_true_not_consumed(self, mocker): - mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE) - mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key)) - mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers")) app = firebase_admin.get_app() app_check_service = app_check._get_app_check_service(app) mock_body = mocker.patch.object( @@ -320,10 +325,8 @@ def test_verify_token_with_consume_true_not_consumed(self, mocker): "post", expected_url, json={"app_check_token": "encoded"} ) + @pytest.mark.usefixtures("mock_jwt") def test_verify_token_with_consume_true_already_consumed(self, mocker): - mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE) - mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key)) - mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers")) app = firebase_admin.get_app() app_check_service = app_check._get_app_check_service(app) mock_body = mocker.patch.object( @@ -343,10 +346,8 @@ def test_verify_token_with_consume_true_already_consumed(self, mocker): "post", expected_url, json={"app_check_token": "encoded"} ) + @pytest.mark.usefixtures("mock_jwt") def test_verify_token_with_consume_true_backend_error(self, mocker): - mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE) - mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key)) - mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers")) app = firebase_admin.get_app() app_check_service = app_check._get_app_check_service(app) @@ -357,16 +358,16 @@ def test_verify_token_with_consume_true_backend_error(self, mocker): app_check.verify_token("encoded", app=app, consume=True) assert "Unknown error while making a remote service call" in str(excinfo.value) + @pytest.mark.usefixtures("mock_jwt") def test_verify_token_with_consume_true_http_error(self, mocker): - mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE) - mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key)) - mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers")) app = firebase_admin.get_app() app_check_service = app_check._get_app_check_service(app) response = requests.Response() response.status_code = 403 - response._content = b'{"error": {"status": "PERMISSION_DENIED", "message": "Permission denied."}}' + response._content = ( + b'{"error": {"status": "PERMISSION_DENIED", "message": "Permission denied."}}' + ) http_exc = requests.exceptions.HTTPError(response=response) mocker.patch.object(app_check_service._http_client, "body", side_effect=http_exc) @@ -374,13 +375,11 @@ def test_verify_token_with_consume_true_http_error(self, mocker): app_check.verify_token("encoded", app=app, consume=True) assert "Permission denied." in str(excinfo.value) + @pytest.mark.usefixtures("mock_jwt") @pytest.mark.parametrize('malformed_body', ['string_response', [1, 2], 123, None]) def test_verify_token_with_consume_true_malformed_response_raises_error( self, mocker, malformed_body ): - mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE) - mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key)) - mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers")) app = firebase_admin.get_app() app_check_service = app_check._get_app_check_service(app) mocker.patch.object(app_check_service._http_client, "body", return_value=malformed_body) @@ -389,10 +388,8 @@ def test_verify_token_with_consume_true_malformed_response_raises_error( app_check.verify_token("encoded", app=app, consume=True) assert 'Unexpected response from App Check service' in str(excinfo.value) + @pytest.mark.usefixtures("mock_jwt") def test_verify_token_with_consume_true_json_decode_error(self, mocker): - mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE) - mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key)) - mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers")) app = firebase_admin.get_app() app_check_service = app_check._get_app_check_service(app) mocker.patch.object( @@ -403,6 +400,7 @@ def test_verify_token_with_consume_true_json_decode_error(self, mocker): with pytest.raises(exceptions.UnknownError) as excinfo: app_check.verify_token("encoded", app=app, consume=True) - assert "Unexpected response from App Check service: Expecting value: line 1 column 1 (char 0)" in str( - excinfo.value + expected_msg = ( + "Unexpected response from App Check service: Expecting value: line 1 column 1 (char 0)" ) + assert expected_msg in str(excinfo.value) From f500ef42feec3ced7a33c20572a20d46ce40a12e Mon Sep 17 00:00:00 2001 From: Yvonne Pan <103622026+yvonnep165@users.noreply.github.com> Date: Wed, 16 Sep 2026 16:36:54 -0400 Subject: [PATCH 14/14] Document ValueError when consume is not a boolean in verify_token docstring --- firebase_admin/app_check.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/firebase_admin/app_check.py b/firebase_admin/app_check.py index efefea33..1117e151 100644 --- a/firebase_admin/app_check.py +++ b/firebase_admin/app_check.py @@ -41,8 +41,8 @@ def verify_token(token: str, app=None, consume: bool = False) -> Dict[str, Any]: previously consumed. Raises: - ValueError: If the app's ``project_id`` is invalid or unspecified, - or if the token's headers or payload are invalid. + ValueError: If ``consume`` is not a boolean, or if the app's ``project_id`` + is invalid or unspecified, or if the token's headers or payload are invalid. FirebaseError: If an error occurs while communicating with the App Check service. PyJWKClientError: If PyJWKClient fails to fetch a valid signing key. """