Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
address PR comments
  • Loading branch information
lahirumaramba committed Nov 28, 2025
commit 23505ebe7fcf39132cc9396d62dac558bf556c21
11 changes: 10 additions & 1 deletion firebase_admin/app_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,10 @@
import jwt
from jwt import PyJWKClient, ExpiredSignatureError, InvalidTokenError, DecodeError
from jwt import InvalidAudienceError, InvalidIssuerError, InvalidSignatureError
import requests
from firebase_admin import _utils
from firebase_admin import _http_client
import requests
from firebase_admin import exceptions

_APP_CHECK_ATTRIBUTE = '_app_check'

Expand Down Expand Up @@ -106,9 +107,17 @@ def _verify_replay_protection(self, token: str) -> bool:
body = {'app_check_token': token}
try:
response = self._http_client.body('post', path, json=body)
if not isinstance(response, dict):
raise exceptions.UnknownError(
'Unexpected response from App Check service. '
f'Expected a JSON object, but got {type(response).__name__}.')
return response.get('alreadyConsumed', False)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The current implementation response.get('alreadyConsumed', False) is not fully robust. If the alreadyConsumed key is present in the API response but its value is not a boolean (e.g., None), this function would return that non-boolean value. This could lead to unexpected behavior for consumers of verify_token.

To ensure this function always returns a boolean, it's safer to explicitly check for the value being True.

Suggested change
return response.get('alreadyConsumed', False)
return response.get('alreadyConsumed') is True

except requests.exceptions.RequestException as error:
raise _utils.handle_platform_error_from_requests(error)
Comment thread
lahirumaramba marked this conversation as resolved.
except ValueError as error:
raise exceptions.UnknownError(
'Unexpected response from App Check service. '
f'Error: {error}')

def _has_valid_token_headers(self, headers: Any) -> None:
"""Checks whether the token has valid headers for App Check."""
Expand Down
55 changes: 51 additions & 4 deletions tests/test_app_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -264,16 +264,63 @@ def test_verify_token_with_consume_network_error(self, mocker):
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"))
mock_http_client = mocker.patch("firebase_admin._http_client.JsonHttpClient")
mock_http_client.return_value.body.side_effect = requests.exceptions.RequestException("Network error")

mock_http_client.return_value.body.side_effect = requests.exceptions.RequestException(
"Network error")

# Use a fresh app to ensure _AppCheckService is re-initialized with the mock
cred = testutils.MockCredential()
app = firebase_admin.initialize_app(
cred, {'projectId': PROJECT_ID}, name='test_consume_error')

try:
with pytest.raises(exceptions.UnknownError) as excinfo:
app_check.verify_token("encoded", app, consume=True)
assert str(excinfo.value) == (
"Unknown error while making a remote service call: Network error")
finally:
firebase_admin.delete_app(app)

def test_verify_token_with_consume_non_dict_response(self, mocker):
"""Test verify_token with consume=True handles non-dict response."""
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"))
mock_http_client = mocker.patch("firebase_admin._http_client.JsonHttpClient")
mock_http_client.return_value.body.return_value = ["not", "a", "dict"]

# Use a fresh app to ensure _AppCheckService is re-initialized with the mock
cred = testutils.MockCredential()
app = firebase_admin.initialize_app(
cred, {'projectId': PROJECT_ID}, name='test_consume_non_dict')

try:
with pytest.raises(exceptions.UnknownError) as excinfo:
app_check.verify_token("encoded", app, consume=True)
assert str(excinfo.value) == (
'Unexpected response from App Check service. '
'Expected a JSON object, but got list.')
finally:
firebase_admin.delete_app(app)

def test_verify_token_with_consume_malformed_json(self, mocker):
"""Test verify_token with consume=True handles malformed JSON response."""
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"))
mock_http_client = mocker.patch("firebase_admin._http_client.JsonHttpClient")
mock_http_client.return_value.body.side_effect = ValueError("Malformed JSON")

# Use a fresh app to ensure _AppCheckService is re-initialized with the mock
cred = testutils.MockCredential()
app = firebase_admin.initialize_app(cred, {'projectId': PROJECT_ID}, name='test_consume_error')
app = firebase_admin.initialize_app(
cred, {'projectId': PROJECT_ID}, name='test_consume_malformed_json')

try:
with pytest.raises(exceptions.UnknownError) as excinfo:
app_check.verify_token("encoded", app, consume=True)
assert str(excinfo.value) == "Unknown error while making a remote service call: Network error"
assert str(excinfo.value) == (
'Unexpected response from App Check service. '
'Error: Malformed JSON')
finally:
firebase_admin.delete_app(app)
Comment thread
lahirumaramba marked this conversation as resolved.
Outdated

Expand Down
Loading