fix: Skip authz if admin is requesting data connectors (#887)

For reprovisioning it is necessary to get all entities. When selecting
data connectors from the db, there was always a call to authz to first
get all identifiers of resources the requesting user has read access
to. For internal service admins, this doesn't work, because they are
not modelled in authz. For admins in general the call can be
prevented, because admins are defined to see everything.

This change skips the authz call if the requesting user is an admin.
Additionally the search reprovisioning code uses an internal service
admin for doing its work when triggered internally (and not by an endpoint).

Author: eikek

bases/renku_data_services/data_api/main.py

@@ -17,7 +17,7 @@
 import renku_data_services.solr.entity_schema as entity_schema
 from renku_data_services.app_config import logging
 from renku_data_services.authz.admin_sync import sync_admins_from_keycloak
-from renku_data_services.base_models.core import APIUser
+from renku_data_services.base_models.core import InternalServiceAdmin, ServiceAdminId
 from renku_data_services.data_api.app import register_all_handlers
 from renku_data_services.data_api.dependencies import DependencyManager
 from renku_data_services.data_api.prometheus import setup_app_metrics, setup_prometheus
@@ -46,7 +46,7 @@ async def _solr_reindex(app: Sanic) -> None:
     """
     config = DependencyManager.from_env()
     reprovision = config.search_reprovisioning
-    admin = APIUser(is_admin=True)
+    admin = InternalServiceAdmin(id=ServiceAdminId.search_reprovision)
     await reprovision.run_reprovision(admin)
 
 

components/renku_data_services/base_models/core.py

@@ -78,6 +78,7 @@ class ServiceAdminId(StrEnum):
     migrations = "migrations"
     secrets_rotation = "secrets_rotation"
     k8s_watcher = "k8s_watcher"
+    search_reprovision = "search_reprovision"
 
 
 @dataclass(kw_only=True, frozen=True)

components/renku_data_services/data_connectors/db.py

@@ -20,6 +20,7 @@
     DataConnectorInProjectPath,
     DataConnectorPath,
     DataConnectorSlug,
+    InternalServiceAdmin,
     NamespacePath,
     NamespaceSlug,
     ProjectPath,
@@ -68,29 +69,27 @@ async def get_data_connectors(
         namespace: ProjectPath | NamespacePath | None = None,
     ) -> tuple[list[models.DataConnector | models.GlobalDataConnector], int]:
         """Get multiple data connectors from the database."""
-        data_connector_ids = await self.authz.resources_with_permission(
-            user, user.id, ResourceType.data_connector, Scope.READ
-        )
 
-        async with self.session_maker() as session:
-            stmt = (
-                select(schemas.DataConnectorORM)
-                .where(schemas.DataConnectorORM.id.in_(data_connector_ids))
-                .options(
-                    joinedload(schemas.DataConnectorORM.slug)
-                    .joinedload(ns_schemas.EntitySlugORM.project)
-                    .joinedload(ProjectORM.slug)
+        async def restrict_by_read(stmt: Select) -> Select:
+            if isinstance(user, InternalServiceAdmin):
+                return stmt
+            else:
+                dc_ids = await self.authz.resources_with_permission(
+                    user, user.id, ResourceType.data_connector, Scope.READ
                 )
+                return stmt.where(schemas.DataConnectorORM.id.in_(dc_ids))
+
+        async with self.session_maker() as session:
+            stmt = (await restrict_by_read(select(schemas.DataConnectorORM))).options(
+                joinedload(schemas.DataConnectorORM.slug)
+                .joinedload(ns_schemas.EntitySlugORM.project)
+                .joinedload(ProjectORM.slug)
             )
             if namespace:
                 stmt = _filter_by_namespace_slug(stmt, namespace)
             stmt = stmt.limit(pagination.per_page).offset(pagination.offset)
             stmt = stmt.order_by(schemas.DataConnectorORM.id.desc())
-            stmt_count = (
-                select(func.count())
-                .select_from(schemas.DataConnectorORM)
-                .where(schemas.DataConnectorORM.id.in_(data_connector_ids))
-            )
+            stmt_count = await restrict_by_read(select(func.count()).select_from(schemas.DataConnectorORM))
             if namespace:
                 stmt_count = _filter_by_namespace_slug(stmt_count, namespace)
             results = await session.scalars(stmt), await session.scalar(stmt_count)

components/renku_data_services/search/blueprints.py

@@ -40,7 +40,7 @@ async def _post(request: Request, user: base_models.APIUser) -> HTTPResponse | J
             reprovisioning = await self.search_reprovision.acquire_reprovision()
 
             request.app.add_task(
-                self.search_reprovision.init_reprovision(requested_by=user, reprovisioning=reprovisioning),
+                self.search_reprovision.init_reprovision(user, reprovisioning=reprovisioning),
                 name=f"reprovisioning-{reprovisioning.id}",
             )
 

components/renku_data_services/search/reprovision.py

@@ -8,6 +8,7 @@
 from renku_data_services.base_models.core import APIUser
 from renku_data_services.data_connectors.db import DataConnectorRepository
 from renku_data_services.data_connectors.models import DataConnector, GlobalDataConnector
+from renku_data_services.errors.errors import ForbiddenError
 from renku_data_services.message_queue.db import ReprovisioningRepository
 from renku_data_services.message_queue.models import Reprovisioning
 from renku_data_services.namespace.db import GroupRepository
@@ -43,10 +44,10 @@ def __init__(
         self._project_repo = project_repo
         self._data_connector_repo = data_connector_repo
 
-    async def run_reprovision(self, requested_by: APIUser) -> int:
+    async def run_reprovision(self, admin: APIUser) -> int:
         """Start a reprovisioning if not already running."""
         reprovision = await self.acquire_reprovision()
-        return await self.init_reprovision(requested_by, reprovision)
+        return await self.init_reprovision(admin, reprovision)
 
     async def acquire_reprovision(self) -> Reprovisioning:
         """Acquire a reprovisioning slot. Throws if already taken."""
@@ -74,7 +75,7 @@ async def _get_all_data_connectors(
             for dc in result[0]:
                 yield dc
 
-    async def init_reprovision(self, requested_by: APIUser, reprovisioning: Reprovisioning) -> int:
+    async def init_reprovision(self, admin: APIUser, reprovisioning: Reprovisioning) -> int:
         """Initiates reprovisioning by inserting documents into the staging table.
 
         Deletes all renku entities in the solr core. Then it goes
@@ -84,6 +85,9 @@ async def init_reprovision(self, requested_by: APIUser, reprovisioning: Reprovis
         solr with these entries.
         """
 
+        if not admin.is_admin:
+            raise ForbiddenError(message="Only Renku administrators are allowed to start search reprovisioning.")
+
         def log_counter(c: int) -> None:
             if c % 50 == 0:
                 logger.info(f"Inserted {c}. entities into staging table...")
@@ -96,20 +100,21 @@ def log_counter(c: int) -> None:
             async with DefaultSolrClient(self._solr_config) as client:
                 await client.delete("_type:*")
 
-            all_users = self._user_repo.get_all_users(requested_by=requested_by)
+            all_users = self._user_repo.get_all_users(requested_by=admin)
             counter = await self.__update_entities(all_users, "user", started, counter, log_counter)
-            logger.info("Done adding user entities to search_updates table.")
+            logger.info(f"Done adding user entities to search_updates table. Record count: {counter}.")
 
-            all_groups = self._group_repo.get_all_groups(requested_by=requested_by)
+            all_groups = self._group_repo.get_all_groups(requested_by=admin)
             counter = await self.__update_entities(all_groups, "group", started, counter, log_counter)
-            logger.info("Done adding group entities to search_updates table.")
+            logger.info(f"Done adding group entities to search_updates table. Record count: {counter}")
 
-            all_projects = self._project_repo.get_all_projects(requested_by=requested_by)
+            all_projects = self._project_repo.get_all_projects(requested_by=admin)
             counter = await self.__update_entities(all_projects, "project", started, counter, log_counter)
-            logger.info("Done adding project entities to search_updates table.")
+            logger.info(f"Done adding project entities to search_updates table. Record count: {counter}")
 
-            all_dcs = self._get_all_data_connectors(requested_by, per_page=20)
+            all_dcs = self._get_all_data_connectors(admin, per_page=20)
             counter = await self.__update_entities(all_dcs, "data connector", started, counter, log_counter)
+            logger.info(f"Done adding dataconnector entities to search_updates table. Record count: {counter}")
 
             logger.info(f"Inserted {counter} entities into the staging table.")
         except Exception as e:

components/renku_data_services/solr/solr_client.py

@@ -655,6 +655,7 @@ async def get_raw(self, id: str) -> Response:
     async def query_raw(self, query: SolrQuery) -> Response:
         """Query documents and return the http response."""
         try:
+            logger.debug(f"Running solr query: {self.config.base_url}/solr/{self.config.core}")
             return await self.delegate.post("/query", params={"wt": "json"}, json=query.to_dict())
         except ConnectError as e:
             raise SolrClientConnectException(e) from e

test/bases/renku_data_services/data_api/conftest.py

@@ -15,7 +15,7 @@
 from renku_data_services.authz.admin_sync import sync_admins_from_keycloak
 from renku_data_services.authz.authz import _AuthzConverter
 from renku_data_services.base_models import Slug
-from renku_data_services.base_models.core import APIUser, NamespacePath
+from renku_data_services.base_models.core import InternalServiceAdmin, NamespacePath, ServiceAdminId
 from renku_data_services.data_api.app import register_all_handlers
 from renku_data_services.data_api.dependencies import DependencyManager
 from renku_data_services.migrations.core import run_migrations_for_app
@@ -261,9 +261,10 @@ async def sanic_client_with_solr(sanic_client: SanicASGITestClient, app_manager)
 
 @pytest_asyncio.fixture
 async def search_reprovision(app_manager_instance: DependencyManager, search_push_updates):
+    admin = InternalServiceAdmin(id=ServiceAdminId.search_reprovision)
+
     async def search_reprovision_helper() -> None:
-        user = APIUser(is_admin=True)
-        await app_manager_instance.search_reprovisioning.run_reprovision(user)
+        await app_manager_instance.search_reprovisioning.run_reprovision(admin)
         await search_push_updates(clear_index=False)
 
     return search_reprovision_helper

test/components/renku_data_services/search/test_reprovision.py

@@ -161,14 +161,14 @@ async def test_get_data_connectors(app_manager_instance) -> None:
 
 
 @pytest.mark.asyncio
-async def test_run_reprovision(app_manager_instance, solr_search) -> None:
+async def test_run_reprovision(app_manager_instance, solr_search, admin_user) -> None:
     setup = make_setup(app_manager_instance, solr_search)
     dcs = await make_data_connectors(setup, 5)
     groups = await make_groups(setup, 4)
     projects = await make_projects(setup, 3)
     users = [item async for item in setup.user_repo.get_all_users(admin)]
 
-    count = await setup.search_reprovision.run_reprovision(admin)
+    count = await setup.search_reprovision.run_reprovision(admin_user)
 
     next = await setup.search_update_repo.select_next(20)