feat(appcheck): Verify one-time tokens for replay protection

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

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)