Merge pull request #44 from invertase/enforce-app-check

feat: support `enforce_app_check` option

Author: taeold

src/firebase_functions/https_fn.py

@@ -25,7 +25,7 @@
 import firebase_functions.core as _core
 from functions_framework import logging as _logging
 
-from firebase_functions.options import HttpsOptions
+from firebase_functions.options import HttpsOptions, _GLOBAL_OPTIONS
 from flask import Request, Response, make_response as _make_response, jsonify as _jsonify
 from flask_cors import cross_origin as _cross_origin
 
@@ -346,7 +346,8 @@ class CallableRequest(_typing.Generic[_core.T]):
 _C2 = _typing.Callable[[CallableRequest[_typing.Any]], _typing.Any]
 
 
-def _on_call_handler(func: _C2, request: Request) -> Response:
+def _on_call_handler(func: _C2, request: Request,
+                     enforce_app_check: bool) -> Response:
     try:
         if not _util.valid_on_call_request(request):
             _logging.error("Invalid request, unable to process.")
@@ -362,23 +363,22 @@ def _on_call_handler(func: _C2, request: Request) -> Response:
             raise HttpsError(FunctionsErrorCode.UNAUTHENTICATED,
                              "Unauthenticated")
 
-        # TODO support for `allowInvalidAppCheckToken`
-        if token_status.app == _util.OnCallTokenState.INVALID:
+        if enforce_app_check and token_status.app in (
+                _util.OnCallTokenState.MISSING, _util.OnCallTokenState.INVALID):
             raise HttpsError(FunctionsErrorCode.UNAUTHENTICATED,
                              "Unauthenticated")
-
-        if token_status.auth_token is not None:
+        if token_status.app == _util.OnCallTokenState.VALID and token_status.app_token is not None:
             context = _dataclasses.replace(
                 context,
-                auth=AuthData(token_status.auth_token["uid"],
-                              token_status.auth_token),
+                app=AppCheckData(token_status.app_token["sub"],
+                                 token_status.app_token),
             )
 
-        if token_status.app_token is not None:
+        if token_status.auth_token is not None:
             context = _dataclasses.replace(
                 context,
-                app=AppCheckData(token_status.app_token["sub"],
-                                 token_status.app_token),
+                auth=AuthData(token_status.auth_token["uid"],
+                              token_status.auth_token),
             )
 
         instance_id = request.headers.get("Firebase-Instance-ID-Token")
@@ -474,13 +474,26 @@ def on_call_inner_decorator(func: _C2):
         if options.cors is not None and options.cors.cors_origins is not None:
             origins = options.cors.cors_origins
 
+        # Default to False.
+        enforce_app_check = False
+        # If the global option is set, use that.
+        if options.enforce_app_check is None and _GLOBAL_OPTIONS.enforce_app_check is not None:
+            enforce_app_check = _GLOBAL_OPTIONS.enforce_app_check
+        # If the global option is not set, use the local option.
+        elif options.enforce_app_check is not None:
+            enforce_app_check = options.enforce_app_check
+
         @_cross_origin(
             methods="POST",
             origins=origins,
         )
         @_functools.wraps(func)
         def on_call_wrapped(request: Request):
-            return _on_call_handler(func, request)
+            return _on_call_handler(
+                func,
+                request,
+                enforce_app_check,
+            )
 
         _util.set_func_endpoint_attr(
             on_call_wrapped,

src/firebase_functions/options.py

@@ -196,6 +196,14 @@ class RuntimeOptions:
     Secrets to bind to a function.
     """
 
+    enforce_app_check: bool | None = None
+    """
+    Determines whether Firebase AppCheck is enforced.
+    When true, requests with invalid tokens auto respond with a 401
+    Unauthorized response.
+    When false, requests with invalid tokens set event.app to None.
+    """
+
     def _asdict_with_global_options(self) -> dict:
         """
         Returns the provider options merged with globally defined options.
@@ -438,7 +446,7 @@ def _asdict_with_global_options(self) -> dict:
         """
         merged_options = super()._asdict_with_global_options()
         # "cors" is only used locally by the functions framework
-        # and is not used in the manifest.
+        # and is not used in the manifest or in global options.
         if "cors" in merged_options:
             del merged_options["cors"]
         return merged_options
@@ -499,6 +507,7 @@ def set_global_options(
     ingress: IngressSetting | _util.Sentinel | None = None,
     labels: dict[str, str] | None = None,
     secrets: list[str] | list[SecretParam] | _util.Sentinel | None = None,
+    enforce_app_check: bool | None = None,
 ):
     """
     Sets default options for all functions.
@@ -518,4 +527,5 @@ def set_global_options(
         ingress=ingress,
         labels=labels,
         secrets=secrets,
+        enforce_app_check=enforce_app_check,
     )

src/firebase_functions/private/util.py

@@ -155,7 +155,7 @@ class _OnCallTokenVerification:
     """
 
     app: OnCallTokenState = OnCallTokenState.INVALID
-    app_token: _typing.Optional[dict] = None
+    app_token: _typing.Optional[dict[str, _typing.Any]] = None
     auth: OnCallTokenState = OnCallTokenState.INVALID
     auth_token: _typing.Optional[dict] = None