feat: Implement before_email_sent and before_sms_sent blocking functions

Author: exaby73

src/firebase_functions/identity_fn.py

@@ -191,7 +191,7 @@ class AdditionalUserInfo:
     The additional user info component of the auth event context.
     """
 
-    provider_id: str
+    provider_id: str | None
     """The provider identifier."""
 
     profile: dict[str, _typing.Any] | None
@@ -206,6 +206,12 @@ class AdditionalUserInfo:
     recaptcha_score: float | None
     """The user's reCAPTCHA score, if available."""
 
+    email: str | None
+    """The user's email, if available."""
+
+    phone_number: str | None
+    """The user's phone number, if available."""
+
 
 @_dataclasses.dataclass(frozen=True)
 class Credential:
@@ -237,14 +243,16 @@ class Credential:
     sign_in_method: str
     """The user's sign-in method."""
 
+EmailType = _typing.Literal["EMAIL_SIGN_IN", "PASSWORD_RESET"]
+SmsType = _typing.Literal["SIGN_IN_OR_SIGN_UP", "MULTI_FACTOR_SIGN_IN", "MULTI_FACTOR_ENROLLMENT"]
 
 @_dataclasses.dataclass(frozen=True)
 class AuthBlockingEvent:
     """
     Defines an auth event for identitytoolkit v2 auth blocking events.
     """
 
-    data: AuthUserRecord
+    data: AuthUserRecord | None # This is None for beforeEmailSent and beforeSmsSent events
     """
     The UserRecord passed to auth blocking functions from the identity platform.
     """
@@ -280,6 +288,12 @@ class AuthBlockingEvent:
     credential: Credential | None
     """An object containing information about the user's credential."""
 
+    email_type: EmailType | None
+    """The type of email event."""
+
+    sms_type: SmsType | None
+    """The type of SMS event."""
+
     timestamp: _dt.datetime
     """
     The time the event was triggered."""
@@ -323,6 +337,22 @@ class BeforeSignInResponse(BeforeCreateResponse, total=False):
     """The user's session claims object if available."""
 
 
+class BeforeEmailSentResponse(_typing.TypedDict, total=False):
+    """
+    The handler response type for 'before_email_sent' blocking events.
+    """
+
+    recaptcha_action_override: RecaptchaActionOptions | None
+
+
+class BeforeSmsSentResponse(BeforeEmailSentResponse, total=False):
+    """
+    The handler response type for 'before_sms_sent' blocking events.
+    """
+
+    recaptcha_action_override: RecaptchaActionOptions | None
+
+
 BeforeUserCreatedCallable = _typing.Callable[[AuthBlockingEvent],
                                              BeforeCreateResponse | None]
 """
@@ -335,6 +365,18 @@ class BeforeSignInResponse(BeforeCreateResponse, total=False):
 The type of the callable for 'before_user_signed_in' blocking events.
 """
 
+BeforeEmailSentCallable = _typing.Callable[[AuthBlockingEvent],
+                                           BeforeEmailSentResponse | None]
+"""
+The type of the callable for 'before_email_sent' blocking events.
+"""
+
+BeforeSmsSentCallable = _typing.Callable[[AuthBlockingEvent],
+                                          BeforeSmsSentResponse | None]
+"""
+The type of the callable for 'before_sms_sent' blocking events.
+"""
+
 
 @_util.copy_func_kwargs(_options.BlockingOptions)
 def before_user_signed_in(
@@ -442,3 +484,110 @@ def before_user_created_wrapped(request: _Request) -> _Response:
         return before_user_created_wrapped
 
     return before_user_created_decorator
+
+@_util.copy_func_kwargs(_options.BlockingOptions)
+def before_email_sent(
+    **kwargs,
+) -> _typing.Callable[[BeforeEmailSentCallable], BeforeEmailSentCallable]:
+    """
+    Handles an event that is triggered before a user's email is sent.
+
+    Example:
+
+    .. code-block:: python
+
+      from firebase_functions import identity_fn
+
+      @identity_fn.before_email_sent()
+      def example(event: identity_fn.AuthBlockingEvent) -> identity_fn.BeforeEmailSentResponse | None:
+          pass
+
+    :param \\*\\*kwargs: Options.
+    :type \\*\\*kwargs: as :exc:`firebase_functions.options.BlockingOptions`
+    :rtype: :exc:`typing.Callable`
+            \\[ \\[ :exc:`firebase_functions.identity_fn.AuthBlockingEvent` \\],
+            :exc:`firebase_functions.identity_fn.BeforeEmailSentResponse` \\| `None` \\]
+            A function that takes a AuthBlockingEvent and optionally returns BeforeEmailSentResponse.
+    """
+    options = _options.BlockingOptions(**kwargs)
+
+    def before_email_sent_decorator(func: BeforeEmailSentCallable):
+        from firebase_functions.private._identity_fn import event_type_before_email_sent
+
+        @_functools.wraps(func)
+        def before_email_sent_wrapped(request: _Request) -> _Response:
+            from firebase_functions.private._identity_fn import before_operation_handler
+            return before_operation_handler(
+                func,
+                event_type_before_email_sent,
+                request,
+            )
+
+        _util.set_func_endpoint_attr(
+            before_email_sent_wrapped,
+            options._endpoint(
+                func_name=func.__name__,
+                event_type=event_type_before_email_sent,
+            ),
+        )
+        _util.set_required_apis_attr(
+            before_email_sent_wrapped,
+            options._required_apis(),
+        )
+        return before_email_sent_wrapped
+
+    return before_email_sent_decorator
+
+
+@_util.copy_func_kwargs(_options.BlockingOptions)
+def before_sms_sent(
+    **kwargs,
+) -> _typing.Callable[[BeforeSmsSentCallable], BeforeSmsSentCallable]:
+    """
+    Handles an event that is triggered before a user's SMS is sent.
+
+    Example:
+
+    .. code-block:: python
+
+      from firebase_functions import identity_fn
+
+      @identity_fn.before_sms_sent()
+      def example(event: identity_fn.AuthBlockingEvent) -> identity_fn.BeforeSmsSentResponse | None:
+          pass
+
+    :param \\*\\*kwargs: Options.
+    :type \\*\\*kwargs: as :exc:`firebase_functions.options.BlockingOptions`
+    :rtype: :exc:`typing.Callable`
+            \\[ \\[ :exc:`firebase_functions.identity_fn.AuthBlockingEvent` \\],
+            :exc:`firebase_functions.identity_fn.BeforeSmsSentResponse` \\| `None` \\]
+            A function that takes a AuthBlockingEvent and optionally returns BeforeSmsSentResponse.
+    """
+    options = _options.BlockingOptions(**kwargs)
+
+    def before_sms_sent_decorator(func: BeforeSmsSentCallable):
+        from firebase_functions.private._identity_fn import event_type_before_sms_sent
+
+        @_functools.wraps(func)
+        def before_sms_sent_wrapped(request: _Request) -> _Response:
+            from firebase_functions.private._identity_fn import before_operation_handler
+            return before_operation_handler(
+                func,
+                event_type_before_sms_sent,
+                request,
+            )
+
+        _util.set_func_endpoint_attr(
+            before_sms_sent_wrapped,
+            options._endpoint(
+                func_name=func.__name__,
+                event_type=event_type_before_sms_sent,
+            ),
+        )
+        _util.set_required_apis_attr(
+            before_sms_sent_wrapped,
+            options._required_apis(),
+        )
+        return before_sms_sent_wrapped
+
+    return before_sms_sent_decorator

src/firebase_functions/options.py

@@ -994,17 +994,24 @@ def _endpoint(
         self,
         **kwargs,
     ) -> _manifest.ManifestEndpoint:
+        from firebase_functions.private._identity_fn import event_type_before_create, event_type_before_sign_in
+
         assert kwargs["event_type"] is not None
 
+        blocking_trigger_options: _manifest.BlockingTriggerOptions
+        
+        if kwargs["event_type"] == event_type_before_create or kwargs["event_type"] == event_type_before_sign_in:
+            blocking_trigger_options = _manifest.BlockingTriggerOptions(
+                idToken=self.id_token if self.id_token is not None else False,
+                accessToken=self.access_token if self.access_token is not None else False,
+                refreshToken=self.refresh_token if self.refresh_token is not None else False,
+            )
+        else:
+            blocking_trigger_options = _manifest.BlockingTriggerOptions()
+
         blocking_trigger = _manifest.BlockingTrigger(
             eventType=kwargs["event_type"],
-            options=_manifest.BlockingTriggerOptions(
-                idToken=self.id_token if self.id_token is not None else False,
-                accessToken=self.access_token
-                if self.access_token is not None else False,
-                refreshToken=self.refresh_token
-                if self.refresh_token is not None else False,
-            ),
+            options=blocking_trigger_options,
         )
 
         kwargs_merged = {

src/firebase_functions/private/_identity_fn.py

@@ -167,6 +167,8 @@ def _additional_user_info_from_token_data(token_data: dict[str, _typing.Any]):
         username=username,
         is_new_user=is_new_user,
         recaptcha_score=token_data.get("recaptcha_score"),
+        email=token_data.get("email"),
+        phone_number=token_data.get("phone_number"),
     )
 
 
@@ -200,22 +202,31 @@ def _credential_from_token_data(token_data: dict[str, _typing.Any],
     )
 
 
-def _auth_blocking_event_from_token_data(token_data: dict[str, _typing.Any]):
-    from firebase_functions.identity_fn import AuthBlockingEvent
+def _auth_blocking_event_from_token_data(token_data: dict[str, _typing.Any], event_type: str):
+    from firebase_functions.identity_fn import AuthBlockingEvent, AuthUserRecord
+
+    data: AuthUserRecord | None = None
+    if event_type == event_type_before_create or event_type == event_type_before_sign_in:
+        data = _auth_user_record_from_token_data(token_data["user_record"])
+
     return AuthBlockingEvent(
-        data=_auth_user_record_from_token_data(token_data["user_record"]),
+        data=data,
         locale=token_data.get("locale"),
         event_id=token_data["event_id"],
         ip_address=token_data["ip_address"],
         user_agent=token_data["user_agent"],
         timestamp=_dt.datetime.fromtimestamp(token_data["iat"]),
         additional_user_info=_additional_user_info_from_token_data(token_data),
         credential=_credential_from_token_data(token_data, _time.time()),
+        email_type=token_data.get("email_type"),
+        sms_type=token_data.get("sms_type"),
     )
 
 
 event_type_before_create = "providers/cloud.auth/eventTypes/user.beforeCreate"
 event_type_before_sign_in = "providers/cloud.auth/eventTypes/user.beforeSignIn"
+event_type_before_email_sent = "providers/cloud.auth/eventTypes/user.beforeSendEmail"
+event_type_before_sms_sent = "providers/cloud.auth/eventTypes/user.beforeSendSms"
 
 
 def _validate_auth_response(
@@ -338,7 +349,7 @@ def before_operation_handler(
     event_type: str,
     request: _Request,
 ) -> _Response:
-    from firebase_functions.identity_fn import BeforeCreateResponse, BeforeSignInResponse
+    from firebase_functions.identity_fn import BeforeCreateResponse, BeforeSignInResponse, BeforeEmailSentResponse, BeforeSmsSentResponse
     try:
         if not _util.valid_on_call_request(request):
             _logging.error("Invalid request, unable to process.")
@@ -351,8 +362,8 @@ def before_operation_handler(
             raise HttpsError(FunctionsErrorCode.INVALID_ARGUMENT, "Bad Request")
         jwt_token = request.json["data"]["jwt"]
         decoded_token = _token_verifier.verify_auth_blocking_token(jwt_token)
-        event = _auth_blocking_event_from_token_data(decoded_token)
-        auth_response: BeforeCreateResponse | BeforeSignInResponse | None = _with_init(
+        event = _auth_blocking_event_from_token_data(decoded_token, event_type)
+        auth_response: BeforeCreateResponse | BeforeSignInResponse | BeforeEmailSentResponse | BeforeSmsSentResponse | None = _with_init(
             func)(event)
         if not auth_response:
             return _jsonify({})