feat: add support for OpenBIS datasets

DEVELOPING.md

@@ -119,6 +119,8 @@ function if you prefer to keep your favorite shell.
 You can run style checks using `make style_checks`.
 To run the test suite, use `make tests` (you likely need to run in the devcontainer for this to work, as it needs
 some surrounding services to run).
+* Run a specific test e.g.: `poetry run pytest -v test/bases/renku_data_services/data_api/test_data_connectors.py::test_create_openbis_data_connector`
+* Also run tests marked with `@pytest.mark.myskip`: `PYTEST_FORCE_RUN_MYSKIPS=1 make tests`
 
 We use [Syrupy](https://github.com/syrupy-project/syrupy) for snapshotting data in tests.
 

components/renku_data_services/data_connectors/blueprints.py

@@ -1,14 +1,15 @@
 """Data connectors blueprint."""
 
 from dataclasses import dataclass
+from datetime import datetime
 from typing import Any
 
 from sanic import Request
 from sanic.response import HTTPResponse, JSONResponse
 from sanic_ext import validate
 from ulid import ULID
 
-from renku_data_services import base_models
+from renku_data_services import base_models, errors
 from renku_data_services.base_api.auth import (
     authenticate,
     only_authenticated,
@@ -38,8 +39,8 @@
     DataConnectorRepository,
     DataConnectorSecretRepository,
 )
-from renku_data_services.errors import errors
 from renku_data_services.storage.rclone import RCloneValidator
+from renku_data_services.utils.core import get_openbis_pat
 
 
 @dataclass(kw_only=True)
@@ -418,8 +419,15 @@ async def _get_secrets(
             secrets = await self.data_connector_secret_repo.get_data_connector_secrets(
                 user=user, data_connector_id=data_connector_id
             )
+            data_connector = await self.data_connector_repo.get_data_connector(
+                user=user, data_connector_id=data_connector_id
+            )
             return validated_json(
-                apispec.DataConnectorSecretsList, [self._dump_data_connector_secret(secret) for secret in secrets]
+                apispec.DataConnectorSecretsList,
+                [
+                    self._dump_data_connector_secret(secret)
+                    for secret in self._adjust_secrets(secrets, data_connector.storage)
+                ],
             )
 
         return "/data_connectors/<data_connector_id:ulid>/secrets", ["GET"], _get_secrets
@@ -435,13 +443,59 @@ async def _patch_secrets(
             user: base_models.APIUser,
             data_connector_id: ULID,
             body: apispec.DataConnectorSecretPatchList,
+            validator: RCloneValidator,
         ) -> JSONResponse:
             unsaved_secrets = validate_data_connector_secrets_patch(put=body)
+            data_connector = await self.data_connector_repo.get_data_connector(
+                user=user, data_connector_id=data_connector_id
+            )
+            storage = data_connector.storage
+            provider = validator.providers[storage.storage_type]
+            sensitive_lookup = [o.name for o in provider.options if o.sensitive]
+            for secret in unsaved_secrets:
+                if secret.name in sensitive_lookup:
+                    continue
+                raise errors.ValidationError(
+                    message=f"The '{secret.name}' property is not marked sensitive and can not be saved in the secret "
+                    f"storage."
+                )
+            expiration_timestamp = None
+
+            if storage.storage_type == "openbis":
+
+                async def openbis_transform_session_token_to_pat() -> (
+                    tuple[list[models.DataConnectorSecretUpdate], datetime]
+                ):
+                    if len(unsaved_secrets) == 1 and unsaved_secrets[0].name == "session_token":
+                        if unsaved_secrets[0].value is not None:
+                            try:
+                                openbis_pat = await get_openbis_pat(
+                                    storage.configuration["host"], unsaved_secrets[0].value
+                                )
+                                return (
+                                    [models.DataConnectorSecretUpdate(name="pass", value=openbis_pat[0])],
+                                    openbis_pat[1],
+                                )
+                            except Exception as e:
+                                raise errors.ProgrammingError(message=str(e)) from e
+                        raise errors.ValidationError(message="The openBIS session token must be a string value.")
+
+                    raise errors.ValidationError(message="The openBIS storage has only one secret: session_token")
+
+                (
+                    unsaved_secrets,
+                    expiration_timestamp,
+                ) = await openbis_transform_session_token_to_pat()
+
             secrets = await self.data_connector_secret_repo.patch_data_connector_secrets(
-                user=user, data_connector_id=data_connector_id, secrets=unsaved_secrets
+                user=user,
+                data_connector_id=data_connector_id,
+                secrets=unsaved_secrets,
+                expiration_timestamp=expiration_timestamp,
             )
             return validated_json(
-                apispec.DataConnectorSecretsList, [self._dump_data_connector_secret(secret) for secret in secrets]
+                apispec.DataConnectorSecretsList,
+                [self._dump_data_connector_secret(secret) for secret in self._adjust_secrets(secrets, storage)],
             )
 
         return "/data_connectors/<data_connector_id:ulid>/secrets", ["PATCH"], _patch_secrets
@@ -508,6 +562,22 @@ def _dump_data_connector_to_project_link(link: models.DataConnectorToProjectLink
             created_by=link.created_by,
         )
 
+    @staticmethod
+    def _adjust_secrets(
+        secrets: list[models.DataConnectorSecret], storage: models.CloudStorageCore
+    ) -> list[models.DataConnectorSecret]:
+        if storage.storage_type == "openbis":
+            for i, secret in enumerate(secrets):
+                if secret.name == "pass":
+                    secrets[i] = models.DataConnectorSecret(
+                        name="session_token",
+                        user_id=secret.user_id,
+                        data_connector_id=secret.data_connector_id,
+                        secret_id=secret.secret_id,
+                    )
+                    break
+        return secrets
+
     @staticmethod
     def _dump_data_connector_secret(secret: models.DataConnectorSecret) -> dict[str, Any]:
         """Dumps a data connector secret for API responses."""

components/renku_data_services/data_connectors/db.py

@@ -4,6 +4,7 @@
 import string
 from collections.abc import AsyncIterator, Callable, Sequence
 from contextlib import suppress
+from datetime import datetime
 from typing import TypeVar
 
 from cryptography.hazmat.primitives.asymmetric import rsa
@@ -886,7 +887,11 @@ async def get_data_connector_secrets(
             return [secret.dump() for secret in secrets]
 
     async def patch_data_connector_secrets(
-        self, user: base_models.APIUser, data_connector_id: ULID, secrets: list[models.DataConnectorSecretUpdate]
+        self,
+        user: base_models.APIUser,
+        data_connector_id: ULID,
+        secrets: list[models.DataConnectorSecretUpdate],
+        expiration_timestamp: datetime | None,
     ) -> list[models.DataConnectorSecret]:
         """Create, update or remove data connector secrets."""
         if user.id is None:
@@ -935,7 +940,9 @@ async def patch_data_connector_secrets(
 
                 if data_connector_secret_orm := existing_secrets_as_dict.get(name):
                     data_connector_secret_orm.secret.update(
-                        encrypted_value=encrypted_value, encrypted_key=encrypted_key
+                        encrypted_value=encrypted_value,
+                        encrypted_key=encrypted_key,
+                        expiration_timestamp=expiration_timestamp,
                     )
                 else:
                     secret_name = f"{data_connector.name[:45]} - {name[:45]}"
@@ -949,6 +956,7 @@ async def patch_data_connector_secrets(
                         encrypted_value=encrypted_value,
                         encrypted_key=encrypted_key,
                         kind=SecretKind.storage,
+                        expiration_timestamp=expiration_timestamp,
                     )
                     data_connector_secret_orm = schemas.DataConnectorSecretORM(
                         name=name,

components/renku_data_services/migrations/versions/aa8f1a39a3bc_add_secret_expiration_timestamp.py

@@ -0,0 +1,30 @@
+"""add secret expiration timestamp
+
+Revision ID: aa8f1a39a3bc
+Revises: d437be68a4fb
+Create Date: 2025-10-21 13:12:56.882429
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "aa8f1a39a3bc"
+down_revision = "d437be68a4fb"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column(
+        "secrets", sa.Column("expiration_timestamp", sa.DateTime(timezone=True), nullable=True), schema="secrets"
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column("secrets", "expiration_timestamp", schema="secrets")
+    # ### end Alembic commands ###

components/renku_data_services/notebooks/api/schemas/cloud_storage.py

@@ -228,22 +228,32 @@ def config_string(self, name: str) -> str:
         if not self.configuration:
             raise ValidationError("Missing configuration for cloud storage")
 
-        # Transform configuration for polybox, switchDrive or sftp
+        # TODO Use RCloneValidator.get_real_configuration(...) instead.
+        # Transform configuration for polybox, switchDrive, openbis or sftp
         storage_type = self.configuration.get("type", "")
         access = self.configuration.get("provider", "")
 
-        if storage_type == "sftp":
-            # Do not allow retries for sftp
-            # Reference: https://rclone.org/docs/#globalconfig
-            self.configuration["override.low_level_retries"] = 1
-
         if storage_type == "polybox" or storage_type == "switchDrive":
             self.configuration["type"] = "webdav"
             self.configuration["provider"] = ""
             # NOTE: Without the vendor field mounting storage and editing files results in the modification
             # time for touched files to be temporarily set to `1999-09-04` which causes the text
             # editor to complain that the file has changed and whether it should overwrite new changes.
             self.configuration["vendor"] = "owncloud"
+        elif storage_type == "s3" and access == "Switch":
+            # Switch is a fake provider we add for users, we need to replace it since rclone itself
+            # doesn't know it
+            self.configuration["provider"] = "Other"
+        elif storage_type == "openbis":
+            self.configuration["type"] = "sftp"
+            self.configuration["port"] = "2222"
+            self.configuration["user"] = "?"
+            self.configuration["pass"] = self.configuration.pop("session_token", self.configuration["pass"])
+
+        if storage_type == "sftp" or storage_type == "openbis":
+            # Do not allow retries for sftp
+            # Reference: https://rclone.org/docs/#globalconfig
+            self.configuration["override.low_level_retries"] = 1
 
         if access == "shared" and storage_type == "polybox":
             self.configuration["url"] = "https://polybox.ethz.ch/public.php/webdav/"
@@ -260,10 +270,6 @@ def config_string(self, name: str) -> str:
             user_identifier = public_link.split("/")[-1]
             self.configuration["user"] = user_identifier
 
-        if self.configuration["type"] == "s3" and self.configuration.get("provider", None) == "Switch":
-            # Switch is a fake provider we add for users, we need to replace it since rclone itself
-            # doesn't know it
-            self.configuration["provider"] = "Other"
         parser = ConfigParser()
         parser.add_section(name)
 

components/renku_data_services/secrets/db.py

@@ -6,12 +6,12 @@
 import random
 import string
 from collections.abc import AsyncGenerator, Callable, Sequence
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from typing import cast
 
 from cryptography.hazmat.primitives.asymmetric import rsa
 from prometheus_client import Counter, Enum
-from sqlalchemy import delete, select
+from sqlalchemy import Select, delete, or_, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 from ulid import ULID
@@ -147,11 +147,23 @@ def __init__(
         self.user_repo = user_repo
         self.secret_service_public_key = secret_service_public_key
 
+    def _get_stmt(self, requested_by: APIUser) -> Select[tuple[SecretORM]]:
+        return (
+            select(SecretORM)
+            .where(SecretORM.user_id == requested_by.id)
+            .where(
+                or_(
+                    SecretORM.expiration_timestamp.is_(None),
+                    SecretORM.expiration_timestamp > datetime.now(UTC) + timedelta(seconds=120),
+                )
+            )
+        )
+
     @only_authenticated
     async def get_user_secrets(self, requested_by: APIUser, kind: SecretKind) -> list[Secret]:
         """Get all user's secrets from the database."""
         async with self.session_maker() as session:
-            stmt = select(SecretORM).where(SecretORM.user_id == requested_by.id).where(SecretORM.kind == kind)
+            stmt = self._get_stmt(requested_by).where(SecretORM.kind == kind)
             res = await session.execute(stmt)
             orm = res.scalars().all()
             return [o.dump() for o in orm]
@@ -160,7 +172,7 @@ async def get_user_secrets(self, requested_by: APIUser, kind: SecretKind) -> lis
     async def get_secret_by_id(self, requested_by: APIUser, secret_id: ULID) -> Secret:
         """Get a specific user secret from the database."""
         async with self.session_maker() as session:
-            stmt = select(SecretORM).where(SecretORM.user_id == requested_by.id).where(SecretORM.id == secret_id)
+            stmt = self._get_stmt(requested_by).where(SecretORM.id == secret_id)
             res = await session.execute(stmt)
             orm = res.scalar_one_or_none()
             if not orm:
@@ -187,11 +199,12 @@ async def insert_secret(self, requested_by: APIUser, secret: UnsavedSecret) -> S
         async with self.session_maker() as session, session.begin():
             secret_orm = SecretORM(
                 name=secret.name,
-                default_filename=default_filename,
                 user_id=requested_by.id,
                 encrypted_value=encrypted_value,
                 encrypted_key=encrypted_key,
                 kind=secret.kind,
+                expiration_timestamp=secret.expiration_timestamp,
+                default_filename=default_filename,
             )
             session.add(secret_orm)
 
@@ -212,9 +225,7 @@ async def update_secret(self, requested_by: APIUser, secret_id: ULID, patch: Sec
         """Update a secret."""
 
         async with self.session_maker() as session, session.begin():
-            result = await session.execute(
-                select(SecretORM).where(SecretORM.id == secret_id).where(SecretORM.user_id == requested_by.id)
-            )
+            result = await session.execute(self._get_stmt(requested_by).where(SecretORM.id == secret_id))
             secret = result.scalar_one_or_none()
             if secret is None:
                 raise errors.MissingResourceError(message=f"The secret with id '{secret_id}' cannot be found")
@@ -239,6 +250,8 @@ async def update_secret(self, requested_by: APIUser, secret_id: ULID, patch: Sec
                     secret_value=patch.secret_value,
                 )
                 secret.update(encrypted_value=encrypted_value, encrypted_key=encrypted_key)
+            if patch.expiration_timestamp is not None:
+                secret.expiration_timestamp = patch.expiration_timestamp
 
             return secret.dump()
 
@@ -247,9 +260,7 @@ async def delete_secret(self, requested_by: APIUser, secret_id: ULID) -> None:
         """Delete a secret."""
 
         async with self.session_maker() as session, session.begin():
-            result = await session.execute(
-                select(SecretORM).where(SecretORM.id == secret_id).where(SecretORM.user_id == requested_by.id)
-            )
+            result = await session.execute(self._get_stmt(requested_by).where(SecretORM.id == secret_id))
             secret = result.scalar_one_or_none()
             if secret is None:
                 return None

components/renku_data_services/secrets/models.py

@@ -8,6 +8,7 @@
 
 from cryptography.hazmat.primitives.asymmetric import rsa
 from kubernetes import client as k8s_client
+from pydantic import Field
 from ulid import ULID
 
 from renku_data_services.app_config import logging
@@ -39,6 +40,7 @@ class Secret:
     encrypted_value: bytes = field(repr=False)
     encrypted_key: bytes = field(repr=False)
     kind: SecretKind
+    expiration_timestamp: datetime | None = Field(default=None)
     modification_date: datetime
 
     session_secret_slot_ids: list[ULID]
@@ -115,15 +117,17 @@ class UnsavedSecret:
     """Model to request the creation of a new user secret."""
 
     name: str
-    default_filename: str | None
     secret_value: str = field(repr=False)
     kind: SecretKind
+    expiration_timestamp: datetime | None = None
+    default_filename: str | None
 
 
 @dataclass(frozen=True, eq=True, kw_only=True)
 class SecretPatch:
     """Model for changes requested on a user secret."""
 
     name: str | None
-    default_filename: str | None
     secret_value: str | None = field(repr=False)
+    expiration_timestamp: datetime | None = None
+    default_filename: str | None