CVE-2024-2035: Insufficient Information (NVD-CWE-noinfo)
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.
CWE description: There is insufficient information about the issue to classify it; details are unkown or unspecified.
Published: 2024-06-06T19:15Z
Repository: https://github.com/zenml-io/zenml
Language: Python
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 date: 2024-03-12 17:09:59 +0100
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>
Diff: diff --git a/examples/quickstart/README.md b/examples/quickstart/README.md
index be1d2a1c6..23c8f24e6 100644
--- a/examples/quickstart/README.md
+++ b/examples/quickstart/README.md
@@ -22,11 +22,11 @@ Along the way we will also show you how to:
 
 ## 🏃 Run on Colab
 
 You can use Google Colab to see ZenML in action, no signup / installation required!
 
-<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 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>
 
 ## :computer: Run Locally
 
 To run locally, install ZenML and pull this quickstart:
 
@@ -206,6 +206,6 @@ goes into more detail as to how to transition these same pipelines into producti
 The best way to get a production ZenML instance up and running with all batteries included is the [ZenML Cloud](https://zenml.io/cloud). Check it out!
 
 Also, make sure to join our <a href="https://zenml.io/slack" target="_blank">
     <img width="15" src="https://cdn3.iconfinder.com/data/icons/logos-and-brands-adobe/512/306_Slack-512.png" alt="Slack"/>
     <b>Slack Community</b> 
-</a> to become part of the ZenML family!
\ No newline at end of file
+</a> to become part of the ZenML family!
diff --git a/src/zenml/cli/user_management.py b/src/zenml/cli/user_management.py
index 2e5a7813b..933d8e790 100644
--- a/src/zenml/cli/user_management.py
+++ b/src/zenml/cli/user_management.py
@@ -118,19 +118,31 @@ def list_users(ctx: click.Context, **kwargs: Any) -> None:
         "activate the account."
     ),
     required=False,
     type=str,
 )
+@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,
+)
 def create_user(
     user_name: str,
     password: Optional[str] = None,
+    is_admin: bool = False,
 ) -> None:
     """Create a new user.
 
     Args:
         user_name: The name of the user to create.
         password: The password of the user to create.
+        is_admin: Whether the user should be an admin.
     """
     client = Client()
     if not password:
         if client.zen_store.type != StoreType.REST:
             password = click.prompt(
@@ -144,11 +156,13 @@ def create_user(
                 default="",
                 hide_input=True,
             )
 
     try:
-        new_user = client.create_user(name=user_name, password=password)
+        new_user = client.create_user(
+            name=user_name, password=password, is_admin=is_admin
+        )
 
         cli_utils.declare(f"Created user '{new_user.name}'.")
     except EntityExistsError as err:
         cli_utils.error(str(err))
     else:
@@ -160,12 +174,11 @@ def create_user(
             )
 
 
 @user.command(
     "update",
-    help="Update user information through the cli. All attributes "
-    "except for password can be updated through the cli.",
+    help="Update user information through the cli.",
 )
 @click.argument("user_name_or_id", type=str, required=True)
 @click.option(
     "--name",
     "-n",
@@ -189,30 +202,84 @@ 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,
+    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.",
+)
 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,
 ) -> None:
-    """Create a new user.
+    """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.
     """
+    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."
+        )
     try:
+        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
         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,
         )
     except (KeyError, IllegalOperationError) as err:
         cli_utils.error(str(err))
 
 
diff --git a/src/zenml/client.py b/src/zenml/client.py
index 0a3af4577..66f144b4e 100644
--- a/src/zenml/client.py
+++ b/src/zenml/client.py
@@ -689,22 +689,26 @@ class Client(metaclass=ClientMetaClass):
 
     def create_user(
         self,
         name: str,
         password: Optional[str] = None,
+        is_admin: bool = False,
     ) -> UserResponse:
         """Create a new user.
 
         Args:
             name: The name of the user.
             password: The password of the user. If not provided, the user will
                 be created with empty password.
+            is_admin: Whether the user should be an admin.
 
         Returns:
             The model of the created user.
         """
-        user = UserRequest(name=name, password=password or None)
+        user = UserRequest(
+            name=name, password=password or None, is_admin=is_admin
+        )
         user.active = (
             password != "" if self.zen_store.type != StoreType.REST else True
         )
         created_user = self.zen_store.create_user(user=user)
 
@@ -799,20 +803,24 @@ class Client(metaclass=ClientMetaClass):
         updated_name: Optional[str] = None,
         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,
+        updated_is_admin: Optional[bool] = None,
     ) -> UserResponse:
         """Update a user.
 
         Args:
             name_id_or_prefix: The name or ID of the user to update.
             updated_name: The new name of the user.
             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.
+            updated_is_admin: Whether the user should be an admin.
 
         Returns:
             The updated user.
         """
         user = self.get_user(
@@ -828,10 +836,14 @@ class Client(metaclass=ClientMetaClass):
             )
         if updated_email_opt_in is not None:
             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 updated_is_admin is not None:
+            user_update.is_admin = updated_is_admin
 
         return self.zen_store.update_user(
             user_id=user.id, user_update=user_update
         )
 
diff --git a/src/zenml/models/v2/core/service_account.py b/src/zenml/models/v2/core/service_account.py
index d051c8c32..ba200711c 100644
--- a/src/zenml/models/v2/core/service_account.py
+++ b/src/zenml/models/v2/core/service_account.py
@@ -145,10 +145,11 @@ class ServiceAccountResponse(
                 active=self.active,
                 is_service_account=True,
                 email_opted_in=False,
                 created=self.created,
                 updated=self.updated,
+                is_admin=False,
             ),
             metadata=UserResponseMetadata(
                 description=self.description,
             ),
         )
diff --git a/src/zenml/models/v2/core/user.py b/src/zenml/models/v2/core/user.py
index c20ef8382..fded4014a 100644
--- a/src/zenml/models/v2/core/user.py
+++ b/src/zenml/models/v2/core/user.py
@@ -24,53 +24,36 @@ from typing import (
     Type,
     Union,
 )
 from uuid import UUID
 
-from pydantic import Field, root_validator
+from pydantic import BaseModel, Field, root_validator
 
 from zenml.constants import STR_FIELD_MAX_LENGTH
 from zenml.models.v2.base.base import (
     BaseDatedResponseBody,
     BaseIdentifiedResponse,
     BaseRequest,
     BaseResponseMetadata,
     BaseResponseResources,
+    BaseZenModel,
 )
 from zenml.models.v2.base.filter import AnyQuery, BaseFilter
-from zenml.models.v2.base.update import update_model
 
 if TYPE_CHECKING:
     from passlib.context import CryptContext
 
     from zenml.models.v2.base.filter import AnySchema
 
-# ------------------ Request Model ------------------
-
+# ------------------ Base 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",
-    ]
+class UserBase(BaseModel):
+    """Base model for users."""
 
     # Fields
-    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,
-    )
+
     email: Optional[str] = Field(
         default=None,
         title="The email address associated with the account.",
         max_length=STR_FIELD_MAX_LENGTH,
     )
@@ -97,21 +80,10 @@ class UserRequest(BaseRequest):
     )
     external_user_id: Optional[UUID] = Field(
         default=None,
         title="The external user ID associated with the account.",
     )
-    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
 
     @classmethod
     def _get_crypt_context(cls) -> "CryptContext":
         """Returns the password encryption context.
 
@@ -163,17 +135,75 @@ class UserRequest(BaseRequest):
         """
         self.activation_token = token_hex(32)
         return self.activation_token
 
 
+# ------------------ 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
+
+
 # ------------------ Update Model ------------------
 
 
-@update_model
-class UserUpdate(UserRequest):
+class UserUpdate(UserBase, BaseZenModel):
     """Update model for users."""
 
+    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."
+    )
+
     @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.
 
         Args:
@@ -229,10 +259,13 @@ class UserResponseBody(BaseDatedResponseBody):
         "`false` if skipped.",
     )
     is_service_account: bool = Field(
         title="Indicates whether this is a service account or a user account."
     )
+    is_admin: bool = Field(
+        title="Whether the account is an administrator.",
+    )
 
 
 class UserResponseMetadata(BaseResponseMetadata):
     """Response metadata for users."""
 
@@ -338,10 +371,19 @@ class UserResponse(
         Returns:
             the value of the property.
         """
         return self.get_body().is_service_account
 
+    @property
+    def is_admin(self) -> bool:
+        """The `is_admin` property.
+
+        Returns:
+            Whether the user is an admin.
+        """
+        return self.get_body().is_admin
+
     @property
     def email(self) -> Optional[str]:
         """The `email` property.
 
         Returns:
diff --git a/src/zenml/models/v2/misc/external_user.py b/src/zenml/models/v2/misc/external_user.py
index a306f878d..4329e5e7e 100644
--- a/src/zenml/models/v2/misc/external_user.py
+++ b/src/zenml/models/v2/misc/external_user.py
@@ -23,10 +23,11 @@ class ExternalUserModel(BaseModel):
     """External user model."""
 
     id: UUID
     email: str
     name: Optional[str] = None
+    is_admin: bool = False
 
     class Config:
         """Pydantic configuration."""
 
         # ignore arbitrary fields
diff --git a/src/zenml/zen_server/auth.py b/src/zenml/zen_server/auth.py
index a80fc3db8..e7113305d 100644
--- a/src/zenml/zen_server/auth.py
+++ b/src/zenml/zen_server/auth.py
@@ -586,10 +586,11 @@ def authenticate_external_user(external_access_token: str) -> AuthContext:
                 name=external_user.email,
                 full_name=external_user.name or "",
                 email_opted_in=True,
                 active=True,
                 email=external_user.email,
+                is_admin=external_user.is_admin,
             ),
         )
     except KeyError:
         logger.info(
             f"External user with ID {external_user.id} not found in ZenML "
@@ -601,10 +602,11 @@ def authenticate_external_user(external_access_token: str) -> AuthContext:
                 full_name=external_user.name or "",
                 external_user_id=external_user.id,
                 email_opted_in=True,
                 active=True,
                 email=external_user.email,
+                is_admin=external_user.is_admin,
             )
         )
 
         with AnalyticsContext() as context:
             context.user_id = user.id
diff --git a/src/zenml/zen_server/rbac/endpoint_utils.py b/src/zenml/zen_server/rbac/endpoint_utils.py
index bc01f1c73..6cc78ddcc 100644
--- a/src/zenml/zen_server/rbac/endpoint_utils.py
+++ b/src/zenml/zen_server/rbac/endpoint_utils.py
@@ -57,11 +57,14 @@ def verify_permissions_and_create_entity(
             raise IllegalOperationError(
                 f"Not allowed to create resource '{resource_type}' for a "
                 "different user."
             )
 
-    verify_permission(resource_type=resource_type, action=Action.CREATE)
+    verify_permission(
+        resource_type=resource_type,
+        action=Action.CREATE,
+    )
     return create_method(request_model)
 
 
 def verify_permissions_and_get_entity(
     id: UUIDOrStr,
diff --git a/src/zenml/zen_server/rbac/utils.py b/src/zenml/zen_server/rbac/utils.py
index a5dc8829c..da64e4178 100644
--- a/src/zenml/zen_server/rbac/utils.py
+++ b/src/zenml/zen_server/rbac/utils.py
@@ -231,11 +231,14 @@ def verify_permission_for_model(model: AnyResponse, action: Action) -> None:
         action: The action the user wants to perform.
     """
     batch_verify_permissions_for_models(models=[model], action=action)
 
 
-def batch_verify_permissions(resources: Set[Resource], action: Action) -> None:
+def batch_verify_permissions(
+    resources: Set[Resource],
+    action: Action,
+) -> None:
     """Batch permission verification.
 
     Args:
         resources: The resources the user wants to perform the action on.
         action: The action the user wants to perform.
diff --git a/src/zenml/zen_server/routers/users_endpoints.py b/src/zenml/zen_server/routers/users_endpoints.py
index b5b796240..f75bc42b5 100644
--- a/src/zenml/zen_server/routers/users_endpoints.py
+++ b/src/zenml/zen_server/routers/users_endpoints.py
@@ -57,10 +57,11 @@ from zenml.zen_server.rbac.utils import (
 )
 from zenml.zen_server.utils import (
     handle_exceptions,
     make_dependable,
     server_config,
+    verify_admin_status_if_no_rbac,
     zen_store,
 )
 
 logger = get_logger(__name__)
 
@@ -110,10 +111,13 @@ def list_users(
     """
     allowed_ids = get_allowed_resource_ids(resource_type=ResourceType.USER)
     if allowed_ids is not None:
         # Make sure users can see themselves
         allowed_ids.add(auth_context.user.id)
+    else:
+        if not auth_context.user.is_admin and not server_config().rbac_enabled:
+            allowed_ids = {auth_context.user.id}
 
     user_filter_model.configure_rbac(
         authenticated_user_id=auth_context.user.id, id=allowed_ids
     )
 
@@ -137,18 +141,19 @@ if server_config().auth_scheme != AuthScheme.EXTERNAL:
         },
     )
     @handle_exceptions
     def create_user(
         user: UserRequest,
-        _: AuthContext = Security(authorize),
+        auth_context: AuthContext = Security(authorize),
     ) -> UserResponse:
         """Creates a user.
 
         # noqa: DAR401
 
         Args:
             user: User to create.
+            auth_context: Authentication context.
 
         Returns:
             The created user.
         """
         # Two ways of creating a new user:
@@ -161,10 +166,14 @@ if server_config().auth_scheme != AuthScheme.EXTERNAL:
             user.active = False
             token = user.generate_activation_token()
         else:
             user.active = True
 
+        verify_admin_status_if_no_rbac(
+            auth_context.user.is_admin, "create user"
+        )
+
         new_user = verify_permissions_and_create_entity(
             request_model=user,
             resource_type=ResourceType.USER,
             create_method=zen_store().create_user,
         )
@@ -200,11 +209,17 @@ def get_user(
     """
     user = zen_store().get_user(
         user_name_or_id=user_name_or_id, hydrate=hydrate
     )
     if user.id != auth_context.user.id:
-        verify_permission_for_model(user, action=Action.READ)
+        verify_admin_status_if_no_rbac(
+            auth_context.user.is_admin, "get other user"
+        )
+        verify_permission_for_model(
+            user,
+            action=Action.READ,
+        )
 
     return dehydrate_response_model(user)
 
 
 # When the auth scheme is set to EXTERNAL, users cannot be updated via the
@@ -233,15 +248,36 @@ if server_config().auth_scheme != AuthScheme.EXTERNAL:
             user_update: the user to use for the update.
             auth_context: Authentication context.
 
         Returns:
             The updated user.
+
+        Raises:
+            IllegalOperationError: if the user tries change admin status,
+                while not an admin
         """
         user = zen_store().get_user(user_name_or_id)
         if user.id != auth_context.user.id:
-            verify_permission_for_model(user, action=Action.UPDATE)
+            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
         updated_user = zen_store().update_user(
             user_id=user.id,
             user_update=user_update,
         )
         return dehydrate_response_model(updated_user)
@@ -277,10 +313,11 @@ if server_config().auth_scheme != AuthScheme.EXTERNAL:
             user_name_or_id=user_name_or_id,
             activation_token=user_update.activation_token,
         )
         user_update.active = True
         user_update.activation_token = None
+        user_update.is_admin = user.is_admin
         return zen_store().update_user(
             user_id=user.id, user_update=user_update
         )
 
     @router.put(
@@ -303,14 +340,25 @@ if server_config().auth_scheme != AuthScheme.EXTERNAL:
             user_name_or_id: Name or ID of the user.
             auth_context: Authentication context.
 
         Returns:
             The generated activation token.
+
+        Raises:
+            IllegalOperationError: if the user is trying to deactivate
+                themselves.
         """
         user = zen_store().get_user(user_name_or_id)
-        if user.id != auth_context.user.id:
-            verify_permission_for_model(user, action=Action.UPDATE)
+        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,
+        )
 
         user_update = UserUpdate(
             name=user.name,
             active=False,
         )
@@ -352,11 +400,17 @@ if server_config().auth_scheme != AuthScheme.EXTERNAL:
                 "to the ZenML server. If you wish to delete this account, "
                 "please authenticate with another account or contact your ZenML "
                 "administrator."
             )
         else:
-            verify_permission_for_model(user, action=Action.DELETE)
+            verify_admin_status_if_no_rbac(
+                auth_context.user.is_admin, "delete user"
+            )
+            verify_permission_for_model(
+                user,
+                action=Action.DELETE,
+            )
 
         zen_store().delete_user(user_name_or_id=user_name_or_id)
 
     @router.put(
         "/{user_name_or_id}" + EMAIL_ANALYTICS,
@@ -400,11 +454,10 @@ if server_config().auth_scheme != AuthScheme.EXTERNAL:
                 email_opt_int(
                     opted_in=user_response.email_opted_in,
                     email=user_response.email,
                     source="zenml server",
                 )
-
             updated_user = zen_store().update_user(
                 user_id=user.id, user_update=user_update
             )
             return dehydrate_response_model(updated_user)
         else:
@@ -458,10 +511,15 @@ if server_config().auth_scheme != AuthScheme.EXTERNAL:
             auth_context: The authentication context.
 
         Returns:
             The updated user.
         """
+        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
+
         updated_user = zen_store().update_user(
             user_id=auth_context.user.id, user_update=user
         )
         return dehydrate_response_model(updated_user)
 
diff --git a/src/zenml/zen_server/utils.py b/src/zenml/zen_server/utils.py
index 9325fe7cf..bc6be68e2 100644
--- a/src/zenml/zen_server/utils.py
+++ b/src/zenml/zen_server/utils.py
@@ -25,11 +25,11 @@ from zenml.config.global_config import GlobalConfiguration
 from zenml.config.server_config import ServerConfiguration
 from zenml.constants import (
     ENV_ZENML_SERVER,
 )
 from zenml.enums import ServerProviderType
-from zenml.exceptions import OAuthError
+from zenml.exceptions import IllegalOperationError, OAuthError
 from zenml.logger import get_logger
 from zenml.plugins.plugin_flavor_registry import PluginFlavorRegistry
 from zenml.zen_server.deploy.deployment import ServerDeployment
 from zenml.zen_server.deploy.local.local_zen_server import (
     LocalServerDeploymentConfig,
@@ -394,5 +394,36 @@ def get_ip_location(ip_address: str) -> Tuple[str, str, str]:
             details.country_name,
         )
     except Exception:
         logger.exception(f"Could not get IP location for {ip_address}.")
         return "", "", ""
+
+
+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
diff --git a/src/zenml/zen_stores/migrations/versions/1a9a9d2a836d_admin_users.py b/src/zenml/zen_stores/migrations/versions/1a9a9d2a836d_admin_users.py
new file mode 100644
index 000000000..e7de45ecd
--- /dev/null
+++ b/src/zenml/zen_stores/migrations/versions/1a9a9d2a836d_admin_users.py
@@ -0,0 +1,56 @@
+"""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 ###
diff --git a/src/zenml/zen_stores/schemas/user_schemas.py b/src/zenml/zen_stores/schemas/user_schemas.py
index 744301b7c..610e45d4c 100644
--- a/src/zenml/zen_stores/schemas/user_schemas.py
+++ b/src/zenml/zen_stores/schemas/user_schemas.py
@@ -75,10 +75,11 @@ class UserSchema(NamedSchema, table=True):
     password: Optional[str] = Field(nullable=True)
     activation_token: Optional[str] = Field(nullable=True)
     hub_token: Optional[str] = Field(nullable=True)
     email_opted_in: Optional[bool] = Field(nullable=True)
     external_user_id: Optional[UUID] = Field(nullable=True)
+    is_admin: bool = Field(default=False)
 
     stacks: List["StackSchema"] = Relationship(back_populates="user")
     components: List["StackComponentSchema"] = Relationship(
         back_populates="user",
     )
@@ -165,10 +166,11 @@ class UserSchema(NamedSchema, table=True):
             activation_token=model.create_hashed_activation_token(),
             external_user_id=model.external_user_id,
             email_opted_in=model.email_opted_in,
             email=model.email,
             is_service_account=False,
+            is_admin=model.is_admin,
         )
 
     @classmethod
     def from_service_account_request(
         cls, model: ServiceAccountRequest
@@ -187,10 +189,11 @@ class UserSchema(NamedSchema, table=True):
             description=model.description or "",
             active=model.active,
             is_service_account=True,
             email_opted_in=False,
             full_name="",
+            is_admin=False,
         )
 
     def update_user(self, user_update: UserUpdate) -> "UserSchema":
         """Update a `UserSchema` from a `UserUpdate`.
 
@@ -269,10 +272,11 @@ class UserSchema(NamedSchema, table=True):
                 full_name=self.full_name,
                 email_opted_in=self.email_opted_in,
                 is_service_account=self.is_service_account,
                 created=self.created,
                 updated=self.updated,
+                is_admin=self.is_admin,
             ),
             metadata=metadata,
         )
 
     def to_service_account_model(
diff --git a/src/zenml/zen_stores/sql_zen_store.py b/src/zenml/zen_stores/sql_zen_store.py
index 93ae87a21..5feb10d93 100644
--- a/src/zenml/zen_stores/sql_zen_store.py
+++ b/src/zenml/zen_stores/sql_zen_store.py
@@ -7512,10 +7512,18 @@ class SqlZenStore(BaseZenStore):
         with Session(self.engine) as session:
             existing_user = self._get_account_schema(
                 user_id, session=session, service_account=False
             )
 
+            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."
+                )
+
             if (
                 user_update.name is not None
                 and user_update.name != existing_user.name
             ):
                 if existing_user.name == self._default_user_name:
@@ -7602,10 +7610,11 @@ class SqlZenStore(BaseZenStore):
             return self.create_user(
                 UserRequest(
                     name=default_user_name,
                     active=True,
                     password=password,
+                    is_admin=True,
                 )
             )
 
     # ----------------------------- Workspaces -----------------------------
 
diff --git a/tests/integration/functional/zen_stores/test_zen_store.py b/tests/integration/functional/zen_stores/test_zen_store.py
index 1c0261b54..b62c163e9 100644
--- a/tests/integration/functional/zen_stores/test_zen_store.py
+++ b/tests/integration/functional/zen_stores/test_zen_store.py
@@ -10,14 +10,16 @@
 #  distributed under the License is distributed on an "AS IS" BASIS,
 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 #  or implied. See the License for the specific language governing
 #  permissions and limitations under the License.
 import os
+import random
 import time
 import uuid
 from contextlib import ExitStack as does_not_raise
 from datetime import datetime
+from string import ascii_lowercase
 from threading import Thread
 from typing import Dict, List, Optional, Tuple
 from uuid import UUID, uuid4
 
 import pytest
@@ -125,10 +127,11 @@ from zenml.models.v2.core.pipeline_run import PipelineRunRequest
 from zenml.models.v2.core.run_metadata import RunMetadataRequest
 from zenml.models.v2.core.step_run import StepRunRequest
 from zenml.models.v2.core.user import UserFilter
 from zenml.utils import code_repository_utils, source_utils
 from zenml.utils.enum_utils import StrEnum
+from zenml.zen_stores.rest_zen_store import RestZenStore
 from zenml.zen_stores.sql_zen_store import SqlZenStore
 
 DEFAULT_NAME = "default"
 
 # .--------------.
@@ -375,10 +378,239 @@ def test_deleting_default_workspace_fails():
 #  .------.
 # | USERS |
 # '-------'
 
 
+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
+
+
 def test_active_user():
     """Tests the active user can be queried with .get_user()."""
     zen_store = Client().zen_store
     active_user = zen_store.get_user()
     assert active_user is not None
@@ -395,18 +627,24 @@ def test_creating_user_with_existing_name_fails():
     zen_store = Client().zen_store
 
     with UserContext() as existing_user:
         with pytest.raises(EntityExistsError):
             zen_store.create_user(
-                UserRequest(name=existing_user.name, password="password")
+                UserRequest(
+                    name=existing_user.name,
+                    password="password",
+                    is_admin=False,
+                )
             )
 
     with ServiceAccountContext() as existing_service_account:
         with does_not_raise():
             user = zen_store.create_user(
                 UserRequest(
-                    name=existing_service_account.name, password="password"
+                    name=existing_service_account.name,
+                    password="password",
+                    is_admin=False,
                 )
             )
             # clean up
             zen_store.delete_user(user.id)
 
@@ -432,11 +670,13 @@ def test_creating_users_in_parallel_do_not_duplicate_fails(
 
     threads: List[Thread] = []
     for _ in range(count):
         t = Thread(
             target=silent_create_user,
-            args=(UserRequest(name=user_name, password=password),),
+            args=(
+                UserRequest(name=user_name, password=password, is_admin=False),
+            ),
         )
         threads.append(t)
     for t in threads:
         t.start()
     for t in threads:
@@ -738,21 +978,23 @@ def test_create_user_no_password():
                 pass
 
         with pytest.raises(AuthorizationException):
             response_body = store.put(
                 f"{USERS}/{str(user.id)}{ACTIVATE}",
-                body=UserUpdate(password="password"),
+                body=UserUpdate(password="password", is_admin=user.is_admin),
             )
 
         with pytest.raises(AuthorizationException):
             with LoginContext(user_name=user.name, password="password"):
                 pass
 
         response_body = store.put(
             f"{USERS}/{str(user.id)}{ACTIVATE}",
             body=UserUpdate(
-                password="password", activation_token=user.activation_token
+                password="password",
+                activation_token=user.activation_token,
+                is_admin=user.is_admin,
             ),
         )
         activated_user = UserResponse.parse_obj(response_body)
         assert activated_user.active
         assert activated_user.name == user.name
@@ -791,11 +1033,13 @@ def test_reactivate_user():
                 pass
 
         with pytest.raises(AuthorizationException):
             response_body = store.put(
                 f"{USERS}/{str(user.id)}{ACTIVATE}",
-                body=UserUpdate(password="newpassword"),
+                body=UserUpdate(
+                    password="newpassword", is_admin=user.is_admin
+                ),
             )
 
         with pytest.raises(AuthorizationException):
             with LoginContext(user_name=user.name, password="password"):
                 pass
@@ -807,10 +1051,11 @@ def test_reactivate_user():
         response_body = store.put(
             f"{USERS}/{str(user.id)}{ACTIVATE}",
             body=UserUpdate(
                 password="newpassword",
                 activation_token=deactivated_user.activation_token,
+                is_admin=user.is_admin,
             ),
         )
         activated_user = UserResponse.parse_obj(response_body)
         assert activated_user.active
         assert activated_user.name == user.name
diff --git a/tests/integration/functional/zen_stores/utils.py b/tests/integration/functional/zen_stores/utils.py
index 2877dae84..b5bd4d5b4 100644
--- a/tests/integration/functional/zen_stores/utils.py
+++ b/tests/integration/functional/zen_stores/utils.py
@@ -189,10 +189,11 @@ class UserContext:
         password: Optional[str] = None,
         inactive: bool = False,
         login: bool = False,
         existing_user: bool = False,
         delete: bool = True,
+        is_admin: bool = True,
     ):
         if existing_user:
             self.user_name = user_name
         elif user_name:
             self.user_name = user_name
@@ -205,15 +206,19 @@ class UserContext:
             self.password = None
         else:
             self.password = password or random_str(32)
         self.existing_user = existing_user
         self.delete = delete
+        self.is_admin = is_admin
 
     def __enter__(self):
         if not self.existing_user:
             new_user = UserRequest(
-                name=self.user_name, password=self.password, active=True
+                name=self.user_name,
+                password=self.password,
+                active=True,
+                is_admin=self.is_admin,
             )
             self.created_user = self.store.create_user(new_user)
         else:
             self.created_user = self.store.get_user(self.user_name)
 
@@ -1009,11 +1014,11 @@ workspace_crud_test_config = CrudTestConfig(
     update_model=WorkspaceUpdate(name=sample_name("updated_sample_workspace")),
     filter_model=WorkspaceFilter,
     entity_name="workspace",
 )
 user_crud_test_config = CrudTestConfig(
-    create_model=UserRequest(name=sample_name("sample_user")),
+    create_model=UserRequest(name=sample_name("sample_user"), is_admin=True),
     update_model=UserUpdate(name=sample_name("updated_sample_user")),
     filter_model=UserFilter,
     entity_name="user",
 )
 flavor_crud_test_config = CrudTestConfig(
diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py
index b84ec5d4b..930d8a489 100644
--- a/tests/unit/conftest.py
+++ b/tests/unit/conftest.py
@@ -370,10 +370,11 @@ def sample_user_model() -> UserResponse:
         name="axl",
         body=UserResponseBody(
             created=datetime.now(),
             updated=datetime.now(),
             is_service_account=False,
+            is_admin=True,
         ),
         metadata=UserResponseMetadata(),
     )
 
 
diff --git a/tests/unit/models/test_user_models.py b/tests/unit/models/test_user_models.py
index 8b1375486..a90739d34 100644
--- a/tests/unit/models/test_user_models.py
+++ b/tests/unit/models/test_user_models.py
@@ -24,17 +24,13 @@ UUID_BASE_STRING = "00000000-0000-0000-0000-000000000000"
 
 def test_user_request_model_fails_with_long_password():
     """Test that the user request model fails with long passwords."""
     long_password = "a" * (STR_FIELD_MAX_LENGTH + 1)
     with pytest.raises(ValidationError):
-        UserRequest(
-            password=long_password,
-        )
+        UserRequest(password=long_password, is_admin=False)
 
 
 def test_user_request_model_fails_with_long_activation_token():
     """Test that the user request model fails with long activation tokens."""
     long_token = "a" * (STR_FIELD_MAX_LENGTH + 1)
     with pytest.raises(ValidationError):
-        UserRequest(
-            activation_token=long_token,
-        )
+        UserRequest(activation_token=long_token, is_admin=False)

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)