feat(auth): Add support for VERIFY_AND_CHANGE_EMAIL out-of-band links

Author: jagadeesh545

firebase_admin/_auth_client.py

@@ -500,6 +500,31 @@ def generate_sign_in_with_email_link(self, email, action_code_settings):
         return self._user_manager.generate_email_action_link(
             'EMAIL_SIGNIN', email, action_code_settings=action_code_settings)
 
+    def generate_verify_and_change_email_link(self, email, new_email, action_code_settings=None):
+        """Generates the out-of-band email action link for email verification and change flows for
+        the specified email address.
+
+        Args:
+            email: The current email of the user.
+            new_email: The new email address of the user to be verified.
+            action_code_settings: ``ActionCodeSettings`` instance (optional). Defines whether
+                the link is to be handled by a mobile app and the additional state information to
+                be passed in the deep link.
+
+        Returns:
+            str: The email verification and change link created by the API
+
+        Raises:
+            ValueError: If the provided arguments are invalid
+            FirebaseError: If an error occurs while generating the link
+        """
+        return self._user_manager.generate_email_action_link(
+            "VERIFY_AND_CHANGE_EMAIL",
+            email,
+            action_code_settings=action_code_settings,
+            new_email=new_email,
+        )
+
     def get_oidc_provider_config(self, provider_id):
         """Returns the ``OIDCProviderConfig`` with the given ID.
 

firebase_admin/_auth_utils.py

@@ -29,7 +29,9 @@
     'acr', 'amr', 'at_hash', 'aud', 'auth_time', 'azp', 'cnf', 'c_hash', 'exp', 'iat',
     'iss', 'jti', 'nbf', 'nonce', 'sub', 'firebase',
 ])
-VALID_EMAIL_ACTION_TYPES = set(['VERIFY_EMAIL', 'EMAIL_SIGNIN', 'PASSWORD_RESET'])
+VALID_EMAIL_ACTION_TYPES = set(
+    ["VERIFY_EMAIL", "EMAIL_SIGNIN", "PASSWORD_RESET", "VERIFY_AND_CHANGE_EMAIL"]
+)
 
 
 class PageIterator:

firebase_admin/_user_mgt.py

@@ -827,29 +827,42 @@ def import_users(self, users, hash_alg=None):
                 'Failed to import users.', http_response=http_resp)
         return body
 
-    def generate_email_action_link(self, action_type, email, action_code_settings=None):
+    def generate_email_action_link(
+        self, action_type, email, action_code_settings=None, new_email=None
+    ):
         """Fetches the email action links for types
 
         Args:
-            action_type: String. Valid values ['VERIFY_EMAIL', 'EMAIL_SIGNIN', 'PASSWORD_RESET']
+            action_type: String. Valid values ['VERIFY_EMAIL', 'EMAIL_SIGNIN', 'PASSWORD_RESET',
+                'VERIFY_AND_CHANGE_EMAIL']
             email: Email of the user for which the action is performed
             action_code_settings: ``ActionCodeSettings`` object or dict (optional). Defines whether
                 the link is to be handled by a mobile app and the additional state information to be
                 passed in the deep link, etc.
+            new_email: The new email address of the user. This is required if ``action_type``
+                is 'VERIFY_AND_CHANGE_EMAIL'.
         Returns:
-            link_url: action url to be emailed to the user
+            str: Action URL to be emailed to the user
 
         Raises:
             UnexpectedResponseError: If the backend server responds with an unexpected message
             FirebaseError: If an error occurs while generating the link
             ValueError: If the provided arguments are invalid
         """
+        if action_type == 'VERIFY_AND_CHANGE_EMAIL' and not new_email:
+            raise ValueError(
+                'new_email must be provided when action_type is VERIFY_AND_CHANGE_EMAIL.'
+            )
+
         payload = {
             'requestType': _auth_utils.validate_action_type(action_type),
-            'email': _auth_utils.validate_email(email),
+            'email': _auth_utils.validate_email(email, required=True),
             'returnOobLink': True
         }
 
+        if new_email:
+            payload['newEmail'] = _auth_utils.validate_email(new_email, required=True)
+
         if action_code_settings:
             payload.update(encode_action_code_settings(action_code_settings))
 

firebase_admin/auth.py

@@ -92,6 +92,7 @@
     'generate_email_verification_link',
     'generate_password_reset_link',
     'generate_sign_in_with_email_link',
+    'generate_verify_and_change_email_link',
     'get_oidc_provider_config',
     'get_saml_provider_config',
     'get_user',
@@ -647,6 +648,30 @@ def generate_sign_in_with_email_link(email, action_code_settings, app=None):
         email, action_code_settings=action_code_settings)
 
 
+def generate_verify_and_change_email_link(email, new_email, action_code_settings=None, app=None):
+    """Generates the out-of-band email action link for email verification and change flows for the
+    specified email address.
+
+    Args:
+        email: The current email of the user.
+        new_email: The new email address of the user to be verified.
+        action_code_settings: ``ActionCodeSettings`` instance (optional). Defines whether
+            the link is to be handled by a mobile app and the additional state information to be
+            passed in the deep link.
+        app: An App instance (optional).
+
+    Returns:
+        str: The email verification and change link created by the API
+
+    Raises:
+        ValueError: If the provided arguments are invalid
+        FirebaseError: If an error occurs while generating the link
+    """
+    client = _get_client(app)
+    return client.generate_verify_and_change_email_link(
+        email, new_email, action_code_settings=action_code_settings)
+
+
 def get_oidc_provider_config(provider_id, app=None):
     """Returns the ``OIDCProviderConfig`` with the given ID.
 

integration/test_auth.py

@@ -689,6 +689,17 @@ def test_email_verification(new_user_email_unverified, api_key):
     assert new_user_email_unverified.email == user_email
     assert auth.get_user(new_user_email_unverified.uid).email_verified
 
+def test_verify_and_change_email(new_user_email_unverified, api_key):
+    _, new_email = _random_id()
+    link = auth.generate_verify_and_change_email_link(new_user_email_unverified.email, new_email)
+    assert isinstance(link, str)
+    query_dict = _extract_link_params(link)
+    user_email = _verify_email(query_dict['oobCode'], api_key)
+    assert new_email == user_email
+    user = auth.get_user(new_user_email_unverified.uid)
+    assert user.email == new_email
+    assert user.email_verified
+
 def test_password_reset_with_settings(new_user_email_unverified, api_key):
     action_code_settings = auth.ActionCodeSettings(ACTION_LINK_CONTINUE_URL)
     link = auth.generate_password_reset_link(new_user_email_unverified.email,
@@ -712,6 +723,20 @@ def test_email_verification_with_settings(new_user_email_unverified, api_key):
     assert new_user_email_unverified.email == user_email
     assert auth.get_user(new_user_email_unverified.uid).email_verified
 
+def test_verify_and_change_email_with_settings(new_user_email_unverified, api_key):
+    _, new_email = _random_id()
+    action_code_settings = auth.ActionCodeSettings(ACTION_LINK_CONTINUE_URL)
+    link = auth.generate_verify_and_change_email_link(new_user_email_unverified.email, new_email,
+                                                 action_code_settings=action_code_settings)
+    assert isinstance(link, str)
+    query_dict = _extract_link_params(link)
+    assert query_dict['continueUrl'] == ACTION_LINK_CONTINUE_URL
+    user_email = _verify_email(query_dict['oobCode'], api_key)
+    assert new_email == user_email
+    user = auth.get_user(new_user_email_unverified.uid)
+    assert user.email == new_email
+    assert user.email_verified
+
 def test_email_sign_in_with_settings(new_user_email_unverified, api_key):
     action_code_settings = auth.ActionCodeSettings(ACTION_LINK_CONTINUE_URL)
     link = auth.generate_sign_in_with_email_link(new_user_email_unverified.email,

integration/test_tenant_mgt.py

@@ -202,6 +202,14 @@ def test_sign_in_with_email_link(sample_tenant, tenant_user):
     assert _tenant_id_from_link(link) == sample_tenant.tenant_id
 
 
+def test_verify_and_change_email_link(sample_tenant, tenant_user):
+    client = tenant_mgt.auth_for_tenant(sample_tenant.tenant_id)
+    new_email = _random_email()
+    link = client.generate_verify_and_change_email_link(
+        tenant_user.email, new_email, ACTION_CODE_SETTINGS
+    )
+    assert _tenant_id_from_link(link) == sample_tenant.tenant_id
+
 def test_import_users(sample_tenant):
     client = tenant_mgt.auth_for_tenant(sample_tenant.tenant_id)
     user = auth.ImportUserRecord(

tests/test_tenant_mgt.py

@@ -754,6 +754,24 @@ def test_generate_sign_in_with_email_link(self, tenant_mgt_app):
             'continueUrl': 'http://localhost',
         })
 
+    def test_generate_verify_and_change_email_link(self, tenant_mgt_app):
+        client = tenant_mgt.auth_for_tenant('tenant-id', app=tenant_mgt_app)
+        recorder = _instrument_user_mgt(client, 200, '{"oobLink":"https://testlink"}')
+        settings = auth.ActionCodeSettings(url='http://localhost')
+
+        link = client.generate_verify_and_change_email_link(
+            "test@test.com", "new@test.com", settings
+        )
+
+        assert link == 'https://testlink'
+        self._assert_request(recorder, '/accounts:sendOobCode', {
+            'email': 'test@test.com',
+            'newEmail': 'new@test.com',
+            'requestType': 'VERIFY_AND_CHANGE_EMAIL',
+            'returnOobLink': True,
+            'continueUrl': 'http://localhost',
+        })
+
     def test_get_oidc_provider_config(self, tenant_mgt_app):
         client = tenant_mgt.auth_for_tenant('tenant-id', app=tenant_mgt_app)
         recorder = _instrument_provider_mgt(client, 200, OIDC_PROVIDER_CONFIG_RESPONSE)

tests/test_user_mgt.py

@@ -1425,6 +1425,16 @@ def test_password_reset_no_settings(self, user_mgt_app):
         assert request['requestType'] == 'PASSWORD_RESET'
         self._validate_request(request)
 
+    def test_verify_and_change_email_no_settings(self, user_mgt_app):
+        _, recorder = _instrument_user_manager(user_mgt_app, 200, '{"oobLink":"https://testlink"}')
+        link = auth._get_client(user_mgt_app)._user_manager.generate_email_action_link(
+            'VERIFY_AND_CHANGE_EMAIL', 'test@test.com', new_email='new@test.com')
+        request = json.loads(recorder[0].body.decode())
+
+        assert link == 'https://testlink'
+        assert request['requestType'] == 'VERIFY_AND_CHANGE_EMAIL'
+        self._validate_request(request, new_email='new@test.com')
+
     def test_email_signin_with_settings(self, user_mgt_app):
         _, recorder = _instrument_user_manager(user_mgt_app, 200, '{"oobLink":"https://testlink"}')
         link = auth.generate_sign_in_with_email_link('test@test.com',
@@ -1458,6 +1468,20 @@ def test_password_reset_with_settings(self, user_mgt_app):
         assert request['requestType'] == 'PASSWORD_RESET'
         self._validate_request(request, MOCK_ACTION_CODE_SETTINGS)
 
+    def test_verify_and_change_email_with_settings(self, user_mgt_app):
+        _, recorder = _instrument_user_manager(user_mgt_app, 200, '{"oobLink":"https://testlink"}')
+        link = auth._get_client(user_mgt_app)._user_manager.generate_email_action_link(
+            "VERIFY_AND_CHANGE_EMAIL",
+            "test@test.com",
+            action_code_settings=MOCK_ACTION_CODE_SETTINGS,
+            new_email="new@test.com",
+        )
+        request = json.loads(recorder[0].body.decode())
+
+        assert link == 'https://testlink'
+        assert request['requestType'] == 'VERIFY_AND_CHANGE_EMAIL'
+        self._validate_request(request, MOCK_ACTION_CODE_SETTINGS, new_email='new@test.com')
+
     @pytest.mark.parametrize('func', [
         auth.generate_sign_in_with_email_link,
         auth.generate_email_verification_link,
@@ -1547,9 +1571,20 @@ def test_bad_action_type(self, user_mgt_app):
                 .generate_email_action_link('BAD_TYPE', 'test@test.com',
                                             action_code_settings=MOCK_ACTION_CODE_SETTINGS)
 
-    def _validate_request(self, request, settings=None):
+    def test_verify_and_change_email_missing_new_email(self, user_mgt_app):
+        with pytest.raises(ValueError) as excinfo:
+            auth._get_client(user_mgt_app)._user_manager.generate_email_action_link(
+                'VERIFY_AND_CHANGE_EMAIL', 'test@test.com')
+        assert (
+            str(excinfo.value)
+            == "new_email must be provided when action_type is VERIFY_AND_CHANGE_EMAIL."
+        )
+
+    def _validate_request(self, request, settings=None, new_email=None):
         assert request['email'] == 'test@test.com'
         assert request['returnOobLink']
+        if new_email:
+            assert request['newEmail'] == new_email
         if settings:
             assert request['continueUrl'] == settings.url
             assert request['canHandleCodeInApp'] == settings.handle_code_in_app