feat(appcheck): Verify one-time tokens for replay protection - #976
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request adds support for stateful token verification in Firebase App Check by introducing a consume parameter to the verify_token function. When enabled, the service calls the App Check backend to mark the token as consumed, providing replay protection, and returns an already_consumed flag. Corresponding unit tests have also been added. The review feedback suggests improving error handling by using _utils.handle_platform_error_from_requests instead of _utils.handle_requests_error to propagate detailed GCP error messages to developers.
lahirumaramba
left a comment
lahirumaramba
left a comment
There was a problem hiding this comment.
Thanks! Looks great!
I added a few comments, let me know what you think.
|
|
||
| verified_claims['app_id'] = verified_claims.get('sub') | ||
|
|
||
| if not isinstance(consume, bool): |
There was a problem hiding this comment.
Let's move this to the beginning of the function (along side of check_string) that way we fail fast for invalid inputs before obtaining the key sets. If you think this check will be reused, take a look at the check_boolean helper in firebase_admin/_messaging_encoder.py
There was a problem hiding this comment.
I think this also missing unit tests
There was a problem hiding this comment.
Done! Added check_boolean to _Validators and moved the validation to the top before doing any JWKS/crypto operations. Also added unit tests for non-boolean consume arguments.
| raise _utils.handle_platform_error_from_requests(error) | ||
|
|
||
| already_consumed = False | ||
| if isinstance(body, dict): |
There was a problem hiding this comment.
What if body is not a dictionary, do we fail silently and return the token as not already consumed?
This could be an issue if the BE response is malformed for some reason and we silently falls back to already_consumed = False. WDYT about throwing an exception here?
if not isinstance(body, dict):
raise exceptions.UnknownError(
'Unexpected response from App Check service. '
f'Expected a JSON object, but got {type(body).__name__}.')There was a problem hiding this comment.
Great catch! I've updated the logic to raise exceptions.UnknownError if body is not a dict, and added unit test coverage for malformed/non-dict responses.
| 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) |
There was a problem hiding this comment.
nit: moving the replay verification to a separate helper method will make the main verify token function less cluttered.
WDYT about something like:
def _verify_replay_protection(self, token: str) -> bool:
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}')
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))| 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}) |
There was a problem hiding this comment.
Add a test to verify when consume=False (the default), no HTTP call is made to _http_client.body, and already_consumed is not present in the returned dictionary.
| 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): |
There was a problem hiding this comment.
nit: can we test for more specific error types instead of checking the more broader FirebaseError?
| 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")) |
There was a problem hiding this comment.
Can these mocker.patch calls be grouped into a pytest fixture and be reused in here, ... already_consumed , and ...backend_error?
| 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. |
There was a problem hiding this comment.
ValueError: If ``consume`` is not a boolean, or if the app's ``project_id``...
5 participants
This PR adds support for App Check one-time token verification for replay protection by adding an optional
consumeparameter toapp_check.verify_token(). The returned claims dictionary will contain analready_consumedboolean key indicating whether the token was previously consumed.