1848 add permission rights to async jobs (#7262)

Co-authored-by: Andrei Neagu <[email protected]>
Co-authored-by: Giancarlo Romeo <[email protected]>

Author: 3 people

packages/models-library/src/models_library/api_schemas_rpc_async_jobs/async_jobs.py

@@ -36,16 +36,15 @@ class AsyncJobResult(BaseModel):
 
 class AsyncJobGet(BaseModel):
     job_id: AsyncJobId
-    job_name: str
 
 
 class AsyncJobAbort(BaseModel):
     result: bool
     job_id: AsyncJobId
 
 
-class AsyncJobAccessData(BaseModel):
+class AsyncJobNameData(BaseModel):
     """Data for controlling access to an async job"""
 
-    user_id: UserID | None
+    user_id: UserID
     product_name: str

packages/models-library/src/models_library/api_schemas_storage/data_export_async_jobs.py

@@ -1,33 +1,33 @@
 # pylint: disable=R6301
-from pathlib import Path
 
 from common_library.errors_classes import OsparcErrorMixin
-from models_library.projects_nodes_io import LocationID
-from models_library.users import UserID
+from models_library.projects_nodes_io import LocationID, StorageFileID
 from pydantic import BaseModel, Field
 
 
 class DataExportTaskStartInput(BaseModel):
-    user_id: UserID
-    product_name: str
     location_id: LocationID
-    paths: list[Path] = Field(..., min_length=1)
+    file_and_folder_ids: list[StorageFileID] = Field(..., min_length=1)
 
 
 ### Exceptions
 
 
-class StorageRpcError(OsparcErrorMixin, RuntimeError):
+class StorageRpcBaseError(OsparcErrorMixin, RuntimeError):
     pass
 
 
-class InvalidFileIdentifierError(StorageRpcError):
+class InvalidLocationIdError(StorageRpcBaseError):
+    msg_template: str = "Invalid location_id {location_id}"
+
+
+class InvalidFileIdentifierError(StorageRpcBaseError):
     msg_template: str = "Could not find the file {file_id}"
 
 
-class AccessRightError(StorageRpcError):
-    msg_template: str = "User {user_id} does not have access to file {file_id}"
+class AccessRightError(StorageRpcBaseError):
+    msg_template: str = "User {user_id} does not have access to file {file_id} with location {location_id}"
 
 
-class DataExportError(StorageRpcError):
+class DataExportError(StorageRpcBaseError):
     msg_template: str = "Could not complete data export job with id {job_id}"

packages/models-library/src/models_library/api_schemas_webserver/storage.py

@@ -12,9 +12,8 @@
 )
 from ..api_schemas_storage.data_export_async_jobs import DataExportTaskStartInput
 from ..progress_bar import ProgressReport
-from ..projects_nodes_io import LocationID
+from ..projects_nodes_io import LocationID, StorageFileID
 from ..rest_pagination import CursorQueryParameters
-from ..users import UserID
 from ._base import InputSchema, OutputSchema
 
 
@@ -27,15 +26,11 @@ class ListPathsQueryParams(InputSchema, CursorQueryParameters):
 
 
 class DataExportPost(InputSchema):
-    paths: list[Path]
+    paths: list[StorageFileID]
 
-    def to_rpc_schema(
-        self, user_id: UserID, product_name: str, location_id: LocationID
-    ) -> DataExportTaskStartInput:
+    def to_rpc_schema(self, location_id: LocationID) -> DataExportTaskStartInput:
         return DataExportTaskStartInput(
-            paths=self.paths,
-            user_id=user_id,
-            product_name=product_name,
+            file_and_folder_ids=self.paths,
             location_id=location_id,
         )
 

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/async_jobs/async_jobs.py

@@ -2,9 +2,9 @@
 
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
     AsyncJobAbort,
-    AsyncJobAccessData,
     AsyncJobGet,
     AsyncJobId,
+    AsyncJobNameData,
     AsyncJobResult,
     AsyncJobStatus,
 )
@@ -23,13 +23,13 @@ async def abort(
     *,
     rpc_namespace: RPCNamespace,
     job_id: AsyncJobId,
-    access_data: AsyncJobAccessData | None
+    job_id_data: AsyncJobNameData
 ) -> AsyncJobAbort:
     result = await rabbitmq_rpc_client.request(
         rpc_namespace,
         _RPC_METHOD_NAME_ADAPTER.validate_python("abort"),
         job_id=job_id,
-        access_data=access_data,
+        job_id_data=job_id_data,
         timeout_s=_DEFAULT_TIMEOUT_S,
     )
     assert isinstance(result, AsyncJobAbort)
@@ -41,13 +41,13 @@ async def get_status(
     *,
     rpc_namespace: RPCNamespace,
     job_id: AsyncJobId,
-    access_data: AsyncJobAccessData | None
+    job_id_data: AsyncJobNameData
 ) -> AsyncJobStatus:
     result = await rabbitmq_rpc_client.request(
         rpc_namespace,
         _RPC_METHOD_NAME_ADAPTER.validate_python("get_status"),
         job_id=job_id,
-        access_data=access_data,
+        job_id_data=job_id_data,
         timeout_s=_DEFAULT_TIMEOUT_S,
     )
     assert isinstance(result, AsyncJobStatus)
@@ -59,26 +59,31 @@ async def get_result(
     *,
     rpc_namespace: RPCNamespace,
     job_id: AsyncJobId,
-    access_data: AsyncJobAccessData | None
+    job_id_data: AsyncJobNameData
 ) -> AsyncJobResult:
     result = await rabbitmq_rpc_client.request(
         rpc_namespace,
         _RPC_METHOD_NAME_ADAPTER.validate_python("get_result"),
         job_id=job_id,
-        access_data=access_data,
+        job_id_data=job_id_data,
         timeout_s=_DEFAULT_TIMEOUT_S,
     )
     assert isinstance(result, AsyncJobResult)
     return result
 
 
 async def list_jobs(
-    rabbitmq_rpc_client: RabbitMQRPCClient, *, rpc_namespace: RPCNamespace, filter_: str
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    rpc_namespace: RPCNamespace,
+    filter_: str,
+    job_id_data: AsyncJobNameData
 ) -> list[AsyncJobGet]:
     result: list[AsyncJobGet] = await rabbitmq_rpc_client.request(
         rpc_namespace,
         _RPC_METHOD_NAME_ADAPTER.validate_python("list_jobs"),
         filter_=filter_,
+        job_id_data=job_id_data,
         timeout_s=_DEFAULT_TIMEOUT_S,
     )
     return result
@@ -88,12 +93,14 @@ async def submit_job(
     rabbitmq_rpc_client: RabbitMQRPCClient,
     *,
     rpc_namespace: RPCNamespace,
-    job_name: str,
+    method_name: str,
+    job_id_data: AsyncJobNameData,
     **kwargs
 ) -> AsyncJobGet:
     result = await rabbitmq_rpc_client.request(
         rpc_namespace,
-        _RPC_METHOD_NAME_ADAPTER.validate_python(job_name),
+        _RPC_METHOD_NAME_ADAPTER.validate_python(method_name),
+        job_id_data=job_id_data,
         **kwargs,
         timeout_s=_DEFAULT_TIMEOUT_S,
     )

services/storage/src/simcore_service_storage/api/rpc/_async_jobs.py

@@ -5,9 +5,9 @@
 from fastapi import FastAPI
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
     AsyncJobAbort,
-    AsyncJobAccessData,
     AsyncJobGet,
     AsyncJobId,
+    AsyncJobNameData,
     AsyncJobResult,
     AsyncJobStatus,
 )
@@ -23,17 +23,19 @@
 
 @router.expose()
 async def abort(
-    app: FastAPI, job_id: AsyncJobId, access_data: AsyncJobAccessData | None
+    app: FastAPI, job_id: AsyncJobId, job_id_data: AsyncJobNameData
 ) -> AsyncJobAbort:
     assert app  # nosec
+    assert job_id_data  # nosec
     return AsyncJobAbort(result=True, job_id=job_id)
 
 
 @router.expose(reraise_if_error_type=(StatusError,))
 async def get_status(
-    app: FastAPI, job_id: AsyncJobId, access_data: AsyncJobAccessData | None
+    app: FastAPI, job_id: AsyncJobId, job_id_data: AsyncJobNameData
 ) -> AsyncJobStatus:
     assert app  # nosec
+    assert job_id_data  # nosec
     progress_report = ProgressReport(actual_value=0.5, total=1.0, attempt=1)
     return AsyncJobStatus(
         job_id=job_id,
@@ -46,14 +48,17 @@ async def get_status(
 
 @router.expose(reraise_if_error_type=(ResultError,))
 async def get_result(
-    app: FastAPI, job_id: AsyncJobId, access_data: AsyncJobAccessData | None
+    app: FastAPI, job_id: AsyncJobId, job_id_data: AsyncJobNameData
 ) -> AsyncJobResult:
     assert app  # nosec
     assert job_id  # nosec
+    assert job_id_data  # nosec
     return AsyncJobResult(result="Here's your result.", error=None)
 
 
 @router.expose()
-async def list_jobs(app: FastAPI, filter_: str) -> list[AsyncJobGet]:
+async def list_jobs(
+    app: FastAPI, filter_: str, job_id_data: AsyncJobNameData
+) -> list[AsyncJobGet]:
     assert app  # nosec
-    return [AsyncJobGet(job_id=AsyncJobId(f"{uuid4()}"), job_name="myjob")]
+    return [AsyncJobGet(job_id=AsyncJobId(f"{uuid4()}"))]

services/storage/src/simcore_service_storage/api/rpc/_data_export.py

@@ -1,7 +1,11 @@
 from uuid import uuid4
 
 from fastapi import FastAPI
-from models_library.api_schemas_rpc_async_jobs.async_jobs import AsyncJobGet, AsyncJobId
+from models_library.api_schemas_rpc_async_jobs.async_jobs import (
+    AsyncJobGet,
+    AsyncJobId,
+    AsyncJobNameData,
+)
 from models_library.api_schemas_storage.data_export_async_jobs import (
     AccessRightError,
     DataExportError,
@@ -10,6 +14,12 @@
 )
 from servicelib.rabbitmq import RPCRouter
 
+from ...datcore_dsm import DatCoreDataManager
+from ...dsm import get_dsm_provider
+from ...exceptions.errors import FileAccessRightError
+from ...modules.datcore_adapter.datcore_adapter_exceptions import DatcoreAdapterError
+from ...simcore_s3_dsm import SimcoreS3DataManager
+
 router = RPCRouter()
 
 
@@ -21,10 +31,28 @@
     )
 )
 async def start_data_export(
-    app: FastAPI, paths: DataExportTaskStartInput
+    app: FastAPI,
+    data_export_start: DataExportTaskStartInput,
+    job_id_data: AsyncJobNameData,
 ) -> AsyncJobGet:
     assert app  # nosec
+
+    dsm = get_dsm_provider(app).get(data_export_start.location_id)
+
+    try:
+        for _id in data_export_start.file_and_folder_ids:
+            if isinstance(dsm, DatCoreDataManager):
+                _ = await dsm.get_file(user_id=job_id_data.user_id, file_id=_id)
+            elif isinstance(dsm, SimcoreS3DataManager):
+                await dsm.can_read_file(user_id=job_id_data.user_id, file_id=_id)
+
+    except (FileAccessRightError, DatcoreAdapterError) as err:
+        raise AccessRightError(
+            user_id=job_id_data.user_id,
+            file_id=_id,
+            location_id=data_export_start.location_id,
+        ) from err
+
     return AsyncJobGet(
         job_id=AsyncJobId(f"{uuid4()}"),
-        job_name=", ".join(str(p) for p in paths.paths),
     )

services/storage/src/simcore_service_storage/simcore_s3_dsm.py

@@ -355,6 +355,12 @@ async def get_file(self, user_id: UserID, file_id: StorageFileID) -> FileMetaDat
         fmd = await self._update_database_from_storage(fmd)
         return convert_db_to_model(fmd)
 
+    async def can_read_file(self, user_id: UserID, file_id: StorageFileID):
+        async with self.engine.connect() as conn:
+            can = await get_file_access_rights(conn, int(user_id), file_id)
+            if not can.read:
+                raise FileAccessRightError(access_right="read", file_id=file_id)
+
     async def create_file_upload_links(
         self,
         user_id: UserID,

services/storage/tests/conftest.py

@@ -61,12 +61,13 @@
 from simcore_service_storage.core.application import create_app
 from simcore_service_storage.core.settings import ApplicationSettings
 from simcore_service_storage.dsm import get_dsm_provider
-from simcore_service_storage.models import S3BucketName
+from simcore_service_storage.models import FileMetaData, FileMetaDataAtDB, S3BucketName
 from simcore_service_storage.modules.long_running_tasks import (
     get_completed_upload_tasks,
 )
 from simcore_service_storage.modules.s3 import get_s3_client
 from simcore_service_storage.simcore_s3_dsm import SimcoreS3DataManager
+from sqlalchemy import literal_column
 from sqlalchemy.ext.asyncio import AsyncEngine
 from tenacity.asyncio import AsyncRetrying
 from tenacity.retry import retry_if_exception_type
@@ -849,3 +850,40 @@ async def with_random_project_with_files(
     faker: Faker,
 ) -> tuple[dict[str, Any], dict[NodeID, dict[SimcoreS3FileID, FileIDDict]],]:
     return await random_project_with_files(project_params)
+
+
+@pytest.fixture()
+async def output_file(
+    user_id: UserID, project_id: str, sqlalchemy_async_engine: AsyncEngine, faker: Faker
+) -> AsyncIterator[FileMetaData]:
+    node_id = "fd6f9737-1988-341b-b4ac-0614b646fa82"
+
+    # pylint: disable=no-value-for-parameter
+
+    file = FileMetaData.from_simcore_node(
+        user_id=user_id,
+        file_id=f"{project_id}/{node_id}/filename.txt",
+        bucket=TypeAdapter(S3BucketName).validate_python("master-simcore"),
+        location_id=SimcoreS3DataManager.get_location_id(),
+        location_name=SimcoreS3DataManager.get_location_name(),
+        sha256_checksum=faker.sha256(),
+    )
+    file.entity_tag = "df9d868b94e53d18009066ca5cd90e9f"
+    file.file_size = ByteSize(12)
+    file.user_id = user_id
+    async with sqlalchemy_async_engine.begin() as conn:
+        stmt = (
+            file_meta_data.insert()
+            .values(jsonable_encoder(FileMetaDataAtDB.model_validate(file)))
+            .returning(literal_column("*"))
+        )
+        result = await conn.execute(stmt)
+        row = result.one()
+        assert row
+
+    yield file
+
+    async with sqlalchemy_async_engine.begin() as conn:
+        result = await conn.execute(
+            file_meta_data.delete().where(file_meta_data.c.file_id == row.file_id)
+        )