Refactor create superadmin mode

backend/app/deps/authorization_deps.py

@@ -1,24 +1,29 @@
 from beanie import PydanticObjectId
 from beanie.operators import Or
-from bson import ObjectId
 from fastapi import Depends, HTTPException
 
 from app.keycloak_auth import get_current_username
 from app.models.authorization import RoleType, AuthorizationDB
 from app.models.datasets import DatasetDB, DatasetStatus
-from app.models.files import FileOut, FileDB
-from app.models.groups import GroupOut, GroupDB
+from app.models.files import FileDB
+from app.models.groups import GroupDB
 from app.models.metadata import MetadataDB
 from app.models.pyobjectid import PyObjectId
 from app.routers.authentication import get_admin
+from app.routers.authentication import get_admin_mode
 
 
 async def get_role(
     dataset_id: str,
     current_user=Depends(get_current_username),
+    admin_mode: bool = Depends(get_admin_mode),
+    admin=Depends(get_admin),
 ) -> RoleType:
     """Returns the role a specific user has on a dataset. If the user is a creator (owner), they are not listed in
     the user_ids list."""
+    if admin and admin_mode:
+        return RoleType.OWNER
+
     authorization = await AuthorizationDB.find_one(
         AuthorizationDB.dataset_id == PyObjectId(dataset_id),
         Or(
@@ -32,7 +37,12 @@ async def get_role(
 async def get_role_by_file(
     file_id: str,
     current_user=Depends(get_current_username),
+    admin_mode: bool = Depends(get_admin_mode),
+    admin=Depends(get_admin),
 ) -> RoleType:
+    if admin and admin_mode:
+        return RoleType.OWNER
+
     if (file := await FileDB.get(PydanticObjectId(file_id))) is not None:
         authorization = await AuthorizationDB.find_one(
             AuthorizationDB.dataset_id == file.dataset_id,
@@ -66,7 +76,12 @@ async def get_role_by_file(
 async def get_role_by_metadata(
     metadata_id: str,
     current_user=Depends(get_current_username),
+    admin_mode: bool = Depends(get_admin_mode),
+    admin=Depends(get_admin),
 ) -> RoleType:
+    if admin and admin_mode:
+        return RoleType.OWNER
+
     if (md_out := await MetadataDB.get(PydanticObjectId(metadata_id))) is not None:
         resource_type = md_out.resource.collection
         resource_id = md_out.resource.resource_id
@@ -97,7 +112,12 @@ async def get_role_by_metadata(
 async def get_role_by_group(
     group_id: str,
     current_user=Depends(get_current_username),
+    admin_mode: bool = Depends(get_admin_mode),
+    admin=Depends(get_admin),
 ) -> RoleType:
+    if admin and admin_mode:
+        return RoleType.OWNER
+
     if (group := await GroupDB.get(group_id)) is not None:
         if group.creator == current_user:
             # Creator can do everything
@@ -148,12 +168,13 @@ async def __call__(
         self,
         dataset_id: str,
         current_user: str = Depends(get_current_username),
+        admin_mode: bool = Depends(get_admin_mode),
         admin: bool = Depends(get_admin),
     ):
         # TODO: Make sure we enforce only one role per user per dataset, or find_one could yield wrong answer here.
 
-        # If the current user is admin, user has access irrespective of any role assigned
-        if admin:
+        # If the current user is admin and has turned on admin_mode, user has access irrespective of any role assigned
+        if admin and admin_mode:
             return True
 
         # Else check role assigned to the user
@@ -204,10 +225,11 @@ async def __call__(
         self,
         file_id: str,
         current_user: str = Depends(get_current_username),
+        admin_mode: bool = Depends(get_admin_mode),
         admin: bool = Depends(get_admin),
     ):
-        # If the current user is admin, user has access irrespective of any role assigned
-        if admin:
+        # If the current user is admin and has turned on admin_mode, user has access irrespective of any role assigned
+        if admin and admin_mode:
             return True
 
         # Else check role assigned to the user
@@ -241,10 +263,11 @@ async def __call__(
         self,
         metadata_id: str,
         current_user: str = Depends(get_current_username),
+        admin_mode: bool = Depends(get_admin_mode),
         admin: bool = Depends(get_admin),
     ):
-        # If the current user is admin, user has access irrespective of any role assigned
-        if admin:
+        # If the current user is admin and has turned on admin_mode, user has access irrespective of any role assigned
+        if admin and admin_mode:
             return True
 
         # Else check role assigned to the user
@@ -307,10 +330,11 @@ async def __call__(
         self,
         group_id: str,
         current_user: str = Depends(get_current_username),
+        admin_mode: bool = Depends(get_admin_mode),
         admin: bool = Depends(get_admin),
     ):
-        # If the current user is admin, user has access irrespective of any role assigned
-        if admin:
+        # If the current user is admin and has turned on admin_mode, user has access irrespective of any role assigned
+        if admin and admin_mode:
             return True
 
         # Else check role assigned to the user
@@ -377,9 +401,14 @@ async def __call__(
             return False
 
 
-def access(user_role: RoleType, role_required: RoleType) -> bool:
-    """Enforce implied role hierarchy OWNER > EDITOR > UPLOADER > VIEWER"""
-    if user_role == RoleType.OWNER:
+def access(
+    user_role: RoleType,
+    role_required: RoleType,
+    admin_mode: bool = Depends(get_admin_mode),
+    admin: bool = Depends(get_admin),
+) -> bool:
+    """Enforce implied role hierarchy ADMIN = OWNER > EDITOR > UPLOADER > VIEWER"""
+    if user_role == RoleType.OWNER or (admin and admin_mode):
         return True
     elif user_role == RoleType.EDITOR and role_required in [
         RoleType.EDITOR,

backend/app/models/users.py

@@ -25,6 +25,7 @@ class UserLogin(BaseModel):
 
 class UserDoc(Document, UserBase):
     admin: bool
+    admin_mode: bool = False
 
     class Settings:
         name = "users"

backend/app/routers/authentication.py

@@ -92,7 +92,10 @@ async def authenticate_user(email: str, password: str):
     return user
 
 
-async def get_admin(dataset_id: str = None, current_username=Depends(get_current_user)):
+@router.get("/users/me/is_admin", response_model=bool)
+async def get_admin(
+    dataset_id: str = None, current_username=Depends(get_current_user)
+) -> bool:
     if (
         current_user := await UserDB.find_one(UserDB.email == current_username.email)
     ) is not None:
@@ -103,11 +106,59 @@ async def get_admin(dataset_id: str = None, current_username=Depends(get_current
         and (dataset_db := await DatasetDB.get(PydanticObjectId(dataset_id)))
         is not None
     ):
+        # TODO: question regarding resource creator is considered as admin of the resource?
         return dataset_db.creator.email == current_username.email
     else:
         return False
 
 
+@router.get("/users/me/admin_mode")
+async def get_admin_mode(current_username=Depends(get_current_user)) -> bool:
+    """Get Admin mode from User Object."""
+    if (
+        current_user := await UserDB.find_one(UserDB.email == current_username.email)
+    ) is not None:
+        if current_user.admin_mode is not None:
+            return current_user.admin_mode
+        else:
+            return False
+    else:
+        raise HTTPException(
+            status_code=404,
+            detail="User doesn't exist.",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+
+@router.post("/users/me/admin_mode", response_model=bool)
+async def set_admin_mode(
+    admin_mode_on: bool,
+    admin=Depends(get_admin),
+    current_username=Depends(get_current_user),
+) -> bool:
+    """Set Admin mode from User Object."""
+    if (
+        current_user := await UserDB.find_one(UserDB.email == current_username.email)
+    ) is not None:
+        # only admin can set admin mode
+        if admin:
+            current_user.admin_mode = admin_mode_on
+            await current_user.replace()
+            return current_user.admin_mode
+        else:
+            raise HTTPException(
+                status_code=403,
+                detail="You are not admin yet. Only admin can set admin mode.",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+    else:
+        raise HTTPException(
+            status_code=404,
+            detail="User doesn't exist.",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+
 @router.post("/users/set_admin/{useremail}", response_model=UserOut)
 async def set_admin(
     useremail: str, current_username=Depends(get_current_user), admin=Depends(get_admin)

backend/app/routers/authorization.py

@@ -3,6 +3,7 @@
 from bson import ObjectId
 from fastapi import APIRouter, Depends
 from fastapi.exceptions import HTTPException
+
 from app.dependencies import get_elasticsearchclient
 from app.deps.authorization_deps import (
     Authorization,
@@ -29,7 +30,7 @@
 from app.models.groups import GroupDB
 from app.models.pyobjectid import PyObjectId
 from app.models.users import UserDB
-from app.routers.authentication import get_admin
+from app.routers.authentication import get_admin, get_admin_mode
 from app.search.index import index_dataset
 
 router = APIRouter()
@@ -69,22 +70,24 @@ async def save_authorization(
 async def get_dataset_role(
     dataset_id: str,
     current_user=Depends(get_current_username),
+    admin_mode: bool = Depends(get_admin_mode),
     admin=Depends(get_admin),
 ):
     """Retrieve role of user for a specific dataset."""
     # Get group id and the associated users from authorization
-    if admin:
-        auth_db = await AuthorizationDB.find_one(
-            AuthorizationDB.dataset_id == PyObjectId(dataset_id)
-        )
-    else:
-        auth_db = await AuthorizationDB.find_one(
-            AuthorizationDB.dataset_id == PyObjectId(dataset_id),
+    criteria = []
+    if not admin or not admin_mode:
+        criteria.append(
             Or(
                 AuthorizationDB.creator == current_user,
                 AuthorizationDB.user_ids == current_user,
-            ),
+            )
         )
+
+    auth_db = await AuthorizationDB.find_one(
+        AuthorizationDB.dataset_id == PyObjectId(dataset_id),
+        *criteria,
+    )
     if auth_db is None:
         if (
             current_dataset := await DatasetDB.get(PydanticObjectId(dataset_id))
@@ -107,18 +110,20 @@ async def get_dataset_role(
         return auth_db.dict()
 
 
-@router.get("/datasets/{dataset_id}/role/viewer")
+@router.get("/datasets/{dataset_id}/role/viewer}")
 async def get_dataset_role_viewer(
-    dataset_id: str, allow: bool = Depends(Authorization("viewer"))
+    dataset_id: str,
+    allow: bool = Depends(Authorization("viewer")),
 ):
     """Used for testing only. Returns true if user has viewer permission on dataset, otherwise throws a 403 Forbidden HTTP exception.
     See `routers/authorization.py` for more info."""
     return {"dataset_id": dataset_id, "allow": allow}
 
 
-@router.get("/datasets/{dataset_id}/role/owner")
+@router.get("/datasets/{dataset_id}/role/owner}")
 async def get_dataset_role_owner(
-    dataset_id: str, allow: bool = Depends(Authorization("owner"))
+    dataset_id: str,
+    allow: bool = Depends(Authorization("owner")),
 ):
     """Used for testing only. Returns true if user has owner permission on dataset, otherwise throws a 403 Forbidden HTTP exception.
     See `routers/authorization.py` for more info."""
@@ -130,25 +135,17 @@ async def get_file_role(
     file_id: str,
     current_user=Depends(get_current_username),
     role: RoleType = Depends(get_role_by_file),
-    admin=Depends(get_admin),
 ):
-    # admin is a superuser and has all the privileges
-    if admin:
-        return RoleType.OWNER
     """Retrieve role of user for an individual file. Role cannot change between file versions."""
     return role
 
 
-@router.get("/metadata/{metadata_id}/role", response_model=AuthorizationMetadata)
+@router.get("/metadata/{metadata_id}/role}", response_model=AuthorizationMetadata)
 async def get_metadata_role(
     metadata_id: str,
     current_user=Depends(get_current_username),
     role: RoleType = Depends(get_role_by_metadata),
-    admin=Depends(get_admin),
 ):
-    # admin is a superuser and has all the privileges
-    if admin:
-        return RoleType.OWNER
     """Retrieve role of user for group. Group roles can be OWNER, EDITOR, or VIEWER (for regular Members)."""
     return role
 
@@ -158,11 +155,7 @@ async def get_group_role(
     group_id: str,
     current_user=Depends(get_current_username),
     role: RoleType = Depends(get_role_by_group),
-    admin=Depends(get_admin),
 ):
-    # admin is a superuser and has all the privileges
-    if admin:
-        return RoleType.OWNER
     """Retrieve role of user on a particular group (i.e. whether they can change group memberships)."""
     return role
 
@@ -342,7 +335,7 @@ async def remove_dataset_user_role(
         raise HTTPException(status_code=404, detail=f"Dataset {dataset_id} not found")
 
 
-@router.get("/datasets/{dataset_id}/roles", response_model=DatasetRoles)
+@router.get("/datasets/{dataset_id}/roles}", response_model=DatasetRoles)
 async def get_dataset_roles(
     dataset_id: str,
     allow: bool = Depends(Authorization("editor")),

backend/app/routers/datasets.py

@@ -55,7 +55,8 @@
 from app.models.users import UserOut
 from app.rabbitmq.listeners import submit_dataset_job
 from app.routers.authentication import get_admin
-from app.routers.files import add_file_entry, add_local_file_entry, remove_file_entry
+from app.routers.authentication import get_admin_mode
+from app.routers.files import add_file_entry, remove_file_entry, add_local_file_entry
 from app.search.connect import (
     delete_document_by_id,
 )
@@ -212,8 +213,9 @@ async def get_datasets(
     limit: int = 10,
     mine: bool = False,
     admin=Depends(get_admin),
+    admin_mode: bool = Depends(get_admin_mode),
 ):
-    if admin:
+    if admin and admin_mode:
         datasets = await DatasetDBViewList.find(
             sort=(-DatasetDBViewList.created),
             skip=skip,
@@ -257,10 +259,10 @@ async def get_dataset_files(
     dataset_id: str,
     folder_id: Optional[str] = None,
     authenticated: bool = Depends(CheckStatus("AUTHENTICATED")),
-    allow: bool = Depends(Authorization("viewer")),
     user_id=Depends(get_user),
     skip: int = 0,
     limit: int = 10,
+    allow: bool = Depends(Authorization("viewer")),
 ):
     if authenticated:
         query = [
@@ -380,10 +382,10 @@ async def get_dataset_folders(
     dataset_id: str,
     parent_folder: Optional[str] = None,
     user_id=Depends(get_user),
-    allow: bool = Depends(Authorization("viewer")),
     authenticated: bool = Depends(CheckStatus("authenticated")),
     skip: int = 0,
     limit: int = 10,
+    allow: bool = Depends(Authorization("viewer")),
 ):
     if (await DatasetDB.get(PydanticObjectId(dataset_id))) is not None:
         if authenticated: