CVE-2024-2213: Improper Authentication (CWE-287)
value: An issue was discovered in zenml-io/zenml versions up to and including 0.55.4. Due to improper authentication mechanisms, an attacker with access to an active user session can change the account password without needing to know the current password. This vulnerability allows for unauthorized account takeover by bypassing the standard password change verification process. The issue was fixed in version 0.56.3.
CWE description: When an actor claims to have a given identity, the product does not prove or insufficiently proves that the claim is correct.
Published: 2024-06-06T19:15Z
Repository: https://github.com/zenml-io/zenml
Language: Python
Files changed: src/zenml/cli/server.py, src/zenml/cli/user_management.py, src/zenml/client.py, src/zenml/models/v2/core/user.py, src/zenml/zen_server/routers/users_endpoints.py, src/zenml/zen_stores/rest_zen_store.py, src/zenml/zen_stores/schemas/user_schemas.py
Commit date: 2024-04-09 09:22:03 +0200
Commit message: Check old password during password change and add missing CLI commands (#2587)

* Check old password during password change and add missing CLI commands

* Fix docstrings

* Ask to re-enter password during CLI password change

* Auto-update of E2E template

* Auto-update of LLM Finetuning template

* Auto-update of NLP template

---------

Co-authored-by: GitHub Actions <actions@github.com>
Diff: diff --git a/src/zenml/cli/server.py b/src/zenml/cli/server.py
index ed76da957b..b96ce0b176 100644
--- a/src/zenml/cli/server.py
+++ b/src/zenml/cli/server.py
@@ -30,11 +30,11 @@ from zenml.cli.cli import cli
 from zenml.cli.web_login import web_login
 from zenml.client import Client
 from zenml.config.global_config import GlobalConfiguration
 from zenml.console import console
 from zenml.enums import ServerProviderType, StoreType
-from zenml.exceptions import IllegalOperationError
+from zenml.exceptions import AuthorizationException, IllegalOperationError
 from zenml.logger import get_logger
 from zenml.utils import terraform_utils, yaml_utils
 from zenml.zen_server.utils import get_active_deployment
 
 logger = get_logger(__name__)
@@ -622,16 +622,10 @@ def status() -> None:
     help="Use an API key to authenticate with a ZenML server. If "
     "omitted, the web login will be used.",
     required=False,
     type=str,
 )
-@click.option(
-    "--workspace",
-    help="The workspace to use when connecting to the ZenML server.",
-    required=False,
-    type=str,
-)
 @click.option(
     "--no-verify-ssl",
     is_flag=True,
     help="Whether to verify the server's TLS certificate",
     default=False,
@@ -659,11 +653,10 @@ def status() -> None:
 def connect(
     url: Optional[str] = None,
     username: Optional[str] = None,
     password: Optional[str] = None,
     api_key: Optional[str] = None,
-    workspace: Optional[str] = None,
     no_verify_ssl: bool = False,
     ssl_ca_cert: Optional[str] = None,
     config: Optional[str] = None,
     raw_config: bool = False,
 ) -> None:
@@ -675,22 +668,26 @@ def connect(
             server.
         password: The password that is used to authenticate with the ZenML
             server.
         api_key: The API key that is used to authenticate with the ZenML
             server.
-        workspace: The active workspace that is used to connect to the ZenML
-            server.
         no_verify_ssl: Whether to verify the server's TLS certificate.
         ssl_ca_cert: A path to a CA bundle to use to verify the server's TLS
             certificate or the CA bundle value itself.
         config: A YAML or JSON configuration or configuration file to use.
         raw_config: Whether to use the configuration without prompting for
             missing fields.
     """
     from zenml.config.store_config import StoreConfiguration
     from zenml.zen_stores.base_zen_store import BaseZenStore
 
+    if password is not None:
+        cli_utils.warning(
+            "Supplying password values in the command line is not safe. "
+            "Please consider using the prompt option."
+        )
+
     # Raise an error if a local server is running when trying to connect to
     # another server
     active_deployment = get_active_deployment(local=True)
     if (
         active_deployment
@@ -765,10 +762,20 @@ def connect(
             store_dict["api_token"] = web_login(url=url, verify_ssl=verify_ssl)
         else:
             username = click.prompt("Username", type=str)
 
     if username:
+        cli_utils.warning(
+            "Connecting to a ZenML server using a username and password is "
+            "not recommended because the password is locally stored on your "
+            "filesystem. You should consider using the web login workflow by "
+            "omitting the `--username` and `--password` flags. An alternative "
+            "for non-interactive environments is to create and use a service "
+            "account API key (see https://docs.zenml.io/user-guide/advanced-guide/configuring-zenml/connecting-to-zenml#using-service-accounts-to-connect-to-a-deployed-zenml-server "
+            "for more information)."
+        )
+
         store_dict["username"] = username
 
         if password is None:
             password = click.prompt(
                 f"Password for user {username} (press ENTER for empty password)",
@@ -788,20 +795,12 @@ def connect(
     except IllegalOperationError:
         cli_utils.warning(
             f"User '{username}' does not have sufficient permissions to "
             f"access the server at '{url}'."
         )
-
-    if workspace:
-        try:
-            Client().set_active_workspace(workspace_name_or_id=workspace)
-        except KeyError:
-            cli_utils.warning(
-                f"The workspace {workspace} does not exist or is not accessible. "
-                f"Please set another workspace by running `zenml "
-                f"workspace set`."
-            )
+    except AuthorizationException as e:
+        cli_utils.warning(f"Authorization error: {e}")
 
 
 @cli.command("disconnect", help="Disconnect from a ZenML server.")
 def disconnect_server() -> None:
     """Disconnect from a ZenML server."""
diff --git a/src/zenml/cli/user_management.py b/src/zenml/cli/user_management.py
index 933d8e7908..6789c791b8 100644
--- a/src/zenml/cli/user_management.py
+++ b/src/zenml/cli/user_management.py
@@ -19,13 +19,18 @@ import click
 
 from zenml.cli import utils as cli_utils
 from zenml.cli.cli import TagGroup, cli
 from zenml.cli.utils import is_sorted_or_filtered, list_options
 from zenml.client import Client
+from zenml.config.global_config import GlobalConfiguration
 from zenml.console import console
 from zenml.enums import CliCategories, StoreType
-from zenml.exceptions import EntityExistsError, IllegalOperationError
+from zenml.exceptions import (
+    AuthorizationException,
+    EntityExistsError,
+    IllegalOperationError,
+)
 from zenml.models import UserFilter
 
 
 @cli.group(cls=TagGroup, tag=CliCategories.IDENTITY_AND_SECURITY)
 def user() -> None:
@@ -154,10 +159,15 @@ def create_user(
                 f"Password for user {user_name}. Leave empty to generate an "
                 f"activation token",
                 default="",
                 hide_input=True,
             )
+    else:
+        cli_utils.warning(
+            "Supplying password values in the command line is not safe. "
+            "Please consider using the prompt option."
+        )
 
     try:
         new_user = client.create_user(
             name=user_name, password=password, is_admin=is_admin
         )
@@ -202,18 +212,10 @@ def create_user(
     "updated_email",
     type=str,
     required=False,
     help="New user email.",
 )
-@click.option(
-    "--password",
-    "-p",
-    "updated_password",
-    type=str,
-    required=False,
-    help="New user password.",
-)
 @click.option(
     "--admin",
     "-a",
     "make_admin",
     is_flag=True,
@@ -228,29 +230,37 @@ def create_user(
     is_flag=True,
     required=False,
     default=None,
     help="Whether the user should be a regular user.",
 )
+@click.option(
+    "--active",
+    "active",
+    type=bool,
+    required=False,
+    default=None,
+    help="Use to activate or deactivate a user account.",
+)
 def update_user(
     user_name_or_id: str,
     updated_name: Optional[str] = None,
     updated_full_name: Optional[str] = None,
     updated_email: Optional[str] = None,
-    updated_password: Optional[str] = None,
     make_admin: Optional[bool] = None,
     make_user: Optional[bool] = None,
+    active: Optional[bool] = None,
 ) -> None:
     """Update an existing user.
 
     Args:
         user_name_or_id: The name of the user to create.
         updated_name: The name of the user to create.
         updated_full_name: The name of the user to create.
         updated_email: The name of the user to create.
-        updated_password: The name of the user to create.
         make_admin: Whether the user should be an admin.
         make_user: Whether the user should be a regular user.
+        active: Use to activate or deactivate a user account.
     """
     if make_admin is not None and make_user is not None:
         cli_utils.error(
             "Cannot set both --admin and --user flags as these are mutually exclusive."
         )
@@ -258,11 +268,12 @@ def update_user(
         current_user = Client().get_user(
             user_name_or_id, allow_name_prefix_match=False
         )
         if current_user.is_admin and make_user:
             confirmation = cli_utils.confirmation(
-                f"Currently user `{current_user.name}` is an admin. Are you sure you want to make them a regular user?"
+                f"Currently user `{current_user.name}` is an admin. Are you "
+                "sure you want to make them a regular user?"
             )
             if not confirmation:
                 cli_utils.declare("User update canceled.")
                 return
 
@@ -274,17 +285,141 @@ def update_user(
         Client().update_user(
             name_id_or_prefix=user_name_or_id,
             updated_name=updated_name,
             updated_full_name=updated_full_name,
             updated_email=updated_email,
-            updated_password=updated_password,
             updated_is_admin=updated_is_admin,
+            active=active,
         )
     except (KeyError, IllegalOperationError) as err:
         cli_utils.error(str(err))
 
 
+@user.command(
+    "change-password",
+    help="Change the password for the current user account.",
+)
+@click.option(
+    "--password",
+    help=(
+        "The new user password. If omitted, a prompt will be shown to enter "
+        "the password."
+    ),
+    required=False,
+    type=str,
+)
+@click.option(
+    "--old-password",
+    help=(
+        "The old user password. If omitted, a prompt will be shown to enter "
+        "the old password."
+    ),
+    required=False,
+    type=str,
+)
+def change_user_password(
+    password: Optional[str] = None, old_password: Optional[str] = None
+) -> None:
+    """Change the password of the current user.
+
+    Args:
+        password: The new password for the current user.
+        old_password: The old password for the current user.
+    """
+    active_user = Client().active_user
+
+    if old_password is not None or password is not None:
+        cli_utils.warning(
+            "Supplying password values in the command line is not safe. "
+            "Please consider using the prompt option."
+        )
+
+    if old_password is None:
+        old_password = click.prompt(
+            f"Current password for user {active_user.name}",
+            hide_input=True,
+        )
+    if password is None:
+        password = click.prompt(
+            f"New password for user {active_user.name}",
+            hide_input=True,
+        )
+        password_again = click.prompt(
+            f"Please re-enter the new password for user {active_user.name}",
+            hide_input=True,
+        )
+        if password != password_again:
+            cli_utils.error("Passwords do not match.")
+
+    try:
+        Client().update_user(
+            name_id_or_prefix=active_user.id,
+            old_password=old_password,
+            updated_password=password,
+        )
+    except (KeyError, IllegalOperationError, AuthorizationException) as err:
+        cli_utils.error(str(err))
+
+    cli_utils.declare(
+        f"Successfully updated password for active user '{active_user.name}'."
+    )
+
+    store = GlobalConfiguration().store_configuration
+    if store.type == StoreType.REST:
+        from zenml.zen_stores.rest_zen_store import RestZenStoreConfiguration
+
+        assert isinstance(store, RestZenStoreConfiguration)
+
+        if store.password is not None:
+            cli_utils.declare(
+                "You may need to log in again with your new password by "
+                "running `zenml connect`."
+            )
+
+
+@user.command(
+    "deactivate",
+    help="Generate an activation token to reset the password for a user account",
+)
+@click.argument("user_name_or_id", type=str, required=True)
+def deactivate_user(
+    user_name_or_id: str,
+) -> None:
+    """Reset the password of a user.
+
+    Args:
+        user_name_or_id: The name or ID of the user to reset the password for.
+    """
+    client = Client()
+
+    store = GlobalConfiguration().store_configuration
+    if store.type != StoreType.REST:
+        cli_utils.error(
+            "Deactivating users is only supported when connected to a ZenML "
+            "server."
+        )
+
+    try:
+        if not client.active_user.is_admin:
+            cli_utils.error(
+                "Only admins can reset the password of other users."
+            )
+
+        user = client.deactivate_user(
+            name_id_or_prefix=user_name_or_id,
+        )
+    except (KeyError, IllegalOperationError) as err:
+        cli_utils.error(str(err))
+
+    cli_utils.declare(
+        f"Successfully deactivated user account '{user.name}'."
+        f"To reactivate the account, please visit the dashboard at the "
+        "following URL:\n"
+        f"{client.zen_store.url}/signup?user={str(user.id)}&username={user.name}&token={user.activation_token}\n"
+    )
+
+
 @user.command("delete")
 @click.argument("user_name_or_id", type=str, required=True)
 def delete_user(user_name_or_id: str) -> None:
     """Delete a user.
 
diff --git a/src/zenml/client.py b/src/zenml/client.py
index bed3a3b5c0..fee1cd5c16 100644
--- a/src/zenml/client.py
+++ b/src/zenml/client.py
@@ -812,11 +812,13 @@ class Client(metaclass=ClientMetaClass):
         updated_full_name: Optional[str] = None,
         updated_email: Optional[str] = None,
         updated_email_opt_in: Optional[bool] = None,
         updated_hub_token: Optional[str] = None,
         updated_password: Optional[str] = None,
+        old_password: Optional[str] = None,
         updated_is_admin: Optional[bool] = None,
+        active: Optional[bool] = None,
     ) -> UserResponse:
         """Update a user.
 
         Args:
             name_id_or_prefix: The name or ID of the user to update.
@@ -824,14 +826,21 @@ class Client(metaclass=ClientMetaClass):
             updated_full_name: The new full name of the user.
             updated_email: The new email of the user.
             updated_email_opt_in: The new email opt-in status of the user.
             updated_hub_token: Update the hub token
             updated_password: The new password of the user.
+            old_password: The old password of the user. Required for password
+                update.
             updated_is_admin: Whether the user should be an admin.
+            active: Use to activate or deactivate the user.
 
         Returns:
             The updated user.
+
+        Raises:
+            ValidationError: If the old password is not provided when updating
+                the password.
         """
         user = self.get_user(
             name_id_or_prefix=name_id_or_prefix, allow_name_prefix_match=False
         )
         user_update = UserUpdate(name=updated_name or user.name)
@@ -846,17 +855,40 @@ class Client(metaclass=ClientMetaClass):
             user_update.email_opted_in = updated_email_opt_in
         if updated_hub_token is not None:
             user_update.hub_token = updated_hub_token
         if updated_password is not None:
             user_update.password = updated_password
+            if old_password is None:
+                raise ValidationError(
+                    "Old password is required to update the password."
+                )
+            user_update.old_password = old_password
         if updated_is_admin is not None:
             user_update.is_admin = updated_is_admin
+        if active is not None:
+            user_update.active = active
 
         return self.zen_store.update_user(
             user_id=user.id, user_update=user_update
         )
 
+    @_fail_for_sql_zen_store
+    def deactivate_user(self, name_id_or_prefix: str) -> "UserResponse":
+        """Deactivate a user and generate an activation token.
+
+        Args:
+            name_id_or_prefix: The name or ID of the user to reset.
+
+        Returns:
+            The deactivated user.
+        """
+        from zenml.zen_stores.rest_zen_store import RestZenStore
+
+        user = self.get_user(name_id_or_prefix, allow_name_prefix_match=False)
+        assert isinstance(self.zen_store, RestZenStore)
+        return self.zen_store.deactivate_user(user_name_or_id=user.name)
+
     def delete_user(self, name_id_or_prefix: str) -> None:
         """Delete a user.
 
         Args:
             name_id_or_prefix: The name or ID of the user to delete.
diff --git a/src/zenml/models/v2/core/user.py b/src/zenml/models/v2/core/user.py
index fded4014ad..8a39e9c528 100644
--- a/src/zenml/models/v2/core/user.py
+++ b/src/zenml/models/v2/core/user.py
@@ -199,10 +199,16 @@ class UserUpdate(UserBase, BaseZenModel):
         title="Whether the account is an administrator.",
     )
     active: Optional[bool] = Field(
         default=None, title="Whether the account is active."
     )
+    old_password: Optional[str] = Field(
+        default=None,
+        title="The previous password for the user. Only relevant for user "
+        "accounts. Required when updating the password.",
+        max_length=STR_FIELD_MAX_LENGTH,
+    )
 
     @root_validator
     def user_email_updates(cls, values: Dict[str, Any]) -> Dict[str, Any]:
         """Validate that the UserUpdateModel conforms to the email-opt-in-flow.
 
diff --git a/src/zenml/zen_server/routers/users_endpoints.py b/src/zenml/zen_server/routers/users_endpoints.py
index f75bc42b5d..d26a94b12c 100644
--- a/src/zenml/zen_server/routers/users_endpoints.py
+++ b/src/zenml/zen_server/routers/users_endpoints.py
@@ -30,10 +30,11 @@ from zenml.constants import (
 from zenml.enums import AuthScheme
 from zenml.exceptions import AuthorizationException, IllegalOperationError
 from zenml.logger import get_logger
 from zenml.models import (
     Page,
+    UserAuthModel,
     UserFilter,
     UserRequest,
     UserResponse,
     UserUpdate,
 )
@@ -251,21 +252,47 @@ if server_config().auth_scheme != AuthScheme.EXTERNAL:
         Returns:
             The updated user.
 
         Raises:
             IllegalOperationError: if the user tries change admin status,
-                while not an admin
+                while not an admin, if the user tries to change the password
+                of another user, or if the user tries to change their own
+                password without providing the old password or providing
+                an incorrect old password.
         """
         user = zen_store().get_user(user_name_or_id)
         if user.id != auth_context.user.id:
             verify_admin_status_if_no_rbac(
                 auth_context.user.is_admin, "update other user"
             )
             verify_permission_for_model(
                 user,
                 action=Action.UPDATE,
             )
+
+            if user_update.password is not None:
+                raise IllegalOperationError(
+                    "Users cannot change the password of other users. Use the "
+                    "account deactivation and activation flow instead."
+                )
+
+        elif user_update.password is not None:
+            # If the user is updating their own password, we need to verify
+            # the old password
+            if user_update.old_password is None:
+                raise IllegalOperationError(
+                    "The current password must be supplied when changing the "
+                    "password."
+                )
+            auth_user = zen_store().get_auth_user(user_name_or_id)
+            if not UserAuthModel.verify_password(
+                user_update.old_password, auth_user
+            ):
+                raise IllegalOperationError(
+                    "The current password is incorrect."
+                )
+
         if (
             user_update.is_admin is not None
             and user.is_admin != user_update.is_admin
             and not auth_context.user.is_admin
         ):
@@ -510,12 +537,31 @@ if server_config().auth_scheme != AuthScheme.EXTERNAL:
             user: the user to use for the update.
             auth_context: The authentication context.
 
         Returns:
             The updated user.
+
+        Raises:
+            IllegalOperationError: if the current password is not supplied when
+                changing the password or if the current password is incorrect.
         """
         current_user = zen_store().get_user(auth_context.user.id)
+
+        if user.password is not None:
+            # If the user is updating their password, we need to verify
+            # the old password
+            if user.old_password is None:
+                raise IllegalOperationError(
+                    "The current password must be supplied when changing the "
+                    "password."
+                )
+            auth_user = zen_store().get_auth_user(auth_context.user.id)
+            if not UserAuthModel.verify_password(user.old_password, auth_user):
+                raise IllegalOperationError(
+                    "The current password is incorrect."
+                )
+
         user.activation_token = current_user.activation_token
         user.active = current_user.active
         user.is_admin = current_user.is_admin
 
         updated_user = zen_store().update_user(
diff --git a/src/zenml/zen_stores/rest_zen_store.py b/src/zenml/zen_stores/rest_zen_store.py
index d03cec9f79..8b90338051 100644
--- a/src/zenml/zen_stores/rest_zen_store.py
+++ b/src/zenml/zen_stores/rest_zen_store.py
@@ -48,10 +48,11 @@ from zenml.constants import (
     ARTIFACT_VISUALIZATIONS,
     ARTIFACTS,
     CODE_REFERENCES,
     CODE_REPOSITORIES,
     CURRENT_USER,
+    DEACTIVATE,
     DEFAULT_HTTP_TIMEOUT,
     DEVICES,
     DISABLE_CLIENT_SERVER_MISMATCH_WARNING,
     ENV_ZENML_DISABLE_CLIENT_SERVER_MISMATCH_WARNING,
     EVENT_SOURCES,
@@ -2941,10 +2942,27 @@ class RestZenStore(BaseZenStore):
             resource_update=user_update,
             route=USERS,
             response_model=UserResponse,
         )
 
+    def deactivate_user(
+        self, user_name_or_id: Union[str, UUID]
+    ) -> UserResponse:
+        """Deactivates a user.
+
+        Args:
+            user_name_or_id: The name or ID of the user to delete.
+
+        Returns:
+            The deactivated user containing the activation token.
+        """
+        response_body = self.put(
+            f"{USERS}/{str(user_name_or_id)}{DEACTIVATE}",
+        )
+
+        return UserResponse.parse_obj(response_body)
+
     def delete_user(self, user_name_or_id: Union[str, UUID]) -> None:
         """Deletes a user.
 
         Args:
             user_name_or_id: The name or ID of the user to delete.
diff --git a/src/zenml/zen_stores/schemas/user_schemas.py b/src/zenml/zen_stores/schemas/user_schemas.py
index 72737ffa9b..51857d431c 100644
--- a/src/zenml/zen_stores/schemas/user_schemas.py
+++ b/src/zenml/zen_stores/schemas/user_schemas.py
@@ -204,10 +204,13 @@ class UserSchema(NamedSchema, table=True):
 
         Returns:
             The updated `UserSchema`.
         """
         for field, value in user_update.dict(exclude_unset=True).items():
+            if field == "old_password":
+                continue
+
             if field == "password":
                 setattr(self, field, user_update.create_hashed_password())
             elif field == "activation_token":
                 setattr(
                     self, field, user_update.create_hashed_activation_token()

Vulnerable code: from zenml.exceptions import IllegalOperationError
@click.option(
    "--workspace",
    help="The workspace to use when connecting to the ZenML server.",
    required=False,
    type=str,
)
    workspace: Optional[str] = None,
        workspace: The active workspace that is used to connect to the ZenML
            server.

    if workspace:
        try:
            Client().set_active_workspace(workspace_name_or_id=workspace)
        except KeyError:
            cli_utils.warning(
                f"The workspace {workspace} does not exist or is not accessible. "
                f"Please set another workspace by running `zenml "
                f"workspace set`."
            )
from zenml.exceptions import EntityExistsError, IllegalOperationError
@click.option(
    "--password",
    "-p",
    "updated_password",
    type=str,
    required=False,
    help="New user password.",
)
    updated_password: Optional[str] = None,
        updated_password: The name of the user to create.
                f"Currently user `{current_user.name}` is an admin. Are you sure you want to make them a regular user?"
            updated_password=updated_password,
                while not an admin
Fixed code: from zenml.exceptions import AuthorizationException, IllegalOperationError
    if password is not None:
        cli_utils.warning(
            "Supplying password values in the command line is not safe. "
            "Please consider using the prompt option."
        )

        cli_utils.warning(
            "Connecting to a ZenML server using a username and password is "
            "not recommended because the password is locally stored on your "
            "filesystem. You should consider using the web login workflow by "
            "omitting the `--username` and `--password` flags. An alternative "
            "for non-interactive environments is to create and use a service "
            "account API key (see https://docs.zenml.io/user-guide/advanced-guide/configuring-zenml/connecting-to-zenml#using-service-accounts-to-connect-to-a-deployed-zenml-server "
            "for more information)."
        )

    except AuthorizationException as e:
        cli_utils.warning(f"Authorization error: {e}")
from zenml.config.global_config import GlobalConfiguration
from zenml.exceptions import (
    AuthorizationException,
    EntityExistsError,
    IllegalOperationError,
)
    else:
        cli_utils.warning(
            "Supplying password values in the command line is not safe. "
            "Please consider using the prompt option."
        )
@click.option(
    "--active",
    "active",
    type=bool,
    required=False,
    default=None,
    help="Use to activate or deactivate a user account.",
)
    active: Optional[bool] = None,
        active: Use to activate or deactivate a user account.
                f"Currently user `{current_user.name}` is an admin. Are you "
                "sure you want to make them a regular user?"
            active=active,
@user.command(
    "change-password",
    help="Change the password for the current user account.",
)
@click.option(
    "--password",
    help=(
        "The new user password. If omitted, a prompt will be shown to enter "
        "the password."
    ),
    required=False,
    type=str,
)
@click.option(
    "--old-password",
    help=(
        "The old user password. If omitted, a prompt will be shown to enter "
        "the old password."
    ),
    required=False,
    type=str,
)
def change_user_password(
    password: Optional[str] = None, old_password: Optional[str] = None
) -> None:
    """Change the password of the current user.

    Args:
        password: The new password for the current user.
        old_password: The old password for the current user.
    """
    active_user = Client().active_user

    if old_password is not None or password is not None:
        cli_utils.warning(
            "Supplying password values in the command line is not safe. "
            "Please consider using the prompt option."
        )

    if old_password is None:
        old_password = click.prompt(
            f"Current password for user {active_user.name}",
            hide_input=True,
        )
    if password is None:
        password = click.prompt(
            f"New password for user {active_user.name}",
            hide_input=True,
        )
        password_again = click.prompt(
            f"Please re-enter the new password for user {active_user.name}",
            hide_input=True,
        )
        if password != password_again:
            cli_utils.error("Passwords do not match.")

    try:
        Client().update_user(
            name_id_or_prefix=active_user.id,
            old_password=old_password,
            updated_password=password,
        )
    except (KeyError, IllegalOperationError, AuthorizationException) as err:
        cli_utils.error(str(err))

    cli_utils.declare(
        f"Successfully updated password for active user '{active_user.name}'."
    )

    store = GlobalConfiguration().store_configuration
    if store.type == StoreType.REST:
        from zenml.zen_stores.rest_zen_store import RestZenStoreConfiguration

        assert isinstance(store, RestZenStoreConfiguration)

        if store.password is not None:
            cli_utils.declare(
                "You may need to log in again with your new password by "
                "running `zenml connect`."
            )


@user.command(
    "deactivate",
    help="Generate an activation token to reset the password for a user account",
)
@click.argument("user_name_or_id", type=str, required=True)
def deactivate_user(
    user_name_or_id: str,
) -> None:
    """Reset the password of a user.

    Args:
        user_name_or_id: The name or ID of the user to reset the password for.
    """
    client = Client()

    store = GlobalConfiguration().store_configuration
    if store.type != StoreType.REST:
        cli_utils.error(
            "Deactivating users is only supported when connected to a ZenML "
            "server."
        )

    try:
        if not client.active_user.is_admin:
            cli_utils.error(
                "Only admins can reset the password of other users."
            )

        user = client.deactivate_user(
            name_id_or_prefix=user_name_or_id,
        )
    except (KeyError, IllegalOperationError) as err:
        cli_utils.error(str(err))

    cli_utils.declare(
        f"Successfully deactivated user account '{user.name}'."
        f"To reactivate the account, please visit the dashboard at the "
        "following URL:\n"
        f"{client.zen_store.url}/signup?user={str(user.id)}&username={user.name}&token={user.activation_token}\n"
    )


        old_password: Optional[str] = None,
        active: Optional[bool] = None,
            old_password: The old password of the user. Required for password
                update.
            active: Use to activate or deactivate the user.

        Raises:
            ValidationError: If the old password is not provided when updating
                the password.
            if old_password is None:
                raise ValidationError(
                    "Old password is required to update the password."
                )
            user_update.old_password = old_password
        if active is not None:
            user_update.active = active
    @_fail_for_sql_zen_store
    def deactivate_user(self, name_id_or_prefix: str) -> "UserResponse":
        """Deactivate a user and generate an activation token.

        Args:
            name_id_or_prefix: The name or ID of the user to reset.

        Returns:
            The deactivated user.
        """
        from zenml.zen_stores.rest_zen_store import RestZenStore

        user = self.get_user(name_id_or_prefix, allow_name_prefix_match=False)
        assert isinstance(self.zen_store, RestZenStore)
        return self.zen_store.deactivate_user(user_name_or_id=user.name)

    old_password: Optional[str] = Field(
        default=None,
        title="The previous password for the user. Only relevant for user "
        "accounts. Required when updating the password.",
        max_length=STR_FIELD_MAX_LENGTH,
    )
    UserAuthModel,
                while not an admin, if the user tries to change the password
                of another user, or if the user tries to change their own
                password without providing the old password or providing
                an incorrect old password.

            if user_update.password is not None:
                raise IllegalOperationError(
                    "Users cannot change the password of other users. Use the "
                    "account deactivation and activation flow instead."
                )

        elif user_update.password is not None:
            # If the user is updating their own password, we need to verify
            # the old password
            if user_update.old_password is None:
                raise IllegalOperationError(
                    "The current password must be supplied when changing the "
                    "password."
                )
            auth_user = zen_store().get_auth_user(user_name_or_id)
            if not UserAuthModel.verify_password(
                user_update.old_password, auth_user
            ):
                raise IllegalOperationError(
                    "The current password is incorrect."
                )


        Raises:
            IllegalOperationError: if the current password is not supplied when
                changing the password or if the current password is incorrect.

        if user.password is not None:
            # If the user is updating their password, we need to verify
            # the old password
            if user.old_password is None:
                raise IllegalOperationError(
                    "The current password must be supplied when changing the "
                    "password."
                )
            auth_user = zen_store().get_auth_user(auth_context.user.id)
            if not UserAuthModel.verify_password(user.old_password, auth_user):
                raise IllegalOperationError(
                    "The current password is incorrect."
                )

    DEACTIVATE,
    def deactivate_user(
        self, user_name_or_id: Union[str, UUID]
    ) -> UserResponse:
        """Deactivates a user.

        Args:
            user_name_or_id: The name or ID of the user to delete.

        Returns:
            The deactivated user containing the activation token.
        """
        response_body = self.put(
            f"{USERS}/{str(user_name_or_id)}{DEACTIVATE}",
        )

        return UserResponse.parse_obj(response_body)

            if field == "old_password":
                continue