refactor: remove verify_token and complete spit implementation

Author: exaby73

src/firebase_functions/https_fn.py

@@ -354,8 +354,7 @@ class CallableRequest(_typing.Generic[_core.T]):
 
 def _on_call_handler(func: _C2,
                      request: Request,
-                     enforce_app_check: bool,
-                     verify_token: bool = True) -> Response:
+                     enforce_app_check: bool) -> Response:
     try:
         if not _util.valid_on_call_request(request):
             _logging.error("Invalid request, unable to process.")
@@ -365,8 +364,7 @@ def _on_call_handler(func: _C2,
             data=_json.loads(request.data)["data"],
         )
 
-        token_status = _util.on_call_check_tokens(request,
-                                                  verify_token=verify_token)
+        token_status = _util.on_call_check_tokens(request)
 
         if token_status.auth == _util.OnCallTokenState.INVALID:
             raise HttpsError(FunctionsErrorCode.UNAUTHENTICATED,
@@ -420,7 +418,7 @@ def _on_call_handler(func: _C2,
 def on_request(**kwargs) -> _typing.Callable[[_C1], _C1]:
     """
     Handler which handles HTTPS requests.
-    Requires a function that takes a ``Request`` and ``Response`` object, 
+    Requires a function that takes a ``Request`` and ``Response`` object,
     the same signature as a Flask app.
 
     Example:

src/firebase_functions/private/util.py

@@ -211,12 +211,9 @@ def as_dict(self) -> dict:
         }
 
 
-def _on_call_check_auth_token(
-    request: _Request,
-    verify_token: bool = True,
-) -> None | _typing.Literal[OnCallTokenState.INVALID] | dict[str, _typing.Any]:
+def _on_call_check_auth_token(request: _Request) -> None | _typing.Literal[OnCallTokenState.INVALID] | dict[str, _typing.Any]:
     """
-        Validates the auth token in a callable request. 
+        Validates the auth token in a callable request.
         If verify_token is False, the token will be decoded without verification.
     """
     authorization = request.headers.get("Authorization")
@@ -227,10 +224,7 @@ def _on_call_check_auth_token(
         return OnCallTokenState.INVALID
     try:
         id_token = authorization.replace("Bearer ", "")
-        if verify_token:
-            auth_token = _auth.verify_id_token(id_token)
-        else:
-            auth_token = _unsafe_decode_id_token(id_token)
+        auth_token = _auth.verify_id_token(id_token)
         return auth_token
     # pylint: disable=broad-except
     except Exception as err:
@@ -273,25 +267,23 @@ def _unsafe_decode_id_token(token: str):
     return payload
 
 
-def on_call_check_tokens(request: _Request,
-                         verify_token: bool = True) -> _OnCallTokenVerification:
+def on_call_check_tokens(request: _Request) -> _OnCallTokenVerification:
     """Check tokens"""
     verifications = _OnCallTokenVerification()
 
-    auth_token = _on_call_check_auth_token(request, verify_token=verify_token)
+    auth_token = _on_call_check_auth_token(request)
     if auth_token is None:
         verifications.auth = OnCallTokenState.MISSING
     elif isinstance(auth_token, dict):
         verifications.auth = OnCallTokenState.VALID
         verifications.auth_token = auth_token
 
-    if verify_token:
-        app_token = _on_call_check_app_token(request)
-        if app_token is None:
-            verifications.app = OnCallTokenState.MISSING
-        elif isinstance(app_token, dict):
-            verifications.app = OnCallTokenState.VALID
-            verifications.app_token = app_token
+    app_token = _on_call_check_app_token(request)
+    if app_token is None:
+        verifications.app = OnCallTokenState.MISSING
+    elif isinstance(app_token, dict):
+        verifications.app = OnCallTokenState.VALID
+        verifications.app_token = app_token
 
     log_payload = {
         **verifications.as_dict(),
@@ -301,7 +293,7 @@ def on_call_check_tokens(request: _Request,
     }
 
     errs = []
-    if verify_token and verifications.app == OnCallTokenState.INVALID:
+    if verifications.app == OnCallTokenState.INVALID:
         errs.append(("AppCheck token was rejected.", log_payload))
 
     if verifications.auth == OnCallTokenState.INVALID:

src/firebase_functions/tasks_fn.py

@@ -16,19 +16,24 @@
 # pylint: disable=protected-access
 import typing as _typing
 import functools as _functools
+import dataclasses as _dataclasses
+import json as _json
 
-from flask import Request, Response
+from flask import Request, Response, make_response as _make_response, jsonify as _jsonify
 
+import firebase_functions.core as _core
 import firebase_functions.options as _options
 import firebase_functions.private.util as _util
-from firebase_functions.https_fn import CallableRequest, _on_call_handler
+from firebase_functions.https_fn import CallableRequest, HttpsError, FunctionsErrorCode
+
+from functions_framework import logging as _logging
 
 _C = _typing.Callable[[CallableRequest[_typing.Any]], _typing.Any]
+_C1 = _typing.Callable[[Request], Response]
+_C2 = _typing.Callable[[CallableRequest[_typing.Any]], _typing.Any]
 
 def _on_call_handler(func: _C2,
-                     request: Request,
-                     enforce_app_check: bool,
-                     verify_token: bool = True) -> Response:
+                     request: Request) -> Response:
     try:
         if not _util.valid_on_call_request(request):
             _logging.error("Invalid request, unable to process.")
@@ -38,33 +43,6 @@ def _on_call_handler(func: _C2,
             data=_json.loads(request.data)["data"],
         )
 
-        token_status = _util.on_call_check_tokens(request,
-                                                  verify_token=verify_token)
-
-        if token_status.auth == _util.OnCallTokenState.INVALID:
-            raise HttpsError(FunctionsErrorCode.UNAUTHENTICATED,
-                             "Unauthenticated")
-
-        if enforce_app_check and token_status.app in (
-                _util.OnCallTokenState.MISSING, _util.OnCallTokenState.INVALID):
-            raise HttpsError(FunctionsErrorCode.UNAUTHENTICATED,
-                             "Unauthenticated")
-        if token_status.app == _util.OnCallTokenState.VALID and token_status.app_token is not None:
-            context = _dataclasses.replace(
-                context,
-                app=AppCheckData(token_status.app_token["sub"],
-                                 token_status.app_token),
-            )
-
-        if token_status.auth_token is not None:
-            context = _dataclasses.replace(
-                context,
-                auth=AuthData(
-                    token_status.auth_token["uid"]
-                    if "uid" in token_status.auth_token else None,
-                    token_status.auth_token),
-            )
-
         instance_id = request.headers.get("Firebase-Instance-ID-Token")
         if instance_id is not None:
             # Validating the token requires an http request, so we don't do it.
@@ -116,10 +94,7 @@ def on_task_dispatched_decorator(func: _C):
 
         @_functools.wraps(func)
         def on_task_dispatched_wrapped(request: Request) -> Response:
-            return _on_call_handler(func,
-                                    request,
-                                    enforce_app_check=False,
-                                    verify_token=False)
+            return _on_call_handler(func, request)
 
         _util.set_func_endpoint_attr(
             on_task_dispatched_wrapped,