Add "airflow users reset-password" command (#37044)

Added support for reset-password command which
can either randomly generate a password, or
take a password from user.

airflow/providers/fab/auth_manager/cli_commands/definition.py

@@ -136,6 +136,27 @@
             "          --email admin@example.org"
         ),
     ),
+    ActionCommand(
+        name="reset-password",
+        help="Reset a user's password",
+        func=lazy_load_command(
+            "airflow.providers.fab.auth_manager.cli_commands.user_command.user_reset_password"
+        ),
+        args=(
+            ARG_USERNAME_OPTIONAL,
+            ARG_EMAIL_OPTIONAL,
+            ARG_PASSWORD,
+            ARG_USE_RANDOM_PASSWORD,
+            ARG_VERBOSE,
+        ),
+        epilog=(
+            "examples:\n"
+            'To reset an user with username equals to "admin", run:\n'
+            "\n"
+            "    $ airflow users reset-password \\\n"
+            "          --username admin"
+        ),
+    ),
     ActionCommand(
         name="delete",
         help="Delete a user",

airflow/providers/fab/auth_manager/cli_commands/user_command.py

@@ -70,16 +70,7 @@ def users_create(args):
         if not role:
             valid_roles = appbuilder.sm.get_all_roles()
             raise SystemExit(f"{args.role} is not a valid role. Valid roles are: {valid_roles}")
-        if args.use_random_password:
-            password = "".join(random.choices(string.printable, k=16))
-        elif args.password:
-            password = args.password
-        else:
-            password = getpass.getpass("Password:")
-            password_confirmation = getpass.getpass("Repeat for confirmation:")
-            if password != password_confirmation:
-                raise SystemExit("Passwords did not match")
-
+        password = _create_password(args)
         if appbuilder.sm.find_user(args.username):
             print(f"{args.username} already exist in the db")
             return
@@ -106,6 +97,32 @@ def _find_user(args):
     return user
 
 
+@cli_utils.action_cli
+@providers_configuration_loaded
+def user_reset_password(args):
+    """Reset user password user from DB."""
+    user = _find_user(args)
+    password = _create_password(args)
+    with get_application_builder() as appbuilder:
+        if appbuilder.sm.reset_password(user.id, password):
+            print(f'User "{user.username}" password reset successfully')
+        else:
+            raise SystemExit("Failed to reset user password")
+
+
+def _create_password(args):
+    if args.use_random_password:
+        password = "".join(random.choices(string.printable, k=16))
+    elif args.password:
+        password = args.password
+    else:
+        password = getpass.getpass("Password:")
+        password_confirmation = getpass.getpass("Repeat for confirmation:")
+        if password != password_confirmation:
+            raise SystemExit("Passwords did not match")
+    return password
+
+
 @cli_utils.action_cli
 @providers_configuration_loaded
 def users_delete(args):

airflow/providers/fab/auth_manager/security_manager/override.py

@@ -458,7 +458,7 @@ def create_jwt_manager(self):
         jwt_manager.init_app(self.appbuilder.app)
         jwt_manager.user_lookup_loader(self.load_user_jwt)
 
-    def reset_password(self, userid, password):
+    def reset_password(self, userid: int, password: str) -> bool:
         """
         Change/Reset a user's password for auth db.
 
@@ -470,7 +470,7 @@ def reset_password(self, userid, password):
         user = self.get_user_by_id(userid)
         user.password = generate_password_hash(password)
         self.reset_user_sessions(user)
-        self.update_user(user)
+        return self.update_user(user)
 
     def reset_user_sessions(self, user: User) -> None:
         if isinstance(self.appbuilder.get_app.session_interface, AirflowDatabaseSessionInterface):
@@ -1536,7 +1536,7 @@ def find_register_user(self, registration_hash):
             .limit(1)
         )
 
-    def update_user(self, user):
+    def update_user(self, user: User) -> bool:
         try:
             self.get_session.merge(user)
             self.get_session.commit()
@@ -1545,6 +1545,7 @@ def update_user(self, user):
             log.error(const.LOGMSG_ERR_SEC_UPD_USER, e)
             self.get_session.rollback()
             return False
+        return True
 
     def del_register_user(self, register_user):
         """

tests/providers/fab/auth_manager/cli_commands/test_definition.py

@@ -25,7 +25,7 @@
 
 class TestCliDefinition:
     def test_users_commands(self):
-        assert len(USERS_COMMANDS) == 7
+        assert len(USERS_COMMANDS) == 8
 
     def test_roles_commands(self):
         assert len(ROLES_COMMANDS) == 7

tests/providers/fab/auth_manager/cli_commands/test_user_command.py

@@ -472,3 +472,55 @@ def test_cli_manage_roles_exceptions(self, create_user_test4, action, role, mess
     def test_cli_import_users_exceptions(self, user, message):
         with pytest.raises(SystemExit, match=re.escape(message)):
             self._import_users_from_file([user])
+
+    def test_cli_reset_user_password(self):
+        args = self.parser.parse_args(
+            [
+                "users",
+                "create",
+                "--username",
+                "test3",
+                "--lastname",
+                "doe",
+                "--firstname",
+                "jon",
+                "--email",
+                "jdoe@example.com",
+                "--role",
+                "Viewer",
+                "--use-random-password",
+            ]
+        )
+        user_command.users_create(args)
+        args = self.parser.parse_args(
+            ["users", "reset-password", "--username", "test3", "--use-random-password"]
+        )
+        with redirect_stdout(StringIO()) as stdout:
+            user_command.user_reset_password(args)
+        assert 'User "test3" password reset successfully' in stdout.getvalue()
+
+    def test_cli_reset_user_password_with_email(self):
+        args = self.parser.parse_args(
+            [
+                "users",
+                "create",
+                "--username",
+                "test3",
+                "--lastname",
+                "doe",
+                "--firstname",
+                "jon",
+                "--email",
+                "jdoe@example.com",
+                "--role",
+                "Viewer",
+                "--use-random-password",
+            ]
+        )
+        user_command.users_create(args)
+        args = self.parser.parse_args(
+            ["users", "reset-password", "--email", "jdoe@example.com", "--password", "s3cr3t"]
+        )
+        with redirect_stdout(StringIO()) as stdout:
+            user_command.user_reset_password(args)
+        assert 'User "test3" password reset successfully' in stdout.getvalue()