feat: add expiration timestamps for secrets

README.md

@@ -27,12 +27,14 @@ details.
 
 1. Write code
 2. Run tests: `make tests`
+   * Run a specific test e.g.: `poetry run pytest test/bases/renku_data_services/data_api/test_storage_v2.py::test_storage_v2_create_openbis_secret`
+   * Also run tests marked with `@pytest.mark.myskip`: `PYTEST_FORCE_RUN_MYSKIPS=1 make tests`
 3. Style checks: `make style_checks`
 
 ### Developing with the container image
 
 The container image can be built to be used as a local development service (for renku_crc):
-`docker build -f projects/renku_data_services/Dockerfile . --build-arg DEV_BUILD=true -t renku-data-service`
+`docker build -f projects/renku_data_service/Dockerfile . --build-arg DEV_BUILD=true -t renku-data-service`
 
 It can then be run as daemon: `docker run -d -e DUMMY_STORES=true --name renku-crc renku-data-service`
 
@@ -48,10 +50,10 @@ metadata for it in the `components/renku_data_services/migrations/env.py` file.
 
 **To create a new migration:**
 
-`DUMMY_STORES=true alembic -c components/renku_data_services/migrations/alembic.ini --name common revision -m "<message>" --autogenerate --version-path components/renku_data_services/migrations/versions`
+`DUMMY_STORES=true poetry run alembic -c components/renku_data_services/migrations/alembic.ini --name common revision -m "<message>" --autogenerate --version-path components/renku_data_services/migrations/versions`
 
 You can specify a different version path if you wish to, just make sure it is listed in `alembic.ini` under
 `version_locations`.
 
 **To run all migrations:**
-`DUMMY_STORES=true alembic -c components/renku_data_services/migrations/alembic.ini --name=common upgrade heads`
+`DUMMY_STORES=true poetry run alembic -c components/renku_data_services/migrations/alembic.ini --name=common upgrade heads`

components/renku_data_services/migrations/versions/7bc32829ed2f_add_secret_expiration_timestamp.py

@@ -0,0 +1,30 @@
+"""add_secret_expiration_timestamp
+
+Revision ID: 7bc32829ed2f
+Revises: 9058bf0a1a12
+Create Date: 2024-08-21 12:38:30.932694
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "7bc32829ed2f"
+down_revision = "9058bf0a1a12"
+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/secrets/db.py

@@ -1,10 +1,10 @@
 """Database repo for secrets."""
 
 from collections.abc import AsyncGenerator, Callable, Sequence
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from typing import cast
 
-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
@@ -25,11 +25,23 @@ def __init__(
     ) -> None:
         self.session_maker = session_maker
 
+    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]
@@ -38,7 +50,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 | None:
         """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 orm is None:
@@ -66,6 +78,7 @@ async def insert_secret(self, requested_by: APIUser, secret: Secret) -> Secret:
                 encrypted_value=secret.encrypted_value,
                 encrypted_key=secret.encrypted_key,
                 kind=secret.kind,
+                expiration_timestamp=secret.expiration_timestamp,
             )
             session.add(orm)
 
@@ -83,29 +96,34 @@ async def insert_secret(self, requested_by: APIUser, secret: Secret) -> Secret:
 
     @only_authenticated
     async def update_secret(
-        self, requested_by: APIUser, secret_id: ULID, encrypted_value: bytes, encrypted_key: bytes
+        self,
+        requested_by: APIUser,
+        secret_id: ULID,
+        encrypted_value: bytes,
+        encrypted_key: bytes,
+        expiration_timestamp: datetime | None,
     ) -> Secret:
         """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")
 
-            secret.update(encrypted_value=encrypted_value, encrypted_key=encrypted_key)
+            secret.update(
+                encrypted_value=encrypted_value,
+                encrypted_key=encrypted_key,
+                expiration_timestamp=expiration_timestamp,
+            )
         return secret.dump()
 
     @only_authenticated
     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

@@ -25,6 +25,7 @@ class Secret(BaseModel):
     id: ULID | None = Field(default=None, init=False)
     modification_date: datetime = Field(default_factory=lambda: datetime.now(UTC).replace(microsecond=0), init=False)
     kind: SecretKind
+    expiration_timestamp: datetime | None = Field(default=None)
 
 
 @dataclass

components/renku_data_services/secrets/orm.py

@@ -35,6 +35,9 @@ class SecretORM(BaseORM):
     encrypted_value: Mapped[bytes] = mapped_column(LargeBinary())
     encrypted_key: Mapped[bytes] = mapped_column(LargeBinary())
     kind: Mapped[models.SecretKind]
+    expiration_timestamp: Mapped[Optional[datetime]] = mapped_column(
+        "expiration_timestamp", DateTime(timezone=True), default=None, nullable=True
+    )
     modification_date: Mapped[datetime] = mapped_column(
         "modification_date", DateTime(timezone=True), default_factory=lambda: datetime.now(UTC).replace(microsecond=0)
     )
@@ -49,6 +52,7 @@ def dump(self) -> models.Secret:
             name=self.name, encrypted_value=self.encrypted_value, encrypted_key=self.encrypted_key, kind=self.kind
         )
         secret.id = self.id
+        secret.expiration_timestamp = self.expiration_timestamp
         secret.modification_date = self.modification_date
         return secret
 
@@ -59,12 +63,14 @@ def load(cls, secret: models.Secret) -> "SecretORM":
             name=secret.name,
             encrypted_value=secret.encrypted_value,
             encrypted_key=secret.encrypted_key,
-            modification_date=secret.modification_date,
             kind=secret.kind,
+            expiration_timestamp=secret.expiration_timestamp,
+            modification_date=secret.modification_date,
         )
 
-    def update(self, encrypted_value: bytes, encrypted_key: bytes) -> None:
+    def update(self, encrypted_value: bytes, encrypted_key: bytes, expiration_timestamp: datetime | None) -> None:
         """Update an existing secret."""
         self.encrypted_value = encrypted_value
         self.encrypted_key = encrypted_key
+        self.expiration_timestamp = expiration_timestamp
         self.modification_date = datetime.now(UTC).replace(microsecond=0)

components/renku_data_services/storage/blueprints.py

@@ -1,6 +1,7 @@
 """Cloud storage app."""
 
 from dataclasses import dataclass
+from datetime import datetime
 from typing import Any
 
 from sanic import HTTPResponse, Request, empty, json
@@ -16,6 +17,7 @@
 from renku_data_services.storage import apispec, models
 from renku_data_services.storage.db import StorageRepository, StorageV2Repository
 from renku_data_services.storage.rclone import RCloneValidator
+from renku_data_services.utils.core import get_openbis_pat
 
 
 def dump_storage_with_sensitive_fields(storage: models.CloudStorage, validator: RCloneValidator) -> dict[str, Any]:
@@ -303,12 +305,49 @@ def upsert_secrets(self) -> BlueprintFactoryResponse:
         """Create/update secrets for a cloud storage."""
 
         @authenticate(self.authenticator)
-        async def _upsert_secrets(request: Request, user: base_models.APIUser, storage_id: ULID) -> JSONResponse:
+        async def _upsert_secrets(
+            request: Request, user: base_models.APIUser, storage_id: ULID, validator: RCloneValidator
+        ) -> JSONResponse:
             # TODO: use @validate once sanic supports validating json lists
             body = apispec.CloudStorageSecretPostList.model_validate(request.json)
+
+            # NOTE: Check that user has proper access to the storage
+            storage = await self.storage_v2_repo.get_storage_by_id(storage_id=storage_id, user=user)
+
             secrets = [models.CloudStorageSecretUpsert.model_validate(s.model_dump()) for s in body.root]
+
+            provider = validator.providers[storage.storage_type]
+            sensitive_lookup = [o.name for o in provider.options if o.sensitive]
+            for secret in secrets:
+                if len(secret.value) > 0 and 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[str, datetime]:
+                    if len(secrets) == 1 and secrets[0].name == "session_token":
+                        try:
+                            return await get_openbis_pat(storage.configuration["host"], secrets[0].value)
+                        except Exception as e:
+                            raise errors.ProgrammingError(message=str(e))
+
+                    raise errors.ValidationError(message="The openBIS storage has only one secret: session_token")
+
+                (
+                    secrets[0].value,
+                    expiration_timestamp,
+                ) = await openbis_transform_session_token_to_pat()
+
             result = await self.storage_v2_repo.upsert_storage_secrets(
-                storage_id=storage_id, user=user, secrets=secrets
+                storage_id=storage_id,
+                user=user,
+                secrets=secrets,
+                expiration_timestamp=expiration_timestamp,
             )
             return json(
                 apispec.CloudStorageSecretGetList.model_validate(result).model_dump(exclude_none=True, mode="json"), 201

components/renku_data_services/storage/db.py

@@ -1,6 +1,7 @@
 """Adapters for storage database classes."""
 
 from collections.abc import Callable
+from datetime import datetime
 from typing import cast
 
 from cryptography.hazmat.primitives.asymmetric import rsa
@@ -165,18 +166,18 @@ async def delete_storage(self, storage_id: ULID, user: base_models.APIUser) -> N
             await session.delete(storage[0])
 
     async def upsert_storage_secrets(
-        self, storage_id: ULID, user: base_models.APIUser, secrets: list[models.CloudStorageSecretUpsert]
+        self,
+        storage_id: ULID,
+        user: base_models.APIUser,
+        secrets: list[models.CloudStorageSecretUpsert],
+        expiration_timestamp: datetime | None,
     ) -> list[models.CloudStorageSecret]:
         """Create/update cloud storage secrets."""
-        # NOTE: Check that user has proper access to the storage
-        storage = await self.get_storage_by_id(storage_id=storage_id, user=user)
-
-        secret_names_values = {s.name: s.value for s in secrets}
         async with self.session_maker() as session, session.begin():
             stmt = (
                 select(schemas.CloudStorageSecretsORM)
                 .where(schemas.CloudStorageSecretsORM.user_id == user.id)
-                .where(schemas.CloudStorageSecretsORM.storage_id == storage.storage_id)
+                .where(schemas.CloudStorageSecretsORM.storage_id == storage_id)
                 .options(selectinload(schemas.CloudStorageSecretsORM.secret))
             )
             result = await session.execute(stmt)
@@ -185,6 +186,7 @@ async def upsert_storage_secrets(
             existing_secrets = {s.name: s for s in existing_storage_secrets_orm}
             stored_secrets = []
 
+            secret_names_values = {s.name: s.value for s in secrets}
             for name, value in secret_names_values.items():
                 encrypted_value, encrypted_key = await encrypt_user_secret(
                     user_repo=self.user_repo,
@@ -194,14 +196,19 @@ async def upsert_storage_secrets(
                 )
 
                 if storage_secret_orm := existing_secrets.get(name):
-                    storage_secret_orm.secret.update(encrypted_value=encrypted_value, encrypted_key=encrypted_key)
+                    storage_secret_orm.secret.update(
+                        encrypted_value=encrypted_value,
+                        encrypted_key=encrypted_key,
+                        expiration_timestamp=expiration_timestamp,
+                    )
                 else:
                     secret_orm = secrets_schemas.SecretORM(
                         name=f"{storage_id}-{name}",
                         user_id=cast(str, user.id),
                         encrypted_value=encrypted_value,
                         encrypted_key=encrypted_key,
                         kind=SecretKind.storage,
+                        expiration_timestamp=expiration_timestamp,
                     )
                     session.add(secret_orm)
 

components/renku_data_services/storage/rclone.py

@@ -161,7 +161,7 @@ def __patch_schema_add_openbis_type(spec: list[dict[str, Any]]) -> None:
                         "Value": None,
                         "ShortOpt": "",
                         "Hide": 0,
-                        "Required": False,
+                        "Required": True,
                         "IsPassword": True,
                         "NoPrefix": False,
                         "Advanced": False,
@@ -416,7 +416,7 @@ class RCloneProviderSchema(BaseModel):
     @property
     def required_options(self) -> list[RCloneOption]:
         """Returns all required options for this provider."""
-        return [o for o in self.options if o.required]
+        return [o for o in self.options if o.required and not o.sensitive]
 
     @property
     def sensitive_options(self) -> list[RCloneOption]: