add MetadataAuthorization dependency

backend/app/deps/authorization_deps.py

@@ -189,6 +189,93 @@ async def __call__(
             raise HTTPException(status_code=404, detail=f"File {file_id} not found")
 
 
+class MetadataAuthorization:
+    """We use class dependency so that we can provide the `permission` parameter to the dependency.
+    For more info see https://fastapi.tiangolo.com/advanced/advanced-dependencies/."""
+
+    def __init__(self, role: str):
+        self.role = role
+
+    async def __call__(
+        self,
+        metadata_id: str,
+        db: MongoClient = Depends(get_db),
+        current_user: str = Depends(get_current_username),
+    ):
+        if (
+            metadata := await db["metadata"].find_one({"_id": ObjectId(metadata_id)})
+        ) is not None:
+            md_out = MetadataOut.from_mongo(metadata)
+            resource_type = md_out.resource.collection
+            resource_id = md_out.resource.resource_id
+            if resource_type == "files":
+                if (
+                    file := await db["files"].find_one({"_id": ObjectId(resource_id)})
+                ) is not None:
+                    file_out = FileOut.from_mongo(file)
+                    if (
+                        authorization_q := await db["authorization"].find_one(
+                            {
+                                "$and": [
+                                    {"dataset_id": ObjectId(file_out.dataset_id)},
+                                    {
+                                        "$or": [
+                                            {"creator": current_user},
+                                            {"user_ids": current_user},
+                                        ]
+                                    },
+                                ]
+                            }
+                        )
+                    ) is not None:
+                        authorization = AuthorizationDB.from_mongo(authorization_q)
+                        if access(authorization.role, self.role):
+                            return True
+
+                    raise HTTPException(
+                        status_code=403,
+                        detail=f"User `{current_user} does not have `{self.role}` permission on metadata {metadata_id}",
+                    )
+                else:
+                    raise HTTPException(
+                        status_code=404, detail=f"Metadata {metadata_id} not found"
+                    )
+            elif resource_type == "datasets":
+                if (
+                    dataset := await db["datasets"].find_one(
+                        {"_id": ObjectId(resource_id)}
+                    )
+                ) is not None:
+                    dataset_out = DatasetOut.from_mongo(dataset)
+                    if (
+                        authorization_q := await db["authorization"].find_one(
+                            {
+                                "$and": [
+                                    {"dataset_id": ObjectId(dataset_out.id)},
+                                    {
+                                        "$or": [
+                                            {"creator": current_user},
+                                            {"user_ids": current_user},
+                                        ]
+                                    },
+                                ]
+                            }
+                        )
+                    ) is not None:
+                        authorization = AuthorizationDB.from_mongo(authorization_q)
+                        if access(authorization.role, self.role):
+                            return True
+
+                    raise HTTPException(
+                        status_code=403,
+                        detail=f"User `{current_user} does not have `{self.role}` permission on metadata {metadata_id}",
+                    )
+                else:
+                    raise HTTPException(
+                        status_code=404, detail=f"Metadata {metadata_id} not found"
+                    )
+
+
 def access(user_role: RoleType, role_required: RoleType) -> bool:
     """Enforce implied role hierarchy OWNER > EDITOR > UPLOADER > VIEWER"""
     if user_role == RoleType.OWNER:

backend/app/routers/datasets.py

@@ -45,6 +45,7 @@
 from app.models.users import UserOut
 from app.rabbitmq.listeners import submit_dataset_job
 from app.routers.files import add_file_entry, remove_file_entry
+
 from app.search.connect import (
     connect_elasticsearch,
     insert_record,
@@ -296,6 +297,7 @@ async def get_dataset_files(
     folder_id: Optional[str] = None,
     user_id=Depends(get_user),
     db: MongoClient = Depends(dependencies.get_db),
+    allow: bool = Depends(Authorization("viewer")),
     skip: int = 0,
     limit: int = 10,
 ):

backend/app/routers/metadata.py

@@ -10,6 +10,7 @@
 from pymongo import MongoClient
 
 from app import dependencies
+from app.deps.authorization_deps import Authorization, MetadataAuthorization
 from app.keycloak_auth import get_user, get_current_user
 from app.models.pyobjectid import PyObjectId
 from app.models.metadata import (
@@ -74,6 +75,7 @@ async def update_metadata(
     metadata_id: str,
     user=Depends(get_current_user),
     db: MongoClient = Depends(dependencies.get_db),
+    allow: bool = Depends(MetadataAuthorization("editor")),
 ):
     """Update metadata. Any fields provided in the contents JSON will be added or updated in the metadata. If context or
     agent should be changed, use PUT.
@@ -97,6 +99,7 @@ async def delete_metadata(
     metadata_id: str,
     user=Depends(get_current_user),
     db: MongoClient = Depends(dependencies.get_db),
+    allow: bool = Depends(MetadataAuthorization("editor")),
 ):
     """Delete metadata by specific ID."""
     if (

backend/app/routers/metadata_datasets.py

@@ -11,6 +11,7 @@
 
 from app import keycloak_auth
 from app import dependencies
+from app.deps.authorization_deps import Authorization
 from app.keycloak_auth import get_user, get_current_user, UserOut
 from app.config import settings
 from app.models.datasets import DatasetOut
@@ -83,6 +84,7 @@ async def add_dataset_metadata(
     user=Depends(get_current_user),
     db: MongoClient = Depends(dependencies.get_db),
     es: Elasticsearch = Depends(dependencies.get_elasticsearchclient),
+    allow: bool = Depends(Authorization("editor")),
 ):
     """Attach new metadata to a dataset. The body must include a contents field with the JSON metadata, and either a
     context JSON-LD object, context_url, or definition (name of a metadata definition) to be valid.
@@ -139,6 +141,7 @@ async def replace_dataset_metadata(
     user=Depends(get_current_user),
     db: MongoClient = Depends(dependencies.get_db),
     es: Elasticsearch = Depends(dependencies.get_elasticsearchclient),
+    allow: bool = Depends(Authorization("editor")),
 ):
     """Update metadata. Any fields provided in the contents JSON will be added or updated in the metadata. If context or
     agent should be changed, use PUT.
@@ -194,6 +197,7 @@ async def update_dataset_metadata(
     user=Depends(get_current_user),
     db: MongoClient = Depends(dependencies.get_db),
     es: Elasticsearch = Depends(dependencies.get_elasticsearchclient),
+    allow: bool = Depends(Authorization("editor")),
 ):
     """Update metadata. Any fields provided in the contents JSON will be added or updated in the metadata. If context or
     agent should be changed, use PUT.
@@ -268,6 +272,7 @@ async def get_dataset_metadata(
     listener_version: Optional[float] = Form(None),
     user=Depends(get_current_user),
     db: MongoClient = Depends(dependencies.get_db),
+    allow: bool = Depends(Authorization("viewer")),
 ):
     if (
         dataset := await db["datasets"].find_one({"_id": ObjectId(dataset_id)})
@@ -303,6 +308,7 @@ async def delete_dataset_metadata(
     user=Depends(get_current_user),
     db: MongoClient = Depends(dependencies.get_db),
     es: Elasticsearch = Depends(dependencies.get_elasticsearchclient),
+    allow: bool = Depends(Authorization("editor")),
 ):
     if (
         dataset := await db["datasets"].find_one({"_id": ObjectId(dataset_id)})