diff --git a/firebase_admin/app_check.py b/firebase_admin/app_check.py index 40d857f4..1117e151 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, exceptions _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]: - """Verifies a Firebase App Check token. +def verify_token(token: str, app=None, consume: bool = False) -> Dict[str, Any]: + """Verifies a Firebase App Check token, optionally consuming limited-use tokens. Args: token: A token from App Check. app: An App instance (optional). + 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. + 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. + 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. """ - 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,11 +77,15 @@ 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]: - """Verifies a Firebase App Check token.""" + 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, @@ -87,8 +100,32 @@ def verify_token(self, token: str) -> Dict[str, Any]: ) from exception verified_claims['app_id'] = verified_claims.get('sub') + + if consume: + 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 @@ -159,3 +196,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 e55ae39d..b7a7bcd3 100644 --- a/tests/test_app_check.py +++ b/tests/test_app_check.py @@ -15,11 +15,12 @@ """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 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] @@ -58,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): @@ -74,6 +84,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) @@ -221,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) @@ -273,3 +287,120 @@ 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 + ): + 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() + + @pytest.mark.usefixtures("mock_jwt") + def test_verify_token_with_consume_true_not_consumed(self, mocker): + 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"} + ) + + @pytest.mark.usefixtures("mock_jwt") + def test_verify_token_with_consume_true_already_consumed(self, mocker): + 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"} + ) + + @pytest.mark.usefixtures("mock_jwt") + def test_verify_token_with_consume_true_backend_error(self, mocker): + 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.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) + + @pytest.mark.usefixtures("mock_jwt") + def test_verify_token_with_consume_true_http_error(self, mocker): + 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.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 + ): + 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) + + @pytest.mark.usefixtures("mock_jwt") + def test_verify_token_with_consume_true_json_decode_error(self, mocker): + 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) + expected_msg = ( + "Unexpected response from App Check service: Expecting value: line 1 column 1 (char 0)" + ) + assert expected_msg in str(excinfo.value)