CVE-2024-2213 (Python, published 2024-06-06T19:15Z)
Description: 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.
Severity: nan (CVSS v3 base score: )
Weakness: CWE-287 \u2014 Improper Authentication
When an actor claims to have a given identity, the product does not prove or insufficiently proves that the claim is correct.
Repository: https://github.com/zenml-io/zenml
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 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>

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