feat: Add fine-grained permission system using scopes

Instead of the unflexible roles and project roles, introduce a
fine-grained permission system.

For OpenID / frontend authentication, the role concept is
still in place and will be expended with customizable roles in the future.

During creation of personal access tokens, the user can select
the permissions that the token should have. This replaces the full-access
of personal access tokens. Existing access tokens will receive all
permissions that the user has access to during migration.

The permission system will be matched the current role and the behaviour
will not change; it's not a breaking change.

This is also the basis for mapping of OAuth scopes.

backend/capellacollab/core/authentication/basic_auth.py

@@ -6,10 +6,12 @@
 
 import fastapi
 from fastapi import security
+from sqlalchemy import orm
 
-from capellacollab.core import database
 from capellacollab.users import crud as user_crud
+from capellacollab.users import models as users_models
 from capellacollab.users.tokens import crud as token_crud
+from capellacollab.users.tokens import models as tokens_models
 
 from . import exceptions
 
@@ -20,29 +22,31 @@ class HTTPBasicAuth(security.HTTPBasic):
     def __init__(self):
         super().__init__(auto_error=True)
 
-    async def __call__(self, request: fastapi.Request) -> str:  # type: ignore
+    async def validate(
+        self, db: orm.Session, request: fastapi.Request
+    ) -> tuple[users_models.DatabaseUser, tokens_models.DatabaseUserToken]:
         credentials: (
             security.HTTPBasicCredentials | None
         ) = await super().__call__(request)
         if not credentials:
             raise exceptions.UnauthenticatedError()
-        with database.SessionLocal() as session:
-            user = user_crud.get_user_by_name(session, credentials.username)
-            db_token = (
-                token_crud.get_token_by_token_and_user(
-                    session, credentials.password, user.id
-                )
-                if user
-                else None
+
+        user = user_crud.get_user_by_name(db, credentials.username)
+        if not user:
+            logger.info(
+                "User with username '%s' not found.", credentials.username
             )
-            if not db_token:
-                logger.info("Token invalid for user %s", credentials.username)
-                raise exceptions.InvalidPersonalAccessTokenError()
+            raise exceptions.InvalidPersonalAccessTokenError()
+        db_token = token_crud.get_token_by_token_and_user(
+            db, credentials.password, user.id
+        )
+
+        if not db_token:
+            logger.info("Token invalid for user %s", credentials.username)
+            raise exceptions.InvalidPersonalAccessTokenError()
 
-            if db_token.expiration_date < datetime.date.today():
-                logger.info("Token expired for user %s", credentials.username)
-                raise exceptions.PersonalAccessTokenExpired()
-        return self.get_username(credentials)
+        if db_token.expiration_date < datetime.date.today():
+            logger.info("Token expired for user %s", credentials.username)
+            raise exceptions.PersonalAccessTokenExpired()
 
-    def get_username(self, credentials: security.HTTPBasicCredentials) -> str:
-        return credentials.username
+        return user, db_token

backend/capellacollab/core/authentication/injectables.py

@@ -22,6 +22,7 @@
 from capellacollab.users import crud as users_crud
 from capellacollab.users import exceptions as users_exceptions
 from capellacollab.users import models as users_models
+from capellacollab.users import permissions as users_permissions
 
 from . import exceptions
 
@@ -59,28 +60,51 @@ class OpenAPIPersonalAccessToken(OpenAPIFakeBase):
     __hash__ = OpenAPIFakeBase.__hash__
 
 
-async def get_username(
+async def validate_authentication_information(
     request: fastapi.Request,
+    db: orm.Session = fastapi.Depends(database.get_db),
     _unused1=fastapi.Depends(OpenAPIPersonalAccessToken()),
-) -> str:
+) -> tuple[users_models.DatabaseUser, users_permissions.UserScope]:
     if request.cookies.get("id_token"):
-        username = await api_key_cookie.JWTAPIKeyCookie()(request)
-        return username
+        user = users_crud.get_user_by_name(
+            db, await api_key_cookie.JWTAPIKeyCookie()(request)
+        )
+        assert user
+        return user, users_models.ROLE_MAPPING[user.role]
 
     authorization = request.headers.get("Authorization")
     scheme, _ = security_utils.get_authorization_scheme_param(authorization)
 
     match scheme.lower():
         case "basic":
-            username = await basic_auth.HTTPBasicAuth()(request)
+            user, token = await basic_auth.HTTPBasicAuth().validate(
+                db, request
+            )
         case "":
             raise exceptions.UnauthenticatedError()
         case _:
             raise exceptions.UnknownScheme(scheme)
 
-    return username
+    return user, users_permissions.UserScope()  # Replace with token scope
+
+
+async def get_username(
+    authentication_information: tuple[
+        users_models.DatabaseUser, users_permissions.UserScope
+    ] = fastapi.Depends(validate_authentication_information),
+) -> str:
+    return authentication_information[0].name
+
+
+async def get_scope(
+    authentication_information: tuple[
+        users_models.DatabaseUser, users_permissions.UserScope
+    ] = fastapi.Depends(validate_authentication_information),
+) -> users_permissions.UserScope:
+    return authentication_information[1]
 
 
+# TODO: Replace all occurrences of RoleVerification with PermissionValidation
 class RoleVerification:
     def __init__(self, required_role: users_models.Role, verify: bool = True):
         self.required_role = required_role

backend/capellacollab/core/logging/__init__.py

@@ -14,6 +14,7 @@
 from starlette.middleware import base
 
 from capellacollab.configuration.app import config
+from capellacollab.core import database
 from capellacollab.core.authentication import injectables as auth_injectables
 
 LOGGING_LEVEL = config.logging.level
@@ -80,7 +81,14 @@ async def dispatch(
         call_next: base.RequestResponseEndpoint,
     ):
         try:
-            username = await auth_injectables.get_username(request)
+            with database.SessionLocal() as session:
+                (
+                    user,
+                    _,
+                ) = await auth_injectables.validate_authentication_information(
+                    request, session
+                )
+            username = user.name
         except fastapi.HTTPException:
             username = "anonymous"
 

backend/capellacollab/users/exceptions.py

@@ -5,6 +5,8 @@
 
 from capellacollab.core import exceptions as core_exceptions
 
+from . import permissions
+
 
 class UserNotFoundError(core_exceptions.BaseError):
     def __init__(
@@ -78,3 +80,18 @@ def __init__(self):
             reason="You do not have permission to enroll yourself in beta testing.",
             err_code="BETA_TESTING_SELF_ENROLLMENT_NOT_ALLOWED",
         )
+
+
+class InsufficientPermissionError(core_exceptions.BaseError):
+    def __init__(
+        self,
+        required_permission: str,
+        required_verbs: set[permissions.UserTokenVerb],
+    ):
+        verbs = ", ".join(verb.value for verb in required_verbs)
+        super().__init__(
+            status_code=status.HTTP_403_FORBIDDEN,
+            title="Permission denied",
+            reason=f"Insufficient permissions: '{required_permission}' permission with verbs {verbs} is required for this transaction.",
+            err_code="INSUFFICIENT_PERMISSION",
+        )

backend/capellacollab/users/injectables.py

@@ -1,13 +1,15 @@
 # SPDX-FileCopyrightText: Copyright DB InfraGO AG and contributors
 # SPDX-License-Identifier: Apache-2.0
 
+import dataclasses
+
 import fastapi
 from sqlalchemy import orm
 
 from capellacollab.core import database
 from capellacollab.core.authentication import injectables as auth_injectables
 
-from . import crud, exceptions, models
+from . import crud, exceptions, models, permissions
 
 
 def get_own_user(
@@ -28,3 +30,27 @@ def get_existing_user(
         return user
 
     raise exceptions.UserNotFoundError(user_id=user_id)
+
+
+@dataclasses.dataclass(eq=False)
+class PermissionValidation:
+    required_scope: permissions.UserScope
+
+    def __call__(
+        self,
+        authentication_information: tuple[
+            models.DatabaseUser, permissions.UserScope
+        ] = fastapi.Depends(
+            auth_injectables.validate_authentication_information
+        ),
+        db: orm.Session = fastapi.Depends(database.get_db),
+    ) -> None:
+        actual_scope = authentication_information[1].model_dump()
+
+        for scope, perms in self.required_scope:
+            for perm, verbs in perms:
+                for verb in verbs:
+                    if verb not in actual_scope[scope][perm]:
+                        raise exceptions.InsufficientPermissionError(
+                            perm, verbs
+                        )

backend/capellacollab/users/models.py

@@ -13,6 +13,8 @@
 from capellacollab.core import database
 from capellacollab.core import pydantic as core_pydantic
 
+from . import permissions
+
 if t.TYPE_CHECKING:
     from capellacollab.events.models import DatabaseUserHistoryEvent
     from capellacollab.projects.users.models import ProjectUserAssociation
@@ -25,6 +27,79 @@ class Role(str, enum.Enum):
     USER = "user"
 
 
+USER_TOKEN_SCOPE = permissions.UserTokenResource(
+    sessions={
+        permissions.UserTokenVerb.GET,
+        permissions.UserTokenVerb.CREATE,
+    },
+    projects={permissions.UserTokenVerb.CREATE},
+    monitoring={permissions.UserTokenVerb.GET},
+    tokens={
+        permissions.UserTokenVerb.GET,
+        permissions.UserTokenVerb.CREATE,
+        permissions.UserTokenVerb.DELETE,
+    },
+)
+
+ROLE_MAPPING = {
+    Role.USER: permissions.UserScope(
+        user=USER_TOKEN_SCOPE,
+    ),
+    Role.ADMIN: permissions.UserScope(
+        user=USER_TOKEN_SCOPE,
+        admin=permissions.AdminTokenResource(
+            users={
+                permissions.UserTokenVerb.GET,
+                permissions.UserTokenVerb.CREATE,
+                permissions.UserTokenVerb.UPDATE,
+                permissions.UserTokenVerb.DELETE,
+            },
+            projects={
+                permissions.UserTokenVerb.GET,
+                permissions.UserTokenVerb.CREATE,
+                permissions.UserTokenVerb.UPDATE,
+                permissions.UserTokenVerb.DELETE,
+            },
+            tools={
+                permissions.UserTokenVerb.GET,
+                permissions.UserTokenVerb.CREATE,
+                permissions.UserTokenVerb.UPDATE,
+                permissions.UserTokenVerb.DELETE,
+            },
+            announcements={
+                permissions.UserTokenVerb.CREATE,
+                permissions.UserTokenVerb.UPDATE,
+                permissions.UserTokenVerb.DELETE,
+            },
+            monitoring={
+                permissions.UserTokenVerb.GET,
+            },
+            configuration={
+                permissions.UserTokenVerb.GET,
+                permissions.UserTokenVerb.UPDATE,
+            },
+            t4c_servers={
+                permissions.UserTokenVerb.GET,
+                permissions.UserTokenVerb.UPDATE,
+                permissions.UserTokenVerb.DELETE,
+            },
+            t4c_repositories={
+                permissions.UserTokenVerb.GET,
+                permissions.UserTokenVerb.UPDATE,
+                permissions.UserTokenVerb.DELETE,
+            },
+            pv_configuration={
+                permissions.UserTokenVerb.UPDATE,
+                permissions.UserTokenVerb.DELETE,
+            },
+            events={
+                permissions.UserTokenVerb.GET,
+            },
+        ),
+    ),
+}
+
+
 class BaseUser(core_pydantic.BaseModel):
     id: int
     name: str