CVE-2024-2035 (Python, published 2024-06-06T19:15Z)
Description: value: An improper authorization vulnerability exists in the zenml-io/zenml repository, specifically within the API PUT /api/v1/users/id endpoint. This vulnerability allows any authenticated user to modify the information of other users, including changing the `active` status of user accounts to false, effectively deactivating them. This issue affects version 0.55.3 and was fixed in version 0.56.2. The impact of this vulnerability is significant as it allows for the deactivation of admin accounts, potentially disrupting the functionality and security of the application.
Severity: nan (CVSS v3 base score: )
Weakness: NVD-CWE-noinfo \u2014 Insufficient Information
There is insufficient information about the issue to classify it; details are unkown or unspecified.
Repository: https://github.com/zenml-io/zenml
Files changed:
examples/quickstart/README.md
src/zenml/cli/user_management.py
src/zenml/client.py
src/zenml/models/v2/core/service_account.py
src/zenml/models/v2/core/user.py
src/zenml/models/v2/misc/external_user.py
src/zenml/zen_server/auth.py
src/zenml/zen_server/rbac/endpoint_utils.py
src/zenml/zen_server/rbac/utils.py
src/zenml/zen_server/routers/users_endpoints.py
src/zenml/zen_server/utils.py
src/zenml/zen_stores/migrations/versions/1a9a9d2a836d_admin_users.py
src/zenml/zen_stores/schemas/user_schemas.py
src/zenml/zen_stores/sql_zen_store.py
tests/integration/functional/zen_stores/test_zen_store.py
tests/integration/functional/zen_stores/utils.py
tests/unit/conftest.py
tests/unit/models/test_user_models.py
Commit message: Add admin users notion (#2494)

* add admin users to OSS

* skip missing methods

* increase readability

* doc string

* lint

* lint

* missing arg

* add some edge-cases

* wip commit to carve out clean_client changes

* revert irrelevant changes

* revert irrelevant changes

* rework tests to run on rest

* Apply suggestions from code review

Co-authored-by: Alex Strick van Linschoten <strickvl@users.noreply.github.com>
Co-authored-by: Stefan Nica <stefan@zenml.io>

* polish test cases

* fix branching

* admin user mgmt CLI/Client

* close activation vulnerability

* revert rbac changes

* verify admin permissions in endpoints

* add `is_admin` to external users

* only reg users will be migrated as admins

* default is always admin

* extend tests

* lint

* default `is_admin` None

* Auto-update of Starter template

* review suggestions

* review suggestions

* calm down linter

* Update src/zenml/cli/user_management.py

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

* Apply suggestions from code review

Co-authored-by: Alex Strick van Linschoten <strickvl@users.noreply.github.com>

* review suggestion

---------

Co-authored-by: Alex Strick van Linschoten <strickvl@users.noreply.github.com>
Co-authored-by: Stefan Nica <stefan@zenml.io>
Co-authored-by: GitHub Actions <actions@github.com>
Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

Vulnerable code:
<a href="https://colab.research.google.com/github/zenml-io/zenml/blob/main/examples/quickstart/run.ipynb" target="_parent"><img src="https://colab.research.google.com/assets/colab-badge.svg" alt="Open In Colab"/></a>
</a> to become part of the ZenML family!
        new_user = client.create_user(name=user_name, password=password)
    help="Update user information through the cli. All attributes "
    "except for password can be updated through the cli.",
    """Create a new user.
        user = UserRequest(name=name, password=password or None)
from pydantic import Field, root_validator
from zenml.models.v2.base.update import update_model
# ------------------ Request Model ------------------

class UserRequest(BaseRequest):
    """Request model for users."""
    # Analytics fields for user request models
    ANALYTICS_FIELDS: ClassVar[List[str]] = [
        "name",
        "full_name",
        "active",
        "email_opted_in",
    ]
    name: str = Field(
        title="The unique username for the account.",
        max_length=STR_FIELD_MAX_LENGTH,
    )
    full_name: str = Field(
        default="",
        title="The full name for the account owner. Only relevant for user "
        "accounts.",
        max_length=STR_FIELD_MAX_LENGTH,
    )
    active: bool = Field(default=False, title="Whether the account is active.")

    class Config:
        """Pydantic configuration class."""

        # Validate attributes when assigning them
        validate_assignment = True

        # Forbid extra attributes to prevent unexpected behavior
        extra = "forbid"
        underscore_attrs_are_private = True
@update_model
class UserUpdate(UserRequest):
    verify_permission(resource_type=resource_type, action=Action.CREATE)
def batch_verify_permissions(resources: Set[Resource], action: Action) -> None:
        _: AuthContext = Security(authorize),
        verify_permission_for_model(user, action=Action.READ)
            verify_permission_for_model(user, action=Action.UPDATE)
        if user.id != auth_context.user.id:
            verify_permission_for_model(user, action=Action.UPDATE)
            verify_permission_for_model(user, action=Action.DELETE)

from zenml.exceptions import OAuthError
                UserRequest(name=existing_user.name, password="password")
                    name=existing_service_account.name, password="password"
            args=(UserRequest(name=user_name, password=password),),
                body=UserUpdate(password="password"),
                password="password", activation_token=user.activation_token
                body=UserUpdate(password="newpassword"),
                name=self.user_name, password=self.password, active=True
    create_model=UserRequest(name=sample_name("sample_user")),
        UserRequest(
            password=long_password,
        )
        UserRequest(
            activation_token=long_token,
        )

Fixed code:
<a href="https://colab.research.google.com/github/zenml-io/zenml/blob/main/examples/quickstart/quickstart.ipynb" target="_parent"><img src="https://colab.research.google.com/assets/colab-badge.svg" alt="Open In Colab"/></a>
</a> to become part of the ZenML family!
@click.option(
    "--is_admin",
    is_flag=True,
    help=(
        "Whether the user should be an admin. If not specified, the user will "
        "be a regular user."
    ),
    required=False,
    default=False,
)
    is_admin: bool = False,
        is_admin: Whether the user should be an admin.
        new_user = client.create_user(
            name=user_name, password=password, is_admin=is_admin
        )
    help="Update user information through the cli.",
@click.option(
    "--password",
    "-p",
    "updated_password",
    type=str,
    required=False,
    help="New user password.",
)
@click.option(
    "--admin",
    "-a",
    "make_admin",
    is_flag=True,
    required=False,
    default=None,
    help="Whether the user should be an admin.",
)
@click.option(
    "--user",
    "-u",
    "make_user",
    is_flag=True,
    required=False,
    default=None,
    help="Whether the user should be a regular user.",
)
    updated_password: Optional[str] = None,
    make_admin: Optional[bool] = None,
    make_user: Optional[bool] = None,
    """Update an existing user.
        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.
    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."
        )
        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?"
            )
            if not confirmation:
                cli_utils.declare("User update canceled.")
                return

        updated_is_admin = None
        if make_admin is True:
            updated_is_admin = True
        elif make_user is True:
            updated_is_admin = False
            updated_password=updated_password,
            updated_is_admin=updated_is_admin,
        is_admin: bool = False,
            is_admin: Whether the user should be an admin.
        user = UserRequest(
            name=name, password=password or None, is_admin=is_admin
        )
        updated_password: Optional[str] = None,
        updated_is_admin: Optional[bool] = None,
            updated_password: The new password of the user.
            updated_is_admin: Whether the user should be an admin.
        if updated_password is not None:
            user_update.password = updated_password
        if updated_is_admin is not None:
            user_update.is_admin = updated_is_admin
                is_admin=False,
from pydantic import BaseModel, Field, root_validator
    BaseZenModel,
# ------------------ Base Model ------------------
class UserBase(BaseModel):
    """Base model for users."""

# ------------------ Request Model ------------------


class UserRequest(UserBase, BaseRequest):
    """Request model for users."""

    # Analytics fields for user request models
    ANALYTICS_FIELDS: ClassVar[List[str]] = [
        "name",
        "full_name",
        "active",
        "email_opted_in",
    ]

    name: str = Field(
        title="The unique username for the account.",
        max_length=STR_FIELD_MAX_LENGTH,
    )
    full_name: str = Field(
        default="",
        title="The full name for the account owner. Only relevant for user "
        "accounts.",
        max_length=STR_FIELD_MAX_LENGTH,
    )
    is_admin: bool = Field(
        title="Whether the account is an administrator.",
    )
    active: bool = Field(default=False, title="Whether the account is active.")

    class Config:
        """Pydantic configuration class."""

        # Validate attributes when assigning them
        validate_assignment = True

        # Forbid extra attributes to prevent unexpected behavior
        extra = "forbid"
        underscore_attrs_are_private = True


class UserUpdate(UserBase, BaseZenModel):
    name: Optional[str] = Field(
        title="The unique username for the account.",
        max_length=STR_FIELD_MAX_LENGTH,
        default=None,
    )
    full_name: Optional[str] = Field(
        default=None,
        title="The full name for the account owner. Only relevant for user "
        "accounts.",
        max_length=STR_FIELD_MAX_LENGTH,
    )
    is_admin: Optional[bool] = Field(
        default=None,
        title="Whether the account is an administrator.",
    )
    active: Optional[bool] = Field(
        default=None, title="Whether the account is active."
    )

    is_admin: bool = Field(
        title="Whether the account is an administrator.",
    )
    @property
    def is_admin(self) -> bool:
        """The `is_admin` property.

        Returns:
            Whether the user is an admin.
        """
        return self.get_body().is_admin

    is_admin: bool = False
                is_admin=external_user.is_admin,
                is_admin=external_user.is_admin,
    verify_permission(
        resource_type=resource_type,
        action=Action.CREATE,
    )
def batch_verify_permissions(
    resources: Set[Resource],
    action: Action,
) -> None:
    verify_admin_status_if_no_rbac,
    else:
        if not auth_context.user.is_admin and not server_config().rbac_enabled:
            allowed_ids = {auth_context.user.id}
        auth_context: AuthContext = Security(authorize),
            auth_context: Authentication context.
        verify_admin_status_if_no_rbac(
            auth_context.user.is_admin, "create user"
        )

        verify_admin_status_if_no_rbac(
            auth_context.user.is_admin, "get other user"
        )
        verify_permission_for_model(
            user,
            action=Action.READ,
        )

        Raises:
            IllegalOperationError: if the user tries change admin status,
                while not an admin
            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.is_admin is not None
            and user.is_admin != user_update.is_admin
            and not auth_context.user.is_admin
        ):
            raise IllegalOperationError(
                "Only admins can change the admin status of other users."
            )
        user_update.activation_token = user.activation_token
        if not auth_context.user.is_admin or user.id == auth_context.user.id:
            user_update.active = user.active
        user_update.is_admin = user.is_admin

        Raises:
            IllegalOperationError: if the user is trying to deactivate
                themselves.
        if user.id == auth_context.user.id:
            raise IllegalOperationError("Cannot deactivate yourself.")
        verify_admin_status_if_no_rbac(
            auth_context.user.is_admin, "deactivate user"
        )
        verify_permission_for_model(
            user,
            action=Action.UPDATE,
        )
            verify_admin_status_if_no_rbac(
                auth_context.user.is_admin, "delete user"
            )
            verify_permission_for_model(
                user,
                action=Action.DELETE,
            )
        current_user = zen_store().get_user(auth_context.user.id)
        user.activation_token = current_user.activation_token
        user.active = current_user.active
        user.is_admin = current_user.is_admin

from zenml.exceptions import IllegalOperationError, OAuthError


def verify_admin_status_if_no_rbac(
    admin_status: Optional[bool],
    action: Optional[str] = None,
) -> None:
    """Validate the admin status for sensitive requests.

    Only add this check in endpoints meant for admin use only.

    Args:
        admin_status: Whether the user is an admin or not. This is only used
            if explicitly specified in the call and even if passed will be
            ignored, if RBAC is enabled.
        action: The action that is being performed, used for output only.

    Raises:
        IllegalOperationError: If the admin status is not valid.
    """
    if not server_config().rbac_enabled:
        if not action:
            action = "this action"
        else:
            action = f"`{action.strip('`')}`"

        if admin_status is False:
            raise IllegalOperationError(
                message=f"Only admin users can perform {action} "
                "without RBAC enabled.",
            )
    return
"""admin users [1a9a9d2a836d].

Revision ID: 1a9a9d2a836d
Revises: 0.55.5
Create Date: 2024-03-04 15:48:16.580871

"""

import sqlalchemy as sa
import sqlmodel
from alembic import op

# revision identifiers, used by Alembic.
revision = "1a9a9d2a836d"
down_revision = "0.55.5"
branch_labels = None
depends_on = None


def upgrade() -> None:
    """Upgrade database schema and/or data, creating a new revision."""
    # ### commands auto generated by Alembic - please adjust! ###
    bind = op.get_bind()
    session = sqlmodel.Session(bind=bind)

    with op.batch_alter_table("user", schema=None) as batch_op:
        batch_op.add_column(
            sa.Column(
                "is_admin",
                sa.Boolean(),
                nullable=False,
                server_default=sa.sql.expression.false(),
            )
        )

    # during migration we treat all users as admin for backward compatibility
    # this should be adjusted by server admins after upgrade
    session.execute(
        sa.text(
            """
            UPDATE user
            SET is_admin = true
            WHERE NOT is_service_account AND external_user_id IS NULL
            """
        )
    )
    # ### end Alembic commands ###


def downgrade() -> None:
    """Downgrade database schema and/or data back to the previous revision."""
    # ### commands auto generated by Alembic - please adjust! ###
    with op.batch_alter_table("user", schema=None) as batch_op:
        batch_op.drop_column("is_admin")

    # ### end Alembic commands ###
    is_admin: bool = Field(default=False)
            is_admin=model.is_admin,
            is_admin=False,
                is_admin=self.is_admin,
            if (
                existing_user.name == self._default_user_name
                and user_update.is_admin is False
            ):
                raise IllegalOperationError(
                    "The default user's admin status cannot be removed."
                )

                    is_admin=True,
import random
from string import ascii_lowercase
from zenml.zen_stores.rest_zen_store import RestZenStore
class TestAdminUser:
    default_pwd = "".join(random.choices(ascii_lowercase, k=10))

    def test_creation_as_admin_and_non_admin(self):
        """Tests creating a user as an admin and as a non-admin."""
        if Client().zen_store.type == StoreType.SQL:
            pytest.skip("SQL ZenStore does not support admin users.")

        with UserContext(login=True, is_admin=False):
            zen_store: RestZenStore = Client().zen_store
            # this is not allowed for non-admin users
            with pytest.raises(IllegalOperationError):
                zen_store.create_user(
                    UserRequest(
                        name=sample_name("test_user"),
                        password=self.default_pwd,
                        is_admin=False,
                    )
                )

    def test_listing_users(self):
        """Tests listing users as an admin and as a non-admin."""
        if Client().zen_store.type == StoreType.SQL:
            pytest.skip("SQL ZenStore does not support admin users.")

        with UserContext(
            password=self.default_pwd, is_admin=False
        ) as test_user:
            zen_store: RestZenStore = Client().zen_store
            users = zen_store.list_users(UserFilter())
            assert users.total >= 2

            # this is limited to self only for non-admin users
            with LoginContext(
                user_name=test_user.name, password=self.default_pwd
            ):
                zen_store = Client().zen_store
                users = zen_store.list_users(UserFilter())
                assert users.total == 1

    def test_get_users(self):
        """Tests getting users as an admin and as a non-admin."""
        if Client().zen_store.type == StoreType.SQL:
            pytest.skip("SQL ZenStore does not support admin users.")

        with UserContext(
            password=self.default_pwd, is_admin=False
        ) as test_user:
            zen_store: RestZenStore = Client().zen_store

            user = zen_store.get_user(test_user.name)
            assert user.id == test_user.id

            # this is not allowed for non-admin users
            with LoginContext(
                user_name=test_user.name, password=self.default_pwd
            ):
                zen_store = Client().zen_store
                with pytest.raises(IllegalOperationError):
                    zen_store.get_user(DEFAULT_USERNAME)

    def test_update_users(self):
        """Tests updating users as an admin and as a non-admin."""
        if Client().zen_store.type == StoreType.SQL:
            pytest.skip("SQL ZenStore does not support admin users.")

        with UserContext(
            password=self.default_pwd, is_admin=False
        ) as test_user:
            zen_store: RestZenStore = Client().zen_store

            user = zen_store.update_user(
                test_user.id,
                UserUpdate(full_name="foo@bar.ai"),
            )
            assert user.full_name == "foo@bar.ai"

            with UserContext(login=True, is_admin=False):
                zen_store = Client().zen_store

                # this is not allowed for non-admin users
                with pytest.raises(IllegalOperationError):
                    zen_store.update_user(
                        test_user.id,
                        UserUpdate(full_name="bar@foo.io"),
                    )

            # user is allowed to update itself
            with LoginContext(
                user_name=test_user.name, password=self.default_pwd
            ):
                zen_store = Client().zen_store
                user = zen_store.update_user(
                    test_user.id,
                    UserUpdate(full_name="bar@foo.io"),
                )

                assert user.full_name == "bar@foo.io"

    def test_deactivate_users(self):
        """Tests deactivating users as an admin and as a non-admin."""
        if Client().zen_store.type == StoreType.SQL:
            pytest.skip("SQL ZenStore does not support admin users.")

        zen_store: RestZenStore = Client().zen_store
        with UserContext(
            password=self.default_pwd, is_admin=False
        ) as test_user:
            with UserContext(is_admin=False) as test_user2:
                # this is not allowed for non-admin users
                with LoginContext(
                    user_name=test_user.name, password=self.default_pwd
                ):
                    new_zen_store: RestZenStore = Client().zen_store
                    with pytest.raises(IllegalOperationError):
                        new_zen_store.put(
                            f"{USERS}/{str(test_user2.id)}{DEACTIVATE}"
                        )

                response_body = zen_store.put(
                    f"{USERS}/{str(test_user2.id)}{DEACTIVATE}",
                )
                deactivated_user = UserResponse.parse_obj(response_body)
                assert deactivated_user.name == test_user2.name

    def test_delete_users(self):
        """Tests deleting users as an admin and as a non-admin."""
        if Client().zen_store.type == StoreType.SQL:
            pytest.skip("SQL ZenStore does not support admin users.")

        zen_store: RestZenStore = Client().zen_store
        with UserContext(
            password=self.default_pwd, is_admin=False
        ) as test_user:
            with UserContext(is_admin=False) as test_user2:
                # this is not allowed for non-admin users
                with LoginContext(
                    user_name=test_user.name, password=self.default_pwd
                ):
                    new_zen_store: RestZenStore = Client().zen_store
                    with pytest.raises(IllegalOperationError):
                        new_zen_store.delete_user(test_user2.id)

            zen_store.delete_user(test_user.id)

    def test_update_self_via_normal_endpoint(self):
        """Tests updating self in admin and non-admin setting."""
        if Client().zen_store.type == StoreType.SQL:
            pytest.skip("SQL ZenStore does not support admin users.")

        zen_store: RestZenStore = Client().zen_store
        default_user = zen_store.get_user(DEFAULT_USERNAME)
        with pytest.raises(IllegalOperationError):
            # cannot update admin status for default user
            zen_store.update_user(
                default_user.id,
                UserUpdate(name=default_user.name, is_admin=False),
            )
        with UserContext(
            password=self.default_pwd, is_admin=False
        ) as test_user:
            # this is not allowed for non-admin users
            with LoginContext(
                user_name=test_user.name, password=self.default_pwd
            ):
                new_zen_store: RestZenStore = Client().zen_store
                with pytest.raises(IllegalOperationError):
                    new_zen_store.update_user(
                        test_user.id,
                        UserUpdate(
                            name=test_user.name,
                            is_admin=True,
                        ),
                    )

        with UserContext(
            password=self.default_pwd, is_admin=False
        ) as test_user:
            zen_store.update_user(
                test_user.id,
                UserUpdate(name=test_user.name, is_admin=True),
            )
            with LoginContext(
                user_name=test_user.name, password=self.default_pwd
            ):
                new_zen_store: RestZenStore = Client().zen_store
                new_zen_store.update_user(
                    test_user.id,
                    UserUpdate(
                        name=test_user.name,
                        is_admin=False,
                    ),
                )

    def test_update_self_via_current_user_endpoint(self):
        """Tests updating self in admin and non-admin setting."""
        if Client().zen_store.type == StoreType.SQL:
            pytest.skip("SQL ZenStore does not support admin users.")

        zen_store: RestZenStore = Client().zen_store
        default_user = zen_store.get_user(DEFAULT_USERNAME)
        assert default_user.is_admin
        # self update cannot change admin status
        zen_store.put(
            "/current-user",
            body=UserUpdate(name=default_user.name, is_admin=False),
        )
        default_user = zen_store.get_user(DEFAULT_USERNAME)
        assert default_user.is_admin
        with UserContext(
            password=self.default_pwd, is_admin=False
        ) as test_user:
            with LoginContext(
                user_name=test_user.name, password=self.default_pwd
            ):
                new_zen_store: RestZenStore = Client().zen_store
                assert not test_user.is_admin
                # self update cannot change admin status
                new_zen_store.put(
                    "/current-user",
                    body=UserUpdate(
                        name=test_user.name,
                        is_admin=True,
                    ),
                )
                user = zen_store.get_user(test_user.id)
                assert not user.is_admin


                UserRequest(
                    name=existing_user.name,
                    password="password",
                    is_admin=False,
                )
                    name=existing_service_account.name,
                    password="password",
                    is_admin=False,
            args=(
                UserRequest(name=user_name, password=password, is_admin=False),
            ),
                body=UserUpdate(password="password", is_admin=user.is_admin),
                password="password",
                activation_token=user.activation_token,
                is_admin=user.is_admin,
                body=UserUpdate(
                    password="newpassword", is_admin=user.is_admin
                ),
                is_admin=user.is_admin,
        is_admin: bool = True,
        self.is_admin = is_admin
                name=self.user_name,
                password=self.password,
                active=True,
                is_admin=self.is_admin,
    create_model=UserRequest(name=sample_name("sample_user"), is_admin=True),
            is_admin=True,
        UserRequest(password=long_password, is_admin=False)
        UserRequest(activation_token=long_token, is_admin=False)