feat: add authentication

src/db/adapter.py

@@ -1,4 +1,11 @@
-"""Adapter based on SQLAlchemy."""
+"""Adapter based on SQLAlchemy.
+
+These adapters currently do a few things (1) generate SQL queries, (2) apply resource access controls,
+(3) fetch the SQL results and (4) format them into a workable representation. In the future and as the code
+grows it is worth looking into separating this functionality into separate classes rather than having
+it all in one place.
+"""
+from functools import wraps
 from typing import List, Optional, Tuple
 
 from alembic import command, config
@@ -31,16 +38,12 @@ def do_migrations(self):
             command.upgrade(cfg, "head")
 
 
-def _get_resource_pool_select_stmt(
+def _resource_pool_access_control(
     api_user: models.APIUser,
+    stmt: Select[Tuple[schemas.ResourcePoolORM]],
     keycloak_id: Optional[str] = None,
-    resource_pool_name: Optional[str] = None,
-    resource_pool_id: Optional[int] = None,
 ) -> Select[Tuple[schemas.ResourcePoolORM]]:
-    """Generates a query to list resource pools based on different criteria and whether the user is logged in or not."""
-    stmt = select(schemas.ResourcePoolORM).options(
-        selectinload(schemas.ResourcePoolORM.quota), selectinload(schemas.ResourcePoolORM.classes)
-    )
+    """Modifies a select query to list resource pools based on whether the user is logged in or not."""
     output = stmt
     match (api_user.is_authenticated, api_user.is_admin):
         case True, False:
@@ -62,89 +65,65 @@ def _get_resource_pool_select_stmt(
         case _:
             # The user is not logged in, they can see only the default resource pools
             output = output.where(schemas.ResourcePoolORM.default == True)  # noqa: E712
-    if resource_pool_name is not None:
-        output = output.where(schemas.ResourcePoolORM.name == resource_pool_name)
-    if resource_pool_id is not None:
-        output = output.where(schemas.ResourcePoolORM.id == resource_pool_id)
     return output
 
 
-def _get_select_for_quota(
-    api_user: models.APIUser, resource_pool_id: Optional[int] = None
-) -> Select[Tuple[schemas.QuotaORM]]:
-    """Adjust the select statement for quota based on whether the user is logged in or not."""
-    output = select(schemas.QuotaORM)
-    match (api_user.is_authenticated, api_user.is_admin):
-        case True, False:
-            # The user is logged in but is not an admin, they can see only a quota from a resource pool they have
-            # been granted access to or from default resource pools.
-            output = (
-                output.join(schemas.ResourcePoolORM, schemas.QuotaORM.resource_pool)
-                .join(schemas.UserORM, schemas.ResourcePoolORM.users)
-                .where(
-                    or_(
-                        schemas.UserORM.keycloak_id == api_user.id,
-                        schemas.ResourcePoolORM.default == True,  # noqa: E712
-                    )
-                )
-            )
-            if resource_pool_id is not None:
-                output = output.where(schemas.ResourcePoolORM.id == resource_pool_id)
-        case True, True:
-            # The user is logged in and is an admin, they can see any quota
-            if resource_pool_id is not None:
-                output = output.join(schemas.ResourcePoolORM, schemas.QuotaORM.resource_pool).where(
-                    schemas.ResourcePoolORM.id == resource_pool_id
-                )
-        case _:
-            # The user is not an admin, they can see the quota of only the default resource pool
-            output = output.join(schemas.ResourcePoolORM, schemas.QuotaORM.resource_pool).where(
-                schemas.ResourcePoolORM.default == True  # noqa: E712
-            )
-            if resource_pool_id is not None:
-                output = output.where(schemas.ResourcePoolORM.id == resource_pool_id)
-    return output
-
-
-def _get_select_for_classes(
+def _classes_user_access_control(
     api_user: models.APIUser,
-    resource_pool_id: Optional[int] = None,
-    resource_class_id: Optional[int] = None,
-    resource_class_name: Optional[str] = None,
+    stmt: Select[Tuple[schemas.ResourceClassORM]],
 ) -> Select[Tuple[schemas.ResourceClassORM]]:
     """Adjust the select statement for classes based on whether the user is logged in or not."""
-    output = select(schemas.ResourceClassORM)
-    match (api_user.is_authenticated, api_user.is_admin):
-        case True, False:
-            # The user is logged in but is not an admin (they have access to resource pools
-            # they have access to and to default resource pools)
-            output = (
-                output.join(schemas.ResourcePoolORM, schemas.ResourceClassORM.resource_pool)
-                .join(schemas.UserORM, schemas.ResourcePoolORM.users)
-                .where(schemas.UserORM.keycloak_id == api_user.id)
-            )
-            if resource_pool_id is not None:
-                output = output.where(schemas.ResourcePoolORM.id == resource_pool_id)
-        case True, True:
-            # The user is logged in and is an admin (they have access to everything)
-            if resource_pool_id is not None:
-                output = output.join(schemas.ResourcePoolORM, schemas.ResourceClassORM.resource_pool).where(
-                    schemas.ResourcePoolORM.id == resource_pool_id
-                )
-        case _:
-            # The user is not logged in at all (they have access only to classes frmo default resource pools)
-            output = output.join(schemas.ResourcePoolORM, schemas.ResourceClassORM.resource_pool).where(
-                schemas.ResourcePoolORM.default == True  # noqa: E712
+    output = stmt
+    if api_user.is_authenticated and not api_user.is_admin:
+        # The user is logged in but is not an admin (they have access to resource pools
+        # they have access to and to default resource pools)
+        output = output.join(schemas.UserORM, schemas.ResourcePoolORM.users).where(
+            schemas.UserORM.keycloak_id == api_user.id
+        )
+    return output
+
+
+def _quota_user_access_control(
+    api_user: models.APIUser, stmt: Select[Tuple[schemas.QuotaORM]]
+) -> Select[Tuple[schemas.QuotaORM]]:
+    """Adjust the select statement for a quota based on whether the user is logged in or not."""
+    output = stmt
+    if api_user.is_authenticated and not api_user.is_admin:
+        # The user is logged in but is not an admin, they can see only a quota from a resource pool they have
+        # been granted access to or from default resource pools.
+        output = output.join(schemas.UserORM, schemas.ResourcePoolORM.users).where(
+            or_(
+                schemas.UserORM.keycloak_id == api_user.id,
+                schemas.ResourcePoolORM.default == True,  # noqa: E712
             )
-            if resource_pool_id is not None:
-                output = output.where(schemas.ResourcePoolORM.id == resource_pool_id)
-    if resource_class_id is not None:
-        output = output.where(schemas.ResourceClassORM.id == resource_class_id)
-    if resource_class_name is not None:
-        output = output.where(schemas.ResourceClassORM.name == resource_class_name)
+        )
     return output
 
 
+def _only_admins(f):
+    """Decorator that errors out if the user is not an admin.
+
+    It expects the APIUser model to be a named parameter in the decorated function or
+    to be the first parameter (after self).
+    """
+
+    @wraps(f)
+    async def decorated_function(self, *args, **kwargs):
+        api_user = None
+        if "api_user" in kwargs:
+            api_user = kwargs["api_user"]
+        elif len(args) >= 1:
+            api_user = args[0]
+        if api_user is None or not api_user.is_admin:
+            raise errors.Unauthorized(message="You do not have the required permissions for this operation.")
+
+        # the user is authenticated and is an admin
+        response = await f(self, *args, **kwargs)
+        return response
+
+    return decorated_function
+
+
 class ResourcePoolRepository(_Base):
     """The adapter used for accessing resource pools with SQLAlchemy."""
 
@@ -153,11 +132,20 @@ async def get_resource_pools(
     ) -> List[models.ResourcePool]:
         """Get resource pools from database."""
         async with self.session_maker() as session:
-            stmt = _get_resource_pool_select_stmt(api_user=api_user, resource_pool_id=id, resource_pool_name=name)
+            stmt = select(schemas.ResourcePoolORM).options(
+                selectinload(schemas.ResourcePoolORM.quota), selectinload(schemas.ResourcePoolORM.classes)
+            )
+            if name is not None:
+                stmt = stmt.where(schemas.ResourcePoolORM.name == name)
+            if id is not None:
+                stmt = stmt.where(schemas.ResourcePoolORM.id == id)
+            # NOTE: The line below ensures that the right users can access the right resources, do not remove.
+            stmt = _resource_pool_access_control(api_user, stmt)
             res = await session.execute(stmt)
             orms = res.scalars().all()
             return [orm.dump() for orm in orms]
 
+    @_only_admins
     async def insert_resource_pool(
         self, api_user: models.APIUser, resource_pool: models.ResourcePool
     ) -> models.ResourcePool:
@@ -174,9 +162,13 @@ async def get_quota(self, api_user: models.APIUser, resource_pool_id: int) -> Op
         """Get a quota for a specific resource pool."""
 
         async with self.session_maker() as session:
-            stmt = _get_select_for_quota(api_user=api_user, resource_pool_id=resource_pool_id)
+            stmt = select(schemas.QuotaORM).join(schemas.ResourcePoolORM, schemas.QuotaORM.resource_pool)
+            # NOTE: The line below ensures that the right users can access the right resources, do not remove.
+            stmt = _quota_user_access_control(api_user, stmt)
+            if resource_pool_id is not None:
+                stmt = stmt.where(schemas.ResourcePoolORM.id == resource_pool_id)
             res = await session.execute(stmt)
-            orm: schemas.QuotaORM = res.scalars().first()
+            orm: Optional[schemas.QuotaORM] = res.scalars().first()
             if not orm:
                 return None
             return orm.dump()
@@ -190,13 +182,22 @@ async def get_classes(
     ) -> List[models.ResourceClass]:
         """Get classes from the database."""
         async with self.session_maker() as session:
-            stmt = _get_select_for_classes(
-                api_user=api_user, resource_pool_id=resource_pool_id, resource_class_id=id, resource_class_name=name
+            stmt = select(schemas.ResourceClassORM).join(
+                schemas.ResourcePoolORM, schemas.ResourceClassORM.resource_pool
             )
+            if resource_pool_id is not None:
+                stmt = stmt.where(schemas.ResourcePoolORM.id == resource_pool_id)
+            if id is not None:
+                stmt = stmt.where(schemas.ResourceClassORM.id == id)
+            if name is not None:
+                stmt = stmt.where(schemas.ResourceClassORM.name == name)
+            # NOTE: The line below ensures that the right users can access the right resources, do not remove.
+            stmt = _classes_user_access_control(api_user, stmt)
             res = await session.execute(stmt)
             orms = res.scalars().all()
             return [orm.dump() for orm in orms]
 
+    @_only_admins
     async def insert_resource_class(
         self, api_user: models.APIUser, resource_class: models.ResourceClass, *, resource_pool_id: Optional[int] = None
     ) -> models.ResourceClass:
@@ -219,6 +220,7 @@ async def insert_resource_class(
                 session.add(cls)
         return cls.dump()
 
+    @_only_admins
     async def update_quota(self, api_user: models.APIUser, resource_pool_id: int, **kwargs) -> models.Quota:
         """Update an existing quota in the database."""
         if not api_user.is_admin:
@@ -242,6 +244,7 @@ async def update_quota(self, api_user: models.APIUser, resource_pool_id: int, **
             raise errors.MissingResourceError(message=f"Resource pool with id {resource_pool_id} cannot be found")
         return orm.dump()
 
+    @_only_admins
     async def update_resource_pool(self, api_user: models.APIUser, id: int, **kwargs) -> models.ResourcePool:
         """Update an existing resource pool in the database."""
         if not api_user.is_admin:
@@ -293,6 +296,7 @@ async def update_resource_pool(self, api_user: models.APIUser, id: int, **kwargs
 
                 return rp.dump()
 
+    @_only_admins
     async def delete_resource_pool(self, api_user: models.APIUser, id: int):
         """Delete a resource pool from the database."""
         if not api_user.is_admin:
@@ -305,6 +309,7 @@ async def delete_resource_pool(self, api_user: models.APIUser, id: int):
                 if rp is not None:
                     await session.delete(rp)
 
+    @_only_admins
     async def delete_resource_class(self, api_user: models.APIUser, resource_pool_id: int, resource_class_id: int):
         """Delete a specific resource class."""
         if not api_user.is_admin:
@@ -322,6 +327,7 @@ async def delete_resource_class(self, api_user: models.APIUser, resource_pool_id
                     return None
                 await session.delete(cls)
 
+    @_only_admins
     async def update_resource_class(
         self, api_user: models.APIUser, resource_pool_id: int, resource_class_id: int, **kwargs
     ) -> models.ResourceClass:
@@ -353,6 +359,7 @@ async def update_resource_class(
 class UserRepository(_Base):
     """The adapter used for accessing users with SQLAlchemy."""
 
+    @_only_admins
     async def get_users(
         self,
         *,
@@ -389,6 +396,7 @@ async def get_users(
                     orms = res.scalars().all()
                     return [orm.dump() for orm in orms]
 
+    @_only_admins
     async def insert_user(self, api_user: models.APIUser, user: models.User) -> models.User:
         """Inser a user in the database."""
         if not api_user.is_admin:
@@ -399,6 +407,7 @@ async def insert_user(self, api_user: models.APIUser, user: models.User) -> mode
                 session.add(orm)
         return orm.dump()
 
+    @_only_admins
     async def delete_user(self, api_user: models.APIUser, id: str):
         """Remove a user from the database."""
         if not api_user.is_admin:
@@ -423,16 +432,20 @@ async def get_user_resource_pools(
         """Get resource pools that a specific user has access to."""
         async with self.session_maker() as session:
             async with session.begin():
-                stmt = _get_resource_pool_select_stmt(
-                    api_user=api_user,
-                    keycloak_id=keycloak_id,
-                    resource_pool_id=resource_pool_id,
-                    resource_pool_name=resource_pool_name,
+                stmt = select(schemas.ResourcePoolORM).options(
+                    selectinload(schemas.ResourcePoolORM.quota), selectinload(schemas.ResourcePoolORM.classes)
                 )
+                if resource_pool_name is not None:
+                    stmt = stmt.where(schemas.ResourcePoolORM.name == resource_pool_name)
+                if resource_pool_id is not None:
+                    stmt = stmt.where(schemas.ResourcePoolORM.id == resource_pool_id)
+                # NOTE: The line below ensures that the right users can access the right resources, do not remove.
+                stmt = _resource_pool_access_control(api_user, stmt, keycloak_id=keycloak_id)
                 res = await session.execute(stmt)
                 rps: List[schemas.ResourcePoolORM] = res.scalars().all()
                 return [rp.dump() for rp in rps]
 
+    @_only_admins
     async def update_user_resource_pools(
         self, api_user: models.APIUser, keycloak_id: str, resource_pool_ids: List[int], append: bool = True
     ) -> List[models.ResourcePool]:
@@ -468,6 +481,7 @@ async def update_user_resource_pools(
                     user.resource_pools = rps_to_add
                 return [rp.dump() for rp in rps_to_add]
 
+    @_only_admins
     async def delete_resource_pool_user(self, api_user: models.APIUser, resource_pool_id: int, keycloak_id: str):
         """Remove a user from a specific resource pool."""
         if not api_user.is_admin:
@@ -483,6 +497,7 @@ async def delete_resource_pool_user(self, api_user: models.APIUser, resource_poo
                 stmt = delete(schemas.resource_pools_users).where(schemas.resource_pools_users.c.user_id.in_(sub))
                 await session.execute(stmt)
 
+    @_only_admins
     async def update_resource_pool_users(
         self, api_user: models.APIUser, resource_pool_id: int, users: List[models.User], append: bool = True
     ) -> List[models.User]:

src/renku_crac/app.py

@@ -361,13 +361,11 @@ def post(self) -> BlueprintFactoryResponse:
         @only_admins
         @validate(json=apispec.UserWithId)
         async def _post(request: Request, body: apispec.UserWithId, user: models.APIUser):
-            if user.access_token is None:
-                raise errors.Unauthorized()
             users_db, user_kc = await asyncio.gather(
                 self.repo.get_users(keycloak_id=body.id, api_user=user),
-                self.user_store.get_user_by_id(body.id, user.access_token),
+                self.user_store.get_user_by_id(body.id, user.access_token),  # type: ignore[arg-type]
             )
-            user_db = users_db[0] if len(users_db) >= 1 else None
+            user_db = next(iter(users_db), None)
             # The user does not exist in keycloak, delete it form the crac database and fail.
             if user_kc is None:
                 await self.repo.delete_user(id=body.id, api_user=user)

src/renku_crac/auth.py

@@ -6,7 +6,7 @@
 from models import APIUser, Authenticator, errors
 
 
-def authenticate(authneticator: Authenticator):
+def authenticate(authenticator: Authenticator):
     """Decorator for a Sanic handler that adds the APIUser model to the context.
 
     The APIUser is present for admins, non-admins and users who are not logged in.
@@ -18,7 +18,7 @@ async def decorated_function(request: Request, *args, **kwargs):
             token = request.headers.get("Authorization")
             user = APIUser()
             if token is not None and len(token) >= 8:
-                user = await authneticator.authenticate(token[7:])
+                user = await authenticator.authenticate(token[7:])
 
             response = await f(request, *args, **kwargs, user=user)
             return response
@@ -33,8 +33,10 @@ def only_admins(f):
 
     @wraps(f)
     async def decorated_function(request: Request, user: APIUser, *args, **kwargs):
-        if user is None or not user.is_admin:
-            raise errors.Unauthorized()
+        if user is None or user.access_token is None:
+            raise errors.Unauthorized(message="Please provide valid access credentials in the Authorization header.")
+        if not user.is_admin:
+            raise errors.Unauthorized(message="You do not have the reuqired permissions for this operation.")
 
         # the user is authenticated and is an admin
         response = await f(request, *args, **kwargs, user=user)

src/users/dummy.py

@@ -33,7 +33,7 @@ async def get_user_by_id(self, id: str, access_token: str) -> Optional[models.Us
 
 @dataclass
 class DummyAuthenticator:
-    """Dummy authenticator that mimics pretends to call Keycloak, not suitable for production."""
+    """Dummy authenticator that pretends to call Keycloak, not suitable for production."""
 
     logged_in: bool = True
     admin: bool = False